test(W28): 意图漂移守卫用例集（414 条）

覆盖： - 「业务词 × 寒暄外壳」漂移用例 330 条（33 个业务关键词 × 10 种寒暄外壳，如「在吗，帮我看看{kw}」「嗯嗯，那{kw}呢」）—— 最易漂移的一族 - 业务动作与职责词 13 条 - 真寒暄反向守卫 16 条（确保不把闲聊拖去查库） - 空串 / 纯符号 5 条 - 事项码正例 24 条 + 反例 6 条 + 属性自洽 14 条 - 下一步动作 26 条（E5b 双段话术） - AST 结构守卫 4 条 - Agent 层端到端 6 条（含「强制分类为 chitchat + 业务问句 ⇒ 必须改走知识出口」）配套实现在上一提交（客服业务层）。
2026-09-22 10:12:22 +08:00
parent 2b408dc602
commit 020f17363f
1 changed files with 484 additions and 0 deletions
@@ -0,0 +1,484 @@
+"""`W28` 意图漂移护栏与业务事项码的单元测试。
+
+## 这份用例集守什么
+
+答辩老师指出「未搞清楚真实业务需求」之后，`W28` 补了**业务事项轴**（`service_topic`）
+与**意图漂移护栏**（`substantive_business_request`）。本文件两件事：
+
+1. **护栏的准与稳**：用**大量真实业务问法**（含把业务问题伪装成寒暄的口语形态）
+   验证「带业务实质 ⇒ 不判闲聊」；用**真寒暄**验证反向不误伤。
+2. **事项码的可证伪性**：每个 `SVC-*` 都有正例（应命中）与**反例**（不应被误命中），
+   并交叉断言 `owner` / `retention` / `next_step_block` 三条业务属性的自洽性。
+
+## 为什么判据层测试比 Agent 层测试更值钱
+
+护栏的判据是**纯函数**（不查库、不调模型、不读上下文），所以这里可以跑几百条用例而
+不搭任何夹具；而 Agent 层的接线由文件末尾的 **AST 结构守卫**兜底（防止后人把护栏
+从 `_route_and_answer` 里摘掉却没有任何测试变红）。
+"""
+
+from __future__ import annotations
+
+import ast
+import pathlib
+
+import pytest
+
+from app.core import customer_service_rules as rules
+from app.core import service_topic as topic
+
+# ---------------------------------------------------------------------------
+# 一、业务关键词 × 寒暄外壳：漂移主战场
+# ---------------------------------------------------------------------------
+
+#: 业务关键词（每个都是真实高频问法的核心词）。
+BUSINESS_KEYWORDS: tuple[str, ...] = (
+    "费率", "申购费", "赎回费", "管理费", "托管费", "起投金额", "门槛", "期限",
+    "净值", "走势", "行情", "风险等级", "持仓", "份额", "分红", "定投", "转换",
+    "开户", "销户", "赎回", "到账", "确认", "协议", "条款", "说明书",
+    "身份证", "银行卡", "密码", "验证码", "投顾", "投诉",
+)
+
+#: 寒暄外壳（把业务问题包在里面 —— 分类器最容易在这里漂移）。
+CHITCHAT_WRAPPERS: tuple[str, ...] = (
+    "{kw}是多少",
+    "你好，我想问下{kw}",
+    "在吗，帮我看看{kw}",
+    "嗯嗯，那{kw}呢",
+    "好的谢谢，另外问一下{kw}",
+    "哈哈哈，你们这个{kw}在哪看",
+    "早上好，麻烦问一下{kw}",
+    "喂，{kw}怎么算",
+    "你好呀，我想了解{kw}",
+    "有人吗，{kw}是多少",
+)
+
+
+@pytest.mark.parametrize("wrapper", CHITCHAT_WRAPPERS)
+@pytest.mark.parametrize("keyword", BUSINESS_KEYWORDS)
+def test_business_question_in_chitchat_wrapper_is_not_treated_as_chitchat(
+    keyword: str, wrapper: str
+) -> None:
+    """**核心用例**：业务关键词套上寒暄外壳，一律必须判为「带业务实质」。
+
+    这是分类器最容易漂移的一族：句子以寒暄开头（「你好」「在吗」「嗯嗯」），
+    前半段与闲聊词表高度相似，后半段才是真诉求。
+    """
+    message = wrapper.format(kw=keyword)
+    assert rules.substantive_business_request(message) is True, message
+    # 前置判据（`L0-a`）与事后复核必须**同向**：两者任一为真，另一处也不得判闲聊。
+    assert rules.is_chitchat_message(message) is False, message
+
+
+@pytest.mark.parametrize(
+    "message",
+    [
+        "怎么办理", "扣款了但没有确认", "到账了没有", "我想查一下",
+        "客服在么", "热线多少", "我要投诉", "我想找人工", "为什么被限制了",
+        "账户被冻结了", "能撤销吗", "变更一下信息", "修改一下",
+    ],
+)
+def test_business_action_terms_are_substantive(message: str) -> None:
+    """**动作与职责词**族：不含产品名词，但显然是服务请求（`BUSINESS_REQUEST_TERMS`）。"""
+    assert rules.substantive_business_request(message) is True, message
+
+
+@pytest.mark.parametrize(
+    "message",
+    [
+        "你好", "在吗", "谢谢你", "嗯嗯", "哈哈哈", "好的", "拜拜", "再见",
+        "你是谁呀", "讲个笑话", "今天天气不错", "你几岁", "你忙吗", "早上好",
+        "谢谢你啊", "好的呀",
+    ],
+)
+def test_real_chitchat_is_not_flagged_as_business(message: str) -> None:
+    """**反向守卫**：真寒暄不得被判成业务 —— 否则闲聊会被拖去查库，把快路径废掉。"""
+    assert rules.substantive_business_request(message) is False, message
+
+
+@pytest.mark.parametrize("message", ["", "   ", "？", "。。。", "🙂"])
+def test_empty_and_symbol_only_input_is_not_business(message: str) -> None:
+    """空串 / 纯符号 / 纯表情：**不算业务实质**。
+
+    这类输入由 `L0-e`（无信息量）承接，不应被护栏抢走 —— 护栏的职责是"别把业务
+    问题当闲聊"，不是"处理所有非闲聊输入"。
+    """
+    assert rules.substantive_business_request(message) is False, message
+
+
+# ---------------------------------------------------------------------------
+# 二、业务事项码：正例 + 反例
+# ---------------------------------------------------------------------------
+
+TOPIC_CASES: tuple[tuple[str, str], ...] = (
+    # 安全事项优先
+    ("我的验证码被人要走了怎么办", topic.SVC_RISK),
+    ("有人打电话让我转账，是不是诈骗", topic.SVC_RISK),
+    # 写操作 / 代办
+    ("帮我买一万块南方稳健增利债券A", topic.SVC_ACCT_OP),
+    ("帮我换一下绑定的银行卡", topic.SVC_ACCT_OP),
+    # 投诉与升级
+    ("我要投诉，让你们经理来找我", topic.SVC_COMPLAINT),
+    ("对处理结果不满意，怎么向监管部门反映", topic.SVC_COMPLAINT),
+    # 账户状态
+    ("我的账户为什么被限制了交易", topic.SVC_STATE),
+    ("风险测评到期了会有什么影响", topic.SVC_STATE),
+    ("身份证过期超过 90 天会怎样", topic.SVC_STATE),
+    # 渠道归属
+    ("我在银行买的基金能在你们这里赎回吗", topic.SVC_CHANNEL),
+    ("你们的产品在支付宝能买吗", topic.SVC_CHANNEL),
+    # 本人数据
+    ("我够哪一档", topic.SVC_MINE),
+    ("我能买什么等级的产品", topic.SVC_MINE),
+    # 适当性
+    ("我是 C1，能买 R3 的产品吗", topic.SVC_SUIT),
+    ("这只基金适合我吗", topic.SVC_SUIT),
+    ("什么情况下需要双录", topic.SVC_SUIT),
+    # 行情
+    ("159382这只ETF最近走势怎么样", topic.SVC_QUOTE),
+    ("最近净值涨了还是跌了", topic.SVC_QUOTE),
+    # 交易时限
+    ("基金赎回几天到账", topic.SVC_TXN_DAY),
+    ("交易日 15:00 后提交什么时候确认", topic.SVC_TXN_DAY),
+    # 产品参数
+    ("南方金利定开债券A的管理费率是多少", topic.SVC_PROD_PARAM),
+    ("各类基金的起投金额分别是多少", topic.SVC_PROD_PARAM),
+    # 自助办理
+    ("怎么开户", topic.SVC_SELF),
+    ("忘记密码了怎么重置", topic.SVC_SELF),
+)
+
+
+@pytest.mark.parametrize(("message", "expected"), TOPIC_CASES)
+def test_topic_classification_matches_business_expectation(
+    message: str, expected: str
+) -> None:
+    """每个事项码都有对应的**真实业务问法**（不是人工造的抽象例句）。"""
+    assert topic.topic_of(message) == expected, message
+
+
+@pytest.mark.parametrize(
+    "message",
+    ["今天天气怎么样", "讲个笑话", "你是谁呀", "", "   ", "。。。"],
+)
+def test_non_business_input_falls_to_unknown(message: str) -> None:
+    """**反例**：非业务输入不得被塞进任何一个业务事项码。
+
+    为什么重要：事项码会进留痕与报表。把寒暄记成 `SVC-SUIT`（适当性）会让
+    「本月适当性咨询量」这个业务指标凭空虚高 —— 假数据比没有数据更危险。
+    """
+    assert topic.topic_of(message) == topic.SVC_UNKNOWN, message
+
+
+def test_all_topics_are_registered_in_every_attribute_table() -> None:
+    """**自洽性**：`ALL_SVC_TOPICS` 里的每个码都能查到 owner 与 retention。
+
+    为什么必须钉住：三张表（码表 / owner / retention）分别维护，漏一行不会报错，
+    只会让某个事项在报表里静默落到兜底值。
+    """
+    for code in topic.ALL_SVC_TOPICS:
+        assert topic.owner_of(code) in (
+            topic.OWNER_BOT,
+            topic.OWNER_SELF_SERVICE,
+            topic.OWNER_HUMAN,
+        ), code
+        assert topic.retention_of(code) in (
+            topic.RETENTION_NONE,
+            topic.RETENTION_TRACE,
+            topic.RETENTION_DISCLOSURE,
+            topic.RETENTION_REQUIRED,
+        ), code
+
+
+def test_every_topic_rule_maps_to_a_known_code() -> None:
+    """判据表里的事项码必须都在 `ALL_SVC_TOPICS` 内（防手写字符串漂移）。"""
+    for code, terms in topic.TOPIC_RULES:
+        assert code in topic.ALL_TOPIC_SET, code
+        assert terms, code
+
+
+def test_human_owned_topics_are_all_required_retention() -> None:
+    """**业务属性交叉断言**：必须人工办的事项，留痕等级不得低于 `required`。
+
+    依据：投诉 / 反洗钱 / 写操作三类在制度上都有强制留痕要求
+    （`D6.3.1` 第二十三条档案保存、`D6.3.3` 第十六条与第十八条）。
+    这一条把"业务上谁办"与"合规上留什么证"绑在一起 —— 两者一旦脱钩，
+    就会出现"交给人工办了但查不到办过"的合规空洞。
+    """
+    for code in topic.ALL_SVC_TOPICS:
+        if topic.owner_of(code) == topic.OWNER_HUMAN:
+            assert topic.retention_of(code) == topic.RETENTION_REQUIRED, code
+
+
+# ---------------------------------------------------------------------------
+# 三、「下一步」动作：必须有动作、且不得触碰零容忍字面
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.parametrize("code", topic.ALL_SVC_TOPICS)
+def test_next_step_block_is_present_and_carries_a_concrete_action(code: str) -> None:
+    """每个事项都必须给得出「下一步」；且不得只是把问题推回客户。
+
+    依据（`W28` 对 `E5b` 的整改）：原话术「换个说法再问我一次就行」在真实业务里
+    等于把问题推回客户 —— 客户既不知道该换什么说法，也不知道有现成的路可走。
+    """
+    text = topic.next_step_block(
+        code, phone=rules.CONTACT_PHONE, hours=rules.CONTACT_HOURS
+    )
+    assert text.strip(), code
+    assert "换个说法再问我一次" not in text, code
+    # 必须给出**至少一个可执行落脚点**：电话号码 / APP 路径 / 线下网点 / 邮件 / 监管渠道。
+    anchors = ("400-889-8899", "APP", "客户服务中心", "@nffund.com", "12386", "96110")
+    assert any(a in text for a in anchors), (code, text)
+
+
+@pytest.mark.parametrize("code", topic.ALL_SVC_TOPICS)
+def test_next_step_block_never_hits_zero_tolerance(code: str) -> None:
+    """**合规守卫**：新话术不得触碰零容忍字面（复用产品自己的判据，不另写一套）。
+
+    为什么必须测：`E5b` 话术是**原文直返**给客户的，若含零容忍字面，会被治理层
+    或输出守护整条替换掉 —— 客户拿到的将是话术而不是内容，且不会有任何报错。
+    """
+    text = topic.next_step_block(
+        code, phone=rules.CONTACT_PHONE, hours=rules.CONTACT_HOURS
+    )
+    assert rules.hits_zero_tolerance(text) is False, (code, text)
+
+
+def test_service_topic_describe_is_consistent_with_the_two_accessors() -> None:
+    """`describe()` 是报表口径的唯一入口，必须与 `owner_of` / `retention_of` 同源。"""
+    for code in topic.ALL_SVC_TOPICS:
+        described = topic.describe(code)
+        assert described == {
+            "topic": code,
+            "owner": topic.owner_of(code),
+            "retention": topic.retention_of(code),
+        }
+
+
+# ---------------------------------------------------------------------------
+# 四、接线守卫（AST）：防「判据写好了但没接上」
+# ---------------------------------------------------------------------------
+
+_IMPL = (
+    pathlib.Path(__file__).resolve().parents[3]
+    / "app"
+    / "service"
+    / "agent"
+    / "implementations"
+    / "customer_service.py"
+)
+
+
+def _function_source(tree: ast.AST, name: str, source: str) -> str:
+    for node in ast.walk(tree):
+        if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) and node.name == name:
+            return ast.get_source_segment(source, node) or ""
+    raise AssertionError(f"未找到函数 {name}")
+
+
+def test_chitchat_branch_is_guarded_by_drift_check() -> None:
+    """**结构守卫**：`_route_and_answer` 的闲聊分支必须调用漂移护栏。
+
+    为什么用源码断言而不是跑一遍 Agent：本项目的教训是「度量工具失真比产品缺陷更危险」
+    （见 `_exit_safety` 的说明）。这条守卫的目的是让**摘掉护栏**这个动作必然变红 ——
+    否则后人重构时把两行删掉，全部金标仍会通过（金标里没有"业务问题被寒暄掉"的用例）。
+    """
+    source = _IMPL.read_text(encoding="utf-8")
+    tree = ast.parse(source)
+    body = _function_source(tree, "_route_and_answer", source)
+    assert "INTENT_CHITCHAT" in body, "闲聊分支不见了"
+    assert "substantive_business_request(request.message)" in body, (
+        "闲聊分支缺少意图漂移护栏 —— 分类器一旦把业务问题判成 chitchat，客户会收到一句寒暄"
+    )
+    # 护栏必须**在**闲聊出口之前（顺序反了等于没护栏）。
+    assert body.index("substantive_business_request(request.message)") < body.index(
+        "self._chitchat(request)"
+    ), "护栏必须在 _chitchat 调用之前"
+
+
+def test_partial_exit_declares_kb_miss_and_next_step() -> None:
+    """**结构守卫**：`E5b` 必须显式声明 `kb_miss`，并按事项给「下一步」。
+
+    为什么要显式声明：`F-2` 的转人工判据原本靠 `result.text in KNOWLEDGE_MISS_TEXTS`
+    比对文本。话术一旦按事项变化，这个判据会**静默失效**（恒为假 ⇒ 该转人工的不转），
+    而失效时没有任何报错 —— 这正是本项目登记过的那类「看着生效、其实没生效」缺陷。
+    """
+    source = _IMPL.read_text(encoding="utf-8")
+    tree = ast.parse(source)
+    body = _function_source(tree, "_exit_partial", source)
+    assert "kb_miss" in body, "E5b 未声明 kb_miss，转人工判据会退回脆弱的文本比对"
+    assert "next_step_block(" in body, "E5b 未按业务事项给出「下一步」动作"
+
+
+def test_knowledge_missed_reads_the_explicit_flag_first() -> None:
+    """**结构守卫**：`_knowledge_missed` 必须先读显式声明，再退回文本比对。"""
+    source = _IMPL.read_text(encoding="utf-8")
+    tree = ast.parse(source)
+    body = _function_source(tree, "_knowledge_missed", source)
+    assert 'result.data.get("kb_miss")' in body, "未读 kb_miss 显式声明"
+    assert body.index('result.data.get("kb_miss")') < body.index(
+        "KNOWLEDGE_MISS_TEXTS"
+    ), "显式声明必须优先于文本比对"
+
+
+def test_service_topic_is_tagged_on_every_result() -> None:
+    """**结构守卫**：`handle()` 必须给每个出口的结果打上事项码（留痕口径的唯一入口）。"""
+    source = _IMPL.read_text(encoding="utf-8")
+    tree = ast.parse(source)
+    body = _function_source(tree, "handle", source)
+    assert "_tag_service_topic(" in body, "handle() 未打事项码，报表取不到业务事项"
+
+
+# ---------------------------------------------------------------------------
+# 五、Agent 层：护栏必须真的改变走向（不只是判据层通过）
+# ---------------------------------------------------------------------------
+#
+# 判据层通过 ≠ 接线生效。这一节用**被强制的意图分类结果**复现最危险的那条漂移通路：
+# 「分类器把一条真业务问句判成 chitchat」，然后验证护栏把它拉回知识出口。
+
+
+def _request(message: str):
+    from app.core.contracts import AgentRequest, AgentRequestMetadata
+
+    return AgentRequest(
+        agent_type="customer_service",
+        message=message,
+        session_id="s-w28-drift",
+        idempotency_key="w28driftkey000001",
+        metadata=AgentRequestMetadata(),
+        history=(),
+    )
+
+
+def _agent(*, intent_code: str | None):
+    from app.core.contracts import IntentResult
+    from app.service.agent.implementations.customer_service import CustomerServiceAgent
+
+    agent = CustomerServiceAgent(CustomerServiceAgent.definition)
+    agent._classified_intent = (
+        None
+        if intent_code is None
+        else IntentResult(intent=intent_code, confidence=0.95)
+    )
+    return agent
+
+
+def _stub_hits(hits: list):
+    async def _call(name, arguments, *, intent, context):
+        del name, arguments, intent, context
+        return {"hits": hits}
+
+    return _call
+
+_KNOWLEDGE_HIT = [
+    {
+        "doc_id": "PROD-009-01",
+        "title": "公募基金与专户产品手册 · 南方稳健增利债券 A · 起投金额",
+        "content": "南方稳健增利债券 A〔示例〕：起投金额 1,000 元。",
+        "score": 0.91,
+        "source_file": "product/个人理财产品手册.md",
+    }
+]
+
+
+def _customer():
+    from app.core.contracts import RequestContext
+
+    return RequestContext(user_id="9001", trace_id="t", roles=("customer",))
+
+
+async def test_chitchat_intent_on_business_question_is_deferred_to_knowledge() -> None:
+    """**核心端到端用例**：分类器判 `chitchat`，问句却含业务实体 ⇒ 不得回寒暄。
+
+    这是护栏要守的那条通路：`L0-a` 用的是确定性判据，但它只在**分类之前**跑一次；
+    分类之后若直接采信 `chitchat`，客户会在一个起投金额问题上收到一句问候。
+    本用例把分类结果**强制**成 `chitchat` 来复现该场景。
+    """
+    impl = __import__(
+        "app.service.agent.implementations.customer_service", fromlist=["x"]
+    )
+    agent = _agent(intent_code="chitchat")
+    agent.call_tool = _stub_hits(_KNOWLEDGE_HIT)  # type: ignore[method-assign]
+
+    result = await agent._route_and_answer(
+        _request("你好，我想问下南方稳健增利债券A的起投金额是多少"), _customer()
+    )
+
+    assert result.exit_code != impl.EXIT_CHITCHAT, "业务问题被寒暄掉了 —— 护栏未生效"
+    assert "起投金额" in result.text, result.text
+
+
+async def test_real_chitchat_still_takes_the_chitchat_exit() -> None:
+    """**反向守卫**：真寒暄仍走寒暄出口 —— 护栏不得把闲聊也拖去查库。
+
+    `_chitchat` 会调模型，这里替换成哨兵，只验证**分支走向**。
+    """
+    impl = __import__(
+        "app.service.agent.implementations.customer_service", fromlist=["x"]
+    )
+    from app.core.contracts import CoreResult
+
+    agent = _agent(intent_code="chitchat")
+
+    async def _sentinel(request):
+        del request
+        return CoreResult(text="哨兵：闲聊出口", exit_code=impl.EXIT_CHITCHAT)
+
+    agent._chitchat = _sentinel  # type: ignore[method-assign]
+    result = await agent._route_and_answer(_request("谢谢你啊"), _customer())
+    assert result.exit_code == impl.EXIT_CHITCHAT
+    assert result.text == "哨兵：闲聊出口"
+
+
+async def test_handle_tags_business_topic_on_the_result() -> None:
+    """端到端确认事项码落到结果对象上（报表取数口径）。"""
+    agent = _agent(intent_code=None)
+
+    async def _fake_route(request, context):
+        del request, context
+
+        from app.core.contracts import CoreResult
+
+        return CoreResult(text="占位答复", exit_code="E5B")
+
+    agent._route_and_answer = _fake_route  # type: ignore[method-assign]
+    result = await agent.handle(_request("我在银行买的基金能在你们这赎回吗"), _customer())
+    assert result.data["svc_topic"] == topic.SVC_CHANNEL
+    assert result.data["svc_owner"] == topic.OWNER_SELF_SERVICE
+
+
+def test_eb5_empty_answer_carries_a_business_next_step() -> None:
+    """`E5b` 空答必须按**业务事项**给出具体下一步，并显式声明 `kb_miss`。"""
+    agent = _agent(intent_code=None)
+    result = agent._exit_partial(
+        [], note="知识库未命中", topic_source="我的账户为什么被限制了交易"
+    )
+    assert result.data["kb_miss"] is True
+    assert agent._knowledge_missed(result) is True
+    # 事项是「账户状态」⇒ 下一步应指向状态查询路径，而不是「换个说法再问我一次」。
+    assert "安全中心" in result.text or "客服热线" in result.text
+    assert "换个说法再问我一次" not in result.text
+
+
+def test_eb5_without_topic_source_keeps_the_legacy_shape() -> None:
+    """**向后兼容**：不传 `topic_source` 时输出与旧版逐字一致（供未改造调用点使用）。"""
+    agent = _agent(intent_code=None)
+    legacy = agent._exit_partial([], note="知识库未命中")
+    from app.service.agent.implementations.customer_service import PARTIAL_EMPTY_TEMPLATE
+
+    assert legacy.text == PARTIAL_EMPTY_TEMPLATE
+    assert agent._knowledge_missed(legacy) is True
+
+
+def test_eb5_partial_answer_does_not_get_a_next_step_block() -> None:
+    """有内容的部分答**不加**「下一步」段 —— 客户已经拿到材料，再挂路径是噪声。"""
+    agent = _agent(intent_code=None)
+    result = agent._exit_partial(
+        [{"score": 0.6, "content": "基金申购费率按金额分档。"}],
+        note="置信度不足",
+        topic_source="费率是多少",
+    )
+    assert result.data["kb_miss"] is False
+    assert agent._knowledge_missed(result) is False
+    assert "基金申购费率按金额分档。" in result.text