lzf_0626
8b1ed70916
实测连问「季季盈90天起投多少」→「C3 客户能买它吗」→「那我能买它吗」，第三问转人工。 原因是上一步刚加的客户自述等级说明（"您提到自己是 C3。…以您在公司留存的测评结果为准"） 占了回答首行，而 _topic_of 只看首行，产品名被挤到第二行，追问于是丢掉了指代对象。 这是 _topic_of 第三次因为"只认固定格式"而咬人（前两次：FAQ 的"问"字、适当性回答没有 冒号）。改成遍历回答前四行、逐行尝试解析，并把单行解析抽成 _topic_in。 同时补上自述等级时的依据说明：客户说"我是 C3"而系统答"您没有测评结果"，在他眼里是 矛盾的，必须点明判断以档案为准、不以自述为准。自述等级只用于这一句说明，**绝不**参与 裁决。拒绝后的指引也从"联系客户经理或拨打热线"改成"请先完成风险测评"。 新增三条单测：自述等级识别（含小写）、自述说明行占首行时仍能取到主语、 "为 R2"开头说明前面没有产品名时返回空。
209 lines
7.3 KiB
Python
209 lines
7.3 KiB
Python
"""客服适当性出口的单元测试。
|
|
|
|
这个出口会给出「能不能买」的结论，所以每条分支都必须锁住——写错一条就是误导客户。
|
|
用例里的数字取自真实链路：南方季季盈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_self_claimed_level_is_detected_case_insensitively() -> None:
|
|
"""客户自述等级只用来在回答里说明判断依据，绝不参与裁决。"""
|
|
assert CustomerServiceAgent._self_claimed_level("C3 客户能买它吗") == "C3"
|
|
assert CustomerServiceAgent._self_claimed_level("我是c1客户，能买吗") == "C1"
|
|
assert CustomerServiceAgent._self_claimed_level("那我能买它吗") == ""
|
|
|
|
|
|
# ---- 产品名来源 ----
|
|
|
|
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
|