Files
group_fqcd_jr/app/service/agent/implementations/customer_service.py
T
张胜宇 c3d57fbf3f feat(cs)+docs: 修掉 P1 错分「风险测评结果」—— 画像问答改由受控工具作答(W15)
背景(用户提问触发):
  §1.2.1「客户能看本人的持仓/交易/账户/画像与风评」与 §1.4.5 P1
  「账户与个人数据(含风险测评结果)Agent 无权限读取」读起来互相矛盾。

实测根因(两处):
  1) route_message() 在画像分支之前,且 P1_KEYWORDS 含裸词「风险测评结果」
     ⇒「我的风险等级是多少」走画像作答,
       「我的风险测评结果是什么」被降级成「无法读取本人账户数据」
       —— 同一诉求两种结论,属「能答而不答」(H-03 同类)。
  2) 画像词在前、账户词在后的混问法漏网
     (「我的风险测评结果和持仓一起给我」落 P3 ⇒ 只答画像、静默忽略账户诉求)。

依据(决定性):D2.2 §1.7 第 21 项「画像问答字段直返」;
  D3.1 §0.3 术语表「画像问答属客服能力,与持仓查询严格区分」。

代码:
  - app/core/customer_service_rules.py:P1_KEYWORDS 移除裸词 + 口径说明;
    P1_PATTERNS 新增混问法守卫(第一人称 + 画像词 + 并列连词 + 账户词)。
  - customer_service.py / profile_projection.py:补「投影层白名单 ⊇ 客服对话
    渲染集」口径(total_asset / behavior_score / risk_tags 刻意不陈述,
    渲染它们等于用画像工具绕过 P1)。

守卫:
  - test_customer_service_rules.py:RT-004 → None;新增 RT-004b → P1;
    SAFETY_CASES 由字面区间改显式名单(RT-004b 字母后缀会落区间外被静默漏掉)。
  - test_customer_service_agent.py:画像端到端 3 条 + 混问法反向守卫。

安全不降:答案只来自 query_customer_profile(self 作用域 + 字段白名单 + 工具
  审计),查不到失败关闭、绝不猜等级;P0/P2 未动、P1 其余字面未动;
  混问法仍走 P1;访客问画像仍引导登录。

文档:D2.2 v2.5→v2.6 / D3.1 v2.4→v2.5 / D2.6 更正 / D4.6 追加 §3(不改正文)
  / D1.1 §24 + 版本位 / D1.6 §4.41 修正误记 + §4.42 / D2.1 v6.29。

实测:pytest 1914 passed / 3 skipped / 0 failed(+5);ruff 20(无新债);
  check_authoritative_docs 54 文档无冲突;_consistency GATE PASS;
  http_probe 11/11;定向真机复验 9/9;portal_api_check 35/0/5;
  e2e_smoke 31/31;fe_boundary 12/12;demo.ps1 五项自检全过。
2026-09-20 16:22:02 +08:00

2257 lines
125 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""客服 Agent:只回答能溯源到公司资料的问题,答不出来时分级回退,而不是一律转人工。
设计取向(金融场景):
- **确定性优先**:安全边界、档位、置信判定全部是确定性规则,不靠模型自律;命中知识块后
**直接返回原文**,不经模型改写——答案的字面内容来自公司已发布的资料,模型不参与事实生成,
因此不存在「编一个看起来合理的答案」的通路。
- **分级回退,而非一刀切**:原实现只有「答得出」与「转人工」两种结局,于是**一切不确定性
都变成转人工**(未命中 / 置信度略低 / 检索降级 / 画像查不到),这是「客服不智能」的主因。
现在改为 E5a 澄清 → E5b 部分答 + 引导 → E5c 转人工,且 **E5c 只在四类白名单原因下发生**。
- **转人工只由「用户显式要求」触发**(`F-2`):意图分类会把「客服电话是多少 / 怎么联系」这类
**问询**也标成 `transfer_human`,旧实现据此**直接建单**——这正是「很多问题强制转人工」的形态。
现在该标签只表示「先按最宽口径检索一次」,仍答不上来才转人工(原因码仍是白名单内的
`explicit_request`);真正的显式要求由 `route_message()` 在检索**之前**确定性拦下。
- **安全与智能不冲突**:放宽的只是「答不上来时的去向」,红线一条没松——零容忍词、提示词注入、
反诈、账户数据、代办争议仍全部由确定性规则在检索之前拦下。
五个出口:
E1 安全出口 ``route_message()`` 确定性拦截(P0 反诈 / 注入 / 合规 / P1 账户 / P2 代办争议)
E2 计算型出口 费率试算与 C—R 通用规则(H-02:只取**当前档位可见**的参数位,纯函数计算)
E3 知识直答 高置信命中 → 原文直返,不经模型
E4 证据约束生成 同一章节的多个小节**合并作答**(H-03:只依据证据包生成,
数字必须可解析到出处;只用包内事实,不推介不承诺不代办)
E5 分级回退 E5a 澄清 → E5b 部分答 + 引导 → E5c 转人工(白名单四类)
出口之前还有一道**访客侧投资建议护栏**(`C-09`):访客答复里出现投资建议类措辞时,
换成边界话术且不建单 —— 只对访客生效,客户侧不套用(两侧红线不同)。
另有**本人画像出口**与**适当性裁决出口**(确定性关键词识别,先于意图分发执行):
画像走 ``query_customer_profile`` 取权威字段,适当性走 ``check_suitability`` 组合产品与客户等级。
"""
import json
import logging
import re
from collections.abc import Sequence
from decimal import Decimal, InvalidOperation
from typing import Any
from app.core.actor import is_visitor
from app.core.contracts import (
AgentDefinition,
AgentRequest,
CoreResult,
IntentResult,
RequestContext,
SourceReference,
)
from app.core.customer_service_rules import (
ADVICE_BOUNDARY_REPLY,
COMPANY,
CONTACT_HOURS,
CONTACT_PHONE,
RISK_LEVEL_CHANGE_REPLY,
TRANSFER_REASON_EXPLICIT,
TRANSFER_REASONS,
SafetyRoute,
hits_zero_tolerance,
is_risk_level_change_request,
route_message,
visitor_advice_violation,
)
from app.core.errors import ForbiddenAgentError
from app.core.fund_fee_rules import (
CATEGORIES,
DISCLOSURE,
FORBIDDEN,
RedemptionTier,
is_fee_question,
is_general_suitability_question,
parse_amount_yuan,
parse_customer_level,
parse_fee_table,
parse_fund_category,
parse_holding_days,
parse_product_level,
parse_redemption_schedule,
parse_suitability_matrix,
purchase_fee_range,
redemption_tier,
)
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:<intent>` 白名单 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)
VISITOR_INTENTS = (INTENT_FAQ, INTENT_PRODUCT, INTENT_POLICY, INTENT_CHITCHAT, INTENT_TRANSFER)
TOOL_NAME = "search_knowledge"
VISITOR_TOOL_NAME = "query_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
#: 「本人客户分层 / 档位」的自述问法(`H-03` 实测漏网)。
#: 为什么要单独一组:知识库答不了「我够哪一档」—— 分层存在**本人档案**里
#: (`customer_tier`),只能走 `query_customer_profile`。旧实现没有这组词,该问句
#: 落到意图分类 `suitability_check` → `E5b-suitability`「给不出这个适当性结论」,
#: 属**能答而不答**(金标 `H-03` 期望出口含画像作答、`expect_profile_tool=True`)。
#: 判据刻意只认**第一人称 + 档位词**,且**不收裸词「等级」**:问「高净值客户是什么档」
#: 仍走知识库;问「我的等级能买 R5 吗」仍走适当性(那说的是产品风险等级)。
PROFILE_TIER_KEYWORDS: tuple[str, ...] = (
"哪一档", "哪个档", "什么档", "几档", "档位", "档次", "分档",
"分层", "星级", "客户层级",
)
#: 第一人称标记("本人"/"咱"是口语常见自称,一并认)。
FIRST_PERSON_MARKERS: tuple[str, ...] = ("我", "本人", "咱")
def is_profile_question(message: str) -> bool:
"""是否在问**本人画像**(确定性判定,模型不参与)。"""
if any(word in message for word in PROFILE_NEGATIVE_KEYWORDS):
return False
if any(word in message for word in PROFILE_KEYWORDS):
return True
return (
any(marker in message for marker in FIRST_PERSON_MARKERS)
and any(word in message for word in PROFILE_TIER_KEYWORDS)
)
#: 画像枚举码 → 客户可读标签。快照里存的是**内部码**:`profile_generation_service`
#: 直接照搬 `fin_customer_profile` 的列值,旧实现把码原样吐给客户 —— 实测 `H-03` 答复
#: 出现「投资期限偏好:short_term。交易频率:low。偏好资产类别:money_fund。」,
#: 客户看不懂,也不像一句客服说的话。
HORIZON_LABELS: dict[str, str] = {
"short_term": "短期(1 年以内)",
"medium_term": "中期(1—3 年)",
"long_term": "长期(3 年以上)",
}
FREQUENCY_LABELS: dict[str, str] = {"low": "较低", "medium": "中等", "high": "较高"}
ASSET_CLASS_LABELS: dict[str, str] = {
"money_fund": "货币基金",
"bond_fund": "债券基金",
"index_fund": "指数基金",
"equity_fund": "权益基金",
"bank_wm": "银行理财",
"private_equity": "私募股权",
"structured": "结构化产品",
}
#: 客户分层标签取语料里的公司口径(`HNW-001` / `FAQ-0014`:普通 / 金卡 / 白金 / 钻石 / 尊享)。
TIER_LABELS: dict[str, str] = {
"normal": "普通",
"ordinary": "普通",
"gold": "金卡",
"platinum": "白金",
"diamond": "钻石",
"exclusive": "尊享",
"premium": "尊享",
}
def _localized(value: object, labels: dict[str, str]) -> str:
"""把画像字段值说成人话;三条口径缺一不可。
1. 已知内部码 → 对应中文标签;
2. 值本身已是中文(另一条生产链写的是「3至5年」「固定收益类」这类标签)→ 原样保留;
3. **未知的非中文值 → 返回空串**:宁可这一句不说,也不把内部枚举码吐给客户
(既不猜它的含义,也不泄露内部编码)。
"""
text = str(value or "").strip()
if not text:
return ""
if text in labels:
return labels[text]
return "" if text.isascii() else text
def render_profile(profile: dict[str, object]) -> str:
"""把画像投影渲染成客户可读的一段话。
**只陈述该客户自己的字段**,不推断、不承诺收益;测评过期时**必须明说**并引导重新测评
(与适当性服务的 `ASSESSMENT_EXPIRED` 失败关闭口径一致)。
## 为什么只渲染这 6 类,而白名单里还有 `total_asset` / `behavior_score`
`profile_projection.PROFILE_FIELD_POLICY` 是**投影层**白名单(HTTP 画像端点与
`query_customer_profile` 工具共用),它决定"哪些字段可以出库";
本函数是**客服对话的渲染层**,它决定"哪些字段可以直接讲给客户听"。两者刻意不等:
- `total_asset`(资产规模)与 `behavior_score`(行为评分)属**账户/资产维度**,
正是 `D2.2` §1.4.5 `FR-CS-023` 的 `P1` 裁定的对象(账户数据 Agent 无权限读取)。
若在这里渲染,等于用画像工具**绕过 `P1`** —— 同一份数据换个入口就能读到,
`P1` 就不再是一条可验证的边界。
- `risk_tags` 是内部风控标签,措辞未经客户沟通话术审核,不对外陈述。
- `assessment_valid_until` 只用于**推导** `assessment_expired`(按当前时间重算),
对客户讲"有效期到某日"价值低、且易与适当性服务的失败关闭口径产生歧义。
⇒ 结论:**白名单 ⊃ 渲染集**是设计意图,不是漏渲染。要放开某个字段,
必须先改 `FR-CS-023` 的 `P1` 口径,并在 `P1` 的反向守卫用例中登记。
"""
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 = _localized(profile.get("investment_horizon"), HORIZON_LABELS)
if horizon:
lines.append(f"投资期限偏好:{horizon}。")
frequency = _localized(profile.get("trading_frequency"), FREQUENCY_LABELS)
if frequency:
lines.append(f"交易频率:{frequency}。")
assets = profile.get("preferred_asset_class")
if isinstance(assets, (list, tuple)):
asset_labels = [
label for label in (_localized(a, ASSET_CLASS_LABELS) for a in assets) if label
]
if asset_labels:
lines.append("偏好资产类别:" + "、".join(asset_labels) + "。")
tier = _localized(profile.get("customer_tier"), TIER_LABELS)
if tier:
lines.append(f"客户分层:{tier}。")
if profile.get("assessment_expired") is True:
lines.append("注意:您的风险测评已过有效期,需要重新完成测评后才能继续匹配产品风险等级。")
if not lines:
# 兜底不再写「建议转人工客服核实」:这条出口按 `D3.6` §4.3 白名单**不建单**,
# 一句话把客户推向人工,正是转人工率虚高的来源;改用与画像查不到同一口径的
# `PROFILE_MISS_TEMPLATE`(引导自助查看 + 给热线,不建单)。
return PROFILE_MISS_TEMPLATE
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 领先次优的最小间隙;领先不足说明有并列候选,不硬答
# 2026-09-19 `乙-19`(`DEC-18`)真实语料校准复核 —— **结论:维持以上三个常量不变**。
# 取证 `group_fqcd_jr/docs/evidence/20260919-t9-threshold-calibration.json`(46 条金标 ·
# 真实工具输出 + 实测出口)。方法:把 46 条按**实测接管分支**分三类,只在「由阈值-间隙
# 路径决定出口」的题上取谷 —— 安全路由 `P0`/`P1`/`P2`、合规、登录引导、计算型 `E2*`、
# 专项闸门 `E5b-*` 的出口由前置分支决定,与分数无关,一律不参与取谷。
# · 直返 `E3`(15 条,零模型调用):gap 最小 **0.0759**(`B-05`)
# · 合并 `E4`(15 条,调 1 次模型):gap 最大 **0.0649**(`B-03`)
# ⇒ `MIN_GAP` 实测可选取值区间 = **(0.0649, 0.0759)**,分界中点 0.0704;
# 现值 **0.07** 正落其中,两侧裕度各约 0.005。**它是唯一承重的常量**:
# 低于 0.0649 会让本该「同章节合并作答」的题退化成只答一节;
# 高于 0.0759 会把 `B-05` 这类只许直返的题推进 `E4`(= 能答而不答)。
# · 残余分支(`gap < MIN_GAP` 且证据包未成形):实测仅 `E-03`,score 0.5252
# ⇒ `HIGH_SCORE=0.75` 裕度 0.22,且无上界约束(没有必须 `E3` 的题依赖这条路径)。
# · `MID_SCORE`/`CLARIFY_SCORE`/`E4_MIN_SCORE` 均**低于**实测下限,属保守侧 —— 放宽才引入风险。
# ⚠️ 取谷必须用**真实工具输出**(`_search_query` 会补主语)与**实测出口**:
# 按原始问句独立检索取谷是错的(`E-02` 原始问句 gap 0.0054、真实查询 0.1784),本轮为此返工两次。
# ⚠️ `E5b` 要再拆:`_answer_from_evidence`(`E4` 分支)内部也会回 `_exit_partial`(记 `E5b`),
# 这类题仍由 `MIN_GAP` 支配 —— 用 `model_calls` 区分(调过模型 = 证据包成形)。
#: 检索条数:**10 而不是 5**。`E4`(`H-03`)要按「同一章节」分组,而同一章节的小节块
#: 常散落在第 5—8 位 —— `C-01` 实测 top5 只回 `HNW-005`/`HNW-006` 两档,放到 10 才把
#: `HNW-004`/`HNW-007` 取齐。放大不影响 `E3`:`hits[0]`/`hits[1]`、排序与保底名额都不变,
#: 只是「更多证据进了候选池」。单集合上限仍由 `search()` 的 `min(top_k, 20)` 兜住。
TOP_K = 10
MAX_ANSWER_CHARS = 1200
REFERENCE_LIMIT = 3
#: `E-06` 截断提示。此前超长答案是**静默截断**:客户看到一段读不到尾的资料,
#: 既不知道后面还有内容,也不会追问 —— 体验上等同于「客服没答完就不管了」。
#: 文案只承诺两件真的会做的事:说明这是摘要、邀请客户指明想看的条目。
TRUNCATION_NOTICE = (
"\n\n(以上为要点摘录,原文较长未全部展开。如需完整条款,请说说您想看哪一条,我来展开。)"
)
# 品牌名(`COMPANY`):**唯一来源是 `app/core/customer_service_rules.py`**,与本文件的
# `HOTLINE` 同理,由上方 import 直接转发。
# 本文件此前自写了一份品牌名常量,与规则文件的「南方基金」不一致 —— 结果是**同一个客服
# 两种自称**:闲聊提示词里一个名字、业务话术里另一个名字(2026-09-18 全库品牌扫描发现,
# 明细见 `docs/evidence/20260918-t2c-compliance-seed.json`)。常量各写一份必然漂移,
# 所以这里不再保留第二份定义。
# 客服热线与工作时间:**唯一来源是 `app/core/customer_service_rules.py`**,这里只做转发。
#
# 为什么必须转发而不是各写一份:这两处曾一度不一致 —— `customer_service_rules.CONTACT_PHONE`
# 是当时在用的客服手机号(安全路由出口在用),而本文件曾写**占位符号码**(兜底出口在用)。
# 后果是**同一个客服给客户两个不同的电话号码**:问"风险等级怎么划分"被安全路由处理时给真号码,
# 问一个知识库答不了的问题走兜底时给假号码 —— 客户按假号码永远打不通。
# 常量各写一份就一定会漂移,所以这里直接引用,改号码只需改 `customer_service_rules` 一处。
HOTLINE = CONTACT_PHONE
SERVICE_HOURS = CONTACT_HOURS
#: E5c 转人工话术。**只在白名单四类原因下使用**(见 `_exit_transfer`)。
TRANSFER_TEMPLATE = (
"这件事需要人工为您处理。"
f"请拨打客服热线 {HOTLINE}({SERVICE_HOURS}),我们会为您核实并跟进。"
)
#: E5b 部分答 + 引导。**不置 `transfer_required`**(不建单):知识未命中、置信度不足、
#: 检索降级、画像查不到都属"这次没查到",不是"必须人工办的事"。原实现把这四类一律渲染成
#: 同一句兜底话术并建单,是转人工率偏高的主因。
PARTIAL_TEMPLATE = (
"这个问题我暂时只能提供以下公开资料供您参考:\n\n{content}\n\n"
"如果需要更确切的答复,可以换个说法再问我一次(例如带上具体产品名称或条款名),"
f"也可以拨打客服热线 {HOTLINE}({SERVICE_HOURS})咨询。"
)
#: E5b 在连"部分命中"都没有时的说法(仍不建单)。
PARTIAL_EMPTY_TEMPLATE = (
"这个问题我没有找到对应的公开资料,也不想凭猜测回答您。"
"您可以换个说法再问我一次(例如带上具体产品名称、费率或条款名),"
f"也可以拨打客服热线 {HOTLINE}({SERVICE_HOURS})咨询。"
)
#: `E5b` **空答**(连部分资料都没给出)的两种确切文案 —— `F-2` 用它判定「答不上来」。
#: 直接比对上面两个模板常量本身:改文案时两边同源移动,不会出现「改了模板、判据静默失效」。
KNOWLEDGE_MISS_TEXTS = (
PARTIAL_EMPTY_TEMPLATE,
PARTIAL_TEMPLATE.format(content=""),
)
#: E5a 澄清:**一次只问一个问题**,候选必须来自当前主体可见档位——命中列表本身已按档位
#: 裁剪,因此这里天然满足"不得暗示不可见条目存在性"。
CLARIFY_TEMPLATE = (
"您的意思我还不确定,方便确认一下您想了解的是哪一项吗?\n{candidates}\n"
"回复序号或直接补充说明都可以。"
)
#: 画像查不到时的说明(E5b,不建单):引导自助查看,不推给人工建单。
PROFILE_MISS_TEMPLATE = (
"我这边暂时读不到您的画像信息,因此不能给您一个等级结论(这种结论不能靠猜)。"
"您可以在「我的账户」页面查看风险测评结果;"
f"如需协助,请拨打客服热线 {HOTLINE}({SERVICE_HOURS})。"
)
#: E5a 的澄清下限:低于此分连"候选"都算不上,直接走 E5b,不拿噪声去问客户。
CLARIFY_SCORE = 0.40
#: E5a 一次最多给的候选数(H-01 要求 2—3 个)。
CLARIFY_LIMIT = 3
#: E5b 部分答的展示下限:低于此分不展示内容,只说"没找到"。
PARTIAL_FLOOR = 0.35
#: 澄清轮次上限。轮次取自 `request.metadata.clarification_round`(由受理服务从会话行
#: 覆写进 outbox 元数据),**不是** `context.clarification_round`——后者在客服链路上没有
#: 写入方、恒为 0,用它会让「同话题上限 2 轮」失去作用。
MAX_CLARIFY_ROUNDS = 2
#: `E-04` 主动确认路径的**要求**文案(合规红线 2:先风险揭示、后客户确认)。
#:
#: 为什么必须有这一段:只写「签署风险揭示书后可以购买」,客户不知道下一步做什么,
#: 实测会转而要求转人工;而**代客户确认**是明令禁止的(「不得输出任何可被理解为
#: 客户已确认的措辞」)。因此这里的做法是:明确告诉他**怎么确认**、由谁来办,
#: 并显式声明客服不代替确认、不代客操作。
#: ⚠️ 文案里刻意不出现「安全」「保证收益」这类零容忍字面(见 `ZERO_TOLERANCE_WORDS`)。
SUITABILITY_CONFIRM_REQUEST = (
"如您已阅读并理解上述风险揭示,且愿意自行承担相应风险,"
"请明确回复「我已知悉风险,确认继续」。收到您的确认后,"
"由您的客户经理为您办理风险揭示书签署与双录。\n"
"说明:在您本人确认之前,本客服不会代替您做任何确认,也不会代您发起任何交易操作。"
)
#: 适当性结论给不出来时的说明(E5b,不建单):结论型判断**绝不猜**,但也不推给人工建单。
SUITABILITY_MISS_TEMPLATE = (
"我暂时给不出这个适当性结论——结论型判断不能靠猜,也不能只凭产品名称推断。"
"您可以登录后在「我的账户」页面查看风险测评结果与可购产品范围;"
f"如需协助,请拨打客服热线 {HOTLINE}({SERVICE_HOURS})。"
)
#: 适当性问题里没识别出具体产品时的澄清话术(E5a)。
SUITABILITY_CLARIFY_TEMPLATE = (
"想帮您核对是否与您的风险承受能力匹配,但我还没看出您指的是哪只产品。"
"请告诉我具体的基金名称或代码,我再为您核对。"
)
# ---------------------------------------------------------------------------
# 计算型出口 `E2`(费率试算 / C—R 通用规则)—— `H-02`
# ---------------------------------------------------------------------------
#
# 三条硬约束,缺一条这个出口就不该存在:
#
# ① 参数**只取自当前档位可见**的参数位:取参数与知识出口走**同一条** `call_tool` 检索,
# 档位由鉴权结果推导(`knowledge_contracts.tiers_for_roles`),本文件里**零个档位字面量**
# —— 于是「参数取不到就去别的档位再取一次」这条通路**在结构上不存在**(`INV-1`/`INV-5`);
# ② 计算是**纯函数**(`app/core/fund_fee_rules.py`):本文件只做「检索取参数 + 组织话术」,
# 不调模型,数字直接来自参数位;解析不出来就降级 `E5b`,**绝不猜一个数**;
# ③ 未指定具体产品时**只给算法与区间**,不给「最终只收 X 元」式的确定结论(DoD ③)——
# 同一类别下各产品费率并不相同(官方直销 1 折 0.15% vs 原费率 1.50%),给单值就是误导。
#
# 访客侧分项开放(`D3.6` §9.1 / `DEC-I8`):**公开产品的费用试算**与 **C—R 通用规则**对访客
# 开放(参数位本身就在 `public` 档);**以访客自身为对象的适当性结论**不在本出口 —— 那要读
# 画像,访客读不到,仍走登录引导。本出口也**不读任何持仓/资产数据**,只用客户自己说出的数字。
#: `E2` 取不到公开参数位时的说明(`E5b`,**不建单**)。算钱不能靠猜。
CALC_MISS_TEMPLATE = (
"这个问题要核对公开的产品参数,但我没有取到可以引用的公开费率资料,"
"所以不能给您一个数字(费率算错比不回答更糟)。"
"您可以换个说法再问我一次(例如带上基金类别或具体产品名称),"
f"也可以拨打客服热线 {HOTLINE}({SERVICE_HOURS})咨询。"
)
#: 申购费算法句 —— `D3.7` `D-01` 要求「给出算法 + 说明以产品说明书为准」。
CALC_PURCHASE_ALGORITHM = (
"申购费算法:申购费 = 申购金额 × 申购费率"
"(各销售渠道的费率优惠不同,最终以产品说明书与销售机构公示为准)。"
)
#: 赎回费算法句(`D6.2.1` §6.3 示例 2 口径:100,000 × 0.50% = 500 元)。
CALC_REDEMPTION_ALGORITHM = "赎回费算法:赎回费 = 赎回金额 × 赎回费率。"
#: 匹配矩阵结论的补充说明:把「通用规则」与「对您本人的结论」**分开**,不越界。
CALC_RULE_FOOTNOTE = (
"这是通用规则说明,不针对某一款产品;您本人的等级以您在公司留存的、"
"在有效期内的风险测评结果为准,本次自述不作为适当性判断依据。"
)
CALC_RULE_NEXT_VISITOR = "如需按您本人的测评结果核对,请先登录客户账户。"
CALC_RULE_NEXT_CUSTOMER = "如需核对您本人与某只具体产品的匹配情况,请告诉我产品名称或代码。"
#: 闲聊出口在模型不可用时的回退(E5b,不建单):礼貌收尾并把话题引回业务。
CHITCHAT_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 字回应,并把话题引导到业务上。"
# ---------------------------------------------------------------------------
# `E4` 证据约束生成(`H-03`)—— 同章节多块合并作答
# ---------------------------------------------------------------------------
#
# 为什么必须有这个出口:`D3.7` `C` 组的四类问题,答案**分散在同一章节的不同小节**里,
# 而检索是按「单块」排序的 —— 于是同一章节的几块分数咬得极紧(实测 `C-01` top1 0.7612 /
# top2 0.7321,gap 0.029)。旧实现只有两条路:原文直返**其中一块**("四档权益"只答一档,
# 金标判失败)或因为"打平"去澄清(资料明明就在库里)。`E4` 把这些块打成一个**证据包**
# 交给模型,要求它**只依据包内事实与数字**合并作答,并回填用到的 `doc_id`。
#
# 三条硬约束(`D3.7` `M-5`/`M-9` 直接盯这里),缺一条这个出口就不该存在:
# ① 输入是**证据包**(块 + `doc_id` + 族标识),不是单块原文;
# ② 约束**同时**落在 prompt 与**输出校验**两处 —— 只写 prompt 等于没有约束;
# ③ 答案里的数字必须能在**包内文本**里解析到出处,否则整条拦回 `E5b`。
#
# 安全边界一条没松:包本身先过了档位裁剪(`E4_MIN_SCORE` 之下的噪声不进包,档位由
# `call_tool` 按鉴权结果推导);生成结果还要过零容忍与访客投资建议两道**确定性**校验;
# 模型"判断答不了"(缺主语 / 过泛 / 证据不足)时**不接管**,交回 E5a/E5b,绝不硬编。
E4_PROMPT_CODE = "customer_service_evidence_answer"
E4_TASK_TYPE = "text_generation"
#: 参与「同章节」分组的分数下限:低于它的命中是噪声,进包只会让模型分心。
E4_MIN_SCORE = 0.5
#: 证据包上限(块)。太多会让模型抓不住重点,也会把"包内数字"的校验面撑得过宽。
E4_MAX_EVIDENCE = 6
#: 证据包在提示词里的标题。
E4_EVIDENCE_HEADER = "【证据】"
#: `F-3` **主体相关性闸门**(`E3`/`E4` 共用):库内已知的「服务名 / 条款名 / 主题词」。
#:
#: 触发条件**只有一条**:问句里出现下表中的某个词。此时要求该词**至少落在一个命中块的
#: `title` 或 `content` 里**;一块都不含 → 判定「证据与问题主体无关」,交回 `E5b`
#: (**不生成、不转人工**)。问句不含本表任何词时**闸门不启用**,行为与改造前逐字相同。
#:
#: 「只在点名主题时生效」这条限定是**刻意**的:把闸门开成"任何问句都要与证据有词面重合"
#: 会把 `C-02`「资产到多少能升级?」这类**没有主题词但检索正确**的问句一并误伤 ——
#: 实测该问句与证据几乎无字面重合,却必须答(金标期望 `E4` 合并作答)。
#:
#: 实测来源(`D1.6` §4.27 五):`B-04`「南方基金投顾服务起点是多少?」召回的是产品手册的
#: 产品参数块(**一块都没提到投顾**),`E4` 却按"同章节多块"接管,答成某只混合基金的整段
#: 参数 —— 题目问 A、答案是 B。
EVIDENCE_SUBJECT_TERMS: tuple[str, ...] = (
"投顾服务", "基金投顾", "投顾",
"高净值", "分层权益", "专属权益", "增值服务",
"专业投资者", "普通投资者",
"申购费", "赎回费", "管理费", "托管费", "销售服务费", "费率",
"起投金额", "起投", "门槛",
"风险等级", "风险测评", "适当性",
"到账", "确认份额", "交易时限",
"定投", "基金转换", "分红",
"开户", "销户", "信息披露", "反洗钱", "投诉",
)
#: `F-3` 闸门拦下时的说明(`E5b`,**不建单、不生成**):宁可说"没找到",也不拿相近条款替答。
#: 措辞有意**不写死档位判断**:闸门只知道"可见资料里没有",不知道它是否在更高档位里。
SUBJECT_MISS_TEMPLATE = (
"关于「{subject}」这一点,我在您当前可见的资料里没有找到对应内容,"
"也不想用相近的条款替您作答。\n"
f"您可以换个说法再问我一次;如果这项内容需要登录后才能查询,登录后我再为您核对,"
f"也可以拨打客服热线 {HOTLINE}({SERVICE_HOURS})咨询。"
)
DEFAULT_EVIDENCE_SYSTEM = (
f"你是{COMPANY}的智能客服助手。你只能依据给出的【证据】回答用户问题,"
"并且只输出一个 JSON 对象。四条红线:① 只使用【证据】里出现过的事实与数字,"
"不得引入证据之外的任何信息、数字、产品名称或推测;② 不得给出投资建议、"
"不得承诺收益、不得代替客户办理任何业务;③ 证据不足以完整回答时不要勉强作答;"
"④ 「我司不经营某项业务」这类**否定事实**只要有证据就是可答的,不得当作证据不足。"
)
DEFAULT_EVIDENCE_TEMPLATE = (
"{evidence}\n\n【用户问题】{message}\n\n"
"请只输出一个 JSON 对象(不要输出任何其它文字),字段为:"
"answer(合并后的中文答复,字符串)、used_chunk_ids(本条答复用到的 doc_id 数组)、"
"confidence(0 到 1 之间的小数)、unanswerable_reason(无法作答时的原因,字符串)。"
"规则:答案里出现的每个数字都必须能在【证据】原文里找到出处;"
"若问题缺少必要主语(例如只有「它」「这个」)、或过于笼统而无法确定要回答哪一项、"
"或证据不足以完整回答,请把 answer 留空,并在 unanswerable_reason 里填"
"missing_subject / too_broad / insufficient_evidence / subject_mismatch 之一"
"(**若【证据】里没有任何一块与问题的主体相关,一律填 subject_mismatch**)。"
"两条**不算答不了**的反例,命中时必须照常作答(都**不要**填 subject_mismatch):"
"① 若【证据】说明我司**不经营/不代销/不销售**问题里那类业务(例:我司只销售本公司"
"管理的产品,不代销其他基金管理人的基金、银行理财产品与保险产品),这是有据可答的"
"**否定事实**:请据实作答并给出可自助的替代路径;"
"② 若问题里的主体或产品名称在【证据】里**完全查不到**(名称被说错、或问了不存在的"
"公司/产品),**不要**回绝、不要反问、**也不要在答复里出现那个名称(一个字都不要重复)**"
":直接据【证据】给出最接近的**正确**事实(如我司全称、简称、业务范围、组织架构),"
"让用户从正确信息里自己看出差异即可。"
"**复述那个名称等于把不实信息再传播一次** —— 实测写「未查询到名为「某科技」的公司」"
"会把不实名称原样印进答复,而禁忌判据正是这个名字;正确的做法是**根本不提它**。"
"③ 答复里**不要出现这些承诺性字面**:「保证收益」「保本」「稳赚」「无风险」「不会亏」"
"「一定安全」。需要表达相同意思时改写成**否定式/不确定式**(如「不承诺收益」"
"「收益并非确定」「可能亏损」)。这条是给「要不要写这几个字」把关,"
"**不是**在放宽上面任何一条红线:输出侧有逐字校验,写出来会被整条拦回 `E5b`,"
"客户只会拿到一段没有被回答的资料。"
)
class CustomerServiceAgent(BaseAgent):
definition = AgentDefinition(
agent_type=AGENT_TYPE,
version="1.0.0",
allowed_roles=("visitor", "customer"),
allowed_portals=("api",),
# 代码上限:实际可用范围由发布配置的意图白名单收窄(两者取交集)
allowed_tools=(TOOL_NAME, VISITOR_TOOL_NAME, SUITABILITY_TOOL, PROFILE_TOOL_NAME),
supported_intents=(
INTENT_FAQ, INTENT_PRODUCT, INTENT_POLICY, INTENT_SUITABILITY,
INTENT_CHITCHAT, INTENT_TRANSFER,
),
# 客服不隐式召回长期画像;已登录用户的画像查询必须显式调用受控工具。
recalls_customer_memory=False,
)
def __init__(self, definition: AgentDefinition | None = None) -> None:
"""Use the class definition for direct tests and factory-created instances alike."""
super().__init__(definition or self.definition)
async def handle(self, request: AgentRequest, context: RequestContext) -> CoreResult:
"""出口入口:跑完所有分支,再做一次**访客侧投资建议护栏**(`C-09`)。"""
result = await self._route_and_answer(request, context)
return self._guard_visitor_advice(result, context)
def _guard_visitor_advice(
self, result: CoreResult, context: RequestContext
) -> CoreResult:
"""访客答复里出现投资建议类措辞时,换成边界话术(`C-09` 输出侧按主体分化)。
**只对访客生效**:客户侧不套用这套规则 —— 两侧红线不同(见
`VISITOR_ADVICE_PATTERNS` 的说明)。触发时**不建单**:内容被换掉不是
「必须人来办的事」,访客也没有工单承接方。
"""
if not is_visitor(context):
return result
hit = visitor_advice_violation(result.text)
if hit is None:
return result
logger.info(
"visitor advice guard replaced answer: agent_type=%s hit=%r", AGENT_TYPE, hit
)
return result.model_copy(update={
"text": ADVICE_BOUNDARY_REPLY,
"transfer_required": False,
"transfer_reason": None,
})
def _exit_safety(self, safety: SafetyRoute) -> CoreResult:
"""安全出口(`P0` 反诈/凭据、注入、合规、推介边界、`P1` 账户、`P2` 代办争议)。
抽成**独立出口方法**而不是内联在 `_route_and_answer` 里,两个理由:
1. 评测探针靠「包装终止型方法」给每条打出口标(`D3.7` §3)。内联分支打不到标,
`F-02`/`F-03` 这类**合规拒答**会被记成「没有任何出口」,评测于是把做对了的
安全行为判成未达标 —— **度量工具失真比产品缺陷更危险**,它会误导后续决策。
2. 这是本模块最该被单点审计的分支(`D3.6` 称「`E1` 安全出口」):独立成方法后
AST 守卫与单测都够得着它,改动这个分支必然留下痕迹。
**行为与内联时逐字等价**(文案 / 意图 / 建单三件套都不变):`text` 取 `route.reply`;
`intent` 恒取 `route.intent` 且置信度 1.0(确定性判定不需要模型打分,也不该被
模型改判);建单与否完全跟随 `route.transfer_required` / `route.transfer_reason`
—— 即 `P0`/`P2` 建单、`P1` 与合规**不建单**(`H-04` 口径)。
"""
return CoreResult(
text=safety.reply,
intent=IntentResult(intent=safety.intent, confidence=1.0),
transfer_required=safety.transfer_required,
transfer_reason=safety.transfer_reason,
)
def _exit_risk_level_change(self) -> CoreResult:
"""`E-08` 红线 1 ④:风险等级不可代办的确定性答复(**不建单**)。
为什么单独成出口而不是并进 `SafetyRoute`:它不是「安全事件」(没有人在受威胁),
也不该占用四类转人工白名单里的任何一类 —— 转人工在这里毫无用处,因为**人工
也不能改**。独立成方法同时让评测探针能给它打标,便于按条复盘。
"""
logger.info("E-guard 风险等级代办请求:确定性拦截(不建单)")
return CoreResult(
text=RISK_LEVEL_CHANGE_REPLY,
intent=IntentResult(intent=INTENT_FAQ, confidence=1.0),
)
async def _route_and_answer(
self, request: AgentRequest, context: RequestContext
) -> CoreResult:
# 先执行确定性的安全与权限边界路由;这些分支不查知识库、不调用模型,
# 从根上阻断账户敏感数据、凭据泄露、诈骗和代办交易等越界请求。
safety = route_message(request.message)
if safety is not None:
return self._exit_safety(safety)
# `E-08` 红线 1 ④:要求「改 / 提 / 降 风险等级」 → **确定性拦截**。
# 位置与安全路由并列、在检索之前:这类请求不对应任何可办事项,
# 交给检索就等于「回答什么取决于召回到哪一段」。
if is_risk_level_change_request(request.message):
return self._exit_risk_level_change()
# 画像问题优先处理(确定性关键词,不走意图分类):知识库答不了"我的风险等级是多少",
# 那需要读该客户的画像数据,必须走 `query_customer_profile` 工具取权威字段。
# 放在意图分发**之前**是有意的:让画像能力不依赖意图分类是否恰好给出 faq。
# 连带效果:该分支的 `intent` 恒为 `faq`,与 `PROFILE_WHITELIST_INTENT` 同源。
if is_profile_question(request.message):
if is_visitor(context):
return self._guide_to_login("访客不能查询个人画像")
return await self._answer_profile(request, context)
# 计算型出口 `E2`(费率试算 / C—R 通用规则):**确定性触发**(词法判定,
# 不经过意图分类),因此对客户与访客**同样生效**。
#
# 为什么放在访客意图白名单**之前**:`suitability_check` 不在 `VISITOR_INTENTS` 里,
# 若按常规顺序,访客问「C1 能买 R3 的产品吗」会被直接推去登录;而 `DEC-I8` 已裁定
# 这类**通用规则**对访客开放(它不需要读任何画像数据)。以访客**自身**为对象的
# 适当性结论不在这里 —— 那要读画像,仍走下面的登录引导。
calculation = await self._answer_calculation(request, context)
if calculation is not None:
return calculation
intent = self._intent_code()
if is_visitor(context) and intent not in VISITOR_INTENTS:
if intent == INTENT_SUITABILITY:
# 例外:**C—R 匹配规则本身**对访客开放(`DEC-I8` / `D3.6` §9.1)。
# 实测意图分类器会把「C1 客户能买什么?C5 呢?」判成 `suitability_check`
# (0.95),而该意图不在 `VISITOR_INTENTS` 里 —— 不加这一条,一条**公开规则题**
# 会被直接推去登录(金标 `C-04` 期望的正是 `E4` 合并作答 C—R 矩阵)。
# 用 `faq` 当工具白名单 key:发布配置里只有 `customer_service:faq` 一个 key
# (与 `PROFILE_WHITELIST_INTENT` 同源),换新 key 会让工具交集为空。
# **以访客自身为对象的适当性结论不在这里** —— 那要读测评结果,仍走登录引导。
return await self._answer_from_knowledge(request, context, INTENT_FAQ)
return self._guide_to_login("访客请求超出公开服务范围")
if intent == INTENT_CHITCHAT:
return await self._chitchat(request)
if intent == INTENT_TRANSFER:
# `F-2`(2026-09-19 裁定):转人工**只由「用户显式要求」触发**,意图标签不再直通。
#
# 实测意图分类器把「南方基金客服现在方便联系吗?」判成 transfer_human(0.9)——
# 它问的是**联系方式**(`COMP-022-01` 含 400-889-8899 / 7:00—22:00),旧实现却直接
# 建单。**分类标签 ≠ 用户诉求**:真正的显式要求(转人工 / 找人工 / 要人工 / 真人)
# 已在 `route_message()` 里确定性拦下(`TRANSFER_REASON_EXPLICIT`)且先于检索生效,
# 所以本分支不必再承担这件事——改为「**先检索一次,答不上来再转**」。
answer = await self._answer_from_knowledge(request, context, INTENT_FAQ)
# 只有知识点**一点可用资料都没给出**(`E5b` 空答)才转人工;给出部分内容或澄清
# 候选的都算答到了一部分,**不建单**(`E5b` 本身也从不建单,见 `D3.7` §5)。
if self._knowledge_missed(answer):
return self._exit_transfer(TRANSFER_REASON_EXPLICIT)
return answer
if intent not in BUSINESS_INTENTS:
# 分类失败或意图未覆盖:**不再直接转人工**。这类问法常常只是措辞口语化,
# 知识库里其实有答案,退回最宽的 faq 检索试一次;确实答不上来由 E5 分级回退处理。
return await self._answer_from_knowledge(request, context, INTENT_FAQ)
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("画像查询失败,改为 E5b 如实告知", exc_info=True)
return self._exit_profile_miss()
profile = output.get("profile") if isinstance(output, dict) else None
if not isinstance(profile, dict) or not profile:
logger.info("画像为空或不可用,改为 E5b 如实告知")
return self._exit_profile_miss()
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:
knowledge_tool = VISITOR_TOOL_NAME if is_visitor(context) else TOOL_NAME
output = await self.call_tool(
knowledge_tool,
{"query": self._search_query(request), "top_k": TOP_K},
intent=intent,
context=context,
)
except ForbiddenAgentError:
# 白名单/权限类失败必须冒泡:那是配置错误,若被兜底话术吞掉,
# 运维会看到"客服一直引导人工"却查不出原因。
raise
except Exception:
return self._exit_partial([], note="知识检索调用失败")
if not isinstance(output, dict):
return self._exit_partial([], note="知识检索返回格式异常")
if output.get("degraded"):
reason = str(output.get("reason") or "未知")
return self._exit_partial([], note=f"知识检索降级:{reason}")
hits = output.get("hits")
if not isinstance(hits, list) or not hits:
return self._exit_partial([], note="知识库未命中")
best = hits[0]
if not isinstance(best, dict):
return self._exit_partial([], note="命中内容格式异常")
# `F-3` 主体相关性闸门:问句点名了某个主题(如「投顾服务」),但**一块命中都没提到它**
# —— 这是"检索拿相近概念凑数",直返原文或合并生成都会答非所问(实测 `B-04`)。
# 放在 `E4` 与置信判定**之前**:这类答复不是"置信度不够",而是"证据与问题无关"。
subject_terms = self._subject_terms_in(request.message)
if not self._subject_covered_by(hits, subject_terms):
return self._exit_subject_miss(subject_terms[0])
score = self._score(best.get("score"))
gap = score - self._second_score(hits)
# `E4` 证据约束生成(`H-03`):同一章节的多个小节分数咬得紧时**合并作答**。
# 放在置信判定**之前**:`C-01` 的 top1 分数 0.7612 已经够"直接答",
# 但直返的只是四档权益里的一档(金标判"只答其中一档 = 失败")——
# 高置信不等于答案完整。模型自述答不了时 `_answer_from_evidence` 回 `None`,
# 下面原有的 E3 / E5a / E5b 判定**原样生效**。
evidence = self._evidence_pack(hits, gap=gap)
if evidence is not None:
generated = await self._answer_from_evidence(request, context, evidence)
if generated is not None:
return generated
# 混合判定:高置信直接答;中置信必须同时满足「分数够」与「领先次优够多」。
# 只满足其一的(分数够但两三个候选并驾齐驱)**先问一句**——金融场景下"答错"
# 不可接受,但"直接推给人工"是最差的一档;问清比推走好,答准比问更好。
confident = score >= HIGH_SCORE
if not confident and not (score >= MID_SCORE and gap >= MIN_GAP):
# E5a:够格当候选、且还没问满两轮 → 问一句,而不是推给人工。
# **只在 `_clarify_reason` 给出的四类触发条件下问**:问错了比不问更伤体验,
# 而"什么不确定都问一句"会把客服变成审问(`H-01` DoD ②⑥)。
rounds = request.metadata.clarification_round
# 变量名不与上方「检索降级」分支的 `reason` 复用:同名会让 mypy 把
# `str | None` 判成对 `str` 的赋值(实测)。
clarify_reason = self._clarify_reason(request, hits, score=score, gap=gap)
# `W6`:候选同属一个文档时**不问**—— 范围已经清楚(理由见 `_same_document`),
# 直接走下面的 `E5b` 部分答,把材料给客户。
if (
clarify_reason is not None
and rounds < MAX_CLARIFY_ROUNDS
and not self._same_document(hits[:CLARIFY_LIMIT])
):
clarified = self._exit_clarify(hits)
if clarified is not None:
logger.info(
"E5a 澄清:reason=%s round=%s score=%.3f gap=%.3f",
clarify_reason, rounds, score, gap,
)
return clarified
return self._exit_partial(
hits, note=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._exit_partial(hits, note="命中内容为空")
# 正文只保留答案本身:固定免责声明由**治理层**统一追加(见文件头 `DISCLAIMER` 说明),
# 业务代码不再拼字符串——否则会出现两条重复声明,且合规文案变成不可配置的硬编码。
#
# 可追溯性不受影响:本次命中哪个知识块仍由审计(agent.tool_executed 的工具调用记录)
# 与消息表留痕,只是不面向客户展示。若将来要把出处给客户看,应当走
# source_references 的 knowledge 类型(需先让 ToolExecutor 登记本次可引用的 doc_id),
# 而不是继续往正文里拼字符串。
return CoreResult(
text=self._clamp_answer(content),
topic=self._declared_topic(content),
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
# ---- 出口一之三:计算型(费率试算 / C—R 通用规则) ----
async def _answer_calculation(
self, request: AgentRequest, context: RequestContext
) -> CoreResult | None:
"""`E2` 计算型出口。返回 `None` 表示「这不是计算型问题」,交回主分发。
触发是**词法判定**(与画像出口同理,不依赖意图分类器):费用类问句要有
「基金类别」,或上文给得出的主语;通用匹配规则要有「客户等级 + 产品等级 +
购买性问句」三件套。**宁可漏触发**(漏了还有 `E3` 原文直答兜着),也不要抢答 ——
把知识型问题截进计算出口,客户会拿到一段答非所问的费率。
"""
message = request.message
if is_general_suitability_question(message):
return await self._answer_suitability_rule(message, context)
if not is_fee_question(message):
return None
category = parse_fund_category(message)
if category is None:
# 类别的两个来源:这一句自己说的,或**上一轮回答的主语**。后者是 `D-03`
# 那种追问 —— 「持有 8 个月赎回要付费吗?」本身没提类别,类别在上文里。
topic = self._previous_topic(request)
if topic in CATEGORIES:
category = topic
elif topic and "赎回" in message:
return await self._answer_product_redemption(request, context, topic)
if category is None:
return None
return await self._answer_category_fee(message, context, category)
async def _search_parameters(
self, query: str, intent: str, context: RequestContext
) -> list[Any] | None:
"""取参数位:与知识出口**同一条**按档位裁剪的检索,命中即返回。
返回 `None` 表示参数拿不到(检索降级 / 调用异常 / 返回格式异常 / 未命中),
调用方据此降级 `E5b`。**绝不**换一个档位或换一个集合再取一次:`tiers` 由鉴权结果
推导,本文件连档位字面量都没有,那条通路在结构上就不存在(`H-02` DoD ⑤)。
"""
try:
knowledge_tool = VISITOR_TOOL_NAME if is_visitor(context) else TOOL_NAME
output = await self.call_tool(
knowledge_tool,
{"query": query[:500], "top_k": TOP_K},
intent=intent,
context=context,
)
except ForbiddenAgentError:
# 白名单/权限类失败必须冒泡(与知识出口同一口径):那是配置错误,
# 用兜底话术吞掉会让「工具没被授权」表现成「资料查不到」。
raise
except Exception:
logger.info("E2 参数检索调用失败", exc_info=True)
return None
if not isinstance(output, dict) or output.get("degraded"):
return None
hits = output.get("hits")
if not isinstance(hits, list) or not hits:
return None
return hits
async def _answer_category_fee(
self, message: str, context: RequestContext, category: str
) -> CoreResult:
"""按**基金类别**回答费用(`D3.7` `D-01`~`D-03`):参数位是 §6.1 费率总表。"""
hits = await self._search_parameters(
f"公募基金费率总表 {category} 申购费率 赎回费率", INTENT_PRODUCT, context
)
if hits is None:
return self._exit_calc_miss("费率参数检索不可用")
fees = None
for hit in hits:
if not isinstance(hit, dict):
continue
table = parse_fee_table(str(hit.get("content") or ""))
if table is not None and category in table.by_category:
fees = table.by_category[category]
break
if fees is None:
return self._exit_calc_miss(f"未取到「{category}」可用的公开费率参数位")
days = parse_holding_days(message)
amount = parse_amount_yuan(message)
wants_redemption = "赎回" in message
wants_purchase = "申购" in message or "认购" in message
if not wants_redemption and not wants_purchase:
# 只说「费率 / 费用怎么收」:两类都给,省得客户再问一遍。
wants_redemption = wants_purchase = True
parts: list[str] = []
if wants_purchase:
rate_range = purchase_fee_range(fees)
if rate_range is not None:
parts.append(self._purchase_text(category, rate_range, amount))
if wants_redemption:
redemption = self._redemption_text(
category, fees.redemption, days=days, amount=amount
)
if redemption is None:
return self._exit_calc_miss(f"「{category}」赎回费率档位覆盖不到客户问的持有期")
parts.append(redemption)
if not parts:
return self._exit_calc_miss(f"「{category}」没有可展示的公开费率参数")
return CoreResult(
text=self._clamp_answer("\n\n".join(parts)),
intent=self._classified_intent, topic=category,
)
@classmethod
def _purchase_text(
cls, category: str, rate_range: tuple[float, float], amount: int | None
) -> str:
"""申购费话术:**区间 + 算法**(DoD ③:未指定具体产品不给确定结论)。"""
low, high = rate_range
text = (
f"{category}:公开申购费率区间为 {cls._format_rate(low)}—{cls._format_rate(high)}。\n"
f"{CALC_PURCHASE_ALGORITHM}"
)
if amount and low != high:
# 只有真的存在区间才给金额区间:低 = 高时那就是一个确定的数,
# 等于 DoD ③ 明令禁止的「最终只收 X 元」式结论(哪怕那个数算对了)。
text += (
f"\n按您提到的 {amount:,} 元估算,申购费大致在 "
f"{cls._money(amount * low)} 元—{cls._money(amount * high)} 元之间。"
)
return text
@classmethod
def _redemption_text(
cls,
category: str,
tiers: Sequence[RedemptionTier],
*,
days: int | None,
amount: int | None,
) -> str | None:
"""赎回费话术。**指定了持有期就必须能套到档**,套不到返回 `None`(不猜)。"""
lines: list[str] = []
if days is not None:
tier = redemption_tier(days, tiers)
if tier is None:
return None
lines.append(
f"{category}:按您提到的持有 {days} 天,对应档位「{tier.label}」,"
f"赎回费率为 {cls._format_rate(tier.rate)}。"
)
if amount:
lines.append(
f"按 {amount:,} 元估算,赎回费约为 {cls._money(amount * tier.rate)} 元"
"(实际以赎回金额与基金合同为准)。"
)
lines.append("")
else:
lines.append(f"{category}:公开赎回费率按持有期分档。")
lines.append("公开档位如下:")
lines.append(cls._render_tiers(tiers))
lines.append(CALC_REDEMPTION_ALGORITHM)
return "\n".join(lines)
@classmethod
def _render_tiers(cls, tiers: Sequence[RedemptionTier]) -> str:
"""档位表渲染成客户可读的行。**标签与费率都照抄参数位原文**,不做换算。"""
def labelled(tier: RedemptionTier) -> str:
if tier.label.startswith("持有"):
return tier.label
# 数字开头的档位补一个空格(`持有 7—30 天`);`<7 天` 这种符号开头的不补,
# 与语料原文的 `持有<7 天` 保持一致。
return f"持有 {tier.label}" if tier.label[:1].isdigit() else f"持有{tier.label}"
return "\n".join(
f"- {labelled(tier)}:{cls._format_rate(tier.rate)}" for tier in tiers
)
@staticmethod
def _format_rate(rate: float) -> str:
"""费率小数 → 百分比文字(`0.0075` → `0.75%`)。"""
return f"{round(rate * 100, 4):g}%"
@staticmethod
def _money(value: float) -> str:
"""金额 → 千分位文字,整数不带小数尾巴(`1500.0` → `1,500`)。
`rstrip("0")` 是个陷阱:`"1,000.00"` 会被削成 `"1,"`。只能削**小数部分**。
"""
whole, _, fraction = f"{value:,.2f}".partition(".")
fraction = fraction.rstrip("0")
return f"{whole}.{fraction}" if fraction else whole
async def _answer_product_redemption(
self, request: AgentRequest, context: RequestContext, product: str
) -> CoreResult:
"""指定产品的赎回费(`E2c`):参数位取自**该产品自己那一块**,不跨块拼。
跨块拼是这条路上最容易犯的错:命中里同时有总表和产品块时,从 A 块取档位标签、
从 B 块取费率,就会编出一张两边都没有的表。所以本方法**只认同一块里解析出来的
整张档位表** —— 解析不出的块直接跳过(解析层是失败关闭的)。
"""
hits = await self._search_parameters(f"{product} 赎回费率", INTENT_PRODUCT, context)
if hits is None:
return self._exit_calc_miss("产品赎回费率检索不可用")
tiers = None
for hit in hits:
if not isinstance(hit, dict):
continue
title = str(hit.get("title") or "")
content = str(hit.get("content") or "")
if product not in title and product not in content:
continue # 不是这只产品的块,不能用
cell = self._redemption_cell(content)
if cell is None:
continue
tiers = parse_redemption_schedule(cell)
if tiers is not None:
break
if tiers is None:
return self._exit_calc_miss(f"未取到「{product}」可引用的赎回费率档位")
text = self._redemption_text(
product,
tiers,
days=parse_holding_days(request.message),
amount=parse_amount_yuan(request.message),
)
if text is None:
return self._exit_calc_miss(f"「{product}」赎回费率档位覆盖不到客户问的持有期")
return CoreResult(
text=self._clamp_answer(text), intent=self._classified_intent,
topic=product,
)
@staticmethod
def _redemption_cell(content: str) -> str | None:
"""从命中正文里取出「赎回费率」**这一格**的原文;取不到返回 `None`。
命中可能是整节块(Markdown 表格里的 `| 赎回费率 | … |` 行),也可能是行级子块
(`产品名:赎回费率 持有<7 天:…`),两种形态都要认。**只取这一格**:把整块正文
交给解析器,会连「管理费/托管费」那些行一起读进来。剩下多个 `|` 的行(例如费率
总表里 `| 赎回费率(<7 天) | 1.50% | … |`)说明这一格后面还有别的列,直接判为
读不懂 —— 宁可降级,也不要把总表的一格当成某只产品的完整档位表。
"""
for line in content.splitlines():
index = line.find("赎回费率")
if index < 0:
continue
cell = line[index + len("赎回费率"):].strip().strip("|").strip()
if cell and "|" not in cell:
return cell
return None
async def _answer_suitability_rule(
self, message: str, context: RequestContext
) -> CoreResult:
"""按第十二条的**匹配矩阵**回答通用规则(`D3.7` `D-04`)。
这一问在问**规则**,不在问「我本人能不能买」:所以答案只由矩阵决定,不读画像、
不调 `check_suitability`。后者要的输入是**具体产品**的风险等级,客户问的却是
「R3 这一类产品」,走那条路会把一句规则问题变成「请问您指哪只产品」。
"""
customer_level = parse_customer_level(message)
product_level = parse_product_level(message)
if customer_level is None or product_level is None:
# 上游 `is_general_suitability_question` 已保证两者都在;这里是失败关闭 ——
# 宁可说「取不到参数」,也不给一个猜出来的结论。
return self._exit_calc_miss("匹配规则问题里没有可解析的等级")
# 查询串**必须带上刚解析出的两个参数**(`C1`/`R3`)。实测证据(2026-09-19):
# 固定的「投资者与产品匹配矩阵 投资者类型 产品等级」top1 是 **C3 那一格**
# (`POL-AST-012-03`,0.7757)—— 矩阵是**按等级切块入库**的,泛问法只会撞上
# 其中一个格子,客户问的等级压根没进查询。带上参数后 top1 变为 `FAQ-0019`
# 「我最多可以购买哪个风险等级的产品?」(0.875,领先 0.177),矩阵本体
# (`PROD-012` 4.2)仍稳在第 2 位,`parse_suitability_matrix` 照旧取得到。
# 尾串是**语料的规范问法**:库里讲的都是"某档最多能买到哪一级",客户的
# 「能买 R3 吗」是同一件事的实例化问法 —— 这是**查询改写**,不是放宽判据。
query = (
f"{message.strip()} C{customer_level} R{product_level}"
" 最多可以购买哪个风险等级"
)
hits = await self._search_parameters(query, INTENT_POLICY, context)
if hits is None:
return self._exit_calc_miss("匹配矩阵检索不可用")
matrix = None
for hit in hits:
if not isinstance(hit, dict):
continue
matrix = parse_suitability_matrix(str(hit.get("content") or ""))
if matrix is not None:
break
if matrix is None:
return self._exit_calc_miss("未取到可引用的投资者与产品匹配矩阵")
verdict = matrix.verdict(customer_level, product_level)
if verdict is None:
return self._exit_calc_miss(
f"匹配矩阵里没有 C{customer_level} 与 R{product_level} 这一格"
)
return CoreResult(
text=self._suitability_rule_text(
customer_level, product_level, verdict, visitor=is_visitor(context)
),
intent=self._classified_intent,
)
@staticmethod
def _suitability_rule_text(
customer_level: int, product_level: int, verdict: str, *, visitor: bool
) -> str:
"""把矩阵的一格说成人话。**不做个性化推断**:只讲规则,并把依据的边界讲清。"""
if verdict == FORBIDDEN:
conclusion = "不可以购买(跨级禁止)"
elif verdict == DISCLOSURE:
conclusion = "需签署产品风险揭示书后可以购买"
else:
conclusion = "可以购买"
customer_label = RISK_LEVEL_LABELS.get(f"C{customer_level}", f"C{customer_level}")
product_label = RISK_LEVEL_NAMES.get(product_level, f"R{product_level}")
next_step = CALC_RULE_NEXT_VISITOR if visitor else CALC_RULE_NEXT_CUSTOMER
return (
"按《个人投资者适当性管理指南》第十二条的投资者与产品匹配矩阵,"
f"{customer_label}的投资者与{product_label}产品的匹配结论是:{conclusion}。\n"
f"{CALC_RULE_FOOTNOTE}\n{next_step}"
)
def _exit_calc_miss(self, note: str) -> CoreResult:
"""`E2` 取不到公开参数位 → `E5b` 如实告知(**不猜、不建单**)。"""
logger.info("E5b 计算出口无参数:%s", note)
return CoreResult(text=CALC_MISS_TEMPLATE, intent=self._classified_intent)
# ---- 出口一之二:适当性裁决(唯一给出"能不能买"结论的出口) ----
async def _answer_suitability(
self, request: AgentRequest, context: RequestContext
) -> CoreResult:
"""回答「以我的风险等级能不能买这只产品」。
为什么不能靠知识检索直接答:客户问「c1客户能买它吗」,答案是**两个事实的组合**
——该产品的风险等级(R2)与客户档案里的等级能不能匹配。检索只能给出"最像的那段
原文",实测给的是 C1 的通用规则,答非所问。
三条硬约束;缺任何一条都**不猜结论**——走 E5a(问清产品)或 E5b(如实告知),不转人工:
1. 产品风险等级**从知识库查出来**,不猜、也不采信问句里出现的"R2"字样;
2. 客户等级**由底座按档案解析**(check_suitability 内部读 fin_risk_assessment,
带测评有效期),**不采信客户自称**——这次实测里客户说"C1",档案其实是 C2;
3. 产品名**只取上一轮回答里的主语**(`_topic_of`,它来自知识块字段,可信)。
客户第一句就直接问"XX 能买吗"时取不到,那就转人工:这个出口会给出"能不能买"
的结论,宁可答不了也不能答错。
"""
product = self._previous_topic(request)
if not product:
return self._exit_suitability_clarify()
risk_level = await self._product_risk_level(product, context)
if risk_level is None:
return self._exit_suitability_miss(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._exit_suitability_miss("适当性校验调用失败")
if not isinstance(decision, dict):
return self._exit_suitability_miss("适当性校验返回格式异常")
text = self._suitability_text(product, risk_level, decision)
# `E-04` 留痕:把这次适当性裁决随消息落库(`CoreResult.data` 已经会被持久化层
# 写进消息的结构化列,零 DDL、零新端点)。`customer_confirmed=False` 是**恒值**,
# 记录的是「客服没有代替客户确认」这条红线 2 的事实,不是为了将来改。
suitability_record = {
"product": product,
"product_risk_level": risk_level,
"customer_risk_level": decision.get("customer_risk_level"),
"allowed": bool(decision.get("allowed")),
"reason_code": decision.get("reason_code"),
"required_disclosure": bool(decision.get("required_disclosure")),
"requires_recording": bool(decision.get("requires_recording")),
"disclosure_first": True,
"customer_confirmed": False,
}
claimed = self._self_claimed_level(request.message)
if claimed:
# 客户自述等级时必须点明判断依据,否则他会觉得"我明明说了我是 C3,
# 你却说我没有任何测评结果"——两句话在他眼里是矛盾的。
# 依据在档案、不在自述,这既是合规要求,也得跟客户讲明白。
text = (
f"您提到自己是 {claimed}。适当性判断以您在公司留存的、在有效期内的"
f"风险测评结果为准,不以本次自述为准。\n{text}"
)
return CoreResult(
text=text,
intent=self._classified_intent,
# `E-04`:适当性裁决随消息落库(`CoreResult.data` 已有持久化通路)。
data={"suitability": suitability_record},
)
@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:
"""上一轮回答里的主语(产品名);取不到返回空串。
`E-05`:与 `_search_query` 同源 —— 优先取出口的**声明值**,存量行回落文本反解。
"""
for turn in reversed(request.history):
if turn.role == "assistant":
return cls._turn_topic(turn)
return ""
async def _product_risk_level(self, product: str, context: RequestContext) -> int | None:
"""查产品的风险等级:问知识库要"风险等级"那一行,不从问句里猜。
产品手册里每个产品都有一行"风险等级 R2(中低风险)",切分后是独立的行级子块,
所以按「{产品名} 风险等级」检索能直接命中。两重校验缺一不可:必须命中**风险等级
行**(否则可能匹配到"C1 可购买 R1、R2"那种列举,把 R1 当成产品等级),而且该行
必须属于**同一个产品**(否则会拿另一个产品的等级去做裁决)。
"""
try:
knowledge_tool = VISITOR_TOOL_NAME if is_visitor(context) else TOOL_NAME
output = await self.call_tool(
knowledge_tool,
{"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")
# 需要客户主动确认的三种情形:要签揭示书 / 要双录 / 矩阵判为「揭示后可买」。
disclosure_required = bool(
decision.get("required_disclosure")
or decision.get("requires_recording")
or decision.get("reason_code") == "SUITABLE_WITH_DISCLOSURE"
)
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("本次购买需进行双录(录音录像)。")
# `E-04`:上面是**风险揭示**,这段是**要求客户主动确认**(顺序不可颠倒)。
if disclosure_required:
lines.append(SUITABILITY_CONFIRM_REQUEST)
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:
# 连续闲聊超过三轮后只做一次自然的业务引导,避免模型无限延续闲聊。
if request.metadata.chitchat_streak == 4:
return CoreResult(
text="您好呀,您是想了解基金产品、申赎规则或其他公开业务信息吗?",
intent=self._classified_intent,
)
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._exit_chitchat_prompt_fail("模型不可用")
text = (execution.text or "").strip()
if not text:
return self._exit_chitchat_prompt_fail("模型返回为空")
return CoreResult(
text=self._clamp_answer(text),
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,
)
# ---- 出口三之补:`E4` 证据约束生成(`H-03`) ----
async def _answer_from_evidence(
self, request: AgentRequest, context: RequestContext, evidence: list[dict[str, Any]]
) -> CoreResult | None:
"""`E4` 主流程。返回 `None` 表示**这次不接管**,交回原有出口判定。
什么时候回 `None`(交回 `E3` 原文直返 / `E5a` 澄清 / `E5b` 部分答):模型不可用、
没按契约返回 JSON、或**模型自己说答不了**(缺主语 / 过泛 / 证据不足)。
什么时候回 `E5b`:模型给了答复但**没过校验**(引用越界 / 数字无出处 / 合规命中)——
这正是 `H-03` DoD ⑤ 要求的拦截面,且**不建单、不转人工**。
"""
message = request.message[:500]
evidence_text = self._render_evidence(evidence)
system, template = await self._evidence_prompt()
try:
prompt = template.format(evidence=evidence_text, message=message)
except (KeyError, IndexError, ValueError):
# 发布配置里的占位符与代码不一致时回落内置模板:配置写错不该让运行期崩
system = DEFAULT_EVIDENCE_SYSTEM
prompt = DEFAULT_EVIDENCE_TEMPLATE.format(evidence=evidence_text, message=message)
full_prompt = f"{system}\n\n{prompt}" if system else prompt
try:
endpoints = await DatabaseModelEndpointResolver().resolve(
agent_type=AGENT_TYPE, task_type=E4_TASK_TYPE
)
# 显式转成 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:
logger.info("E4 模型不可用,交回原有出口判定", exc_info=True)
return None
payload = self._parse_evidence_output(execution.text or "")
if payload is None:
logger.info("E4 未按契约返回 JSON,交回原有出口判定")
return None
answer = str(payload.get("answer") or "").strip()
reason = str(payload.get("unanswerable_reason") or "").strip()
if reason or not answer:
# 模型自述答不了 —— 那是"该澄清 / 该给部分"的信号,不是"生成失败",
# 所以交回原路径而不是降级成部分答(`H-01` 的澄清出口仍然生效)。
logger.info("E4 模型判断本条不可作答:%s", reason or "答案为空")
return None
used = [
str(item).strip()
for item in (payload.get("used_chunk_ids") or [])
if str(item).strip()
]
allowed = {str(hit.get("doc_id") or "") for hit in evidence}
outside = [item for item in used if item not in allowed]
if not used or outside:
logger.info("E4 引用越界,拦回 E5b:%s", outside)
return self._exit_partial(evidence, note="证据约束生成引用了证据包外的来源")
ungrounded = self._ungrounded_numbers(answer, evidence, request.message)
if ungrounded:
logger.info("E4 出现无出处数字,拦回 E5b:%s", ungrounded)
return self._exit_partial(
evidence, note=f"证据约束生成出现无出处数字:{ungrounded}"
)
if hits_zero_tolerance(answer) or (
is_visitor(context) and visitor_advice_violation(answer) is not None
):
logger.info("E4 未过合规校验,拦回 E5b")
return self._exit_partial(evidence, note="证据约束生成未过合规校验")
logger.info("E4 证据约束生成命中:块数=%s 引用=%s", len(evidence), used)
return CoreResult(
text=self._clamp_answer(answer), intent=self._classified_intent
)
@classmethod
def _declared_topic(cls, content: str) -> str:
"""`E-05`:出口**显式声明**本轮在说哪个主语(产品 / 类目 / 条款)。
口径与旧的读侧反解**完全同源**(`_topic_of`),差别只在**求值时机与对象**:
现在在出口、对**本次实际返回的那个块**求一次并存进消息;读侧直接取用,
不再对每一轮历史回答重新反解整段文本。因此本方法对 E3/E5b 是行为等价的
(有等价性单测守着);对 E2 这类出口则改为直接声明它**真正在说的主语**
(类目 / 产品名),不经过任何文本解析。
"""
return cls._topic_of(content)
@staticmethod
def _clamp_answer(text: str, limit: int = MAX_ANSWER_CHARS) -> str:
"""超长时在**句末**收口,并显式告知「这是摘要」。
`E3` 原文直返不需要这一步(单块本身就短);`E4` 的合并答复会撞到
`MAX_ANSWER_CHARS` —— 硬切会把最后一个句子劈成两半(实测 `C-01` 的四档权益
合起来超过 1200 字,切在「私募股权/创投基金」中间)。宁可少说半句,
也不给客户一句读不通的话。
"""
if len(text) <= limit:
return text
window = text[:limit]
cut = max(window.rfind("。"), window.rfind("\n"))
head = window[: cut + 1] if cut > limit // 2 else window
# `E-06`:截断必须**说出来**(`TRUNCATION_NOTICE` 的说明)。
return head + TRUNCATION_NOTICE
async def _evidence_prompt(self) -> tuple[str, str]:
"""读取当前发布版本的证据约束提示词;未发布或读取失败时回落代码内置默认值。"""
try:
row = await load_active_prompt(E4_PROMPT_CODE, E4_TASK_TYPE, AGENT_TYPE)
except Exception:
row = None
if row is None:
return DEFAULT_EVIDENCE_SYSTEM, DEFAULT_EVIDENCE_TEMPLATE
return (
row.system_prompt or DEFAULT_EVIDENCE_SYSTEM,
row.user_prompt_template or DEFAULT_EVIDENCE_TEMPLATE,
)
@classmethod
def _evidence_pack(cls, hits: list[Any], *, gap: float) -> list[dict[str, Any]] | None:
"""`E4` 该不该接管、证据包里放哪些块;`None` = 不接管。
**先决条件**:`gap < MIN_GAP`(分数接近)。领先明显时仍走 `E3` 原文直返
—— 原文直返没有幻觉面,能不用模型就不用。
满足先决后,命中里**有哪一种形态**决定证据包怎么取:
1. **同章节多块**(原判据)→ 取该章节的成员组。为什么按章节而不是按 `family_id`:
`family_id` 的粒度是**父块 / 行级子块**(`D2.4` v1.4),同一次检索里同一
`family_id` 的多块只会是「父块 + 它的子块」—— 父块本就包含子块内容,合并
**没有增量**,原文直返父块即可。真正需要合并的是**同一章节的不同小节**:
实测 `C-01` 的"四档权益"分属 `HNW-004`~`007`(四个不同 `family_id`,
同一章节「二、各层级专属权益」)。
2. 🆕 **跨章节近分**(`W2`,2026-09-19 `H-06` 实测补齐)→ 取 TopK 里
≥ `E4_MIN_SCORE` 的块。**为什么补这一条**:`H-06` 首跑 46 条里有 **14 条落在
澄清(`E1`)**,其中 9 条 top1 已 ≥0.55、`I-01` 的 top1 **就是期望证据家族**
—— 澄清成了新的兜底;而 `D2.4` 附录F.3 的原话是「TopK 内同族多块且**分数接近**
→ 合并为一个答案(E4),**不澄清**」。旧实现只认"同章节"这一种形态,
比设计窄了一档。
3. 🆕 **组内 + 组外合并**(`W6`,2026-09-19 `W5` 复跑实测补齐)→ 先保章节组,
再用 TopK 里其余 ≥ `E4_MIN_SCORE` 的块**补足**到 `E4_MAX_EVIDENCE`。**为什么**:
实测 `A-04`(「风险等级越高的产品是不是收益越高?」)的答案在 `FAQ-0018`
(检索 rank 3),而「成员最多的章节组」是「第四章 产品分类与风险等级对应」的
三块适当性条款 —— 组内**没有**风险↔收益的表述,模型只能判「证据不足」退澄清;
正确答案的互补证据恰好在组外,**二选一必然丢一边**。
**安全边界一条没松**:包仍先过档位裁剪(`E4_MIN_SCORE` 之下的噪声不进包,
档位由鉴权结果决定),生成结果仍要过 `_answer_from_evidence` 的三道校验
(引用越界 / 数字无出处 / 合规命中);模型判断"缺主语 / 过泛 / 证据不足"时
按契约回 `None`,仍退澄清 —— 真正该问回去的问句不会被硬答。
"""
if gap >= MIN_GAP:
return None
groups: dict[tuple[str, str], list[dict[str, Any]]] = {}
for hit in hits:
if not isinstance(hit, dict):
continue
if cls._score(hit.get("score")) < E4_MIN_SCORE:
continue
key = cls._chapter_group_of(hit)
if key is None:
continue
groups.setdefault(key, []).append(hit)
candidates = [members for members in groups.values() if len(members) >= 2]
if not candidates:
# `W2`:近分但凑不出"同章节组"(跨章节)→ 取 TopK 里达标的块做证据包。
# 交回置信判定只会变成"澄清"(`gap < MIN_GAP` 必然不满足直接答的条件),
# 而附录F.3 明确要求这种情况**合并作答、不澄清**。
fallback = [
hit for hit in hits
if isinstance(hit, dict) and cls._score(hit.get("score")) >= E4_MIN_SCORE
][:E4_MAX_EVIDENCE]
return fallback or None
# 取成员最多的组;并列时取组内最高分更大的那个(分数高的那组更贴题)。
chosen = max(
candidates,
key=lambda members: (
len(members),
max(cls._score(hit.get("score")) for hit in members),
),
)
pack = sorted(chosen, key=lambda hit: cls._score(hit.get("score")), reverse=True)
pack = pack[:E4_MAX_EVIDENCE]
# `W6`:章节组之外的高分块**也要进包**(合并,不是二选一),见 docstring 第 3 条。
seen = {str(hit.get("doc_id") or "") for hit in pack}
for hit in hits:
if len(pack) >= E4_MAX_EVIDENCE:
break
if not isinstance(hit, dict):
continue
if cls._score(hit.get("score")) < E4_MIN_SCORE:
continue
doc_id = str(hit.get("doc_id") or "")
if doc_id and doc_id in seen:
continue
seen.add(doc_id)
pack.append(hit)
top = hits[0]
if isinstance(top, dict):
top_id = str(top.get("doc_id") or "")
if top_id and all(str(item.get("doc_id") or "") != top_id for item in pack):
# 检索认为最贴题的那一块必须进包:它只是"不属于成员最多的那一组",
# 丢了它等于把最佳证据排除在外(`C-03` 实测:答案在 `POL-AST-005`,
# 而 top1 是一条 FAQ)。补位补完仍未进包 ⇔ 章节组自己就占满了上限,
# 此时让末位给 top1(组内其余成员仍全部保留)。
pack = [*pack[: E4_MAX_EVIDENCE - 1], top]
return pack
@staticmethod
def _chapter_group_of(hit: object) -> tuple[str, str] | None:
"""命中的「章节」分组键;取不到返回 `None`(该块**不参与**同章节合并)。
题面由切片器用 ` · ` 连接成「文档名 · 章节 · 小节 …」,所以**必须 ≥3 段**才认第二段
是章节 —— 只有两段时第二段是小节,把它当章节会把同一文档里不相干的小节并成一组。
FAQ/公司信息这类"标题就是问句"的块天然取不到章节,于是**永远不参与合并**。
为什么不用库里已有的 `chapter` 字段:知识工具的返回契约里没有它
(`app/service/knowledge_tool.py` 属**底座会签件**,本步不得触碰);而 `title` 与
`source_file` 都在契约里,派生出的键与库内 `chapter` 同源可信。
"""
if not isinstance(hit, dict):
return None
parts = str(hit.get("title") or "").split(" · ")
if len(parts) < 3 or not parts[1].strip():
return None
source_file = str(hit.get("source_file") or "").strip()
if not source_file:
return None
return source_file, parts[1].strip()
@classmethod
def _render_evidence(cls, evidence: Sequence[dict[str, Any]]) -> str:
"""证据包 → 提示词文本:逐块编号,并带 `doc_id` 与**族标识**(`H-03` DoD ①)。"""
lines: list[str] = []
for index, hit in enumerate(evidence, start=1):
doc_id = str(hit.get("doc_id") or "")
family = str(hit.get("family_id") or "") or "未知"
title = str(hit.get("title") or "") or "未命名"
content = str(hit.get("content") or "").strip()
lines.append(f"[{index}] doc_id={doc_id} 族标识={family} 标题={title}\n{content}")
return E4_EVIDENCE_HEADER + "\n" + "\n\n".join(lines)
@staticmethod
def _parse_evidence_output(text: str) -> dict[str, Any] | None:
"""从模型输出里取出 JSON 对象;取不到返回 `None`(调用方据此**不接管**)。
容错三件事,都是为了"模型多说了两句"不至于让整条链路退化成兜底:
去掉代码块围栏与语言标记、允许 JSON 前后有解释性文字、只认第一个对象。
"""
raw = text.strip()
if raw.startswith("```"):
_, _, raw = raw.strip("`").partition("\n")
start = raw.find("{")
end = raw.rfind("}")
if start < 0 or end <= start:
return None
try:
payload = json.loads(raw[start : end + 1])
except (TypeError, ValueError):
return None
return payload if isinstance(payload, dict) else None
#: 数值匹配:先认千分位,再认普通数字;单位只收**会改变数值含义**的那几个。
_NUMBER_PATTERN = re.compile(
r"(?P<value>\d{1,3}(?:,\d{3})+(?:\.\d+)?|\d+(?:\.\d+)?)"
r"\s*(?P<unit>%|%|万元|亿元|万|亿)?"
)
_UNIT_SCALE = {"万": 10000, "万元": 10000, "亿": 100000000, "亿元": 100000000}
_FULLWIDTH_DIGITS = str.maketrans("0123456789", "0123456789")
@classmethod
def _number_tokens(cls, text: str) -> set[str]:
"""把文本里的**数值**归一成可比较的字符串集合。
归一化三件事(`M-9` 要能把「10 万元」与「100,000 元」认成同一个数):
全角数字/百分号转半角、去掉千分位逗号、「万 / 亿」按 10^4 / 10^8 展开成元。
**不带单位、不足 3 位、且没有小数点的整数不入集**:那是序号(`R1`、`四项`、
列表编号、`7×24` 里的 7),不是产品要素 —— 把它们也当"数字"校验,正常答复会被整条拦下。
"""
normalised = text.translate(cls._FULLWIDTH_DIGITS).replace("%", "%")
tokens: set[str] = set()
for match in cls._NUMBER_PATTERN.finditer(normalised):
literal = match.group("value")
unit = match.group("unit") or ""
digits = literal.replace(",", "")
if not unit and "," not in literal and "." not in digits and len(digits) < 3:
continue
try:
value = Decimal(digits)
except InvalidOperation:
continue
if unit in cls._UNIT_SCALE:
tokens.add(cls._plain_number(value * cls._UNIT_SCALE[unit]))
elif unit == "%":
tokens.add(f"{cls._plain_number(value)}%")
else:
tokens.add(cls._plain_number(value))
return tokens
@staticmethod
def _plain_number(value: Decimal) -> str:
"""数值 → 稳定字面(`100000` 而不是 `1E+5`,`0.5` 而不是 `0.50`)。"""
integral = value.to_integral_value()
if value == integral:
return str(int(integral))
return format(value.normalize(), "f")
@classmethod
def _ungrounded_numbers(
cls, answer: str, evidence: Sequence[dict[str, Any]], question: str = ""
) -> list[str]:
"""答复里**解析不到出处**的数字(`D3.7` `M-9`:必须为 0)。
出处 = 证据包全文 ∪ **用户自己这句话**。收用户那句话是有意的:模型转述
「您说的 10 万元」并不是幻觉;不收它,正常复述会被判成"无出处数字"。
"""
haystack = "\n".join(
f"{hit.get('title') or ''}\n{hit.get('content') or ''}"
for hit in evidence
if isinstance(hit, dict)
)
allowed = cls._number_tokens(f"{haystack}\n{question}")
return sorted(token for token in cls._number_tokens(answer) if token not in allowed)
# 客户这一句里出现这些词,说明主语要靠上文补全("这个产品""该基金")。
# 刻意不收"它/他":单字指代在中文里极易误伤("其他产品""其它"都含"他/它"),
# 而这类短句已经由 _MIN_STANDALONE_CHARS 覆盖,不需要靠它兜。
_REFERRING_WORDS = (
"这个", "那个", "这款", "这只", "该产品", "该基金", "上述", "前面提", "刚刚说",
)
#: 🆕 `W3`(2026-09-19 `H-06` 实测):**单字「它」开头的追问**必须靠上文补主语。
#: 为什么不用 ``_REFERRING_WORDS`` 做子串:中文里「其他 / 其它」都含「它」,
#: 子串匹配会把「其他产品怎么买」也当成指代。用**句首锚定**的正则即可两全:
#: 实测 `H-01` 第二问「那它风险等级呢?」共 8 字,恰好不满足 `< _MIN_STANDALONE_CHARS`
#: (严格小于 8 不成立)⇒ 旧实现既没进指代分支、也没进短句分支,**主语整条丢**,
#: 检索退化成「风险等级」泛问 → 落澄清。
_REFERRING_PATTERNS = (
re.compile(r"^(那|这)?它"),
)
#: 基金类别词(`D-03`/`H-02` 的"类目切换式追问"要用)。
_CATEGORY_TOKENS = ("货币", "债券", "混合", "股票", "QDII")
#: 「<类目>呢?」式**纯追问**:自己不带谓词,谓词只能从上一轮**客户问句**里继承。
_CATEGORY_FOLLOWUP = re.compile(r"^(货币|债券|混合|股票|QDII)[^。;!?,,]{0,4}呢")
# 短到这种程度的问题,自己通常不构成完整意图("起投多少""风险高吗")
_MIN_STANDALONE_CHARS = 8
@classmethod
def _turn_topic(cls, turn: Any) -> str:
"""`E-05`:取一轮历史答复的**主语**。
优先出口声明(`ConversationTurn.subject`);存量行没有该字段时回落到
`_topic_of` 的文本反解 —— 回落分支是**刻意保留**的:老消息不该因为新增字段
就丢掉主语,否则多轮追问会突然答不上来。
"""
declared = str(getattr(turn, "subject", "") or "").strip()
if declared:
return declared
return cls._topic_of(str(getattr(turn, "content", "") or ""))
@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()
# `H-02` 反向考点("混合基金呢?"):本句**没有自己的谓词**,谓词只能从上一位
# 客户问句里继承 —— 所以拼的不是主语,而是"把上一位问句的类目换成新类目"。
# 这里若走下面的通用分支会答错:旧实现拼出来的查询只有「混合基金呢?」,
# 检索命中的是一只具体的混合基金产品块,而不是"混合基金到账时限"。
switched = cls._category_switch_query(message, request)
if switched:
return switched[:500]
needs_context = (
len(message) < cls._MIN_STANDALONE_CHARS
or any(word in message for word in cls._REFERRING_WORDS)
or any(pattern.search(message) for pattern in cls._REFERRING_PATTERNS)
)
if not needs_context:
return message[:500]
for turn in reversed(request.history):
if turn.role == "assistant":
topic = cls._turn_topic(turn)
if topic:
return f"{topic} {message}"[:500]
break
return message[:500]
@classmethod
def _category_switch_query(cls, request_message: str, request: AgentRequest) -> str:
"""「<类目>呢?」式纯追问 → 用**上一位客户问句**的谓词 + 新类目拼查询。
为什么不能只补主语(`D3.6` §1.4 的两条实测教训):上文只补**主语**、不补**内容**。
但「混合基金呢?」这句**自己没有任何内容**(既没有"到账"也没有"费率"),
唯一的内容来源就是上一位客户问句 —— 而这正是"类目切换"要考的:**换主语、留谓词**。
判据刻意收得很窄:① 本句以类目词**开头**;② 紧跟「呢」(最多夹 4 字);
③ 上一位客户问句里**确实有另一个类目词**。三条不同时成立就返回空串、退回原逻辑,
所以「货币基金的费率是多少」(自己带谓词)不会走这条路。
"""
match = cls._CATEGORY_FOLLOWUP.match(request_message)
if match is None:
return ""
new_token = match.group(1)
for turn in reversed(request.history):
if turn.role != "user":
continue
for old_token in cls._CATEGORY_TOKENS:
if old_token != new_token and old_token in turn.content:
return turn.content.replace(old_token, new_token, 1).strip()
return ""
return ""
# ---- 出口五:分级回退 E5(E5a 澄清 → E5b 部分答 + 引导 → E5c 转人工) ----
#
# 为什么必须分级:原实现只有"答得出"与"转人工"两种结局,一切不确定性都变成转人工。
# 现在:能问清就问清(E5a),能给部分就给部分(E5b),只有白名单四类才转人工(E5c)。
def _exit_transfer(self, reason_code: str) -> CoreResult:
"""E5c 转人工:**唯一**允许置 ``transfer_required=True`` 的出口。
`reason_code` 必须落在 ``TRANSFER_REASONS`` 白名单内,越界即抛错——
让"白名单外发生转人工"在开发期就暴露,而不是等验收时才发现。
"""
if reason_code not in TRANSFER_REASONS:
raise ValueError(f"转人工原因码不在白名单内:{reason_code}")
return CoreResult(
text=TRANSFER_TEMPLATE,
intent=self._classified_intent,
transfer_required=True,
transfer_reason=reason_code,
)
def _exit_partial(self, hits: list[Any], *, note: str = "") -> CoreResult:
"""E5b 部分答 + 引导。**不建单**:这是"这次没查到",不是"必须人工办的事"。
有达到 ``PARTIAL_FLOOR`` 的命中时展示其正文(原文直返,不经模型);
没有时只说"没找到"并给改写建议——两者都不置 `transfer_required`。
"""
if note:
logger.info("E5b 部分答:%s", note)
# 展示**分数最高**的那一块,而不是 `hits[0]`。`B-06` 实测:`E4` 的三条降级
# 路径把**证据包**(章节组成员在前、补位的高分块在尾)交到这里,取 `hits[0]`
# 等于把"章节组里分最高的那一块"当成最佳证据 —— 客户看到的是
# 「其他费用:认购费 认购时一次性收取」,而真正贴题的 `PROD-018`
# (6.3 费用计算示例,全局 top1)排在包尾,问「举个例子说明费率怎么查」
# 却拿到一段答非所问的碎片。降级时客户手里只剩这一块,更要给最好的那块。
best: dict[str, Any] | None = None
best_score = -1.0
for hit in hits:
if not isinstance(hit, dict):
continue
score = self._score(hit.get("score"))
if score > best_score:
best, best_score = hit, score
content = ""
if best is not None and best_score >= PARTIAL_FLOOR:
content = self._clamp_answer(str(best.get("content") or "").strip())
return CoreResult(
text=PARTIAL_TEMPLATE.format(content=content) if content else PARTIAL_EMPTY_TEMPLATE,
intent=self._classified_intent,
topic=self._declared_topic(content) if content else "",
)
@staticmethod
def _knowledge_missed(result: CoreResult) -> bool:
"""知识出口是否**什么都没答上来**(`E5b` 空答)—— `F-2` 的转人工判据。
只有「连部分资料都没有」才算答不上来:`E5b` 带内容的、给出澄清候选的,都算答到了
一部分,**不转人工**——`E5b` 与澄清本身也从不建单(`D3.7` §5「部分作答即合格」)。
"""
return result.text in KNOWLEDGE_MISS_TEXTS
@staticmethod
def _same_document(hits: Sequence[object]) -> bool:
"""候选是否**同属一个文档**(`W6`:澄清的反向判据)。
**为什么需要它**:`E5a` 澄清的语义是「你说的我分不清,请指明是哪一项」。但当
top-K 候选**全部来自同一份文档**时,客户的提问范围其实已经清楚了 —— 就是在问
这份文档所辖的主题,此时反问「您想了解哪一项」并没有真的减少歧义,只是把本该
给出的材料推回给客户(`I-01` 实测:三个候选全是《企业信息》,却仍在反问)。
取不到 `source_file` 的命中(空串/缺失)一律**不算同文档** —— 判据宁严勿宽:
澄清多问一句只是体验差,漏问一句则可能答错主体。
"""
sources: set[str] = set()
for hit in hits:
if not isinstance(hit, dict):
return False
source = str(hit.get("source_file") or "").strip()
if not source:
return False
sources.add(source)
return len(sources) == 1
@staticmethod
def _subject_terms_in(message: str) -> tuple[str, ...]:
"""问句里出现的**受控主题词**(`F-3` 闸门的触发条件;空元组 = 闸门不启用)。"""
return tuple(term for term in EVIDENCE_SUBJECT_TERMS if term in message)
@classmethod
def _subject_covered_by(cls, hits: Sequence[object], terms: Sequence[str]) -> bool:
"""主题词是否**至少有一个**落在某个命中块的 `title` 或 `content` 里。
只看块内文本,不看 `doc_id` 之类的身份字段:`title` 里的章节名("六、费率说明")
正是"这一块在讲什么"最直接的证据。`terms` 为空时恒为 `True`(闸门不启用)。
"""
if not terms:
return True
for hit in hits:
if not isinstance(hit, dict):
continue
blob = f"{hit.get('title') or ''}\n{hit.get('content') or ''}"
if any(term in blob for term in terms):
return True
return False
def _exit_subject_miss(self, subject: str) -> CoreResult:
"""`F-3` 闸门拦下:该主题在当前可见资料里没有对应内容 → `E5b`(**不建单**)。"""
logger.info("E5b 证据与问句主体无关:subject=%s", subject)
return CoreResult(
text=SUBJECT_MISS_TEMPLATE.format(subject=subject),
intent=self._classified_intent,
)
def _exit_profile_miss(self) -> CoreResult:
"""画像查不到:如实告知 + 引导自助(E5b,不建单)。绝不猜一个等级出来。"""
return CoreResult(text=PROFILE_MISS_TEMPLATE, intent=self._classified_intent)
def _exit_clarify(self, hits: list[Any]) -> CoreResult | None:
"""E5a 澄清:给 2—3 个可见档位内的候选,一次只问一个问题。
返回 ``None`` 表示候选不足(分数低于 ``CLARIFY_SCORE`` 或没有可用标题),
应交由 E5b 处理——**不拿噪声去问客户**,那只会让客户觉得客服在装傻。
"""
candidates: list[str] = []
for hit in hits[:CLARIFY_LIMIT]:
if not isinstance(hit, dict):
continue
if self._score(hit.get("score")) < CLARIFY_SCORE:
break
title = str(hit.get("title") or "").strip()[:40]
if title and title not in candidates:
candidates.append(title)
if not candidates:
return None
lines = "\n".join(
f"{index}. {title}" for index, title in enumerate(candidates, start=1)
)
return CoreResult(
text=CLARIFY_TEMPLATE.format(candidates=lines),
intent=self._classified_intent,
clarification_required=True,
)
def _guide_to_login(self, reason: str) -> CoreResult:
"""Keep customer-only capabilities explicit at the public boundary."""
return CoreResult(
text="该服务需要登录后才能查询您的个人信息或适当性结果,请先登录客户账户。",
intent=self._classified_intent,
transfer_required=False,
transfer_reason=reason[:200],
)
def _exit_suitability_clarify(self) -> CoreResult:
"""适当性问题里还没看出具体产品 → E5a 问清,不转人工。"""
return CoreResult(
text=SUITABILITY_CLARIFY_TEMPLATE,
intent=self._classified_intent,
clarification_required=True,
)
def _exit_suitability_miss(self, note: str) -> CoreResult:
"""给不出适当性结论 → E5b 如实告知(不猜、不建单)。"""
logger.info("E5b 适当性无结论:%s", note)
return CoreResult(text=SUITABILITY_MISS_TEMPLATE, intent=self._classified_intent)
def _exit_chitchat_prompt_fail(self, note: str) -> CoreResult:
"""闲聊出口模型不可用 → E5b 礼貌收尾并引回业务(不建单)。"""
logger.info("E5b 闲聊回退:%s", note)
return CoreResult(text=CHITCHAT_FALLBACK_TEMPLATE, intent=self._classified_intent)
# ---- 辅助 ----
def _intent_code(self) -> str:
classified = self._classified_intent
return classified.intent if classified is not None else ""
def _intent_needs_clarification(self) -> bool:
"""意图分类给出的 `needs_clarification`(`H-01` DoD ①:它此前**全仓无消费方**)。
分类器算出来、契约里有字段、就是没人读——等于白算。它表达的是"这句话我拿不准
属于哪一类",正是 E5a 要问一句的场景之一。注意它**不单独构成拦截**:只有检索
也不够确定时才会走到这里,所以它不会把"答得出"的问题变成"问一句"。
"""
classified = self._classified_intent
return bool(classified is not None and classified.needs_clarification)
@staticmethod
def _family_of(hit: object) -> str:
"""命中的知识族标识(`family_id`,`D2.4` v1.4:同族 = 同一个父块)。
取不到时返回空串,由调用方按"族未知"处理——**不当作同族**。
"""
if not isinstance(hit, dict):
return ""
return str(hit.get("family_id") or "").strip()
@classmethod
def _missing_subject(cls, request: AgentRequest) -> bool:
"""这一句缺主语、**且上文也接不上**(`H-01` DoD ② 的「缺主语」)。
判据与 `_search_query()` 同源(过短 or 指代词),但结论相反:那边问"能不能补上
主语",这边问"补不上就得问回去"。上文能接上时**不问**——客户说"这个产品"只是
用了指代,检索接得住,再问一句就是打扰。
"""
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 False
for turn in reversed(request.history):
if turn.role == "assistant":
return not cls._turn_topic(turn)
return True # 前面没有助手轮次,指代无处可指
def _clarify_reason(
self, request: AgentRequest, hits: list[Any], *, score: float, gap: float
) -> str | None:
"""E5a 该不该问一句:返回理由码,或 `None` 表示**不该问**(`H-01` DoD ②⑥)。
四类触发条件(DoD ②):
- `cross_family_tie`:候选**跨族并列**(top1 与次优咬得紧,族还不同);
- `weak_cross_family`:分数不足**且跨族**(没到能答的档,候选又分散在多个族);
- `missing_subject`:**缺主语**且上文接不上;
- `low_intent_confidence`:意图分类自己就说了拿不准(`needs_clarification`)。
**同族并列不澄清**(DoD ⑥):同一族的多个块是"同一个话题的不同细节",该合并
作答(`H-03`/`E4`);对同族候选问"你要哪一个"是伪问题。族标缺失时按"跨族"
处理——宁可多问一句,也不要把两个不同族的答案混着答。
"""
candidates = [
hit for hit in hits[:CLARIFY_LIMIT]
if isinstance(hit, dict) and self._score(hit.get("score")) >= CLARIFY_SCORE
]
if not candidates:
return None
families = {self._family_of(hit) for hit in candidates}
known_single_family = len(families) == 1 and "" not in families
if known_single_family and gap < MIN_GAP:
return None
if not known_single_family:
if gap < MIN_GAP:
return "cross_family_tie"
if score < MID_SCORE:
return "weak_cross_family"
if self._missing_subject(request):
return "missing_subject"
if self._intent_needs_clarification():
return "low_intent_confidence"
return None
@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, ...]:
"""**保留待用的死代码 —— 本期(MVP)不向客户展示来源引用**(`C-10` 乙·降级)。
为什么在位却不调用:`governance.review_output` 只认可 memory / tool 两类来源
(`app/service/agent/governance.py:314-316`),这里产出的是
`source_type="knowledge"` —— 一旦被知识出口调用,会被判为「引用未来自本次已授权
召回结果」,后果是**整个 run 失败**(不是降级、也不是少一个字段),见红线 `S-8`。
本期口径(`C-10` 乙):
- **不向客户展示来源引用**;可追溯性由**审计**承接(`agent.tool_executed` 的
工具调用记录含命中 doc_id 与分数);
- 护栏断言钉死「无调用点」:`tests/unit/service/test_customer_service_agent.py`
用 AST 守本方法**不得被调用**(防误启用);
- **启用前提**(需底座方会签):让 `ToolExecutor` 把工具返回的知识 doc_id 登记为
本次可引用来源,并在治理层放行 `knowledge` 来源 —— 详见 `D3.6` 与 `D2.2`
`FR-CS-010`(已标「本期降级」)。
"""
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)