zhanghongyu_0626
647c07062e
- Introduced `pending_trade` handling in the chat API to manage trade requests more effectively. - Updated the `submit_trade_api` to allow advisors to access customer trades based on assigned roles. - Added new methods in `GatewayRepository` for managing core holdings during trade subscriptions and redemptions. - Implemented context-aware trade dialogue management in the customer service layer to improve user experience during multi-turn interactions. - Enhanced the tool service to support trade actions and suitability checks, ensuring accurate processing of user requests. This update significantly improves the trade interaction flow, providing a more robust and user-friendly experience for customers engaging in trading activities.
421 lines
17 KiB
Python
421 lines
17 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 get_style_tags_lazy(self, cid):
|
||
return self.get_style_tags(cid)
|
||
|
||
def invalidate(self, cid):
|
||
pass
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 意图路由（关键词快路由，零 LLM）
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def test_route_reject_advice_keyword(env):
|
||
reply, disc, intent, transfer, _pending = 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, _pending = 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, _pending = 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, _pending = 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, _pending = 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, _pending = 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, _pending = 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, _pending = cs.run_customer_chat(_ctx(), "我的持仓怎么样", "s1", "CUST-9527")
|
||
assert transfer is False
|
||
assert "3 只产品" in reply
|
||
assert "无法提供投资建议" not 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, _pending = 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, _pending = 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, _pending = cs.run_customer_chat(_ctx(), "怎么修改手机号", "s1", "CUST-9527")
|
||
assert intent == "fallback"
|
||
assert reply == cs.FALLBACK_TEXT
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 闲聊分支（画像语境注入）
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def test_greeting_chitchat_without_llm(env, monkeypatch):
|
||
"""「你好」走关键词 chit_chat；DeepSeek 不可用时仍返回闲聊降级话术。"""
|
||
monkeypatch.setattr(
|
||
cs,
|
||
"_invoke",
|
||
lambda _s, _u: (_ for _ in ()).throw(RuntimeError("no deepseek")),
|
||
)
|
||
reply, disc, intent, transfer, _pending = cs.run_customer_chat(_ctx(), "你好", "s1", "CUST-9527")
|
||
assert intent == "chit_chat"
|
||
assert "暂时无法回答" not in reply
|
||
assert "您好" in reply or "财富助手" in reply
|
||
assert transfer is False
|
||
|
||
|
||
def test_chitchat_with_profile_context(env):
|
||
env["llm"].responses = ["您好呀，很高兴为您服务。"]
|
||
reply, disc, intent, transfer, _pending = cs.run_customer_chat(_ctx(), "你好呀", "s1", "CUST-9527")
|
||
assert intent == "chit_chat"
|
||
assert "您好" in reply
|
||
assert disc is False
|
||
user_prompt = env["llm"].calls[0][1]["content"]
|
||
assert "所在城市：上海" in user_prompt
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# C-05 / C-11 数据查询扩展
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def test_nav_query_keyword(env, monkeypatch):
|
||
monkeypatch.setitem(
|
||
cs._TOOL_BY_INTENT, "nav_query",
|
||
lambda cid, product_keyword=None, repo=None: {
|
||
"tool": "nav_query", "ok": True, "facts": [],
|
||
"fact_text": "单位净值 1.2345（净值日期 2026-09-09）",
|
||
},
|
||
)
|
||
env["llm"].responses = ["nav_query", "单位净值 1.2345（净值日期 2026-09-09）"]
|
||
reply, disc, intent, transfer, _pending = cs.run_customer_chat(
|
||
_ctx(), "005827的净值是多少", "s1", "CUST-9527",
|
||
)
|
||
assert intent == "nav_query"
|
||
assert transfer is False
|
||
assert "1.2345" in reply
|
||
|
||
|
||
def test_eligible_products_routes_suitability_not_reject(env, monkeypatch):
|
||
monkeypatch.setitem(
|
||
cs._TOOL_BY_INTENT, "suitability_check",
|
||
lambda cid, product_keyword=None, risk_level=None, repo=None: {
|
||
"tool": "suitability_check", "ok": True, "facts": [],
|
||
"fact_text": "按您当前风评，在售产品适当性匹配汇总：",
|
||
},
|
||
)
|
||
env["llm"].responses = [
|
||
"suitability_check",
|
||
"按您当前风评，在售产品适当性匹配汇总：",
|
||
]
|
||
reply, disc, intent, transfer, _pending = cs.run_customer_chat(
|
||
_ctx(), "我能买什么产品", "s1", "CUST-9527",
|
||
)
|
||
assert intent == "suitability_check"
|
||
assert transfer is False
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 记忆保存 / 画像抽槽节流 / 归档
|
||
# ---------------------------------------------------------------------------
|
||
|
||
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 = ["您好。"]
|
||
|
||
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 = ["您好。"]
|
||
|
||
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, _pending = 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, None)
|
||
|
||
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
|