"""客服 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:` 白名单 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 要求的是「绝对阈值 AND（相对间隙 OR 分布优势）」混合判定， # 只做绝对阈值会把口语化问法误判成"答不了"（这是实测踩到的坑）。 # # 数值按本模型（qwen3.7-text-embedding-flash）**实测校准**，不是照搬经验值： # · 库内问法：口语「我们公司叫什么名字」top1=0.592、标准「公司全称是什么」0.671、 # 带品牌名「南方科技的全称」0.855 —— 同样的正确答案，口语问法相似度天然更低； # · 库外/越界：「你们公司什么时候上市」top1 最高 0.500、「今天天气怎么样」0.416， # 且这些问题的 top1 与次优间隙都在 0.046 以内，而库内命中普遍在 0.09 以上。 # 于是：库内最低 0.579 / 库外最高 0.500，绝对分两侧都有余量；间隙 0.07 又能挡住 # 「存在并列候选」的不确定情形，避免因为一次高相似度的巧合就硬答。 HIGH_SCORE = 0.75 # ≥ 直接答（高置信不再要求间隙：分数已足够说明问题） MID_SCORE = 0.55 # 需与 MIN_GAP 同时满足才答，答时附"信息可能不完整"提示 MIN_GAP = 0.07 # top1 领先次优的最小间隙；领先不足说明有并列候选，不硬答 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}）转人工客服咨询。" ) # 客户侧只展示这一句固定话术（业务方确定）。原中置信的"信息可能不完整"提示已按此移除， # 即中置信回答不再对客户标注不确定性——这是调整时知情的取舍，不是遗漏。 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")) gap = score - self._second_score(hits) # 混合判定：高置信直接答；中置信必须同时满足「分数够」与「领先次优够多」。 # 只满足其一的（分数够但两三个候选并驾齐驱）宁可引导人工——金融场景下 # "答不了"可接受，"答错"不可接受。 confident = score >= HIGH_SCORE if not confident and not (score >= MID_SCORE and gap >= MIN_GAP): return self._guide_to_human(f"置信度不足：score={score:.3f} gap={gap:.3f}") content = str(best.get("content") or "").strip() if not content: return self._guide_to_human("命中内容为空") # 正文只保留「答案 + 固定免责声明」：业务方要求客户侧只看到这一句固定话术， # 因此不确定性提示与出处行都不再出现在正文里。 # # 可追溯性不受影响：本次命中哪个知识块仍由审计（agent.tool_executed 的工具调用记录） # 与消息表留痕，只是不面向客户展示。若将来要把出处给客户看，应当走 # source_references 的 knowledge 类型（需先让 ToolExecutor 登记本次可引用的 doc_id）， # 而不是继续往正文里拼字符串。 return CoreResult( text=f"{content[:MAX_ANSWER_CHARS]}\n{DISCLAIMER}", 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)) def _second_score(self, hits: list[object]) -> float: """次优命中分数，用于「绝对阈值 + 相对间隙」的混合判定。 只有一个命中时返回 0：此时间隙最大，是否回答交由绝对阈值把关， 而不是仅凭"没有竞争者"就认定可信。 """ if len(hits) < 2 or not isinstance(hits[1], dict): return 0.0 return self._score(hits[1].get("score")) 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)