"""客服 Agent：只回答能溯源到公司资料的问题，答不了就引导客户致电人工客服。 设计取向（金融场景，由业务方确定）： - **确定性优先**：命中知识块后**直接返回原文**，不经模型改写。答案的字面内容来自公司 已发布的资料，模型不参与事实生成，因此不存在"编一个看起来合理的答案"的通路。 - **答不了就引导**：检索降级、未命中、置信度不足、意图未覆盖、模型异常——一律返回 引导客户拨打客服热线的固定话术，并置 `transfer_required=True` 留痕； 绝不用模型猜测答案。 - **本体保持薄**：只有意图分发与四条出口，不做多轮推理、不自主决策。这是业务方明确 要求的取向——"客服 Agent 本身不会很多内容，不会的就转人工"。 四条出口： `faq` → 检索直返（不经模型）｜`product_inquiry` / `policy_explain` → 检索直返 + 来源引用 ｜`chitchat` → 模型生成（提示词走发布配置）｜其余与异常 → 引导人工客服 另有**出口零：本人画像**（确定性关键词识别，先于意图分发执行，见 `is_profile_question`）。 """ import logging from typing import Any from app.core.contracts import ( AgentDefinition, AgentRequest, CoreResult, IntentResult, 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 logger = logging.getLogger(__name__) 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_SUITABILITY = "suitability_check" INTENT_CHITCHAT = "chitchat" INTENT_TRANSFER = "transfer_human" BUSINESS_INTENTS = (INTENT_FAQ, INTENT_PRODUCT, INTENT_POLICY, INTENT_SUITABILITY) TOOL_NAME = "search_knowledge" # 适当性裁决工具。为什么必须走它而不是自己比大小：客户的风险等级只有底座能给出权威值 # （来自 fin_risk_assessment，且带测评有效期），Agent 自己判分等于绕开合规链路。 SUITABILITY_TOOL = "check_suitability" # 风险等级名称：国标五级，稳定不变，只用于把裁决结果说成人话。 RISK_LEVEL_NAMES = { 1: "R1（低风险）", 2: "R2（中低风险）", 3: "R3（中风险）", 4: "R4（中高风险）", 5: "R5（高风险）", } # 客户风险测评等级（C 级）到中文表述。与上面的 `RISK_LEVEL_NAMES`（产品 R 级）是两套编码： # 产品讲 R1–R5、客户测评讲 C1–C5，回答里**不能混用**，否则客户会误读自己的等级。 RISK_LEVEL_LABELS: dict[str, str] = { "C1": "保守型（C1）", "C2": "稳健型（C2）", "C3": "平衡型（C3）", "C4": "成长型（C4）", "C5": "进取型（C5）", } # --------------------------------------------------------------------------- # 画像查询（确定性关键词识别，不经过 LLM 意图分类） # --------------------------------------------------------------------------- # # 为什么用**确定性关键词**而不是交给意图分类器：画像问题要求"要么给出该客户的真实数据、 # 要么明确说查不到"。走分类器会引入"模型把「风险等级怎么划分」也判成画像问题"的风险， # 而那一类问的是**规则**、该由知识库回答（见下面的负向词表）。 PROFILE_TOOL_NAME = "query_customer_profile" PROFILE_KEYWORDS: tuple[str, ...] = ( "我的风险等级", "我的风险测评", "我是什么风险", "我的投资者类型", "我的投资类型", "我的风险承受", "测评过期", "测评到期", "我的测评", "我的画像", "我的投资偏好", "我的偏好", "我的投资期限", "我的交易频率", ) #: 问题里出现这些词时**不**按画像处理（问的是政策规则，不是本人数据）。 PROFILE_NEGATIVE_KEYWORDS: tuple[str, ...] = ( "怎么划分", "如何划分", "什么标准", "分类标准", "怎么分", "如何分", "有哪些等级", "等级划分", "怎么定义", "如何定义", ) #: 调用画像工具时用的意图 key。必须复用**已发布**的 `faq`： #: `call_tool(intent=...)` 决定"当前意图允许哪些工具"，而发布配置里只有 #: `agent_tools/customer_service:faq` 一个 key；换新意图码会让交集为空 → #: `AGENT_PERMISSION_DENIED`。这与"知识检索用哪个意图"是两个层面，不冲突。 PROFILE_WHITELIST_INTENT = INTENT_FAQ def is_profile_question(message: str) -> bool: """是否在问**本人画像**（确定性判定，模型不参与）。""" if any(word in message for word in PROFILE_NEGATIVE_KEYWORDS): return False return any(word in message for word in PROFILE_KEYWORDS) def render_profile(profile: dict[str, object]) -> str: """把画像投影渲染成客户可读的一段话。 **只陈述该客户自己的字段**，不推断、不承诺收益；测评过期时**必须明说**并引导重新测评 （与适当性服务的 `ASSESSMENT_EXPIRED` 失败关闭口径一致）。 """ lines: list[str] = [] investor_type = str(profile.get("investor_type") or "").strip().upper() if investor_type: lines.append(f"您的风险测评等级是 {RISK_LEVEL_LABELS.get(investor_type, investor_type)}。") horizon = profile.get("investment_horizon") if horizon: lines.append(f"投资期限偏好：{horizon}。") frequency = profile.get("trading_frequency") if frequency: lines.append(f"交易频率：{frequency}。") assets = profile.get("preferred_asset_class") if isinstance(assets, (list, tuple)) and assets: lines.append("偏好资产类别：" + "、".join(str(a) for a in assets) + "。") tier = profile.get("customer_tier") if tier: lines.append(f"客户分层：{tier}。") if profile.get("assessment_expired") is True: lines.append("注意：您的风险测评已过有效期，需要重新完成测评后才能继续匹配产品风险等级。") if not lines: return "暂时查不到您的画像信息，建议转人工客服核实。" return "\n".join(lines) # 三档置信阈值：方案 §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}）转人工客服咨询。" ) # 免责声明**只由治理层注入**（`PlatformGovernance.review` → `review_output`，话术取自 # `agent_reply_template` 的 `TPL_DISCLAIMER`，取不到时退回 `FALLBACK_DISCLAIMER`）。 # # 这里曾自行拼一句 `DISCLAIMER`，合并后与治理层的话术**同时出现**，客户会看到两条 # 意思重复的声明的（实测复现）。更重要的是分工问题：话术是合规文案，属于**发布配置**， # 改文案不该改代码；Agent 自己拼等于把可配置的合规文案硬编码进业务逻辑， # 而且治理层无法判断"业务是不是已经加过了"（它只认自己追加过的那个形状）。 # 因此本文件不再定义、也不再引用任何免责声明常量。 # 客服热线：正式号码确定后改这里（或改为读配置项，避免改代码） 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, SUITABILITY_TOOL, PROFILE_TOOL_NAME), supported_intents=( INTENT_FAQ, INTENT_PRODUCT, INTENT_POLICY, INTENT_SUITABILITY, INTENT_CHITCHAT, INTENT_TRANSFER, ), ) async def handle(self, request: AgentRequest, context: RequestContext) -> CoreResult: # 画像问题优先处理（确定性关键词，不走意图分类）：知识库答不了"我的风险等级是多少"， # 那需要读该客户的画像数据，必须走 `query_customer_profile` 工具取权威字段。 # 放在意图分发**之前**是有意的：让画像能力不依赖意图分类是否恰好给出 faq。 # 连带效果：该分支的 `intent` 恒为 `faq`，与 `PROFILE_WHITELIST_INTENT` 同源。 if is_profile_question(request.message): return await self._answer_profile(request, context) 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 '未识别'}") if intent == INTENT_SUITABILITY: # 适当性裁决是唯一会给出"能不能买"结论的出口，走独立实现： # 它要组合「产品风险等级 + 客户档案等级」，不是知识检索能算出来的 return await self._answer_suitability(request, context) return await self._answer_from_knowledge(request, context, intent) # ---- 出口零：本人画像（确定性识别 + 权威字段） ---- async def _answer_profile( self, request: AgentRequest, context: RequestContext ) -> CoreResult: """查本人画像并渲染。 **只查 `context.user_id`**，不从用户消息里取编号（否则客户可以靠一句话读别人的画像）。 `customer_id` 仍然显式传给工具：底座会把它落进工具审计，事后能追溯"谁读了谁的画像"。 工具内部还会再做一次数据范围校验（`self` / `own_customers` / `all`）。 查不到时**失败关闭为转人工**，绝不猜一个等级出来。 """ del request try: output = await self.call_tool( PROFILE_TOOL_NAME, {"customer_id": str(context.user_id)}, intent=PROFILE_WHITELIST_INTENT, context=context, ) except ForbiddenAgentError: # 权限/白名单类失败必须冒泡（与知识出口同一口径）：那是配置错误， # 用兜底话术吞掉会让"工具没被授权"表现成"客户画像查不到"。 raise except Exception: logger.warning("画像查询失败，改为引导人工客服", exc_info=True) return self._guide_to_human("画像查询失败") profile = output.get("profile") if isinstance(output, dict) else None if not isinstance(profile, dict) or not profile: logger.info("画像为空或不可用，改为引导人工客服") return self._guide_to_human("画像为空或不可用") return CoreResult( text=render_profile(profile), intent=IntentResult(intent=PROFILE_WHITELIST_INTENT, confidence=1.0), ) # ---- 出口一：知识直返（faq / 产品 / 政策） ---- async def _answer_from_knowledge( self, request: AgentRequest, context: RequestContext, intent: str ) -> CoreResult: try: output = await self.call_tool( TOOL_NAME, {"query": self._search_query(request), "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}") # 命中的是行级子块时，先分辨客户问的是"某个字段"还是"整个产品"： # 子块让「起投多少」拿到聚焦答案，但「介绍一下」会被某一行抢答。 best = self._prefer_section(request.message, best, hits) content = str(best.get("content") or "").strip() if not content: return self._guide_to_human("命中内容为空") # 正文只保留答案本身：固定免责声明由**治理层**统一追加（见文件头 `DISCLAIMER` 说明）， # 业务代码不再拼字符串——否则会出现两条重复声明，且合规文案变成不可配置的硬编码。 # # 可追溯性不受影响：本次命中哪个知识块仍由审计（agent.tool_executed 的工具调用记录） # 与消息表留痕，只是不面向客户展示。若将来要把出处给客户看，应当走 # source_references 的 knowledge 类型（需先让 ToolExecutor 登记本次可引用的 doc_id）， # 而不是继续往正文里拼字符串。 return CoreResult( text=content[:MAX_ANSWER_CHARS], intent=self._classified_intent, ) @staticmethod def _prefer_section(message: str, best: dict[str, Any], hits: list[Any]) -> dict[str, Any]: """客户问整个产品时，用整节块替换掉抢答的那一行。 行级子块是为了让「起投多少」拿到聚焦答案，但「介绍一下」会被某一行抢答 （实测返回了"产品期限 90天封闭期"，而客户要的是整个产品）。 判据不用问句分类器，而是看问句与子块标签是否真的对得上。标签取自子块的 section 末段（"起投金额""风险等级"）：「起投多少」含"起投"、「风险高吗」含"风险"， 都算对得上；「介绍一下」与任何标签都不重合，说明客户要的是整节。 父块由检索层按子块分数的 0.9 折算后一并带回。万一没带回来就仍用子块—— 宁可答得窄一点，也不要拿不相干的块去搪塞。 """ doc_id = str(best.get("doc_id") or "") # 末段恰为 2 位数字才是行级子块（PROD-007-04）；整节块自己的编号形如 PROD-901， # 用"含连字符"判断会把整节块误判成子块。 parent_id, _, tail = doc_id.rpartition("-") if not (parent_id and tail.isdigit() and len(tail) == 2): return best # 命中的本来就是整节 # 标签取自子块 title 的末段（title 由 " · " 连接），如"起投金额""风险等级" label = str(best.get("title") or "").split(" · ")[-1].strip() if label and label[:2] in message: return best # 客户问的正是这个字段 for hit in hits: if isinstance(hit, dict) and str(hit.get("doc_id") or "") == parent_id: return hit return best # ---- 出口一之二：适当性裁决（唯一给出"能不能买"结论的出口） ---- async def _answer_suitability( self, request: AgentRequest, context: RequestContext ) -> CoreResult: """回答「以我的风险等级能不能买这只产品」。 为什么不能靠知识检索直接答：客户问「c1客户能买它吗」，答案是**两个事实的组合** ——该产品的风险等级（R2）与客户档案里的等级能不能匹配。检索只能给出"最像的那段 原文"，实测给的是 C1 的通用规则，答非所问。 三条硬约束，缺任何一条都转人工： 1. 产品风险等级**从知识库查出来**，不猜、也不采信问句里出现的"R2"字样； 2. 客户等级**由底座按档案解析**（check_suitability 内部读 fin_risk_assessment， 带测评有效期），**不采信客户自称**——这次实测里客户说"C1"，档案其实是 C2； 3. 产品名**只取上一轮回答里的主语**（`_topic_of`，它来自知识块字段，可信）。 客户第一句就直接问"XX 能买吗"时取不到，那就转人工：这个出口会给出"能不能买" 的结论，宁可答不了也不能答错。 """ product = self._previous_topic(request) if not product: return self._guide_to_human("适当性问题里没识别出具体产品") risk_level = await self._product_risk_level(product, context) if risk_level is None: return self._guide_to_human(f"未查到「{product}」的风险等级") try: decision = await self.call_tool( SUITABILITY_TOOL, {"customer_id": context.user_id, "product_risk_level": risk_level}, intent=INTENT_SUITABILITY, context=context, ) except ForbiddenAgentError: # 与知识检索一致：白名单/权限类失败必须冒泡，那是配置错误， # 被兜底话术吞掉的话运维只会看到"客服一直引导人工"却查不出原因 raise except Exception: return self._guide_to_human("适当性校验调用失败") if not isinstance(decision, dict): return self._guide_to_human("适当性校验返回格式异常") text = self._suitability_text(product, risk_level, decision) claimed = self._self_claimed_level(request.message) if claimed: # 客户自述等级时必须点明判断依据，否则他会觉得"我明明说了我是 C3， # 你却说我没有任何测评结果"——两句话在他眼里是矛盾的。 # 依据在档案、不在自述，这既是合规要求，也得跟客户讲明白。 text = ( f"您提到自己是 {claimed}。适当性判断以您在公司留存的、在有效期内的" f"风险测评结果为准，不以本次自述为准。\n{text}" ) return CoreResult( text=text, intent=self._classified_intent, ) @staticmethod def _self_claimed_level(message: str) -> str: """客户在问句里自述的风险等级（"我是 C3""C3 客户能买吗"），没有则返回空串。 只用来在回答里说明判断依据，**绝不**参与裁决——客户等级只能来自档案。 """ for level in range(1, 6): if f"C{level}" in message.upper(): return f"C{level}" return "" @classmethod def _previous_topic(cls, request: AgentRequest) -> str: """上一轮回答里的主语（产品名）；取不到返回空串。""" for turn in reversed(request.history): if turn.role == "assistant": return cls._topic_of(turn.content) return "" async def _product_risk_level(self, product: str, context: RequestContext) -> int | None: """查产品的风险等级：问知识库要"风险等级"那一行，不从问句里猜。 产品手册里每个产品都有一行"风险等级 R2（中低风险）"，切分后是独立的行级子块， 所以按「{产品名} 风险等级」检索能直接命中。两重校验缺一不可：必须命中**风险等级 行**（否则可能匹配到"C1 可购买 R1、R2"那种列举，把 R1 当成产品等级），而且该行 必须属于**同一个产品**（否则会拿另一个产品的等级去做裁决）。 """ try: output = await self.call_tool( TOOL_NAME, {"query": f"{product} 风险等级", "top_k": 3}, intent=INTENT_SUITABILITY, context=context, ) except ForbiddenAgentError: raise except Exception: return None if not isinstance(output, dict) or output.get("degraded"): return None hits = output.get("hits") if not isinstance(hits, list): return None for hit in hits: if not isinstance(hit, dict): continue title = str(hit.get("title") or "") content = str(hit.get("content") or "") if "风险等级" not in title and "风险等级" not in content: continue if product not in title and product not in content: continue for level in range(1, 6): if f"R{level}" in content: return level return None @staticmethod def _suitability_text(product: str, risk_level: int, decision: dict[str, Any]) -> str: """把裁决结果说成人话。 只说裁决本身与依据，不复述产品资料——客户问的是"我能不能买"，资料在前一问 已经给过了。拒绝对原因下断言：reason_code 可能是等级不匹配、测评过期或未测评， 统一说成"超出风险承受能力"是错的；需要签揭示书时也**不写具体持仓比例**， 那是豁免条款里的业务参数，让客户照着一个数字去操作容易出偏差，留给人工讲。 """ level_name = RISK_LEVEL_NAMES.get(risk_level, f"R{risk_level}") customer_level = decision.get("customer_risk_level") if isinstance(customer_level, int): lines = [f"您当前的风险测评等级为 C{customer_level}。"] else: # 档案里没有在有效期内的测评结果。这是合规上的"不能卖"，但话要说清楚是 # "还没测评/已过期"，不能写成"您的等级为无"这种客户看不懂的句子。 lines = ["您目前没有在有效期内的风险测评结果。"] if decision.get("allowed"): # "在您的风险承受能力范围内"只对 C ≥ R 成立。矩阵允许的越级档（C1→R2、 # C2→R3）和豁免档（C3→R4、C4→R5）都**超出**了客户等级，一律说成"范围内" # 是把监管口径讲错：客户会以为自己的测评等级本来就覆盖这只产品。 if isinstance(customer_level, int) and customer_level >= risk_level: lines.insert( 0, f"{product}为 {level_name}，在您的风险承受能力范围内，可以购买。", ) elif decision.get("reason_code") == "SUITABLE_WITH_DISCLOSURE": lines.insert( 0, f"{product}为 {level_name}，高于您的风险承受能力等级。" "按照投资者适当性管理规定，签署产品风险揭示书后可以购买。", ) else: lines.insert( 0, f"{product}为 {level_name}，虽然高于您的风险测评等级，" "但仍在《个人投资者适当性管理指南》匹配矩阵允许购买的范围内。", ) if decision.get("required_disclosure"): lines.append("购买前需签署产品风险揭示书，具体请咨询您的客户经理。") if decision.get("requires_recording"): lines.append("本次购买需进行双录（录音录像）。") else: lines.insert( 0, f"{product}为 {level_name}，与您当前的风险测评结果不匹配，" "按照投资者适当性管理规定暂时无法购买。", ) lines.append("请先联系您的客户经理完成风险测评，之后即可查询可购买的产品范围。") valid_until = str(decision.get("assessment_valid_until") or "") if valid_until: lines.append(f"风险测评有效期至 {valid_until[:10]}，过期需重新测评。") return "\n".join(lines) # ---- 出口二：闲聊（提示词走发布配置） ---- 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=text[:MAX_ANSWER_CHARS], 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, ) # 客户这一句里出现这些词，说明主语要靠上文补全（"这个产品""该基金"）。 # 刻意不收"它/他"：单字指代在中文里极易误伤（"其他产品""其它"都含"他/它"）， # 而这类短句已经由 _MIN_STANDALONE_CHARS 覆盖，不需要靠它兜。 _REFERRING_WORDS = ( "这个", "那个", "这款", "这只", "该产品", "该基金", "上述", "前面提", "刚刚说", ) # 短到这种程度的问题，自己通常不构成完整意图（"起投多少""风险高吗"） _MIN_STANDALONE_CHARS = 8 @classmethod def _topic_of(cls, answer: str) -> str: """从客服上一轮的回答里取出"这一轮在说哪个产品"，取不出来返回空串。 遍历前几行而不是只看首行：适当性回答在客户自述等级时会先插一行 "您提到自己是 C3。…以您在公司留存的测评结果为准"，产品名被挤到第二行—— 只看首行会让紧随其后的追问丢掉指代对象（实测直接转人工）。 """ for line in answer.splitlines()[:4]: topic = cls._topic_in(line.strip().lstrip("#").strip()) if topic: return topic return "" @staticmethod def _topic_in(first: str) -> str: """解析单行里可能的主语；解析不出返回空串。 回答是我们自己组装的，主语只可能出现在三种形状里： - 行级块：「南方季季盈90天：起投金额 1万元」——首个冒号之前； - 整节块：「### 2.1 南方季季盈90天」——标题行； - 适当性：「南方季季盈90天为 R2（中低风险），…」——"为 R"之前。 """ if first.startswith("|"): return "" # Markdown 表格行不是主语（整节块去掉标题行之后就是表格） for level in range(1, 6): index = first.find(f"为 R{level}") if index >= 0: candidate = first[:index].strip() return candidate if 0 < len(candidate) <= 20 else "" topic = first.split("：", 1)[0].strip() if "：" in first else first # FAQ 型回答的首行是"问：C1 客户能买什么产品?"，冒号前只有一个"问"字； # 拿它当主语只会把检索词污染成"问 那它风险高吗"。兜底话术同理，它含"，"， # 会被下面的长度/标点校验挡掉。 if not topic or topic in {"问", "答"}: return "" # 政策条款不是产品：客户问一条规定、再追问"那它…"时，指代的是条款本身， # 把"第十二条 投资者与产品匹配矩阵"当成产品名只会把追问带到别处。 if topic.startswith("第") and "条" in topic[:6]: return "" if len(topic) > 20 or "，" in topic or "。" in topic: return "" # 整节块的标题带章节号（"2.1 南方季季盈90天"），必须剥掉：留着会让下游的 # "同一产品"校验失配——`_product_risk_level` 要求产品名是该块标题或正文的子串， # 而"2.1 南方季季盈90天"并不是"南方季季盈90天：风险等级 R2…"的子串， # 于是查不到风险等级、静默转人工。 head, _, rest = topic.partition(" ") if rest and head and all(part.isdigit() for part in head.split(".")): topic = rest.strip() return topic @classmethod def _search_query(cls, request: AgentRequest) -> str: """构造交给检索的查询串。 什么时候带上文（实测决定，两条都不能少）： 1. **客户这一句自己说不清楚时**才带。同一会话先问「季季盈90天的起投金额是多少」、 再问「基金赎回几天到账」，若无条件带上文，第二问会命中季季盈的产品块、 答出产品介绍——**答非所问**。在金融场景里这比"引导转人工"糟得多。 2. **带上文时只带"在说哪个产品"，不带上一轮的原话**。把上一轮整句拼进来会让 检索词语义变"宽"，反而只能命中粗粒度的整节块：实测「季季盈90天起投多少 那它风险高吗」命中的是整个产品小节，客户问的"风险"完全没有被聚焦； 换成「南方季季盈90天 那它风险高吗」才命中"风险等级"那一行。 换句话说：上文的作用是**补主语**，不是**补内容**。 只取客户的话、不取 Agent 自己的回答当内容：把 Agent 的措辞也拼进来会让检索偏向 自己上一轮的说法，而客户的真实意图可能已经在下一句里被修正过。这里从回答里取的 只有产品名这一个"主语"，不是它的论述。 """ message = request.message.strip() needs_context = ( len(message) < cls._MIN_STANDALONE_CHARS or any(word in message for word in cls._REFERRING_WORDS) ) if not needs_context: return message[:500] for turn in reversed(request.history): if turn.role == "assistant": topic = cls._topic_of(turn.content) if topic: return f"{topic} {message}"[:500] break return message[:500] # ---- 出口三：引导人工客服（不做工单，只回话并留痕） ---- 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)