zhanghongyu_0626
b841f68295
- Added a new visitor chat API endpoint (`/api/chat/visitor`) to allow unauthenticated users to engage in conversations without requiring customer data. - Introduced a visitor context dependency to manage visitor interactions seamlessly. - Enhanced the chat API to support explicit session termination and improved response handling for customer service interactions. - Updated the database configuration to include Redis client support for caching visitor data. - Added a new customer note repository to persist user notes independently of the L1 profile slots. This update significantly improves the customer service experience by enabling visitor interactions and ensuring efficient data handling for both registered and unregistered users.
363 lines
14 KiB
Python
363 lines
14 KiB
Python
"""CS Wave 3：客户 Agent LangGraph 编排测试（mock LLM/RAG/工具/Redis）。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import pytest
|
||
|
||
from app.model.schemas import AuthContext
|
||
from app.service import customer_service as cs
|
||
from app.service.profile_service import CustomerMemoryService
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 测试替身
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class FakeMsg:
|
||
def __init__(self, content: str) -> None:
|
||
self.content = content
|
||
|
||
|
||
class FakeLLM:
|
||
def __init__(self, responses: list[str]) -> None:
|
||
self.responses = list(responses)
|
||
self.calls: list[list[dict]] = []
|
||
|
||
def invoke(self, messages):
|
||
self.calls.append(messages)
|
||
return FakeMsg(self.responses.pop(0) if self.responses else "{}")
|
||
|
||
|
||
class FakeRag:
|
||
def __init__(self, context: str, sources: list[dict] | None = None) -> None:
|
||
self._context = context
|
||
self._sources = sources or []
|
||
|
||
def retrieve(self, intent, query, top_k=5):
|
||
return self._context, self._sources
|
||
|
||
|
||
def _ctx() -> AuthContext:
|
||
return AuthContext(
|
||
sub="CUST-9527",
|
||
token_type="customer",
|
||
roles=["customer"],
|
||
trace_id="t1",
|
||
agent_type="customer",
|
||
jti="j1",
|
||
)
|
||
|
||
|
||
@pytest.fixture
|
||
def env(monkeypatch, fake_redis):
|
||
"""统一替身环境：LLM / 记忆 / 热缓存 / 后台线程同步化 / 归档。"""
|
||
llm = FakeLLM([])
|
||
recorded = {"intents": [], "extract": [], "archive": [], "idle": []}
|
||
|
||
monkeypatch.setattr(cs, "_build_llm", lambda: llm)
|
||
monkeypatch.setattr(cs, "CustomerMemoryService", lambda: CustomerMemoryService(redis_client=fake_redis))
|
||
monkeypatch.setattr(cs, "ProfileHotCache", lambda repo=None: _FakeHotCache())
|
||
monkeypatch.setattr(cs, "record_intent", lambda cid, intent: recorded["intents"].append((cid, intent)))
|
||
monkeypatch.setattr(cs, "_spawn", lambda fn, *a: fn(*a)) # 同步执行，便于断言
|
||
monkeypatch.setattr(
|
||
cs, "extract_profile",
|
||
lambda cid, window, tid="": recorded["extract"].append((cid, window, tid)) or [],
|
||
)
|
||
monkeypatch.setattr(
|
||
cs, "archive_session",
|
||
lambda sid, cid, tid, reason: recorded["archive"].append((sid, cid, tid, reason)) or True,
|
||
)
|
||
monkeypatch.setattr(
|
||
cs, "archive_idle_sessions",
|
||
lambda tid, exclude_sid="", limit=None: recorded["idle"].append((tid, exclude_sid)) or [],
|
||
)
|
||
monkeypatch.setattr("app.config.database.get_redis_client", lambda: fake_redis)
|
||
return {"llm": llm, "fake": fake_redis, "recorded": recorded}
|
||
|
||
|
||
class _FakeHotCache:
|
||
def get_style_tags(self, cid):
|
||
return {"basic": {"city": {"value": "上海", "source": "user_declared", "confidence": 0.9}}}
|
||
|
||
def invalidate(self, cid):
|
||
pass
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 意图路由（关键词快路由，零 LLM）
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def test_route_reject_advice_keyword(env):
|
||
reply, disc, intent, transfer = cs.run_customer_chat(_ctx(), "帮我推荐个基金", "s1", "CUST-9527")
|
||
assert intent == "reject"
|
||
assert "投资建议" in reply
|
||
assert transfer is False
|
||
assert env["llm"].calls == [] # 未消耗 LLM
|
||
|
||
|
||
def test_route_transfer_human(env):
|
||
reply, disc, intent, transfer = cs.run_customer_chat(_ctx(), "转人工", "s1", "CUST-9527")
|
||
assert intent == "transfer_human"
|
||
assert transfer is True
|
||
assert "人工客服" in reply
|
||
|
||
|
||
def test_route_fallback_on_invalid_llm_label(env):
|
||
env["llm"].responses = ["what_is_this"]
|
||
reply, disc, intent, transfer = cs.run_customer_chat(_ctx(), "今天天气不错", "s1", "CUST-9527")
|
||
assert intent == "fallback"
|
||
assert reply == cs.FALLBACK_TEXT
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 数据查询分支：param_extract → tool_call → interpret
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def test_holding_query_end_to_end(env, monkeypatch):
|
||
captured: dict = {}
|
||
|
||
def fake_holdings(cid, repo=None):
|
||
captured["cid"] = cid
|
||
return {"tool": "holding_query", "ok": True, "facts": [], "error": None,
|
||
"fact_text": "您当前持有 3 只产品，合计市值 149,940.00 元。"}
|
||
|
||
monkeypatch.setitem(cs._TOOL_BY_INTENT, "holding_query", fake_holdings)
|
||
env["llm"].responses = ["您当前持有 3 只产品，合计市值 149,940.00 元，如需了解单只产品可继续询问。"]
|
||
|
||
reply, disc, intent, transfer = cs.run_customer_chat(_ctx(), "我的持仓怎么样", "s1", "CUST-9527")
|
||
|
||
assert intent == "holding_query"
|
||
assert captured["cid"] == "CUST-9527"
|
||
assert "149,940.00" in reply
|
||
assert disc is False and transfer is False
|
||
|
||
|
||
def test_transaction_query_month_extraction(env, monkeypatch):
|
||
captured: dict = {}
|
||
|
||
def fake_trades(cid, months=None, repo=None):
|
||
captured["months"] = months
|
||
return {"tool": "transaction_query", "ok": True, "facts": [], "error": None,
|
||
"fact_text": "近 6 个月共 2 笔交易。"}
|
||
|
||
monkeypatch.setitem(cs._TOOL_BY_INTENT, "transaction_query", fake_trades)
|
||
env["llm"].responses = [
|
||
'{"months": 6, "product_keyword": null, "risk_level": null}',
|
||
"您近 6 个月共 2 笔交易，均为申购。",
|
||
]
|
||
|
||
reply, disc, intent, transfer = cs.run_customer_chat(_ctx(), "最近半年的流水", "s1", "CUST-9527")
|
||
|
||
assert intent == "transaction_query"
|
||
assert captured["months"] == 6
|
||
assert "2 笔交易" in reply
|
||
|
||
|
||
def test_suitability_risk_level_normalized(env, monkeypatch):
|
||
captured: dict = {}
|
||
|
||
def fake_suit(cid, product_keyword=None, risk_level=None, repo=None):
|
||
captured["risk_level"] = risk_level
|
||
captured["product_keyword"] = product_keyword
|
||
return {"tool": "suitability_check", "ok": True, "facts": [], "error": None,
|
||
"fact_text": "R3 汇总：匹配 2 只，需双录 5 只，不匹配 1 只。"}
|
||
|
||
monkeypatch.setitem(cs._TOOL_BY_INTENT, "suitability_check", fake_suit)
|
||
env["llm"].responses = [
|
||
'{"months": null, "product_keyword": null, "risk_level": "r3"}',
|
||
"R3 产品中匹配 2 只、需双录 5 只、不匹配 1 只。",
|
||
]
|
||
|
||
reply, disc, intent, transfer = cs.run_customer_chat(_ctx(), "我能买R3产品吗", "s1", "CUST-9527")
|
||
|
||
assert intent == "suitability_check"
|
||
assert captured["risk_level"] == "R3"
|
||
assert captured["product_keyword"] is None
|
||
|
||
|
||
def test_tool_error_returns_data_error_text(env, monkeypatch):
|
||
monkeypatch.setitem(
|
||
cs._TOOL_BY_INTENT, "holding_query",
|
||
lambda cid, repo=None: {"tool": "holding_query", "ok": False, "facts": None,
|
||
"fact_text": "", "error": "db_down"},
|
||
)
|
||
reply, disc, intent, transfer = cs.run_customer_chat(_ctx(), "我的持仓怎么样", "s1", "CUST-9527")
|
||
assert intent == "holding_query"
|
||
assert reply == cs.DATA_ERROR_TEXT
|
||
|
||
|
||
def test_interpret_sanitized_on_forbidden_reply(env, monkeypatch):
|
||
monkeypatch.setitem(
|
||
cs._TOOL_BY_INTENT, "holding_query",
|
||
lambda cid, repo=None: {"tool": "holding_query", "ok": True, "facts": [], "error": None,
|
||
"fact_text": "您当前持有 3 只产品。"},
|
||
)
|
||
env["llm"].responses = ["我建议您买入更多高风险产品。"]
|
||
|
||
reply, disc, intent, transfer = cs.run_customer_chat(_ctx(), "我的持仓怎么样", "s1", "CUST-9527")
|
||
assert transfer is True
|
||
assert "无法提供投资建议" in reply
|
||
|
||
|
||
def test_interpret_falls_back_to_fact_text_when_llm_down(env, monkeypatch):
|
||
monkeypatch.setitem(
|
||
cs._TOOL_BY_INTENT, "holding_query",
|
||
lambda cid, repo=None: {"tool": "holding_query", "ok": True, "facts": [], "error": None,
|
||
"fact_text": "您当前持有 3 只产品。"},
|
||
)
|
||
|
||
def _boom(*a, **kw):
|
||
raise RuntimeError("llm down")
|
||
|
||
monkeypatch.setattr(cs, "_build_llm", _boom)
|
||
reply, disc, intent, transfer = cs.run_customer_chat(_ctx(), "我的持仓怎么样", "s1", "CUST-9527")
|
||
assert "3 只产品" in reply # 数值不经 LLM 也能返回
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# RAG 分支
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def test_rag_generate_with_disclaimer(env, monkeypatch):
|
||
monkeypatch.setattr(cs, "VisitorRagService", lambda: FakeRag("债券基金以债券为主要投资标的。"))
|
||
env["llm"].responses = ["product_consult", "债券基金以债券为主要投资标的，风险相对较低。"]
|
||
|
||
reply, disc, intent, transfer = cs.run_customer_chat(_ctx(), "债券基金是什么", "s1", "CUST-9527")
|
||
|
||
assert intent == "product_consult"
|
||
assert disc is True
|
||
assert "风险提示" in reply
|
||
# 画像注入 prompt（红线：仅语气参考）
|
||
assert "所在城市：上海" in env["llm"].calls[1][1]["content"]
|
||
|
||
|
||
def test_rag_empty_falls_back(env, monkeypatch):
|
||
monkeypatch.setattr(cs, "VisitorRagService", lambda: FakeRag(""))
|
||
env["llm"].responses = ["faq"]
|
||
reply, disc, intent, transfer = cs.run_customer_chat(_ctx(), "怎么修改手机号", "s1", "CUST-9527")
|
||
assert intent == "fallback"
|
||
assert reply == cs.FALLBACK_TEXT
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 闲聊分支（画像语境注入）
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def test_chitchat_with_profile_context(env):
|
||
env["llm"].responses = ["chit_chat", "您好呀，很高兴为您服务。"]
|
||
reply, disc, intent, transfer = cs.run_customer_chat(_ctx(), "你好呀", "s1", "CUST-9527")
|
||
assert intent == "chit_chat"
|
||
assert "您好" in reply
|
||
assert disc is False
|
||
user_prompt = env["llm"].calls[1][1]["content"]
|
||
assert "所在城市：上海" in user_prompt
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 记忆保存 / 画像抽槽节流 / 归档
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def test_save_memory_writes_consult_line(env, monkeypatch):
|
||
monkeypatch.setitem(
|
||
cs._TOOL_BY_INTENT, "holding_query",
|
||
lambda cid, repo=None: {"tool": "holding_query", "ok": True, "facts": [], "error": None,
|
||
"fact_text": "您持有 1 只产品。"},
|
||
)
|
||
env["llm"].responses = ["您持有 1 只产品。"]
|
||
reply, _, intent, _ = cs.run_customer_chat(_ctx(), "我的持仓", "s1", "CUST-9527")
|
||
|
||
mem = CustomerMemoryService(redis_client=env["fake"])
|
||
consult = mem.recall("s1", "consult")
|
||
assert [m["content"] for m in consult] == ["我的持仓", "您持有 1 只产品。"]
|
||
assert env["recorded"]["intents"] == [("CUST-9527", "holding_query")]
|
||
|
||
|
||
def test_profile_extract_triggered_every_5_rounds(env):
|
||
env["fake"].strings["customer:s1:rounds"] = "4" # 已累计 4 轮
|
||
env["llm"].responses = ["chit_chat", "您好。"]
|
||
|
||
cs.run_customer_chat(_ctx(), "你好", "s1", "CUST-9527")
|
||
|
||
# 轮数达到 5 → 触发抽槽（env 中已同步化并记录调用），计数清零
|
||
assert env["fake"].strings.get("customer:s1:rounds") is None
|
||
assert len(env["recorded"]["extract"]) == 1
|
||
cid, window, tid = env["recorded"]["extract"][0]
|
||
assert cid == "CUST-9527"
|
||
assert tid == "t1"
|
||
assert "你好" in window # 抽槽窗口含本轮对话（双线合并）
|
||
|
||
|
||
def test_profile_extract_not_triggered_before_threshold(env):
|
||
env["fake"].strings["customer:s1:rounds"] = "1"
|
||
env["llm"].responses = ["chit_chat", "您好。"]
|
||
|
||
cs.run_customer_chat(_ctx(), "你好", "s1", "CUST-9527")
|
||
assert env["fake"].strings.get("customer:s1:rounds") == "2" # 仅计数
|
||
assert env["recorded"]["extract"] == []
|
||
|
||
|
||
def test_archive_on_end_session(env):
|
||
reply, disc, intent, transfer = cs.run_customer_chat(
|
||
_ctx(), "转人工", "s1", "CUST-9527", end_session=True
|
||
)
|
||
assert intent == "transfer_human"
|
||
assert env["recorded"]["archive"] == [("s1", "CUST-9527", "t1", "explicit")]
|
||
# 懒扫描始终执行，排除当前会话
|
||
assert env["recorded"]["idle"] == [("t1", "s1")]
|
||
|
||
|
||
def test_archive_idle_scan_every_round(env):
|
||
cs.run_customer_chat(_ctx(), "转人工", "s1", "CUST-9527")
|
||
assert env["recorded"]["idle"] == [("t1", "s1")]
|
||
assert env["recorded"]["archive"] == [] # 未显式结束不归档当前会话
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# /api/chat 分流（X-Agent-Type: customer）
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def test_chat_endpoint_customer_branch(client, monkeypatch):
|
||
captured: dict = {}
|
||
|
||
def fake_run(ctx, message, session_id, customer_id, end_session=False):
|
||
captured["customer_id"] = customer_id
|
||
captured["end_session"] = end_session
|
||
return ("您好，我是您的客服助手。", False, "chit_chat", False)
|
||
|
||
monkeypatch.setattr("app.api.chat.run_customer_chat", fake_run)
|
||
|
||
login = client.post("/api/auth/login", json={"actor_id": "CUST-9527", "token_type": "customer"})
|
||
token = login.json()["data"]["access_token"]
|
||
resp = client.post(
|
||
"/api/chat",
|
||
json={"message": "你好", "end_session": True},
|
||
headers={"Authorization": f"Bearer {token}", "X-Agent-Type": "customer"},
|
||
)
|
||
|
||
assert resp.status_code == 200
|
||
body = resp.json()
|
||
assert body["agent_type"] == "customer"
|
||
assert body["intent"] == "chit_chat"
|
||
assert body["transfer_to_human"] is False
|
||
assert captured["customer_id"] == "CUST-9527"
|
||
assert captured["end_session"] is True
|
||
|
||
|
||
def test_chat_endpoint_non_customer_unchanged(client, monkeypatch):
|
||
"""非 customer 分流仍走 agent_service；仅断言响应不含客服专属字段。"""
|
||
monkeypatch.setattr(
|
||
"app.api.chat.agent_service.chat",
|
||
lambda *a, **kw: {"reply": "ok", "has_disclaimer": False},
|
||
)
|
||
login = client.post("/api/auth/login", json={"actor_id": "STAFF-20001", "token_type": "staff"})
|
||
token = login.json()["data"]["access_token"]
|
||
resp = client.post(
|
||
"/api/chat",
|
||
json={"message": "hello"},
|
||
headers={"Authorization": f"Bearer {token}", "X-Agent-Type": "analyst"},
|
||
)
|
||
assert resp.status_code == 200
|
||
body = resp.json()
|
||
assert body.get("intent") is None
|
||
assert body.get("transfer_to_human") is False
|