qyqy
57c4add7d8
- customer_service.py 以架构师实现为骨架（三档置信/适当性/会话记忆/话题矩阵），嫁接本人画像出口 - 知识检索契约合并两条链路：架构师 search_knowledge(KnowledgeSearchInput) + 本线 query_knowledge 链路所需常量（ALLOWED_CONSTANTS/VECTOR_DIM/intent_for_qa_id） - bootstrap 保留架构师 6 工具/3 Agent，补回 query_customer_profile 与 get_milvus_knowledge_writer - model_gateway 能力映射修正 intent_classification→chat，保留空集回退兜底 - governance 免责声明限定面向客户 Agent（agent_type 由定义透传），风控结构化输出不再被追加 - 修 JWT 密钥路径(config/jwt/dev)、文档 21 号撞号→25 - 测试基线 934 passed / 1 failed（既有空集缺陷）
174 lines
6.8 KiB
Python
174 lines
6.8 KiB
Python
"""客服 Agent 的**画像问答**出口单元测试。
|
||
|
||
为什么这个文件只剩画像：知识检索、适当性裁决、话题矩阵三个出口各有专门测试
|
||
（`test_customer_service_search_query.py` / `test_customer_service_suitability.py` /
|
||
`test_customer_service_topic_matrix.py`），本文件原先那些重复用例在合并时已删除——
|
||
两套测试各自锁同一份实现只会让"改一处要改两遍"。
|
||
|
||
本文件不连数据库、不调模型：`call_tool` 用替身替换。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from typing import Any
|
||
from unittest.mock import AsyncMock
|
||
|
||
import pytest
|
||
|
||
from app.core.contracts import AgentRequest, RequestContext
|
||
from app.service.agent.implementations import customer_service as cs
|
||
from app.service.agent.implementations.customer_service import CustomerServiceAgent
|
||
|
||
CONTEXT = RequestContext(user_id="9001", trace_id="t", roles=("customer",))
|
||
|
||
PROFILE_OUTPUT: dict[str, Any] = {
|
||
"customer_id": "9001",
|
||
"profile": {
|
||
"investor_type": "C3",
|
||
"investment_horizon": "medium_term",
|
||
"trading_frequency": "medium",
|
||
"preferred_asset_class": ["bond_fund"],
|
||
"customer_tier": "platinum",
|
||
"assessment_expired": False,
|
||
},
|
||
}
|
||
|
||
EXPIRED_OUTPUT: dict[str, Any] = {
|
||
"customer_id": "9001",
|
||
"profile": {
|
||
"investor_type": "C5",
|
||
"investment_horizon": "long_term",
|
||
"trading_frequency": "high",
|
||
"preferred_asset_class": ["equity_fund"],
|
||
"customer_tier": "gold",
|
||
"assessment_expired": True,
|
||
},
|
||
}
|
||
|
||
|
||
def make_request(message: str) -> AgentRequest:
|
||
return AgentRequest(
|
||
agent_type="customer_service", message=message, session_id="session-profile",
|
||
idempotency_key="k" * 16,
|
||
)
|
||
|
||
|
||
def build_agent(output: object) -> tuple[CustomerServiceAgent, AsyncMock]:
|
||
"""构造 Agent 并注入工具替身（不触达任何真实依赖）。"""
|
||
agent = CustomerServiceAgent(CustomerServiceAgent.definition)
|
||
call_tool = AsyncMock(return_value=output)
|
||
agent.call_tool = call_tool # type: ignore[method-assign]
|
||
return agent, call_tool
|
||
|
||
|
||
# --- ① 确定性识别：问"本人数据"走画像，问"规则"走知识 ------------------------------
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
("message", "expected"),
|
||
[
|
||
("我的风险等级是多少", True),
|
||
("我的风险测评什么时候到期", True),
|
||
("测评过期了吗", True),
|
||
("我是什么风险类型的投资者", True),
|
||
("我的投资偏好是什么", True),
|
||
("我的画像", True),
|
||
# 对照：问"规则"而不是"本人数据" → 不该走画像（应走知识检索）
|
||
("风险等级怎么划分", False),
|
||
("投资者分类标准是什么", False),
|
||
("基金申购后多久确认", False),
|
||
("有什么债券基金", False),
|
||
],
|
||
)
|
||
def test_profile_question_detection(message: str, expected: bool) -> None:
|
||
"""确定性识别：问"本人数据"才走画像；问"规则"走知识检索。
|
||
|
||
这是本出口存在的理由——知识库答不了"我的风险等级是多少"，而把"风险等级怎么划分"
|
||
误判成画像问题会让政策解读类问题拿不到答案。
|
||
"""
|
||
assert cs.is_profile_question(message) is expected
|
||
|
||
|
||
# --- ② 取权威字段作答 ------------------------------------------------------------
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_profile_question_answers_from_own_profile() -> None:
|
||
"""问本人画像 → 调画像工具、用权威字段作答，且**只查自己**。"""
|
||
agent, call_tool = build_agent(PROFILE_OUTPUT)
|
||
|
||
result = await agent.handle(make_request("我的风险等级是多少"), CONTEXT)
|
||
|
||
assert result.intent is not None and result.intent.intent == cs.INTENT_FAQ
|
||
assert "平衡型" in result.text and "C3" in result.text
|
||
assert result.transfer_required is False
|
||
# 只查本人：customer_id 取自 context.user_id，不从用户消息里解析
|
||
args = call_tool.await_args
|
||
assert args is not None
|
||
assert args.args[0] == cs.PROFILE_TOOL_NAME
|
||
assert args.args[1]["customer_id"] == "9001"
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_expired_assessment_is_stated_not_hidden() -> None:
|
||
"""测评过期必须**明说**并引导重新测评，不能给出一个看起来有效的等级就完事。"""
|
||
agent, _ = build_agent(EXPIRED_OUTPUT)
|
||
|
||
result = await agent.handle(make_request("我的风险等级是多少"), CONTEXT)
|
||
|
||
assert "测评已过有效期" in result.text
|
||
assert "重新完成测评" in result.text
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_profile_lookup_failure_falls_back_to_transfer() -> None:
|
||
"""工具失败/画像为空时**失败关闭为转人工**，绝不猜一个等级出来。"""
|
||
for output in ({}, {"profile": {}}, {"profile": None}):
|
||
agent, _ = build_agent(output)
|
||
result = await agent.handle(make_request("我的画像"), CONTEXT)
|
||
assert result.transfer_required is True, output
|
||
assert result.text == cs.FALLBACK_TEMPLATE, output
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_policy_question_still_goes_to_knowledge_not_profile() -> None:
|
||
"""规则类问题不走画像工具（对照组：确保上面的关键词表没有过度捕获）。"""
|
||
agent, call_tool = build_agent({"hits": []})
|
||
agent._classified_intent = None # type: ignore[assignment]
|
||
|
||
result = await agent.handle(make_request("风险等级怎么划分"), CONTEXT)
|
||
|
||
# 未识别意图 → 引导人工；关键是**没有**去调画像工具
|
||
assert result.transfer_required is True
|
||
if call_tool.await_args is not None:
|
||
assert call_tool.await_args.args[0] != cs.PROFILE_TOOL_NAME
|
||
|
||
|
||
# --- ③ 契约守卫 ------------------------------------------------------------------
|
||
|
||
|
||
def test_definition_declares_profile_tool_in_code_ceiling() -> None:
|
||
"""代码上限必须声明画像工具，否则发布配置开了白名单也调不到（两段式取交集）。"""
|
||
assert cs.PROFILE_TOOL_NAME in CustomerServiceAgent.definition.allowed_tools
|
||
|
||
|
||
def test_profile_whitelist_intent_reuses_published_key() -> None:
|
||
"""画像工具调用复用的意图 key 必须是**已发布**的那个。
|
||
|
||
换新意图码会让 `allowed_tools_by_intent` 缺 key → 交集为空 → `AGENT_PERMISSION_DENIED`，
|
||
画像出口会以"权限被拒"的形式整体失效。
|
||
"""
|
||
assert cs.PROFILE_WHITELIST_INTENT == cs.INTENT_FAQ
|
||
|
||
|
||
def test_agent_does_not_override_governance_methods() -> None:
|
||
"""治理方法是底座的责任，业务 Agent 覆盖会被 `BaseAgent.__init_subclass__` 拒绝。
|
||
|
||
这里断言类字典里没有这些名字，防止将来有人"顺手实现一个"而让合规/审计被绕过。
|
||
"""
|
||
overridden = {
|
||
"check_compliance", "_execute_governed", "call_tool", "generate_with_model",
|
||
"recall_memory", "classify_intent", "bind_governance",
|
||
} & set(CustomerServiceAgent.__dict__)
|
||
assert overridden == set()
|