diff --git a/app/service/agent/implementations/customer_service.py b/app/service/agent/implementations/customer_service.py index 4bcb605..fc7dfa0 100644 --- a/app/service/agent/implementations/customer_service.py +++ b/app/service/agent/implementations/customer_service.py @@ -37,11 +37,21 @@ AGENT_TYPE = "customer_service" 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) +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（高风险）", +} # 三档置信阈值：方案 §2.3 要求的是「绝对阈值 AND（相对间隙 OR 分布优势）」混合判定， # 只做绝对阈值会把口语化问法误判成"答不了"（这是实测踩到的坑）。 @@ -91,9 +101,10 @@ class CustomerServiceAgent(BaseAgent): allowed_roles=("customer",), allowed_portals=("api",), # 代码上限：实际可用范围由发布配置的意图白名单收窄（两者取交集） - allowed_tools=(TOOL_NAME,), + allowed_tools=(TOOL_NAME, SUITABILITY_TOOL), supported_intents=( - INTENT_FAQ, INTENT_PRODUCT, INTENT_POLICY, INTENT_CHITCHAT, INTENT_TRANSFER, + INTENT_FAQ, INTENT_PRODUCT, INTENT_POLICY, INTENT_SUITABILITY, + INTENT_CHITCHAT, INTENT_TRANSFER, ), ) @@ -106,6 +117,10 @@ class CustomerServiceAgent(BaseAgent): 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) # ---- 出口一：知识直返（faq / 产品 / 政策） ---- @@ -196,6 +211,132 @@ class CustomerServiceAgent(BaseAgent): 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("适当性校验返回格式异常") + return CoreResult( + text=f"{self._suitability_text(product, risk_level, decision)}\n{DISCLAIMER}", + intent=self._classified_intent, + ) + + @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"): + 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: diff --git a/tools/chat_console.py b/tools/chat_console.py index 9efe74f..f3fe235 100644 --- a/tools/chat_console.py +++ b/tools/chat_console.py @@ -36,7 +36,9 @@ from app.worker.runtime import WorkerRuntime sys.stdout.reconfigure(errors="replace") AGENT_TYPE = "customer_service" -FALLBACK_MARK = "客服热线" +# 兜底话术的固定开头。不能用"客服热线"这类词判断——正常的适当性回答里也会建议客户 +# 拨打客服热线，那样会把"已经给出了结论"误报成"已引导人工"。 +FALLBACK_MARK = "抱歉，这个问题我暂时无法给出准确答复" PAGE = """ diff --git a/tools/publish_customer_service_config.py b/tools/publish_customer_service_config.py index a9eac63..e636eab 100644 --- a/tools/publish_customer_service_config.py +++ b/tools/publish_customer_service_config.py @@ -33,9 +33,16 @@ from app.main import create_app ADMIN = "9003" AGENT_TYPE = "customer_service" TOOL_NAME = "search_knowledge" +SUITABILITY_TOOL = "check_suitability" # 只有会调用工具的意图才需要白名单；chitchat（模型生成）与 transfer_human（引导人工） # 都不查知识库。给它们配空白名单反而会掩盖"配置漏配"，因此不发布这两条。 -INTENTS = ("faq", "product_inquiry", "policy_explain") +INTENT_TOOLS: dict[str, tuple[str, ...]] = { + "faq": (TOOL_NAME,), + "product_inquiry": (TOOL_NAME,), + "policy_explain": (TOOL_NAME,), + # 适当性裁决要两步：先从知识库拿到产品的风险等级，再由底座按档案里的客户等级裁决 + "suitability_check": (TOOL_NAME, SUITABILITY_TOOL), +} def token(subject: str) -> str: @@ -104,12 +111,88 @@ async def etag_of(client: httpx.AsyncClient, path: str, auth: dict[str, str]) -> return (await client.get(path, headers=auth)).headers.get("ETag") +SUITABILITY_INTENT = "suitability_check" +# description 与 examples 是**给分类器看的**：意图码本身只是个名字，真正让模型分辨 +# "能买吗"和"这个产品是什么"的是这几个例子。所以 examples 全部取客户真实说法。 +SUITABILITY_INTENT_SPEC: dict[str, Any] = { + "intent_name": "投资者适当性判断", + "description": "客户询问以自己的风险承受能力能否购买某只产品，或询问自身风险等级与产品的匹配情况", + "examples": [ + "c1客户能买它吗", "我能买这个产品吗", "那它我能买吗", + "这只基金适合我吗", "我的风险等级能买吗", + ], + "confidence_threshold": "0.6000", +} + + +async def ensure_suitability_intent(client: httpx.AsyncClient, auth: dict[str, str]) -> int: + """确保 `suitability_check` 意图在运行期生效（返回 0 成功、1 失败）。 + + 为什么必须做这一步：意图码要三处对齐（见 `customer_service.py` 的注释），而运行期 + 只读 `agent_intent_config` 里 status='active' 的行。少了这一行，分类链路看不到这个 + 意图，"能买吗"会被分到别的意图里去，客户拿到的就是"C1 的通用规则"而不是结论。 + """ + path = "/api/v1/admin/agent-intent-configs" + listed = await client.get(f"{path}?limit=100", headers=auth) + rows = listed.json().get("data", []) if listed.status_code == 200 else [] + existing = next( + (row for row in rows + if row.get("agent_type") == AGENT_TYPE and row.get("intent_code") == SUITABILITY_INTENT), + None, + ) + if existing is not None and str(existing.get("status")) == "active": + print(f"[意图配置] id={existing['id']} 已生效，跳过") + return 0 + if existing is not None and str(existing.get("status")) in {"draft", "approved"}: + config_id = int(existing["id"]) + else: + # 没有历史行或历史行已归档：新版本号（该表 agent_type+intent_code+version 唯一） + version = int(existing.get("version", 0)) + 1 if existing else 1 + created = await post(client, path, auth=auth, payload={ + "agent_type": AGENT_TYPE, + "intent_code": SUITABILITY_INTENT, + **SUITABILITY_INTENT_SPEC, + "allowed_tools": [TOOL_NAME, SUITABILITY_TOOL], + "version": version, + }) + if created.status_code != 201: + print(f"[意图配置] 创建失败：{created.status_code} {created.text[:200]}") + return 1 + config_id = int(created.json()["data"]["id"]) + print(f"[意图配置] 已创建 id={config_id}（{SUITABILITY_INTENT} v{version}）") + + base = f"{path}/{config_id}" + # 幂等：重跑脚本时某一步可能已经推进过（"已经审过了"不该 409 让整个脚本失败） + settled = {"reviews": {"approved", "active"}, "activations": {"active"}} + for action, payload in ( + ("reviews", {"decision": "approved", "comment": "创建人自审"}), + # 激活端点要求 body 是对象；传 None 时 httpx 根本不发 body，会被判 422 + ("activations", {}), + ): + current = (await client.get(base, headers=auth)).json().get("data", {}) + if str(current.get("status")) in settled[action]: + print(f"[意图配置] {action} 已在目标状态（{current.get('status')}），跳过") + continue + response = await post( + client, f"{base}/{action}", auth=auth, payload=payload, + if_match=await etag_of(client, base, auth), + ) + if response.status_code != 200: + print(f"[意图配置] {action} 失败：{response.status_code} {response.text[:200]}") + return 1 + print(f"[意图配置] id={config_id} 已生效（运行期按 status='active' 读取）") + return 0 + + async def main() -> int: app = create_app() auth = {"Authorization": f"Bearer {token(ADMIN)}"} async with httpx.AsyncClient( transport=httpx.ASGITransport(app=app), base_url="http://test", timeout=60 ) as client: + if await ensure_suitability_intent(client, auth) != 0: + return 1 + inherited = await active_config_items() print(f"当前生效版本的配置项：{len(inherited)} 条（将原样继承）") for item in inherited: @@ -119,10 +202,10 @@ async def main() -> int: { "namespace": "agent_tools", "item_key": f"{AGENT_TYPE}:{intent}", - "value_json": {"allowed_tools": [TOOL_NAME]}, + "value_json": {"allowed_tools": list(tools)}, "schema_version": "1", } - for intent in INTENTS + for intent, tools in INTENT_TOOLS.items() ] inherited_keys = {(str(i["namespace"]), str(i["item_key"])) for i in inherited} pending = [