test(customer-service): 锁住适当性出口的每条分支
这个出口会给出"能不能买"的结论，写错一条就是误导客户，所以把话术与取数逻辑都用 单测钉住（14 条）： - 话术：允许购买时报出产品等级与客户等级；需签揭示书/双录时明说；**拒绝时不下断言** （原因可能是等级不匹配、测评过期或未测评，统一说成"超出承受能力"是错的）； 档案里没有测评结果时写"您目前没有在有效期内的风险测评结果"而不是"您的等级为未记录"； 有效期截成日期。 - 产品名来源：只从上一轮回答的主语取，客户问句不算数，上一轮是兜底话术时返回空 （调用方据此转人工，而不是瞎猜一个产品）。 - 产品风险等级：只认"风险等级"那一行；命中别的产品的等级行必须放弃（否则拿别人的 等级给客户做裁决）；C1 的 FAQ 里"可购买 R1、R2"那种列举必须拒绝；检索降级或工具 抛异常时返回 None 而不是一个等级——那会让裁决建立在"其实没查到"之上。
This commit is contained in:
@@ -0,0 +1,199 @@
|
||||
"""客服适当性出口的单元测试。
|
||||
|
||||
这个出口会给出「能不能买」的结论，所以每条分支都必须锁住——写错一条就是误导客户。
|
||||
用例里的数字取自真实链路：南方季季盈90天是 R2，来自产品手册的「风险等级」行。
|
||||
"""
|
||||
|
||||
from typing import Any, cast
|
||||
|
||||
import pytest
|
||||
|
||||
from app.core.contracts import AgentRequest, ConversationTurn, RequestContext
|
||||
from app.service.agent.implementations.customer_service import CustomerServiceAgent
|
||||
|
||||
FALLBACK = "抱歉，这个问题我暂时无法给出准确答复。建议您拨打客服热线 400-XXX-XXXX 转人工客服咨询。"
|
||||
|
||||
|
||||
def _decision(**overrides: Any) -> dict[str, Any]:
|
||||
base: dict[str, Any] = {
|
||||
"allowed": True, "reason_code": "MATCHED", "required_disclosure": False,
|
||||
"requires_confirmation": False, "requires_recording": False,
|
||||
"customer_risk_level": 2, "assessment_valid_until": None,
|
||||
}
|
||||
base.update(overrides)
|
||||
return base
|
||||
|
||||
|
||||
def _agent() -> CustomerServiceAgent:
|
||||
"""跳过 __init__ 造实例：本组用例只碰话术与 call_tool 替身。"""
|
||||
return object.__new__(CustomerServiceAgent)
|
||||
|
||||
|
||||
def _request(*history: tuple[str, str]) -> AgentRequest:
|
||||
turns = tuple(
|
||||
ConversationTurn(role=cast(Any, role), content=content) for role, content in history
|
||||
)
|
||||
return AgentRequest(
|
||||
agent_type="customer_service", session_id="s", message="c1客户能买它吗",
|
||||
idempotency_key="1234567890123456", history=turns,
|
||||
)
|
||||
|
||||
|
||||
def _hit(title: str, content: str) -> dict[str, Any]:
|
||||
return {"doc_id": "PROD-007-03", "title": title, "content": content, "score": 0.9}
|
||||
|
||||
|
||||
# ---- 话术 ----
|
||||
|
||||
def test_allowed_reply_names_product_level_and_customer_level() -> None:
|
||||
text = CustomerServiceAgent._suitability_text("南方季季盈90天", 2, _decision())
|
||||
|
||||
assert "南方季季盈90天" in text
|
||||
assert "R2（中低风险）" in text
|
||||
assert "可以购买" in text
|
||||
assert "C2" in text
|
||||
|
||||
|
||||
def test_disclosure_and_recording_are_disclosed_when_required() -> None:
|
||||
text = CustomerServiceAgent._suitability_text(
|
||||
"南方稳健增利180天", 4, _decision(required_disclosure=True, requires_recording=True)
|
||||
)
|
||||
|
||||
assert "风险揭示书" in text
|
||||
assert "双录" in text
|
||||
|
||||
|
||||
def test_rejection_does_not_assert_a_reason() -> None:
|
||||
"""拒绝的原因可能是等级不匹配、测评过期或未测评，不能一律说成"超出承受能力"。"""
|
||||
text = CustomerServiceAgent._suitability_text(
|
||||
"南方季季盈90天", 3, _decision(allowed=False)
|
||||
)
|
||||
|
||||
assert "暂时无法购买" in text
|
||||
assert "超出" not in text
|
||||
|
||||
|
||||
def test_missing_assessment_is_said_in_plain_words() -> None:
|
||||
"""档案里没有测评结果时不能写"您的等级为未记录"——客户看不懂。"""
|
||||
text = CustomerServiceAgent._suitability_text(
|
||||
"南方季季盈90天", 2, _decision(allowed=False, customer_risk_level=None)
|
||||
)
|
||||
|
||||
assert "没有在有效期内的风险测评结果" in text
|
||||
assert "未记录" not in text
|
||||
|
||||
|
||||
def test_validity_is_shortened_to_a_plain_date() -> None:
|
||||
text = CustomerServiceAgent._suitability_text(
|
||||
"南方季季盈90天", 2, _decision(assessment_valid_until="2027-01-15T00:00:00+00:00")
|
||||
)
|
||||
|
||||
assert "2027-01-15" in text
|
||||
assert "T00:00" not in text
|
||||
|
||||
|
||||
def test_unknown_level_falls_back_to_the_code() -> None:
|
||||
text = CustomerServiceAgent._suitability_text("某产品", 9, _decision())
|
||||
|
||||
assert "R9" in text
|
||||
|
||||
|
||||
# ---- 产品名来源 ----
|
||||
|
||||
def test_previous_topic_takes_the_subject_of_the_last_answer() -> None:
|
||||
topic = CustomerServiceAgent._previous_topic(_request(
|
||||
("user", "季季盈90天起投多少"),
|
||||
("assistant", "南方季季盈90天：起投金额 1万元"),
|
||||
))
|
||||
|
||||
assert topic == "南方季季盈90天"
|
||||
|
||||
|
||||
def test_previous_topic_ignores_user_turns() -> None:
|
||||
"""客户的问句不是可信的产品名来源，只能从回答里取。"""
|
||||
assert CustomerServiceAgent._previous_topic(
|
||||
_request(("user", "季季盈90天起投多少"))
|
||||
) == ""
|
||||
|
||||
|
||||
def test_previous_topic_is_empty_after_a_fallback_answer() -> None:
|
||||
"""上一轮是兜底话术时取不出主语，调用方据此转人工（而不是瞎猜一个产品）。"""
|
||||
assert CustomerServiceAgent._previous_topic(
|
||||
_request(("user", "随便问点什么"), ("assistant", FALLBACK))
|
||||
) == ""
|
||||
|
||||
|
||||
# ---- 产品风险等级 ----
|
||||
|
||||
async def _level(hits: list[dict[str, Any]]) -> int | None:
|
||||
agent = _agent()
|
||||
|
||||
async def fake_call_tool(name: str, arguments: dict[str, Any], **kwargs: Any) -> Any:
|
||||
return {"hits": hits, "degraded": False}
|
||||
|
||||
agent.call_tool = fake_call_tool # type: ignore[method-assign]
|
||||
return await agent._product_risk_level(
|
||||
"南方季季盈90天", RequestContext(user_id="9001", trace_id="t")
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_level_is_read_from_the_risk_level_row() -> None:
|
||||
level = await _level([
|
||||
_hit("手册 · 2.1 南方季季盈90天 · 风险等级", "南方季季盈90天：风险等级 R2（中低风险）")
|
||||
])
|
||||
|
||||
assert level == 2
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_row_about_another_product_is_rejected() -> None:
|
||||
"""命中别的产品的等级行时必须放弃，否则会拿别人的等级去给客户做裁决。"""
|
||||
level = await _level([
|
||||
_hit("手册 · 3.1 南方稳健增利180天 · 风险等级", "南方稳健增利180天：风险等级 R3（中风险）")
|
||||
])
|
||||
|
||||
assert level is None
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_enumeration_row_is_rejected() -> None:
|
||||
"""C1 的 FAQ 里写着"可购买 R1、R2"这种列举，那不是产品等级。"""
|
||||
level = await _level([
|
||||
_hit(
|
||||
"C1 客户能买什么产品?",
|
||||
"C1（保守型）客户可以购买 R1（低风险）、R2（中低风险）等级的产品。",
|
||||
)
|
||||
])
|
||||
|
||||
assert level is None
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_degraded_search_yields_no_level() -> None:
|
||||
"""检索降级时不能返回一个等级——那会让裁决建立在"其实没查到"之上。"""
|
||||
agent = _agent()
|
||||
|
||||
async def fake_call_tool(name: str, arguments: dict[str, Any], **kwargs: Any) -> Any:
|
||||
return {"hits": [], "degraded": True, "reason": "search_failed"}
|
||||
|
||||
agent.call_tool = fake_call_tool # type: ignore[method-assign]
|
||||
|
||||
assert await agent._product_risk_level(
|
||||
"南方季季盈90天", RequestContext(user_id="9001", trace_id="t")
|
||||
) is None
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_tool_failure_yields_no_level() -> None:
|
||||
"""工具抛异常时返回 None（由调用方转人工），不能让异常冒到主链路。"""
|
||||
agent = _agent()
|
||||
|
||||
async def fake_call_tool(name: str, arguments: dict[str, Any], **kwargs: Any) -> Any:
|
||||
raise RuntimeError("milvus 挂了")
|
||||
|
||||
agent.call_tool = fake_call_tool # type: ignore[method-assign]
|
||||
|
||||
assert await agent._product_risk_level(
|
||||
"南方季季盈90天", RequestContext(user_id="9001", trace_id="t")
|
||||
) is None
|
||||
Reference in New Issue
Block a user