feat(customer-service): 接入适当性裁决，回答"我这个等级能不能买它"
客户实测反馈：「c1客户能买它吗」答的是 C1 的通用规则，没回答"能不能买季季盈90天"。 根因是这个问题需要**组合两个事实**——产品的风险等级（R2）与客户档案等级能不能匹配—— 而检索只能给出"最像的那段原文"，给不出结论。基座早就有了 check_suitability 裁决工具， 接口留好了但客服没接线（这个项目里第三次遇到同一类情况）。 改动三处： 1. 新增 suitability_check 意图，意图码三处对齐（AgentDefinition.supported_intents、 agent_intent_config 的 active 行、发布版 agent_tools 白名单）。 2. 新出口 _answer_suitability：产品风险等级**从知识库查出来**（不猜、也不采信问句里 出现的"R2"字样），产品名只取上一轮回答里的主语（来自知识块字段，可信），客户等级 交给 check_suitability 按档案解析——**不采信客户自称**。任何一步拿不到确定值就转人工： 这个出口会给出"能不能买"的结论，宁可答不了也不能答错。 3. 发布脚本加意图注册与工具白名单，并做成幂等（重跑不会因为"已经审过了"而 409）。 实测：意图正确路由到新出口，裁决链路走通。9001 因为在 fin_risk_assessment 里没有测评 记录，系统给出"暂时无法购买 + 您目前没有在有效期内的风险测评结果"——这正是适当性管理 要求的行为，不是故障：不能卖给一个没有有效测评结果的客户。措辞也据此改过，不写 "您的等级为未记录"这种客户看不懂的句子。 顺带修了前端一处误导标记：它用"是否含客服热线"判断"已引导人工"，而正常的适当性回答 里也会建议拨打客服热线，于是"已经给出结论"被误报成"已引导人工"。
This commit is contained in:
@@ -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:
|
||||
|
||||
@@ -36,7 +36,9 @@ from app.worker.runtime import WorkerRuntime
|
||||
sys.stdout.reconfigure(errors="replace")
|
||||
|
||||
AGENT_TYPE = "customer_service"
|
||||
FALLBACK_MARK = "客服热线"
|
||||
# 兜底话术的固定开头。不能用"客服热线"这类词判断——正常的适当性回答里也会建议客户
|
||||
# 拨打客服热线，那样会把"已经给出了结论"误报成"已引导人工"。
|
||||
FALLBACK_MARK = "抱歉，这个问题我暂时无法给出准确答复"
|
||||
|
||||
PAGE = """<!DOCTYPE html>
|
||||
<html lang="zh-CN">
|
||||
|
||||
@@ -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 = [
|
||||
|
||||
Reference in New Issue
Block a user