xsl
7bcbb8d40b
CI / lint / typecheck / import-linter / tests / secrets (push) Canceled after 0s
- agent: ag_conversation/ag_message 表(迁移0007),上下文组装注入最近报告画像摘要 (report.profile.latest_profile_summary 经 apps 层闭包注入,同层不互调),历史带最近 AGENT_HISTORY_TURNS=10 轮;POST/GET /agent/conversations 与消息历史接口 - agent 修复:ChatRequest.conversation_id 改 str(雪花ID超2^53,JS Number丢精度); _build_context 挪进 reserve commit 后同一 try,异常走 refund 兜底 - report: rp_feedback(维度反馈 upsert,认同率/修正率 SQL 可算,T-P3-21 最小版—— L1 记忆未建,不回写结论)+ rp_reservation(预约留资幂等,T-P3-22,预约率口径 行数/full_report.card_view 去重曝光,门槛≥8%),迁移0008 - 测试:全量 606 passed;ruff/mypy/lint-imports 全绿;E2E 真实 deepseek-v4-flash 验证 画像注入与历史携带
239 lines
9.3 KiB
Python
Executable File
239 lines
9.3 KiB
Python
Executable File
"""Agent 对话编排（架构 §7.2）。
|
||
|
||
① 输入审核 ← NOT implemented (compliance.moderate, CR-06)
|
||
② 意图识别 ← NOT implemented (FR-AGT-02)
|
||
③ 上下文组装 ← IMPLEMENTED：画像摘要 + 最近 N 轮历史（context.py）
|
||
④ 混合检索 ← NOT implemented (retrieval-svc, FR-AGT-04)
|
||
⑤ LLM 生成 ← IMPLEMENTED (non-streaming)
|
||
⑥ 生成后校验 ← NOT implemented (FR-AGT-05)
|
||
⑦ 输出审核 ← NOT implemented (CR-06)
|
||
⑧ 落库 + 成本 ← IMPLEMENTED：ag_message 双行 + ob_ai_call 成本行
|
||
⑨ 异步偏好抽取 ← NOT implemented (FR-MEM-03)
|
||
|
||
会话持久化（FR-AGT-09 / T-P4-01）：会话与消息落 ``ag_conversation`` /
|
||
``ag_message``；assistant 行的 ``call_id`` 关联 ``ob_ai_call``，每条回答可
|
||
复盘到模型、token 与成本。检索轨迹与规则命中列已建，等步骤④落地后填充。
|
||
|
||
画像注入：用户最近一份 succeeded 报告的画像摘要以独立 system 消息排在固定
|
||
前缀之后（不进缓存前缀，理由见 context.py）。摘要的读取由 apps 层以
|
||
``profile_provider`` 闭包注入——agent 与 report 在 import-linter 分层里是
|
||
同层，不允许直接互调。
|
||
|
||
Transaction discipline: the reservation is committed **before** the LLM call.
|
||
If the process dies mid-call, the reservation survives and can be reconciled;
|
||
if the call fails, the refund is committed before the error propagates, so the
|
||
API layer's rollback cannot undo it.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from collections.abc import Awaitable, Callable
|
||
from dataclasses import dataclass
|
||
from decimal import Decimal
|
||
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from packages.agent import repo
|
||
from packages.agent.context import build_context_messages
|
||
from packages.billing.service import BillingService, Reservation
|
||
from packages.domain.ids import next_id
|
||
from packages.llm.gateway import LLMGateway
|
||
from packages.llm.types import Message, Scene
|
||
from packages.shared.errors import NotFoundError
|
||
from packages.shared.logging import get_logger
|
||
from packages.shared.models import AgConversation, AgMessage
|
||
|
||
logger = get_logger(__name__)
|
||
|
||
QUOTA_KEY_AGENT_CHAT = "agent_chat"
|
||
REF_TYPE = "agent_chat"
|
||
|
||
#: 画像提供者：user_id → 画像摘要素材（无报告用户返回 None）。由 apps 层接线。
|
||
ProfileProvider = Callable[[int], Awaitable[str | None]]
|
||
|
||
|
||
def _parse_conversation_id(raw: str) -> int:
|
||
"""wire 侧雪花 ID 是字符串（超 2^53,JS Number 会丢精度）。
|
||
|
||
非法值与「会话不存在」同口径 40400：不区分「格式错」与「不存在」，
|
||
避免给遍历者任何额外的探测信号。
|
||
"""
|
||
try:
|
||
return int(raw)
|
||
except (TypeError, ValueError) as exc:
|
||
raise NotFoundError("会话不存在或已删除") from exc
|
||
|
||
|
||
@dataclass(frozen=True, slots=True)
|
||
class ChatResult:
|
||
conversation_id: int
|
||
reply: str
|
||
provider: str
|
||
model: str
|
||
prompt_id: str | None
|
||
prompt_version: str | None
|
||
call_id: int
|
||
tokens_in: int
|
||
tokens_out: int
|
||
tokens_cached: int
|
||
latency_ms: int
|
||
cost_cents: Decimal
|
||
quota_remaining: int
|
||
reserve_tx_id: int
|
||
|
||
|
||
class AgentService:
|
||
def __init__(
|
||
self,
|
||
session: AsyncSession,
|
||
*,
|
||
gateway: LLMGateway,
|
||
billing: BillingService,
|
||
profile_provider: ProfileProvider | None = None,
|
||
history_turns: int = 10,
|
||
) -> None:
|
||
self._session = session
|
||
self._gateway = gateway
|
||
self._billing = billing
|
||
self._profile_provider = profile_provider
|
||
self._history_turns = history_turns
|
||
|
||
async def create_conversation(self, *, user_id: int) -> AgConversation:
|
||
conversation = await repo.create_conversation(
|
||
self._session, conversation_id=next_id(), user_id=user_id
|
||
)
|
||
await self._session.commit()
|
||
return conversation
|
||
|
||
async def list_conversations(self, *, user_id: int) -> list[AgConversation]:
|
||
return await repo.list_conversations(self._session, user_id=user_id)
|
||
|
||
async def list_messages(self, *, user_id: int, conversation_id: int) -> list[AgMessage]:
|
||
await self.get_owned_conversation(user_id=user_id, conversation_id=conversation_id)
|
||
return await repo.list_messages(self._session, conversation_id=conversation_id)
|
||
|
||
async def get_owned_conversation(self, *, user_id: int, conversation_id: int) -> AgConversation:
|
||
conversation = await repo.get_conversation(self._session, conversation_id)
|
||
if conversation is None:
|
||
raise NotFoundError("会话不存在或已删除")
|
||
if conversation.user_id != user_id:
|
||
# 不暴露「存在但属于别人」：与 NOT_FOUND 同口径，防会话 ID 遍历
|
||
raise NotFoundError("会话不存在或已删除")
|
||
return conversation
|
||
|
||
async def chat(
|
||
self,
|
||
*,
|
||
user_id: int,
|
||
message: str,
|
||
request_key: str,
|
||
conversation_id: str | None = None,
|
||
) -> ChatResult:
|
||
if conversation_id is None:
|
||
conversation = await repo.create_conversation(
|
||
self._session, conversation_id=next_id(), user_id=user_id
|
||
)
|
||
else:
|
||
conversation = await self.get_owned_conversation(
|
||
user_id=user_id, conversation_id=_parse_conversation_id(conversation_id)
|
||
)
|
||
|
||
reservation = await self._billing.reserve(
|
||
user_id=user_id,
|
||
quota_key=QUOTA_KEY_AGENT_CHAT,
|
||
request_key=request_key,
|
||
amount=1,
|
||
reason="agent chat turn",
|
||
ref_type=REF_TYPE,
|
||
ref_id=request_key,
|
||
)
|
||
await self._session.commit()
|
||
|
||
# 预扣已 commit，自此到 commit/reserve 冲销之间的任何异常都必须退还——
|
||
# 包括上下文组装（历史消息/画像查询同样是 DB 调用，也会抛）。
|
||
try:
|
||
context = await self._build_context(user_id=user_id, conversation=conversation)
|
||
completion = await self._gateway.complete(
|
||
scene=Scene.AGENT_CHAT,
|
||
messages=context,
|
||
prompt_vars={"user_input": message},
|
||
user_id=user_id,
|
||
ref_type=REF_TYPE,
|
||
ref_id=request_key,
|
||
)
|
||
except Exception as exc:
|
||
await self._refund(reservation, reason=type(exc).__name__)
|
||
raise
|
||
|
||
await self._billing.commit(reservation)
|
||
await repo.append_message(
|
||
self._session,
|
||
message_id=next_id(),
|
||
conversation=conversation,
|
||
role=repo.ROLE_USER,
|
||
content=message,
|
||
)
|
||
await repo.append_message(
|
||
self._session,
|
||
message_id=next_id(),
|
||
conversation=conversation,
|
||
role=repo.ROLE_ASSISTANT,
|
||
content=completion.text,
|
||
call_id=completion.call_id,
|
||
model=completion.model,
|
||
tokens_in=completion.usage.tokens_in,
|
||
tokens_out=completion.usage.tokens_out,
|
||
tokens_cached=completion.usage.tokens_cached,
|
||
latency_ms=completion.latency_ms,
|
||
cost_cents=completion.cost_cents,
|
||
)
|
||
await self._session.commit()
|
||
|
||
return ChatResult(
|
||
conversation_id=conversation.conversation_id,
|
||
reply=completion.text,
|
||
provider=completion.provider,
|
||
model=completion.model,
|
||
prompt_id=completion.prompt_id,
|
||
prompt_version=completion.prompt_version,
|
||
call_id=completion.call_id,
|
||
tokens_in=completion.usage.tokens_in,
|
||
tokens_out=completion.usage.tokens_out,
|
||
tokens_cached=completion.usage.tokens_cached,
|
||
latency_ms=completion.latency_ms,
|
||
cost_cents=completion.cost_cents,
|
||
quota_remaining=reservation.remaining,
|
||
reserve_tx_id=reservation.tx_id,
|
||
)
|
||
|
||
async def _build_context(self, *, user_id: int, conversation: AgConversation) -> list[Message]:
|
||
# 画像取不到（无报告 / 读取异常）不阻断对话：降级为通用建议（FR-AGT-10）。
|
||
# NFR-SEC-03：摘要内容本身不进日志，只记有无与长度。
|
||
profile_summary: str | None = None
|
||
if self._profile_provider is not None:
|
||
try:
|
||
profile_summary = await self._profile_provider(user_id)
|
||
except Exception:
|
||
logger.warning("agent_profile_load_failed", user_id=user_id)
|
||
history: list[Message] = []
|
||
if conversation.message_count > 0:
|
||
recent = await repo.list_recent_messages(
|
||
self._session,
|
||
conversation_id=conversation.conversation_id,
|
||
limit=self._history_turns * 2,
|
||
)
|
||
history = [Message(role=item.role, content=item.content) for item in recent]
|
||
return build_context_messages(
|
||
profile_summary=profile_summary,
|
||
history=history,
|
||
history_turns=self._history_turns,
|
||
)
|
||
|
||
async def _refund(self, reservation: Reservation, *, reason: str) -> None:
|
||
try:
|
||
await self._billing.refund(reservation, reason=reason)
|
||
await self._session.commit()
|
||
except Exception:
|
||
await self._session.rollback()
|
||
logger.exception("quota_refund_failed", tx_id=reservation.tx_id)
|