lzf_0626
13bab7c3d0
按业务方确定的取向实现：金融场景确定性优先，能溯源到公司资料的才答，答不了就 引导客户拨打客服热线，绝不用模型猜答案。端到端验收 8/8 通过。 新增： - app/service/knowledge_search_service.py：知识检索。未复用记忆的 VectorMemoryAdapter 是因为它只返回 (memory_uuid, score)，会丢掉知识块的标题与正文，而客服回答必须能把 原文与出处一起交付。检索失败一律返回 degraded 而不抛异常，由 Agent 走兜底。 - app/service/knowledge_tool.py + app/core/knowledge_contracts.py：只读工具 search_knowledge。 走 ToolExecutor 而不是让 Agent 直接持有检索服务，是为了让白名单、权限、审计、超时 都归基座统一管理；工具只读也符合 ToolRegistry 的硬约束。复用既有权限码 knowledge:reference:read（customer 角色已具备），不新增权限点。 - app/service/agent/implementations/customer_service.py：Agent 本体，刻意保持薄—— 意图分发 + 四条出口（faq/产品/政策直返、闲聊走模型、其余与异常引导人工）。 直接返回知识原文而不经模型改写，答案的字面内容全部来自公司已发布资料。 - tools/publish_customer_service_config.py：发布意图工具白名单。 - tools/customer_service_check.py：端到端验收（8 个用例，含越界请求与知识库外问题）。 装配： - bootstrap 新增 get_knowledge_search_service 工厂，注册 search_knowledge 工具与 customer_service Agent。 - runtime_config_service 新增 load_active_prompt：提示词绑定 release_id，按当前生效 版本读取，未发布时回落代码默认值。闲聊话术因此可审核、可回滚，不必改代码发版。 过程中发现并处理的三个问题： 1. 自造 source_references 被基座合规闸门拒绝。governance.review_output 只接受 「本次召回的记忆」与「本次成功调用的工具」两类引用（用于防止伪造来源）， knowledge 类型会被判非法并使整个 run 失败。处理方式是**不放开那道校验**， 而把知识出处（文件标题与内部编号）写进正文，source_references 交给基座自动附加。 2. 发布配置是整版本替换语义：新版本会清空旧版本的全部配置项。若只发客服白名单， 示例 Agent 的 fund_query_demo:fund_quote 会被静默清空。故发布脚本先读取当前生效 版本的全部配置项并原样继承，再追加新增项。 3. 验收脚本自身两处自伤：打印 emoji 触发 GBK UnicodeEncodeError、以及读错结果字段 （RunQueryService 返回的答案键是 content 不是 text）。 已知缺口（未修，已记录）： - CoreResult.transfer_required 未持久化：conversation_message 不存该标记， API 读不到"本次是否引导了人工"。当前靠正文里的固定话术判断。 - 知识块引用（source_type=knowledge）尚未启用，需先让 ToolExecutor 把工具返回的 doc_id 登记为本次可引用来源。 验证：ruff 通过、mypy 107 文件无错、unit+contract 447 passed； tools/customer_service_check.py 8/8 通过（含越界请求、投诉、知识库外问题三类 必须引导人工的场景，以及 7 个零容忍负面词零命中）。
252 lines
11 KiB
Python
252 lines
11 KiB
Python
"""客服 Agent：只回答能溯源到公司资料的问题，答不了就引导客户致电人工客服。
|
||
|
||
设计取向（金融场景，由业务方确定）：
|
||
|
||
- **确定性优先**：命中知识块后**直接返回原文**，不经模型改写。答案的字面内容来自公司
|
||
已发布的资料，模型不参与事实生成，因此不存在"编一个看起来合理的答案"的通路。
|
||
- **答不了就引导**：检索降级、未命中、置信度不足、意图未覆盖、模型异常——一律返回
|
||
引导客户拨打客服热线的固定话术，并置 `transfer_required=True` 留痕；
|
||
绝不用模型猜测答案。
|
||
- **本体保持薄**：只有意图分发与四条出口，不做多轮推理、不自主决策。这是业务方明确
|
||
要求的取向——"客服 Agent 本身不会很多内容，不会的就转人工"。
|
||
|
||
四条出口：
|
||
`faq` → 检索直返（不经模型）｜`product_inquiry` / `policy_explain` → 检索直返 + 来源引用
|
||
｜`chitchat` → 模型生成（提示词走发布配置）｜其余与异常 → 引导人工客服
|
||
"""
|
||
|
||
from app.core.contracts import (
|
||
AgentDefinition,
|
||
AgentRequest,
|
||
CoreResult,
|
||
RequestContext,
|
||
SourceReference,
|
||
)
|
||
from app.core.errors import ForbiddenAgentError
|
||
from app.service.agent.base import BaseAgent
|
||
from app.service.model_gateway import DatabaseModelEndpointResolver
|
||
from app.service.runtime_config_service import load_active_prompt
|
||
|
||
AGENT_TYPE = "customer_service"
|
||
|
||
# 意图码必须三处对齐：AgentDefinition.supported_intents、agent_intent_config 的
|
||
# (agent_type, intent_code)、以及发布版 config_release 里 agent_tools 的
|
||
# `customer_service:<intent>` 白名单 key。缺任一处即失败关闭，这是底座的有意设计。
|
||
INTENT_FAQ = "faq"
|
||
INTENT_PRODUCT = "product_inquiry"
|
||
INTENT_POLICY = "policy_explain"
|
||
INTENT_CHITCHAT = "chitchat"
|
||
INTENT_TRANSFER = "transfer_human"
|
||
BUSINESS_INTENTS = (INTENT_FAQ, INTENT_PRODUCT, INTENT_POLICY)
|
||
|
||
TOOL_NAME = "search_knowledge"
|
||
|
||
# 三档置信阈值（方案 §2.3，业务方已确认先按此跑通、后续用语料校准）。
|
||
# 分数为 COSINE 相似度，落在 [0, 1]。
|
||
HIGH_SCORE = 0.75 # ≥ 直接答
|
||
MID_SCORE = 0.60 # ≥ 且 < 高置信：答 + 提示可能不完整；< 该值：不硬答，引导人工
|
||
|
||
TOP_K = 5
|
||
MAX_ANSWER_CHARS = 1200
|
||
REFERENCE_LIMIT = 3
|
||
|
||
COMPANY = "南方科技"
|
||
# 客服热线：正式号码确定后改这里（或改为读配置项，避免改代码）
|
||
HOTLINE = "400-XXX-XXXX"
|
||
SERVICE_HOURS = "每日 7:00-22:00"
|
||
|
||
FALLBACK_TEMPLATE = (
|
||
"抱歉，这个问题我暂时无法给出准确答复。为避免给您错误信息，"
|
||
f"建议您拨打客服热线 {HOTLINE}（{SERVICE_HOURS}）转人工客服咨询。"
|
||
)
|
||
INCOMPLETE_NOTICE = "（以上信息可能不完整，具体以产品说明书与公司制度为准）"
|
||
DISCLAIMER = "（以上内容由智能客服依据公司公开资料整理，不构成投资建议）"
|
||
|
||
CHITCHAT_PROMPT_CODE = "customer_service_chitchat"
|
||
CHITCHAT_TASK_TYPE = "chat"
|
||
DEFAULT_CHITCHAT_SYSTEM = (
|
||
f"你是{COMPANY}的智能客服助手。回应要简短、礼貌，并自然引导用户提出与基金、理财、"
|
||
"账户相关的问题。禁止承诺收益，禁止出现「保本」「稳赚」「无风险」「保证收益」"
|
||
"「预期收益率」「年化收益率」「安全」等表述。"
|
||
)
|
||
DEFAULT_CHITCHAT_TEMPLATE = "用户说：{message}\n请用不超过 40 字回应，并把话题引导到业务上。"
|
||
|
||
|
||
class CustomerServiceAgent(BaseAgent):
|
||
definition = AgentDefinition(
|
||
agent_type=AGENT_TYPE,
|
||
version="1.0.0",
|
||
allowed_roles=("customer",),
|
||
allowed_portals=("api",),
|
||
# 代码上限：实际可用范围由发布配置的意图白名单收窄（两者取交集）
|
||
allowed_tools=(TOOL_NAME,),
|
||
supported_intents=(
|
||
INTENT_FAQ, INTENT_PRODUCT, INTENT_POLICY, INTENT_CHITCHAT, INTENT_TRANSFER,
|
||
),
|
||
)
|
||
|
||
async def handle(self, request: AgentRequest, context: RequestContext) -> CoreResult:
|
||
intent = self._intent_code()
|
||
if intent == INTENT_CHITCHAT:
|
||
return await self._chitchat(request)
|
||
if intent == INTENT_TRANSFER:
|
||
return self._guide_to_human("客户主动要求转人工")
|
||
if intent not in BUSINESS_INTENTS:
|
||
# 分类失败或意图未覆盖：不猜，直接引导人工
|
||
return self._guide_to_human(f"意图未覆盖：{intent or '未识别'}")
|
||
return await self._answer_from_knowledge(request, context, intent)
|
||
|
||
# ---- 出口一：知识直返（faq / 产品 / 政策） ----
|
||
|
||
async def _answer_from_knowledge(
|
||
self, request: AgentRequest, context: RequestContext, intent: str
|
||
) -> CoreResult:
|
||
try:
|
||
output = await self.call_tool(
|
||
TOOL_NAME,
|
||
{"query": request.message[:500], "top_k": TOP_K},
|
||
intent=intent,
|
||
context=context,
|
||
)
|
||
except ForbiddenAgentError:
|
||
# 白名单/权限类失败必须冒泡：那是配置错误，若被兜底话术吞掉，
|
||
# 运维会看到"客服一直引导人工"却查不出原因。
|
||
raise
|
||
except Exception:
|
||
return self._guide_to_human("知识检索调用失败")
|
||
|
||
if not isinstance(output, dict):
|
||
return self._guide_to_human("知识检索返回格式异常")
|
||
if output.get("degraded"):
|
||
reason = str(output.get("reason") or "未知")
|
||
return self._guide_to_human(f"知识检索降级：{reason}")
|
||
hits = output.get("hits")
|
||
if not isinstance(hits, list) or not hits:
|
||
return self._guide_to_human("知识库未命中")
|
||
|
||
best = hits[0]
|
||
if not isinstance(best, dict):
|
||
return self._guide_to_human("命中内容格式异常")
|
||
score = self._score(best.get("score"))
|
||
if score < MID_SCORE:
|
||
return self._guide_to_human(f"置信度不足：{score:.3f}")
|
||
|
||
content = str(best.get("content") or "").strip()
|
||
if not content:
|
||
return self._guide_to_human("命中内容为空")
|
||
|
||
answer = content[:MAX_ANSWER_CHARS]
|
||
if score < HIGH_SCORE:
|
||
answer = f"{answer}\n{INCOMPLETE_NOTICE}"
|
||
# 知识出处写进正文，而不是塞进 source_references：
|
||
# `governance.review_output` 只接受「本次召回的记忆」与「本次成功调用的工具」两类引用
|
||
# （用于防止 Agent 伪造来源），知识块的 doc_id 不属于这两类，会被判为非法引用。
|
||
# 把文件标题与内部文件编号写进正文，客户与人工同样能核对，且不必放开那道校验。
|
||
source_note = self._source_note(best)
|
||
text = (
|
||
f"{answer}\n{source_note}\n{DISCLAIMER}"
|
||
if source_note
|
||
else f"{answer}\n{DISCLAIMER}"
|
||
)
|
||
return CoreResult(text=text, intent=self._classified_intent)
|
||
|
||
# ---- 出口二：闲聊（提示词走发布配置） ----
|
||
|
||
async def _chitchat(self, request: AgentRequest) -> CoreResult:
|
||
system, template = await self._chitchat_prompt()
|
||
message = request.message[:500]
|
||
try:
|
||
prompt = template.format(message=message, company=COMPANY)
|
||
except (KeyError, IndexError, ValueError):
|
||
# 发布配置里的占位符与代码不一致时回落默认模板：配置写错不应该让运行期崩
|
||
system = DEFAULT_CHITCHAT_SYSTEM
|
||
prompt = DEFAULT_CHITCHAT_TEMPLATE.format(message=message, company=COMPANY)
|
||
full_prompt = f"{system}\n\n{prompt}" if system else prompt
|
||
try:
|
||
endpoints = await DatabaseModelEndpointResolver().resolve(
|
||
agent_type=AGENT_TYPE, task_type="text_generation"
|
||
)
|
||
# 显式转成 list[object]：generate_with_model 形参是 list[object]，
|
||
# 而 list 不变型（invariant），直接传 list[ModelEndpointConfig] 过不了类型检查
|
||
endpoint_list: list[object] = list(endpoints)
|
||
execution = await self.generate_with_model(endpoint_list, full_prompt)
|
||
except Exception:
|
||
return self._guide_to_human("模型不可用")
|
||
text = (execution.text or "").strip()
|
||
if not text:
|
||
return self._guide_to_human("模型返回为空")
|
||
return CoreResult(
|
||
text=f"{text[:MAX_ANSWER_CHARS]}\n{DISCLAIMER}",
|
||
intent=self._classified_intent,
|
||
)
|
||
|
||
async def _chitchat_prompt(self) -> tuple[str, str]:
|
||
"""读取当前发布版本的闲聊提示词；未发布或读取失败时回落代码内置默认值。"""
|
||
try:
|
||
row = await load_active_prompt(CHITCHAT_PROMPT_CODE, CHITCHAT_TASK_TYPE, AGENT_TYPE)
|
||
except Exception:
|
||
row = None
|
||
if row is None:
|
||
return DEFAULT_CHITCHAT_SYSTEM, DEFAULT_CHITCHAT_TEMPLATE
|
||
return (
|
||
row.system_prompt or DEFAULT_CHITCHAT_SYSTEM,
|
||
row.user_prompt_template or DEFAULT_CHITCHAT_TEMPLATE,
|
||
)
|
||
|
||
# ---- 出口三：引导人工客服（不做工单，只回话并留痕） ----
|
||
|
||
def _guide_to_human(self, reason: str) -> CoreResult:
|
||
return CoreResult(
|
||
text=FALLBACK_TEMPLATE,
|
||
intent=self._classified_intent,
|
||
transfer_required=True,
|
||
transfer_reason=reason[:200],
|
||
)
|
||
|
||
# ---- 辅助 ----
|
||
|
||
def _intent_code(self) -> str:
|
||
classified = self._classified_intent
|
||
return classified.intent if classified is not None else ""
|
||
|
||
@staticmethod
|
||
def _score(value: object) -> float:
|
||
"""把命中分数夹到 [0,1]：SourceReference.score 有 ge=0/le=1 约束。"""
|
||
try:
|
||
number = float(value) # type: ignore[arg-type]
|
||
except (TypeError, ValueError):
|
||
return 0.0
|
||
return min(1.0, max(0.0, number))
|
||
|
||
@staticmethod
|
||
def _source_note(hit: dict[str, object]) -> str:
|
||
"""把知识出处写成一行正文（文件标题 + 内部文件编号），便于客户与人工核对。"""
|
||
title = str(hit.get("title") or "").strip()
|
||
doc_no = str(hit.get("doc_no") or "").strip()
|
||
if not title:
|
||
return ""
|
||
return f"（依据：{title}{'，' + doc_no if doc_no else ''}）"
|
||
|
||
def _references(self, hits: list[object]) -> tuple[SourceReference, ...]:
|
||
"""保留待用：等基座支持 knowledge 类型引用后再启用。
|
||
|
||
目前 `governance.review_output` 只认可 memory / tool 两类来源，直接返回
|
||
knowledge 引用会被判为「引用未来自本次已授权召回结果」而让整个 run 失败。
|
||
启用前提是让 ToolExecutor 把工具返回的知识 doc_id 登记为本次可引用来源。
|
||
"""
|
||
references: list[SourceReference] = []
|
||
for hit in hits[:REFERENCE_LIMIT]:
|
||
if not isinstance(hit, dict):
|
||
continue
|
||
doc_id = str(hit.get("doc_id") or "").strip()
|
||
if not doc_id:
|
||
continue # 没有标识的命中无法回溯，丢弃而不是造一个假来源
|
||
title = str(hit.get("title") or "").strip()
|
||
references.append(SourceReference(
|
||
source_type="knowledge",
|
||
source_id=doc_id,
|
||
title=title[:200] or None,
|
||
score=self._score(hit.get("score")),
|
||
))
|
||
return tuple(references)
|