feat: 短期会话记忆(多轮指代可解析)
一、此前的缺口 方案 §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。
This commit is contained in:
@@ -13,6 +13,20 @@ class AgentRequestMetadata(BaseModel):
|
|||||||
ui_entry: str | None = None
|
ui_entry: str | None = None
|
||||||
|
|
||||||
|
|
||||||
|
class ConversationTurn(BaseModel):
|
||||||
|
"""会话中的一轮对话(短期记忆)。
|
||||||
|
|
||||||
|
只保留角色与正文:模型用它解析指代("那它风险高吗"里的"它"指哪只基金),
|
||||||
|
不需要意图、置信度这类内部字段——把内部字段一并喂给模型既增加噪声,
|
||||||
|
也扩大了"模型看到不该看的东西"的面。
|
||||||
|
"""
|
||||||
|
|
||||||
|
model_config = ConfigDict(frozen=True)
|
||||||
|
|
||||||
|
role: Literal["user", "assistant"]
|
||||||
|
content: str
|
||||||
|
|
||||||
|
|
||||||
class AgentRequest(BaseModel):
|
class AgentRequest(BaseModel):
|
||||||
model_config = ConfigDict(extra="forbid", frozen=True)
|
model_config = ConfigDict(extra="forbid", frozen=True)
|
||||||
|
|
||||||
@@ -21,6 +35,9 @@ class AgentRequest(BaseModel):
|
|||||||
session_id: str
|
session_id: str
|
||||||
idempotency_key: str
|
idempotency_key: str
|
||||||
metadata: AgentRequestMetadata = Field(default_factory=AgentRequestMetadata)
|
metadata: AgentRequestMetadata = Field(default_factory=AgentRequestMetadata)
|
||||||
|
# 本会话中**本次之前**的对话,按时间正序(旧 → 新)。
|
||||||
|
# 默认空元组:既有构造点(API 受理路径、单测、验收脚本)无需改动即可继续工作。
|
||||||
|
history: tuple[ConversationTurn, ...] = ()
|
||||||
|
|
||||||
@field_validator("message")
|
@field_validator("message")
|
||||||
@classmethod
|
@classmethod
|
||||||
|
|||||||
@@ -114,7 +114,7 @@ class CustomerServiceAgent(BaseAgent):
|
|||||||
try:
|
try:
|
||||||
output = await self.call_tool(
|
output = await self.call_tool(
|
||||||
TOOL_NAME,
|
TOOL_NAME,
|
||||||
{"query": request.message[:500], "top_k": TOP_K},
|
{"query": self._search_query(request), "top_k": TOP_K},
|
||||||
intent=intent,
|
intent=intent,
|
||||||
context=context,
|
context=context,
|
||||||
)
|
)
|
||||||
@@ -205,6 +205,20 @@ class CustomerServiceAgent(BaseAgent):
|
|||||||
row.user_prompt_template or DEFAULT_CHITCHAT_TEMPLATE,
|
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:
|
def _guide_to_human(self, reason: str) -> CoreResult:
|
||||||
|
|||||||
@@ -243,6 +243,43 @@ class WorkerRuntime:
|
|||||||
finally:
|
finally:
|
||||||
await client.aclose()
|
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:
|
async def _cleanup_projection(self, payload: dict[str, Any], *, session: Any) -> None:
|
||||||
"""幂等清理一条记忆的派生投影(Milvus 向量、Neo4j 关系)。
|
"""幂等清理一条记忆的派生投影(Milvus 向量、Neo4j 关系)。
|
||||||
|
|
||||||
@@ -463,10 +500,16 @@ class WorkerRuntime:
|
|||||||
DomainEventOutbox.event_type == "agent.run_requested").limit(1))
|
DomainEventOutbox.event_type == "agent.run_requested").limit(1))
|
||||||
if message is None or idem is None:
|
if message is None or idem is None:
|
||||||
raise ValueError("run input missing")
|
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(
|
request = AgentRequest(
|
||||||
agent_type=run.agent_type, message=message.content, session_id=run.session_id,
|
agent_type=run.agent_type, message=message.content, session_id=run.session_id,
|
||||||
idempotency_key=idem.idempotency_key,
|
idempotency_key=idem.idempotency_key,
|
||||||
metadata=event.payload.get("metadata", {}) if event else {},
|
metadata=event.payload.get("metadata", {}) if event else {},
|
||||||
|
history=history,
|
||||||
)
|
)
|
||||||
identity = RequestContext(user_id=str(run.user_id), trace_id=run.trace_id)
|
identity = RequestContext(user_id=str(run.user_id), trace_id=run.trace_id)
|
||||||
# Re-check account and permissions at execution time, including delayed jobs.
|
# Re-check account and permissions at execution time, including delayed jobs.
|
||||||
|
|||||||
Reference in New Issue
Block a user