From dbe7285c1c1ae1ca75218fc099962c2500ad0bb4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=8D=BF=E4=BA=91=E7=A7=8B=E6=9C=88?= <15273589815@163.com> Date: Thu, 10 Sep 2026 22:01:43 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E7=9F=AD=E6=9C=9F=E4=BC=9A=E8=AF=9D?= =?UTF-8?q?=E8=AE=B0=E5=BF=86=EF=BC=88=E5=A4=9A=E8=BD=AE=E6=8C=87=E4=BB=A3?= =?UTF-8?q?=E5=8F=AF=E8=A7=A3=E6=9E=90=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 一、此前的缺口 方案 §2.2 要求会话短期记忆，但底座**没有任何加载历史消息的代码**：conversation_message 存了全部消息、svc_conversation_session 只在计数，而 run 执行时只拿到当前这一条消息。 后果是客户问"那它风险高吗"时"它"无从对应，向量检索落到无关内容、整条回答走兜底—— 多轮对话事实上不可用。 二、实现 1. 契约：AgentRequest 新增 `history: tuple[ConversationTurn, ...] = ()`（默认空元组， 既有构造点无需改动）。ConversationTurn 只保留 role 与正文，不把意图/置信度等内部字段 喂给模型——既减少噪声，也收窄"模型看到不该看的东西"的面。 2. 加载：WorkerRuntime._execute_claimed 构造 AgentRequest 时加载本会话此前的对话 （上限 10 轮，按 id 正序）。`before_message_id` 排除本轮请求消息本身，否则模型会在 上下文里看到自己的问题被重复一遍。 3. 使用：客服 Agent 构造检索查询时，把最近两轮**客户**消息与当前问题拼接。只取客户的 话、不取 Agent 自己的回答——把后者拼进来会让检索偏向自己上一轮的说法，而客户的真实 意图可能已经在下一句里被修正。 三、两个刻意的取舍 · **不引入 Redis 双写**：方案 §2.2 设想用 Redis 列表，但消息在受理时已落库，再同步一份 只会带来不一致与 TTL 管理成本，换来的仅是一次索引查询的节省。这里取等价语义 （同样"最近若干轮、超出即截断"）而不复制存储。 · **按条数截断而非 token**：没有与模型一致的分词器，按 token 截断只能估算、边界会随实现 漂移；按条数是确定性的，宁可少给几轮，也不给一个不稳定的边界。 四、实测（同一会话两轮） · 第 1 轮"南方季季盈90天的起投金额是多少" → 正确返回该产品表格（R2、起投 1 万元等）； · 第 2 轮只说"那它风险高吗"（不含任何产品名）→ 仍正确检索到同一产品并答出风险等级 R2、 业绩比较基准与投资范围；此前这类提问必然走兜底； · ruff 通过、mypy 113 文件无错、unit+contract 447 passed。 --- app/core/contracts.py | 17 ++++++++ .../agent/implementations/customer_service.py | 16 ++++++- app/worker/runtime.py | 43 +++++++++++++++++++ 3 files changed, 75 insertions(+), 1 deletion(-) diff --git a/app/core/contracts.py b/app/core/contracts.py index 55f30d7..75071a5 100644 --- a/app/core/contracts.py +++ b/app/core/contracts.py @@ -13,6 +13,20 @@ class AgentRequestMetadata(BaseModel): ui_entry: str | None = None +class ConversationTurn(BaseModel): + """会话中的一轮对话（短期记忆）。 + + 只保留角色与正文：模型用它解析指代（"那它风险高吗"里的"它"指哪只基金）， + 不需要意图、置信度这类内部字段——把内部字段一并喂给模型既增加噪声， + 也扩大了"模型看到不该看的东西"的面。 + """ + + model_config = ConfigDict(frozen=True) + + role: Literal["user", "assistant"] + content: str + + class AgentRequest(BaseModel): model_config = ConfigDict(extra="forbid", frozen=True) @@ -21,6 +35,9 @@ class AgentRequest(BaseModel): session_id: str idempotency_key: str metadata: AgentRequestMetadata = Field(default_factory=AgentRequestMetadata) + # 本会话中**本次之前**的对话，按时间正序（旧 → 新）。 + # 默认空元组：既有构造点（API 受理路径、单测、验收脚本）无需改动即可继续工作。 + history: tuple[ConversationTurn, ...] = () @field_validator("message") @classmethod diff --git a/app/service/agent/implementations/customer_service.py b/app/service/agent/implementations/customer_service.py index dc5f1bb..3a5a8a2 100644 --- a/app/service/agent/implementations/customer_service.py +++ b/app/service/agent/implementations/customer_service.py @@ -114,7 +114,7 @@ class CustomerServiceAgent(BaseAgent): try: output = await self.call_tool( TOOL_NAME, - {"query": request.message[:500], "top_k": TOP_K}, + {"query": self._search_query(request), "top_k": TOP_K}, intent=intent, context=context, ) @@ -205,6 +205,20 @@ class CustomerServiceAgent(BaseAgent): row.user_prompt_template or DEFAULT_CHITCHAT_TEMPLATE, ) + @staticmethod + def _search_query(request: AgentRequest) -> str: + """构造交给检索的查询串：把最近几轮**客户**说过的话与当前问题拼在一起。 + + 为什么必须带上文：检索是向量匹配，只看当前这一句时，"那它风险高吗"里的"它" + 无从对应，检索会落到无关内容上、进而整条回答走兜底。把上文一并向量化， + 检索才能落到上文提到的那个产品上。 + + 为什么只取客户的话、不取 Agent 自己的回答：把 Agent 的措辞也拼进来会让检索 + 偏向自己上一轮的说法，而客户的真实意图可能已经在下一句里被修正过。 + """ + recent_user = [turn.content for turn in request.history if turn.role == "user"][-2:] + return " ".join([*recent_user, request.message])[:500] + # ---- 出口三：引导人工客服（不做工单，只回话并留痕） ---- def _guide_to_human(self, reason: str) -> CoreResult: diff --git a/app/worker/runtime.py b/app/worker/runtime.py index 4a0dc8c..e63858b 100644 --- a/app/worker/runtime.py +++ b/app/worker/runtime.py @@ -243,6 +243,43 @@ class WorkerRuntime: finally: await client.aclose() + @staticmethod + async def _conversation_history( + session: Any, *, session_id: str, user_id: int, before_message_id: int | None, + limit: int = 10, + ) -> tuple[Any, ...]: + """取该会话最近若干轮对话，按时间正序（旧 → 新）返回。 + + 以 MySQL 的会话消息为**唯一来源**，不引入 Redis 双写：消息在受理时已经落库， + 再同步一份到 Redis 只会带来不一致与 TTL 管理成本，换来的仅是一次索引查询的节省。 + 方案 §2.2 设想的是 Redis 列表，这里取等价语义（同样"最近若干轮、超出即截断"） + 而不复制存储。 + + `before_message_id` 排除本轮请求消息本身：它刚写入库，若也算进历史， + 模型会在上下文里看到自己的问题被重复一遍。 + + 截断按**条数**而非 token：这里没有与模型一致的分词器，按 token 截断只能靠估算、 + 边界会随实现漂移；按条数是确定性的，宁可少给几轮，也不给一个不稳定的边界。 + """ + from app.core.contracts import ConversationTurn + from app.repository.conversation_repository import ConversationRepository + + rows = await ConversationRepository(session).messages( + session_id, user_id, limit + 1, before=before_message_id + ) + # repository 按 id DESC 返回（最新在前），这里翻正为旧 → 新 + ordered = list(reversed(rows))[-limit:] + turns: list[Any] = [] + for row in ordered: + content = str(row.content or "").strip() + if not content: + continue + turns.append(ConversationTurn( + role="assistant" if str(row.role) == "assistant" else "user", + content=content, + )) + return tuple(turns) + async def _cleanup_projection(self, payload: dict[str, Any], *, session: Any) -> None: """幂等清理一条记忆的派生投影（Milvus 向量、Neo4j 关系）。 @@ -463,10 +500,16 @@ class WorkerRuntime: DomainEventOutbox.event_type == "agent.run_requested").limit(1)) if message is None or idem is None: raise ValueError("run input missing") + # 短期会话记忆：加载本次之前的对话，模型靠它解析指代。 + history = await self._conversation_history( + session, session_id=run.session_id, user_id=int(run.user_id), + before_message_id=run.request_message_id, + ) request = AgentRequest( agent_type=run.agent_type, message=message.content, session_id=run.session_id, idempotency_key=idem.idempotency_key, metadata=event.payload.get("metadata", {}) if event else {}, + history=history, ) identity = RequestContext(user_id=str(run.user_id), trace_id=run.trace_id) # Re-check account and permissions at execution time, including delayed jobs.