张胜宇
3e24033f72
设计与依据：新增 D3.9-客服Agent智能路由与行情出口设计-2026-09-21.md（CS-ARCH-2026-024）
DEC-W27-1~12 全部批准并落地。
## 新增（3 个源文件 + 3 个测试文件）
- app/core/exit_codes.py：出口码注册表（E0/E1/E2a~E2e/E3/E4/E5a~E5c/E6/E8/CHAT/LOGIN/CONTACT）
+ NON_BUSINESS_EXIT_CODES 白名单（免责分档的判据面）
- app/service/fund_trend_service.py（E6）：summarize_trend() 纯函数 + query_fund_trend 工具
读 fin_nav_history；按净值日个数（5/20/60/120）给区间涨跌 / 区间高低 / 来源与区间
- tests/unit/{service,tools} 新增 45 条守卫单测（E6 出口 / L0 表层判定 / 工具配置）
## 改动（重点）
- customer_service.py：
· _route_surface()（L0-a 闲聊 / L0-b 行情 / L0-e 无信息量），位置在安全路由之后
· drop_yield_claims() 重写为「入口闸 × 槽位白名单」+ fail-closed
（旧黑名单会误删权重：A-06 业绩基准公式整行消失）
· allowed_tools 补 query_fund_trend —— 修掉发布被拒 422「配置超出 Agent 工具上限」
（该 422 是**子集**校验而非数量上限，根因即此项漏配）
· 25 处 CoreResult 全部补 exit_code（AST 守卫保证零遗漏）
- customer_service_rules.py：闲聊判定重写（业务实体边界 + 词表 + 特征串 + 语气词）
+ is_low_information_message() + is_contact_inquiry()
- governance.py：免责声明按 exit_code 分档（业务档完整 / 非业务档轻型）
- actor.py：VISITOR_PERMISSIONS 补 fund:quote:read（E6 对访客开放，DEC-W27-11）
- tools/：seed_compliance_baseline 新增 TPL_DISCLAIMER_LIGHT；publish 工具改为
「从生效版本派生原列表再追加」（_inherit_tools 取不到返回 None，防静默改窄）
## 实测
- 金标扩容 46 → 55（新增 Q 组行情 5 条 + C 组闲聊 4 条）
- M-1 55/55、M-4 55/55；M-5 与四项零容忍（M-7/M-8/M-9/M-10）全 0
- 原 46 条可比基线逐项不变（46/46）；M-6 9.1%（5 条全在转人工白名单内）
- 全量 pytest 2094 passed / 3 skipped；ruff 零新增（余 5 条与 HEAD 逐条对应）
- 配置版本 244（cs-tools-75813de45421）已发布并激活
## 文档
- 新增 D3.9（设计专册）；D1.1 §34 补「代码实施已同轮完成」并把旧表述作废
- D3.7：新增 Q 组判据表 + §6.4 第四次实测 + M-3/M-9 口径澄清
- D2.9：新增 §2.11（9 条新增用例的实测答复原文）+ 汇总表改双列口径
- D2.1：新增 v6.41 执行条目
- 权威副本（D:\桌面\金融\）→ 仓库镜像 全量比对一致（0 缺失 / 0 不一致）
## 诚实未做项
- 知识出口（E3/E4）侧的实体锚点闸门未做（故金标扩容 9 条而非 12 条，A-11~A-13 未成集）
- 阈值未重标（D3.9 §3.4 只给方法）；M-3 分母未改（仅明示口径）；
D1.1 §0 与 §4.0 历史计数偏移（差 1）仍沿用
469 lines
23 KiB
Python
469 lines
23 KiB
Python
"""免责声明强制注入与 `applicable_agents` 门控的单元测试。
|
||
|
||
两条需求来源：
|
||
1. 上线门禁 F5「免责声明强制注入，面向客户输出 **100%** 附固定话术」。所以**正常输出**与
|
||
**被负面词拦截后替换的输出**两个分支都必须带上免责声明——被拦截的回复同样是面向客户的输出。
|
||
2. Task 1 审查发现的 fail-open 缺陷：`applicable_agents` 为 `[]`/`NULL` 时旧的
|
||
`if agents and ...` 判据不执行过滤，规则**溢出到所有 Agent**，与项目"失败关闭"原则相反。
|
||
|
||
本文件不连数据库：`PlatformGovernance` 的取数会话用替身替换（与
|
||
`tests/unit/service/test_memory_extraction_service.py` 同一手法），
|
||
真实 MySQL 上的门控行为由 `tests/integration/test_compliance_seed_mysql.py` 固定。
|
||
"""
|
||
|
||
from typing import Any
|
||
|
||
import pytest
|
||
|
||
from app.core.contracts import (
|
||
AgentDefinition,
|
||
AgentResult,
|
||
CoreResult,
|
||
RecalledMemory,
|
||
RequestContext,
|
||
ResolvedAgentConfig,
|
||
)
|
||
from app.service.agent import governance
|
||
from app.service.agent.governance import (
|
||
FALLBACK_DISCLAIMER,
|
||
FALLBACK_LIGHT_NOTICE,
|
||
review_output,
|
||
)
|
||
|
||
CONTEXT = RequestContext(user_id="9001", trace_id="t", roles=("customer",))
|
||
CONFIG = ResolvedAgentConfig(config_version="v", prompt_version="p", model_endpoint="m")
|
||
|
||
#: 固定话术的插入位置：`\n\n` 分隔，附在正文末尾。
|
||
DISCLAIMER = "\n\n" + FALLBACK_DISCLAIMER
|
||
|
||
|
||
def _result(text: str) -> AgentResult:
|
||
return AgentResult(run_id="r", result=CoreResult(text=text))
|
||
|
||
|
||
def _review(
|
||
text_or_result: Any, config: ResolvedAgentConfig | None = None,
|
||
memories: tuple[RecalledMemory, ...] = (), **kwargs: Any,
|
||
) -> AgentResult:
|
||
"""面向客户的治理调用（带 `agent_type`）。
|
||
|
||
门禁 F5 的适用范围是**已确认面向客户**的 Agent（`agent_type` 由 `PlatformGovernance`
|
||
从发布版本反查）。本文件的断言全部针对该场景，所以统一在这里显式声明 agent 类型，
|
||
避免每处都写一遍；`test_internal_agent_gets_no_disclaimer` 是对照组。
|
||
"""
|
||
result = text_or_result if isinstance(text_or_result, AgentResult) \
|
||
else _result(text_or_result)
|
||
return review_output(result, CONTEXT, config or CONFIG, memories,
|
||
agent_type="customer_service", **kwargs)
|
||
|
||
|
||
def test_disclaimer_is_appended_to_normal_output() -> None:
|
||
result = _review("基金申购后 T+1 确认份额。")
|
||
assert FALLBACK_DISCLAIMER in result.result.text
|
||
assert result.result.text.startswith("基金申购后 T+1 确认份额。")
|
||
assert result.result.text.endswith(DISCLAIMER)
|
||
|
||
|
||
def test_internal_agent_gets_no_disclaimer() -> None:
|
||
"""对照组：内部 Agent（风控分析）的结构化输出**不得**被追加面向客户的免责声明。
|
||
|
||
判据是"输出形态"：风控输出是字段化摘要（预警编号/级别/建议动作），追加一句面向
|
||
投资者的免责声明会破坏其字段契约，下游解析与 `tests/contract/test_risk_agent_contract.py`
|
||
都会因此失败（合并时实测复现过）。`agent_type` 为空（未声明）时同样不注入。
|
||
"""
|
||
for agent_type in ("risk", ""):
|
||
result = review_output(_result("工单标题：测试工单"), CONTEXT, CONFIG, (),
|
||
agent_type=agent_type)
|
||
assert result.result.text == "工单标题：测试工单", agent_type
|
||
|
||
|
||
def test_disclaimer_is_not_duplicated() -> None:
|
||
"""幂等：同一结果被治理两次（重放/重试）不得把免责声明叠成两条。"""
|
||
once = _review("答案。")
|
||
twice = _review(once)
|
||
assert twice.result.text.count(FALLBACK_DISCLAIMER) == 1
|
||
assert twice.result.text == once.result.text
|
||
|
||
|
||
def test_disclaimer_inside_the_answer_is_still_appended_at_the_end() -> None:
|
||
"""幂等判断不认"文本里出现过话术"，只认"末尾是不是我们自己追加的那个形状"。
|
||
|
||
子串包含判断下，正文中间出现同样的文字就不会追加（实测过），于是"客户拿到的话术"取决于
|
||
正文里恰好有没有这段话——门禁 F5 要的是 100% 覆盖，正文里出现过不算"已附话术"，
|
||
必须落在**末尾**。
|
||
"""
|
||
middle = f"前半句。{FALLBACK_DISCLAIMER}后半句。"
|
||
result = _review(middle)
|
||
assert result.result.text == middle + DISCLAIMER
|
||
assert result.result.text.endswith(DISCLAIMER)
|
||
assert result.result.text.count(FALLBACK_DISCLAIMER) == 2
|
||
|
||
|
||
def test_answer_already_carrying_the_appended_shape_is_not_duplicated() -> None:
|
||
"""幂等的另一半语义：正文已经以"我们追加过的形状"结尾时，不重复追加（话术只出现一次）。"""
|
||
result = _review("答案。" + DISCLAIMER)
|
||
assert result.result.text.count(FALLBACK_DISCLAIMER) == 1
|
||
assert result.result.text.endswith(DISCLAIMER)
|
||
|
||
|
||
def test_degenerate_published_wording_is_still_appended() -> None:
|
||
"""修复轮 2 的核心动机：**退化文案下也必须追加**。
|
||
|
||
管理员若把免责声明发布成"。"（退化数据，`agent_reply_template.content_text` 没有任何校验），
|
||
答案恰好是"您好。"时，`endswith(disclaimer_text)` 会判定"已经以话术结尾" → **不追加**，
|
||
客户一条话术都拿不到。只认我们自己追加过的形状（"\\n\\n" + 文案）后，这种文本必然被追加。
|
||
代价是极端情况下可能多追加一次——方向正确：门禁 F5 要的是 100% 附话术，宁可多一次不可漏。
|
||
"""
|
||
result = _review("您好。", disclaimer="。")
|
||
assert result.result.text == "您好。\n\n。"
|
||
assert result.result.text.endswith("\n\n。")
|
||
|
||
|
||
def test_degenerate_published_wording_is_appended_only_once() -> None:
|
||
"""退化文案下幂等仍成立：第二次治理不再追加（认的正是第一次追加出来的那个形状）。"""
|
||
once = _review("您好。", disclaimer="。")
|
||
twice = _review(once, disclaimer="。")
|
||
assert twice.result.text == once.result.text
|
||
assert twice.result.text == "您好。\n\n。"
|
||
|
||
|
||
def test_fallback_disclaimer_survives_the_hardcoded_hard_patterns() -> None:
|
||
"""代码常量 `FALLBACK_DISCLAIMER` 的行为守卫：库侧 6 条话术有守卫，它此前没有。
|
||
|
||
若该常量命中硬编码 `hard_patterns`，治理层会先把回复判定为"需人工核实"并替换成安全话术，
|
||
也就是**平台自己的免责声明被平台自己的规则拦下**（Task 1 修过一次的"自绊"）。这条断言是哨兵。
|
||
|
||
⚠️ 为什么**不能**按"输入正文 = 兜底文案 + `assert FALLBACK_DISCLAIMER in 输出`"来写：
|
||
那是恒真的空转——输入本身就是该文案，且命中硬编码词时文案会在替换**之后**被追加回末尾，
|
||
两种情况下 `in` 都成立（已实测：把常量改成含「保证收益」后那条断言仍为真）。
|
||
真正会失败的是下面两条：①没被判定为拦截；②正文原样保留（未被安全话术替换）。
|
||
入参刻意用"正文 + 兜底文案"而不是只有文案，避免"输入即断言对象"造成的自证。
|
||
"""
|
||
answer = "基金申购后 T+1 确认份额。" + FALLBACK_DISCLAIMER
|
||
result = _review(answer)
|
||
assert result.result.transfer_required is False, (
|
||
"兜底免责声明被平台自己的硬编码 hard_patterns 拦下了（自绊）")
|
||
# 正文原样保留（未被替换成安全话术），并按规范形状补一次"\n\n"分隔的追加：
|
||
# 裸文案结尾不算"已追加"，这正是修复轮 2 收紧后的语义。
|
||
assert result.result.text == answer + DISCLAIMER, "正文被替换成了安全话术，说明话术命中了规则"
|
||
assert result.result.text.endswith(DISCLAIMER)
|
||
|
||
|
||
def test_disclaimer_is_appended_after_negative_word_replacement() -> None:
|
||
"""裁定 2：命中负面词被替换成安全话术的分支同样要追加免责声明（F5 要求 100%）。
|
||
|
||
顺序固定为「校验 → 负面词替换 → 脱敏 → 追加免责声明」：免责声明必须落在**替换后**的
|
||
文本之后，否则拦截分支就没有话术。
|
||
"""
|
||
config = CONFIG.model_copy(update={"negative_rules": (("contains", "测试禁止词"),)})
|
||
result = _review("这只基金测试禁止词", config=config)
|
||
assert result.result.transfer_required is True
|
||
assert "测试禁止词" not in result.result.text
|
||
assert result.result.text.startswith("该内容需要人工核实")
|
||
assert result.result.text.endswith(DISCLAIMER)
|
||
|
||
|
||
def test_disclaimer_is_appended_after_hard_pattern_replacement() -> None:
|
||
"""硬编码 `hard_patterns`（不依赖库配置）走的是同一分支，同样必须带话术。"""
|
||
result = _review("这只产品保证收益")
|
||
assert "保证收益" not in result.result.text
|
||
assert result.result.text.endswith(DISCLAIMER)
|
||
|
||
|
||
def test_explicit_disclaimer_argument_wins_over_code_fallback() -> None:
|
||
"""异步层读到的库内话术优先于代码兜底。"""
|
||
result = _review("答案。", disclaimer="库内免责声明。")
|
||
assert result.result.text == "答案。\n\n库内免责声明。"
|
||
assert FALLBACK_DISCLAIMER not in result.result.text
|
||
|
||
|
||
def test_blank_disclaimer_argument_still_gets_the_fixed_wording() -> None:
|
||
"""显式传入空串不得成为"绕过强制注入"的后门：空串一律退回代码兜底。
|
||
|
||
门禁 F5 要求 100% 覆盖，"传了但传空"不算注入话术。
|
||
"""
|
||
result = _review("答案。", disclaimer=" ")
|
||
assert FALLBACK_DISCLAIMER in result.result.text
|
||
|
||
|
||
def test_existing_callers_without_disclaimer_still_get_wording() -> None:
|
||
"""既有调用点按旧签名调用（不传 disclaimer）必须照旧拿到固定话术。"""
|
||
result = _review("答案。")
|
||
assert result.result.text == "答案。" + DISCLAIMER
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# `applicable_agents` 门控：fail-closed
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class _FakeRows:
|
||
def __init__(self, rows: list[dict[str, Any]]) -> None:
|
||
self._rows = rows
|
||
|
||
def mappings(self) -> "_FakeRows":
|
||
return self
|
||
|
||
def all(self) -> list[dict[str, Any]]:
|
||
return self._rows
|
||
|
||
|
||
class _FakeSession:
|
||
"""`resolve()` 用到的替身：`scalar` 取发布版本/话术，`execute` 取规则。
|
||
|
||
这里**不**按 SQL 文本分派：`resolve()` 自己那条 `scalar` 就是取 release，改动它会
|
||
让"是否命中发布配置"的语义漂移，把门控测试变成假绿。`review()` 需要区分两次查询，
|
||
因此另有 `_FakeReviewSession`。
|
||
"""
|
||
|
||
def __init__(self, *, rows: list[dict[str, Any]] = (), template: str | None = None) -> None:
|
||
self._rows = rows
|
||
self._template = template
|
||
|
||
async def scalar(self, _statement: Any, _params: Any = None) -> Any:
|
||
# 返回话术文本（话术查询）或 None（没有 active release，工具分支整段跳过）。
|
||
return self._template
|
||
|
||
async def execute(self, _statement: Any) -> _FakeRows:
|
||
return _FakeRows(self._rows)
|
||
|
||
async def __aenter__(self) -> "_FakeSession":
|
||
return self
|
||
|
||
async def __aexit__(self, *args: object) -> None:
|
||
return None
|
||
|
||
|
||
class _FakeReviewSession(_FakeSession):
|
||
"""`PlatformGovernance.review()` 用到的替身：`scalar` 只回话术文本。
|
||
|
||
`agent_type` **不再**从库里反查——它由 `BaseAgent` 从 Agent 定义直接传给 `review()`
|
||
（`definition.agent_type` 是定义的一部分，没有理由为它多付一次查询）。
|
||
因此替身只需回答"话术是什么"。
|
||
"""
|
||
|
||
def __init__(self, *, template: str | None = None) -> None:
|
||
super().__init__(template=template)
|
||
|
||
|
||
def _rows(applicable_agents: Any) -> list[dict[str, Any]]:
|
||
"""一行规则：MySQL JSON 列经驱动回来是字符串，`NULL` 回来是 None（实测口径）。"""
|
||
return [{"match_type": "contains", "word_pattern": "测试禁止词",
|
||
"applicable_agents": applicable_agents}]
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
("applicable_agents", "customer_service_rules", "advisor_rules"),
|
||
[
|
||
# (客服加载数, 投顾加载数)：与派发说明里的真值表逐格对应，第 2、3 行是本次修复的缺陷。
|
||
('["customer_service"]', 1, 0), # 正常门控：只对声明的 Agent 生效
|
||
([], 0, 0), # 空数组：不适用任何 Agent（fail-closed；原来溢出成 1/1）
|
||
('[]', 0, 0), # 同上，但按驱动真实口径写成 JSON 文本（实测回 str）
|
||
(None, 0, 0), # NULL：同上
|
||
('["advisor"]', 0, 1), # 指向别的 Agent：客服一条都拿不到（原来的静默失效）
|
||
],
|
||
)
|
||
async def test_applicable_agents_gating_fails_closed(
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
applicable_agents: Any,
|
||
customer_service_rules: int,
|
||
advisor_rules: int,
|
||
) -> None:
|
||
"""三种坏取值都不得让规则溢出：空数组/NULL 是"不适用任何 Agent"，不是"适用全部"。"""
|
||
monkeypatch.setattr(governance, "SessionFactory",
|
||
lambda: _FakeSession(rows=_rows(applicable_agents)))
|
||
service = governance.PlatformGovernance()
|
||
customer = await service.resolve(
|
||
AgentDefinition(agent_type="customer_service", version="1"), CONTEXT)
|
||
advisor = await service.resolve(AgentDefinition(agent_type="advisor", version="1"), CONTEXT)
|
||
assert len(customer.negative_rules) == customer_service_rules
|
||
assert len(advisor.negative_rules) == advisor_rules
|
||
|
||
|
||
async def test_empty_applicable_agents_rules_are_not_loaded_at_all(
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
) -> None:
|
||
"""空取值的反证断言：不是"条数变少"，而是该规则对**任何** Agent 都不存在。"""
|
||
monkeypatch.setattr(governance, "SessionFactory",
|
||
lambda: _FakeSession(rows=_rows([])))
|
||
service = governance.PlatformGovernance()
|
||
for agent_type in ("customer_service", "advisor", "fund_query_demo"):
|
||
resolved = await service.resolve(
|
||
AgentDefinition(agent_type=agent_type, version="1"), CONTEXT)
|
||
assert resolved.negative_rules == ()
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 同步/异步分离（裁定 1）：读库在异步层，追加在同步层
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
CONFIG_WITH_RELEASE = CONFIG.model_copy(update={"release_id": 186})
|
||
|
||
|
||
async def test_review_reads_template_from_database(monkeypatch: pytest.MonkeyPatch) -> None:
|
||
"""走 `PlatformGovernance.review()`：话术取库，`agent_type` 由调用方透传。"""
|
||
monkeypatch.setattr(governance, "SessionFactory",
|
||
lambda: _FakeReviewSession(template="库内固定免责声明。"))
|
||
reviewed = await governance.PlatformGovernance().review(
|
||
_result("答案。"), CONTEXT, CONFIG_WITH_RELEASE, (), agent_type="customer_service")
|
||
assert reviewed.result.text == "答案。\n\n库内固定免责声明。"
|
||
|
||
|
||
async def test_review_skips_disclaimer_for_internal_agent(
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
) -> None:
|
||
"""内部 Agent（风控）不追加话术：它的输出是字段化摘要，加话术会破坏字段契约。
|
||
|
||
`agent_type` 由 `BaseAgent` 从定义传入，与发布配置无关（没有发布版本也要能判定）。
|
||
"""
|
||
monkeypatch.setattr(governance, "SessionFactory",
|
||
lambda: _FakeReviewSession(template="库内固定免责声明。"))
|
||
reviewed = await governance.PlatformGovernance().review(
|
||
_result("工单标题：测试工单"), CONTEXT, CONFIG_WITH_RELEASE, (), agent_type="risk")
|
||
assert reviewed.result.text == "工单标题：测试工单"
|
||
|
||
|
||
async def test_review_without_agent_type_declaration_does_not_inject(
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
) -> None:
|
||
"""治理替身按旧签名调用（未传 `agent_type`）时不注入，而不是猜成面向客户。
|
||
|
||
这条是给**测试替身**定的口径：它们刻意不连库、也不知道 agent 类型，若默认注入，
|
||
每个替身测试都会被塞进一句话术，等于用测试噪声换一个假的安全感。生产路径恒有类型
|
||
（`BaseAgent` 从定义传），所以 F5 的覆盖不受影响。
|
||
"""
|
||
monkeypatch.setattr(governance, "SessionFactory",
|
||
lambda: _FakeReviewSession(template="库内固定免责声明。"))
|
||
reviewed = await governance.PlatformGovernance().review(
|
||
_result("答案。"), CONTEXT, CONFIG_WITH_RELEASE, ())
|
||
assert reviewed.result.text == "答案。"
|
||
|
||
|
||
@pytest.mark.parametrize("template", [None, "", " "])
|
||
async def test_review_falls_back_to_code_constant_when_template_missing(
|
||
monkeypatch: pytest.MonkeyPatch, template: str | None,
|
||
) -> None:
|
||
"""查不到/查到空白话术时退回代码常量：门禁 F5 不依赖数据库可用性。"""
|
||
monkeypatch.setattr(governance, "SessionFactory", lambda: _FakeReviewSession(template=template))
|
||
reviewed = await governance.PlatformGovernance().review(
|
||
_result("答案。"), CONTEXT, CONFIG_WITH_RELEASE, (), agent_type="customer_service")
|
||
assert reviewed.result.text == "答案。" + DISCLAIMER
|
||
|
||
|
||
async def test_review_survives_database_failure(monkeypatch: pytest.MonkeyPatch) -> None:
|
||
"""取数抛异常不得让回复失败，也不得让回复**没有**免责声明。
|
||
|
||
场景取"会话可建、查询必炸"（比"建会话就炸"更贴近真实的连接池故障）。
|
||
"""
|
||
|
||
class _ExplodingTemplateSession(_FakeSession):
|
||
async def scalar(self, _statement: Any, _params: Any = None) -> Any:
|
||
raise RuntimeError("database unavailable")
|
||
|
||
monkeypatch.setattr(governance, "SessionFactory", lambda: _ExplodingTemplateSession())
|
||
reviewed = await governance.PlatformGovernance().review(
|
||
_result("答案。"), CONTEXT, CONFIG_WITH_RELEASE, (), agent_type="customer_service")
|
||
assert reviewed.result.text == "答案。" + DISCLAIMER
|
||
|
||
|
||
def test_recalled_memories_do_not_affect_disclaimer() -> None:
|
||
"""带记忆的正常路径同样注入（确认追加在引用校验之后、不受 memories 影响）。"""
|
||
memories = (RecalledMemory(memory_uuid="m1", customer_id="9001", content="偏好低风险"),)
|
||
reviewed = _review("答案。", memories=memories)
|
||
assert reviewed.result.text == "答案。" + DISCLAIMER
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# `W27` / `DEC-W27-10`：门禁 F5 **分两档**（判据是出口码）
|
||
# ---------------------------------------------------------------------------
|
||
|
||
LIGHT = "\n\n" + FALLBACK_LIGHT_NOTICE
|
||
|
||
|
||
def _result_with_exit(text: str, exit_code: str | None) -> AgentResult:
|
||
return AgentResult(run_id="r", result=CoreResult(text=text, exit_code=exit_code))
|
||
|
||
|
||
@pytest.mark.parametrize("exit_code", ["CHAT", "LOGIN", "CONTACT", "E1", "E5a"])
|
||
def test_non_business_exits_get_the_light_notice_only(exit_code: str) -> None:
|
||
"""非业务档只附轻量提示：寒暄/联系方式/登录引导/澄清挂投资风险提示是挂错话术。
|
||
|
||
这也是"客服在背模板"的体感来源之一 —— 门禁 F5 原口径要求 100% 附完整话术，
|
||
于是「讲个笑话」后面跟一段投资风险提示。分档**不减少业务答复的披露**。
|
||
"""
|
||
reviewed = _review(_result_with_exit("您好呀，想了解基金产品吗？", exit_code))
|
||
assert reviewed.result.text.endswith(LIGHT)
|
||
assert FALLBACK_DISCLAIMER not in reviewed.result.text
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
"exit_code", ["E2e", "E3", "E4", "E5b", "E5c", "E6", "P0", "P1", "P2", None]
|
||
)
|
||
def test_business_exits_keep_the_full_disclaimer(exit_code: str | None) -> None:
|
||
"""业务档（含 `E5b` 空答、`E5c` 转人工、`E6` 行情与安全出口）照旧附**完整**话术。
|
||
|
||
`E5b`（"这一条我暂时没找到对应的公开资料…"）是本条最要紧的一格：它的正文里
|
||
**一个数字都没有**，如果用"文本里有没有数字"这类间接判据分档，它会被判成非业务并
|
||
静默掉档 —— 而它恰恰是"业务上答不上来"，投资风险提示一个字都不能少。
|
||
"""
|
||
reviewed = _review(_result_with_exit("关于这一点，公开资料里的口径是：…", exit_code))
|
||
assert reviewed.result.text.endswith(DISCLAIMER)
|
||
assert FALLBACK_LIGHT_NOTICE not in reviewed.result.text
|
||
|
||
|
||
def test_light_notice_is_idempotent_per_tier() -> None:
|
||
"""轻量提示同样只追加一次（重放/重试不得叠成两条）。"""
|
||
once = _review(_result_with_exit("在的，请问需要了解什么？", "CHAT"))
|
||
twice = _review(once)
|
||
assert twice.result.text.count(FALLBACK_LIGHT_NOTICE) == 1
|
||
assert twice.result.text == once.result.text
|
||
|
||
|
||
def test_tier_switch_appends_the_missing_notice() -> None:
|
||
"""已经带了轻量提示的答复再被判成业务档时，**完整话术照加**（不重复轻量提示）。
|
||
|
||
这是"幂等判据按档分别生效"的边界：幂等只认**本次该用的那一份**的形状，
|
||
而不是"末尾已经有任何一句话术就不加"。方向取"宁可多披露一句"。
|
||
"""
|
||
result = _result_with_exit("答案。" + LIGHT, "E3")
|
||
reviewed = _review(result)
|
||
assert reviewed.result.text.endswith(DISCLAIMER)
|
||
# 两头都在：轻量提示是上一次追加的，完整话术是这一次的
|
||
assert FALLBACK_LIGHT_NOTICE in reviewed.result.text
|
||
assert FALLBACK_DISCLAIMER in reviewed.result.text
|
||
|
||
|
||
async def test_review_loads_the_light_template_by_its_own_code(
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
) -> None:
|
||
"""`review()` 按 `template_code` 分别取两份话术 —— 两个 code 都得查一次。"""
|
||
from app.service.agent.governance import DISCLAIMER_TEMPLATE_CODE, LIGHT_NOTICE_TEMPLATE_CODE
|
||
|
||
seen: list[str] = []
|
||
|
||
class _CodeAwareSession(_FakeReviewSession):
|
||
async def scalar(self, statement: Any, params: Any = None) -> Any:
|
||
code = str((params or {}).get("code") or "")
|
||
seen.append(code)
|
||
return {
|
||
DISCLAIMER_TEMPLATE_CODE: "库内完整免责话术。",
|
||
LIGHT_NOTICE_TEMPLATE_CODE: "库内轻量提示。",
|
||
}.get(code)
|
||
|
||
monkeypatch.setattr(governance, "SessionFactory", lambda: _CodeAwareSession())
|
||
service = governance.PlatformGovernance()
|
||
|
||
business = await service.review(
|
||
_result_with_exit("答案。", "E3"), CONTEXT, CONFIG_WITH_RELEASE, (),
|
||
agent_type="customer_service",
|
||
)
|
||
assert business.result.text.endswith("\n\n库内完整免责话术。")
|
||
|
||
small_talk = await service.review(
|
||
_result_with_exit("在的。", "CHAT"), CONTEXT, CONFIG_WITH_RELEASE, (),
|
||
agent_type="customer_service",
|
||
)
|
||
assert small_talk.result.text.endswith("\n\n库内轻量提示。")
|
||
assert DISCLAIMER_TEMPLATE_CODE in seen
|
||
assert LIGHT_NOTICE_TEMPLATE_CODE in seen
|