diff --git a/app/service/agent_service.py b/app/service/agent_service.py index 0e607f8..48b9f28 100644 --- a/app/service/agent_service.py +++ b/app/service/agent_service.py @@ -1 +1,158 @@ -"""Agent 编排:LangGraph StateGraph + DeepSeek;Tool 节点;四 Agent 能力边界。""" +"""Agent 编排(T-07):LangGraph StateGraph 骨架 + DeepSeek 对话。 + +链路(FLOW §2 最小闭环):compose(角色 system + 历史窗口转 messages) +→ llm(DeepSeek)→ guard(合规护栏)。四 Agent 能力边界(FLOW §4 与 +MEMORY 禁止项)由 system prompt 固化:客户无投资建议/收益承诺/自动下单; +代理人草稿不外发;分析只读;风控辅助不自动处置。 + +LLM 未配置(DEEPSEEK_API_KEY 为空)时降级固定提示回复(演示可跑且明确 +标注非模型生成);单测经 FakeLLM 注入,不依赖外网。 + +阶段 C(开发计划 C1/C2)将在 risk 分支接入 intent→tool→respond 与 +chat_tools(agent_tool_call 落库随 Tool 节点一起);本骨架保证图编排 +可测、消息落库与 trace 贯通由 T-06 api/chat 承担。 +""" + +from __future__ import annotations + +import threading +from typing import Annotated, Any, TypedDict + +from langchain_core.messages import AIMessage, BaseMessage, HumanMessage, SystemMessage +from langgraph.graph import END, START, StateGraph +from operator import add + +from app.config.settings import settings + +# 客户/对外口径的固定免责声明(随回复文本尾部输出;G-08 阻断响应另有两要素) +CHAT_DISCLAIMER = "以上内容由 AI 生成,仅供业务参考,不构成投资建议。" + +# LLM 未配置时的降级回复(明确标注非模型生成,演示链路不断) +_DEGRADED_PREFIX = "(LLM 未配置:请在 .env 设置 DEEPSEEK_API_KEY 后重启)" + +# 四 Agent 能力边界 system prompt(FLOW §4;MEMORY 禁止项固化) +_SYSTEM_PROMPTS: dict[str, str] = { + "customer": ( + "你是客户财富智能管家。仅服务当前登录客户本人;不得提供投资建议、" + "收益承诺或自动下单;事实性数据须注明来源,无法回答时如实告知。" + ), + "advisor": ( + "你是代理人助手,服务理财顾问。仅可讨论名下客户(归属经平台校验);" + "生成的草稿内容不会自动外发给客户;引用数据须可溯源。" + ), + "analyst": ( + "你是数据分析助手。仅做只读查询与聚合分析,不写任何画像/预警/业务数据。" + ), + "risk": ( + "你是风控监测助手。可查询预警台账与客户监测信息,输出仅供参考," + "最终处置须经风控专员通过处置接口人工完成;不得自动处置预警。" + ), +} + + +class ChatState(TypedDict): + """对话图状态(messages 由各节点以 add 归并,避免覆盖历史)。""" + + agent_type: str + history: list[dict] + user_message: str + messages: Annotated[list[BaseMessage], add] + reply: str + has_disclaimer: bool + + +def _compose_messages(state: ChatState) -> list[BaseMessage]: + """system(角色边界)+ 历史窗口(role→message)+ 本轮用户消息。""" + msgs: list[BaseMessage] = [ + SystemMessage(content=_SYSTEM_PROMPTS.get(state["agent_type"], _SYSTEM_PROMPTS["customer"])) + ] + for m in state["history"]: + role = m.get("role") + content = m.get("content", "") + if role == "user": + msgs.append(HumanMessage(content=content)) + elif role == "assistant": + msgs.append(AIMessage(content=content)) + msgs.append(HumanMessage(content=state["user_message"])) + return msgs + + +def llm_node(state: ChatState) -> dict[str, Any]: + """组装 messages 后调 LLM;未配置 key 时降级(不抛异常,演示链路不断)。""" + messages = _compose_messages(state) + if not settings.deepseek_api_key: + reply = f"{_DEGRADED_PREFIX}已收到您的消息:{state['user_message']}" + return {"messages": messages + [AIMessage(content=reply)], "reply": reply} + llm = _get_llm() + result = llm.invoke(messages) + return {"messages": messages + [result], "reply": result.content} + + +def guard_node(state: ChatState) -> dict[str, Any]: + """合规护栏:对外角色(customer/risk,未知类型 fail-safe 按最严口径)附免责声明。""" + external = state["agent_type"] in ("customer", "risk") or state["agent_type"] not in _SYSTEM_PROMPTS + if external: + return {"reply": f"{state['reply']}\n\n{CHAT_DISCLAIMER}", "has_disclaimer": True} + return {"has_disclaimer": False} + + +def build_graph(): + """StateGraph 骨架:START → llm → guard → END(阶段 C 扩展 tool 节点)。""" + graph = StateGraph(ChatState) + graph.add_node("llm", llm_node) + graph.add_node("guard", guard_node) + graph.add_edge(START, "llm") + graph.add_edge("llm", "guard") + graph.add_edge("guard", END) + return graph.compile() + + +_graph = None +_graph_lock = threading.Lock() +_llm: Any | None = None + + +def _get_graph(): + global _graph + with _graph_lock: + if _graph is None: + _graph = build_graph() + return _graph + + +def _get_llm() -> Any: + """DeepSeek 经 langchain-openai 兼容接口(懒构造;测试注入 _llm)。""" + global _llm + if _llm is None: + from langchain_openai import ChatOpenAI + + _llm = ChatOpenAI( + model="deepseek-chat", + api_key=settings.deepseek_api_key, + base_url=settings.deepseek_base_url, + temperature=0.3, + ) + return _llm + + +def chat(agent_type: str, history: list[dict], user_message: str) -> dict[str, Any]: + """对话入口(T-06 api/chat 调用):返回 {reply, has_disclaimer}。""" + final = _get_graph().invoke( + { + "agent_type": agent_type, + "history": history, + "user_message": user_message, + "messages": [], + "reply": "", + "has_disclaimer": False, + } + ) + return {"reply": final["reply"], "has_disclaimer": final["has_disclaimer"]} + + +def reset_cache() -> None: + """测试隔离出口:清空图与 LLM 单例缓存。""" + global _graph, _llm + with _graph_lock: + _graph = None + _llm = None diff --git a/tests/test_agent_service.py b/tests/test_agent_service.py new file mode 100644 index 0000000..da40f1c --- /dev/null +++ b/tests/test_agent_service.py @@ -0,0 +1,92 @@ +"""T-07 agent_service StateGraph 骨架(mock LLM,不依赖外网/真 key)。 + +FakeLLM 捕获收到的 messages 供断言(system 边界 + 历史窗口 + 本轮消息); +降级路径(无 key)与免责声明护栏单独覆盖。图/LLM 缓存经 reset_cache 隔离。 +""" + +from __future__ import annotations + +import pytest +from langchain_core.messages import AIMessage + +from app.config.settings import settings +from app.service import agent_service + + +class FakeLLM: + def __init__(self, reply: str = "模拟回复"): + self.reply = reply + self.calls: list[list] = [] + + def invoke(self, messages): + self.calls.append(list(messages)) + return AIMessage(content=self.reply) + + +@pytest.fixture() +def fake_llm(monkeypatch): + llm = FakeLLM() + monkeypatch.setattr(agent_service, "_llm", llm) + monkeypatch.setattr(settings, "deepseek_api_key", "test-key") + yield llm + agent_service.reset_cache() + + +HISTORY = [ + {"role": "user", "content": "你好"}, + {"role": "assistant", "content": "您好,有什么可以帮您?"}, +] + + +def test_chat_customer_appends_disclaimer(fake_llm): + out = agent_service.chat("customer", HISTORY, "查一下我的持仓") + assert out["has_disclaimer"] is True + assert out["reply"].endswith(agent_service.CHAT_DISCLAIMER) + assert "模拟回复" in out["reply"] + + +def test_chat_risk_also_disclaimer(fake_llm): + out = agent_service.chat("risk", [], "今天有多少待审预警") + assert out["has_disclaimer"] is True + + +def test_chat_advisor_no_disclaimer(fake_llm): + out = agent_service.chat("advisor", [], "总结客户需求") + assert out["has_disclaimer"] is False + assert agent_service.CHAT_DISCLAIMER not in out["reply"] + + +def test_messages_compose_system_history_and_user(fake_llm): + agent_service.chat("customer", HISTORY, "最新问题") + msgs = fake_llm.calls[0] + assert msgs[0].__class__.__name__ == "SystemMessage" + assert "不得提供投资建议" in msgs[0].content # customer 边界注入 + assert [m.__class__.__name__ for m in msgs[1:]] == ["HumanMessage", "AIMessage", "HumanMessage"] + assert msgs[-1].content == "最新问题" + + +def test_role_boundary_per_agent_type(fake_llm): + agent_service.chat("analyst", [], "聚合一下") + system = fake_llm.calls[0][0] + assert "只读" in system.content + + +def test_degraded_reply_without_api_key(monkeypatch): + """key 缺失 → 降级固定提示(不抛异常,演示链路不断)。""" + monkeypatch.setattr(settings, "deepseek_api_key", "") + agent_service.reset_cache() + out = agent_service.chat("customer", [], "在吗") + assert "LLM 未配置" in out["reply"] and "在吗" in out["reply"] + assert out["has_disclaimer"] is True # 降级回复同样经 guard 护栏 + + +def test_graph_skeleton_nodes(): + graph = agent_service.build_graph() + assert {"llm", "guard"} <= set(graph.nodes) # __start__ 为 LangGraph 内部节点 + + +def test_unknown_agent_type_falls_back_customer_boundary(fake_llm): + """未知 agent_type 兜底 customer 边界(fail-safe:最严声明口径)。""" + out = agent_service.chat("unknown", [], "hi") + assert out["has_disclaimer"] is True + assert "不得提供投资建议" in fake_llm.calls[0][0].content