diff --git a/app/api/chat.py b/app/api/chat.py index 7bc5cf0..955fabd 100644 --- a/app/api/chat.py +++ b/app/api/chat.py @@ -3,11 +3,12 @@ 链路：X-Agent-Type 分流 + Agent 准入（deps.assert_agent_access，手册 §5.4） → customer 归属固定本人 / 代理人等指定客户走 G-01 归属校验（A-01 语义） → SessionGuard（会话存在、actor/agent_type 一致 AUTH_403_SESSION_AGENT、 - active 状态，手册 §9）→ memory_service 窗口 → agent_service（T-07 图） -→ user/assistant 双消息落 MySQL + Redis 窗口 → 响应 {session_id, reply, -has_disclaimer, trace_id}。 -落库：agent_session/agent_message（同 trace_id）；agent_tool_call 随阶段 C -Tool 节点接入（开发计划 C1）。审计：鉴权失败/越权经 deps.deny 双写留痕。 + active 状态，手册 §9）→ memory_service 窗口 → agent_service（T-07 图 + + T-04 Tool 节点：意图→归属校验→Core RO 只读查询）→ user/assistant 双消息 +落 MySQL + Redis 窗口 → 响应 {session_id, reply, has_disclaimer, trace_id}。 +落库：agent_session/agent_message（同 trace_id）；agent_tool_call 由 Tool +节点落（T-04，success/blocked/error 全留痕）。审计：鉴权失败/越权经 +deps.deny 双写留痕。 """ from __future__ import annotations @@ -131,7 +132,15 @@ def chat_api(req: ChatRequest, request: Request, auth: AuthContext = Depends(get ) history = memory_service.get_recent(agent_type, sid) - result = agent_service.chat(agent_type, history, message) + result = agent_service.chat( + agent_type, + history, + message, + session_id=sid, + trace_id=current_trace(), + actor={"actor_id": auth.actor_id, "roles": auth.roles, "token_type": auth.token_type}, + customer_id=customer_id, + ) # 落盘：user + assistant 同步写（异步化归后续）；同 trace_id 贯通 trace_id = current_trace() diff --git a/app/repository/session_repository.py b/app/repository/session_repository.py index bd2dde5..6537443 100644 --- a/app/repository/session_repository.py +++ b/app/repository/session_repository.py @@ -3,6 +3,8 @@ 会话权威副本在 MySQL（FLOW §6 / redis-keys §2.1：Redis 只保滑动窗口，丢失 可重建）；seq_no 由调用方经 next_seq_no 取号（同会话并发写防重锁归 sess:{agent}:{id}:lock，最小闭环单演示进程暂未启用，留痕）。 +T-04 增 agent_tool_call 落库（对话域第三张表；Tool 调用审计，与 +audit_log/input_guard_log 合规审计表区分——本表由对话链路读写）。 """ from __future__ import annotations @@ -109,6 +111,50 @@ class SessionRepository: }, ) + # ---------- agent_tool_call ---------- + + def insert_tool_call( + self, + *, + session_id: str, + trace_id: str, + tool_name: str, + tool_input: str, + tool_output: str | None, + status: str, + error_code: str | None = None, + latency_ms: int | None = None, + ) -> None: + """Tool 调用留痕（表设计 §01 共用底座；只 INSERT，不更新）。 + + message_id 一期 NULL：Tool 节点先于 LLM 执行，user 消息尚未落库， + 经 session_id + trace_id 已可审计还原；消息级关联归后续增强。 + tool_input/tool_output 由调用方序列化为 JSON 字符串（列类型 JSON）。 + """ + sql = text( + """ + INSERT INTO agent_tool_call + (session_id, trace_id, message_id, tool_name, tool_input, + tool_output, status, error_code, latency_ms) + VALUES (:sid, :trace_id, NULL, :tool_name, :tool_input, + :tool_output, :status, :error_code, :latency_ms) + """ + ) + with self._engine.begin() as conn: + conn.execute( + sql, + { + "sid": session_id, + "trace_id": trace_id, + "tool_name": tool_name, + "tool_input": tool_input, + "tool_output": tool_output, + "status": status, + "error_code": error_code, + "latency_ms": latency_ms, + }, + ) + def list_messages(self, session_id: str, limit: int = 20) -> list[dict[str, Any]]: """最近 limit 条（seq_no 升序返回，供 LLM 窗口直用）。 diff --git a/app/service/agent_service.py b/app/service/agent_service.py index 903bcd7..10c241a 100644 --- a/app/service/agent_service.py +++ b/app/service/agent_service.py @@ -1,16 +1,16 @@ -"""Agent 编排（T-07）：LangGraph StateGraph 骨架 + DeepSeek 对话。 +"""Agent 编排（T-07 骨架 / T-04 Tool 节点）：LangGraph StateGraph + DeepSeek 对话。 -链路（FLOW §2 最小闭环）：compose（角色 system + 历史窗口转 messages） -→ llm（DeepSeek）→ guard（合规护栏）。四 Agent 能力边界（FLOW §4 与 -MEMORY 禁止项）由 system prompt 固化：客户无投资建议/收益承诺/自动下单； -代理人草稿不外发；分析只读；风控辅助不自动处置。 +链路（FLOW §2）：tool（意图匹配→归属校验→只读查询→agent_tool_call 落库） +→ llm（DeepSeek，工具结果注入上下文）→ guard（合规护栏）。四 Agent 能力 +边界（FLOW §4 与 MEMORY 禁止项）由 system prompt 固化：客户无投资建议/ +收益承诺/自动下单；代理人草稿不外发；分析只读；风控辅助不自动处置。 -LLM 未配置（DEEPSEEK_API_KEY 为空）时降级固定提示回复（演示可跑且明确 -标注非模型生成）；单测经 FakeLLM 注入，不依赖外网。 +Tool 节点（T-04）：一期关键词意图（tool_service.match_intent）仅对 +customer/advisor 分支查 Core RO（持仓/流水/L0）；customer_id 由会话注入， +不来自 LLM。risk 分支四个风控 Tool 与 LLM intent 归 C1/C2（开发计划）。 -阶段 C（开发计划 C1/C2）将在 risk 分支接入 intent→tool→respond 与 -chat_tools（agent_tool_call 落库随 Tool 节点一起）；本骨架保证图编排 -可测、消息落库与 trace 贯通由 T-06 api/chat 承担。 +LLM 未配置（DEEPSEEK_API_KEY 为空）时降级：回复携带 Tool 查询摘要 +（演示链路不断且查询不白跑）；单测经 FakeLLM 注入，不依赖外网。 """ from __future__ import annotations @@ -23,6 +23,7 @@ from langgraph.graph import END, START, StateGraph from operator import add from app.config.settings import settings +from app.service import tool_service # 客户/对外口径的固定免责声明（随回复文本尾部输出；G-08 阻断响应另有两要素） CHAT_DISCLAIMER = "以上内容由 AI 生成，仅供业务参考，不构成投资建议。" @@ -51,7 +52,12 @@ _SYSTEM_PROMPTS: dict[str, str] = { class ChatState(TypedDict): - """对话图状态（messages 由各节点以 add 归并，避免覆盖历史）。""" + """对话图状态（messages 由各节点以 add 归并，避免覆盖历史）。 + + T-04 Tool 上下文：session_id/trace_id 供 agent_tool_call 落库；actor 为 + api/chat 展开的鉴权字段（{actor_id, roles, token_type}）；customer_id 为 + 会话绑定客户（归属校验后的值，Tool 查询主体恒取此处）。 + """ agent_type: str history: list[dict] @@ -59,13 +65,21 @@ class ChatState(TypedDict): messages: Annotated[list[BaseMessage], add] reply: str has_disclaimer: bool + session_id: str | None + trace_id: str | None + actor: dict[str, Any] | None + customer_id: str | None + tool_results: list[dict] def _compose_messages(state: ChatState) -> list[BaseMessage]: - """system（角色边界）+ 历史窗口（role→message）+ 本轮用户消息。""" + """system（角色边界）+ 工具结果（有则注入）+ 历史窗口 + 本轮用户消息。""" msgs: list[BaseMessage] = [ SystemMessage(content=_SYSTEM_PROMPTS.get(state["agent_type"], _SYSTEM_PROMPTS["customer"])) ] + tool_text = tool_service.context_text(state.get("tool_results") or []) + if tool_text: + msgs.append(SystemMessage(content=tool_text)) for m in state["history"]: role = m.get("role") content = m.get("content", "") @@ -77,11 +91,42 @@ def _compose_messages(state: ChatState) -> list[BaseMessage]: return msgs +def tool_node(state: ChatState) -> dict[str, Any]: + """Tool 节点（T-04）：意图匹配 → tool_service.run_tool（校验+落库）。 + + 守卫：无会话上下文（session_id/actor/customer_id 任一缺失，如单测直调 + chat 或会话未绑定客户）不触发 Tool；risk/analyst 分支无意图规则， + 自然空转（C1/C2 接风控 Tool 时扩展 match_intent）。 + """ + if not (state.get("session_id") and state.get("actor") and state.get("customer_id")): + return {"tool_results": []} + tool_name = tool_service.match_intent(state["agent_type"], state["user_message"]) + if tool_name is None: + return {"tool_results": []} + record = tool_service.run_tool( + tool_name=tool_name, + agent_type=state["agent_type"], + actor=state["actor"], + customer_id=state["customer_id"], + session_id=state["session_id"], + trace_id=state.get("trace_id"), + ) + return {"tool_results": [record]} + + def llm_node(state: ChatState) -> dict[str, Any]: - """组装 messages 后调 LLM；未配置 key 时降级（不抛异常，演示链路不断）。""" + """组装 messages 后调 LLM；未配置 key 时降级（不抛异常，演示链路不断）。 + + 降级回复携带 Tool 查询摘要（T-04：无 LLM 时查询不白跑，结果直出）。 + """ messages = _compose_messages(state) if not settings.deepseek_api_key: - reply = f"{_DEGRADED_PREFIX}已收到您的消息：{state['user_message']}" + summaries = [tool_service.summarize(r) for r in (state.get("tool_results") or [])] + reply = ( + f"{_DEGRADED_PREFIX}\n" + "\n".join(summaries) + if summaries + else f"{_DEGRADED_PREFIX}已收到您的消息：{state['user_message']}" + ) return {"messages": messages + [AIMessage(content=reply)], "reply": reply} llm = _get_llm() result = llm.invoke(messages) @@ -97,11 +142,13 @@ def guard_node(state: ChatState) -> dict[str, Any]: def build_graph(): - """StateGraph 骨架：START → llm → guard → END（阶段 C 扩展 tool 节点）。""" + """StateGraph：START → tool → llm → guard → END（T-04 接入 Tool 节点）。""" graph = StateGraph(ChatState) + graph.add_node("tool", tool_node) graph.add_node("llm", llm_node) graph.add_node("guard", guard_node) - graph.add_edge(START, "llm") + graph.add_edge(START, "tool") + graph.add_edge("tool", "llm") graph.add_edge("llm", "guard") graph.add_edge("guard", END) return graph.compile() @@ -137,8 +184,21 @@ def _get_llm() -> Any: return _llm -def chat(agent_type: str, history: list[dict], user_message: str) -> dict[str, Any]: - """对话入口（T-06 api/chat 调用）：返回 {reply, has_disclaimer}。""" +def chat( + agent_type: str, + history: list[dict], + user_message: str, + *, + session_id: str | None = None, + trace_id: str | None = None, + actor: dict[str, Any] | None = None, + customer_id: str | None = None, +) -> dict[str, Any]: + """对话入口（T-06 api/chat 调用）：返回 {reply, has_disclaimer}。 + + T-04：session 上下文（session_id/trace_id/actor/customer_id）可选传入； + 缺省时 Tool 节点空转（既有用例与纯闲聊不受影响）。 + """ final = _get_graph().invoke( { "agent_type": agent_type, @@ -147,9 +207,18 @@ def chat(agent_type: str, history: list[dict], user_message: str) -> dict[str, A "messages": [], "reply": "", "has_disclaimer": False, + "session_id": session_id, + "trace_id": trace_id, + "actor": actor, + "customer_id": customer_id, + "tool_results": [], } ) - return {"reply": final["reply"], "has_disclaimer": final["has_disclaimer"]} + return { + "reply": final["reply"], + "has_disclaimer": final["has_disclaimer"], + "tool_results": final.get("tool_results") or [], + } def reset_cache() -> None: diff --git a/app/service/tool_service.py b/app/service/tool_service.py new file mode 100644 index 0000000..dc56ffd --- /dev/null +++ b/app/service/tool_service.py @@ -0,0 +1,227 @@ +"""对话 Tool 编排（T-04）：意图匹配 → 归属校验 → 执行 → agent_tool_call 落库。 + +分层：Tool 定义在 app/tool/core_tools.py（纯查询）；本模块做业务编排 +（FRAMEWORK「tool 不写业务流程」），由 agent_service 的 tool 节点调用。 + +归属校验（A-01 语义 · G-01 同口径）：customer 仅本人、advisor 需 +customer_advisor_rel active、risk_officer 全量、其余角色 fail-closed 拒绝； +顺序对齐 api/deps.assert_customer_access（risk_officer 优先，customer 分支 +先于 advisor）。与 API 层差异：Tool 归属拒绝**不抛 403**，落 +agent_tool_call.status='blocked'（error_code 与 deps 同码）后由对话呈现—— +agent_tool_call 即留痕位（手册 P-05「403 必经 deny」约束 API 层出口； +对话内拒绝以 blocked 行留痕，注释登记为 T-04 口径）。 + +customer_id 恒由会话注入（api/chat 归属校验后的值），不来自 LLM/用户输入。 +落库失败降级 warning 不阻塞对话（口径同 T-02 审计降级；本地日志留底）。 +""" + +from __future__ import annotations + +import json +import logging +import time +from typing import Any + +from app.repository.core_ro import CoreReadOnlyRepository +from app.repository.session_repository import SessionRepository +from app.tool import core_tools +from app.utils.exceptions import PermissionDenied +from app.utils.trace import current_trace + +logger = logging.getLogger(__name__) + +STATUS_SUCCESS = "success" +STATUS_BLOCKED = "blocked" +STATUS_ERROR = "error" + +# LLM 上下文中单条 Tool 结果（summary + data JSON）的截断上限 +_RESULT_CONTEXT_MAX_CHARS = 2000 + + +def _session_repo() -> SessionRepository: + """仓储入口（测试 monkeypatch 点，与 api 模式一致）。""" + return SessionRepository() + + +def _core_ro() -> CoreReadOnlyRepository: + """Core 只读仓储入口（Tool 函数共享同一实例，测试注入点）。""" + return CoreReadOnlyRepository() + + +# ---------- 意图匹配（一期关键词规则；C2 risk 分支换 LLM intent） ---------- + +# (tool_name, 关键词)；顺序即优先级，命中即停。仅对可绑定客户的 +# customer/advisor 分支生效；risk/analyst 的 Tool 归 C1（chat_tools）/T-10。 +_INTENT_KEYWORDS: list[tuple[str, tuple[str, ...]]] = [ + ("query_holdings", ("持仓", "持有", "仓位")), + ("query_customer_profile", ("风险测评", "风险等级", "测评")), + ("query_recent_trades", ("流水", "交易记录", "交易明细", "买卖记录")), +] +INTENT_AGENT_TYPES = ("customer", "advisor") + + +def match_intent(agent_type: str, message: str) -> str | None: + """消息 → tool_name；不命中返回 None（纯闲聊不查库）。""" + if agent_type not in INTENT_AGENT_TYPES: + return None + for name, keywords in _INTENT_KEYWORDS: + if any(k in message for k in keywords): + return name + return None + + +# ---------- 归属校验（A-01；口径对齐 api/deps.assert_customer_access） ---------- + + +def assert_tool_access(actor: dict[str, Any], customer_id: str, core_ro: CoreReadOnlyRepository) -> None: + """Tool 层归属断言（纵深防御：api 层已校验，此处防 state 污染/未来 LLM 传参）。 + + actor = {actor_id, roles, token_type}（api/chat 从 AuthContext 展开）。 + 失败 raise PermissionDenied(code)；由 run_tool 捕获转 blocked 留痕。 + """ + roles = actor.get("roles") or [] + actor_id = actor.get("actor_id") or "" + if "risk_officer" in roles: + return + if "customer" in roles: + if actor_id == customer_id: + return + raise PermissionDenied("AUTH_403_NOT_OWNER", "tool blocked: not owner") + if "advisor" in roles: + if core_ro.is_advisor_assigned(actor_id, customer_id): + return + raise PermissionDenied("AUTH_403_NOT_ASSIGNED", "tool blocked: customer not assigned") + raise PermissionDenied("AUTH_403_SCOPE", "tool blocked: role not allowed") + + +# ---------- 执行与落库 ---------- + + +def run_tool( + *, + tool_name: str, + agent_type: str, + actor: dict[str, Any], + customer_id: str, + tool_input: dict[str, Any] | None = None, + session_id: str, + trace_id: str | None = None, +) -> dict[str, Any]: + """执行单个对话 Tool 并落 agent_tool_call（success/blocked/error 全留痕）。 + + 返回 {tool_name, status, error_code, data, latency_ms}；本函数不抛异常 + （对话链路不因 Tool 故障中断），归属拒绝转 blocked 而非 403（见模块注释）。 + """ + trace_id = trace_id or current_trace() or "" + params = dict(tool_input or {}) + started = time.monotonic() + status, error_code, data = STATUS_ERROR, "TOOL_ERROR", None + try: + spec = core_tools.get_tool(tool_name) + if spec is None: + status, error_code = STATUS_BLOCKED, "TOOL_UNKNOWN" + logger.warning("tool not in registry: %s", tool_name) + elif spec["requires_customer"] and not customer_id: + status, error_code = STATUS_BLOCKED, "TOOL_BLOCKED_NO_CUSTOMER" + else: + core_ro = _core_ro() + assert_tool_access(actor, customer_id, core_ro) + data = spec["func"](customer_id=customer_id, core_ro=core_ro, **params) + status, error_code = STATUS_SUCCESS, None + except PermissionDenied as exc: + status, error_code = STATUS_BLOCKED, exc.code + except Exception: + logger.exception("tool execution failed: %s", tool_name) + status, error_code = STATUS_ERROR, "TOOL_ERROR" + + latency_ms = int((time.monotonic() - started) * 1000) + record = { + "tool_name": tool_name, + "status": status, + "error_code": error_code, + "data": data, + "latency_ms": latency_ms, + } + _audit_tool_call( + session_id=session_id, + trace_id=trace_id, + tool_name=tool_name, + tool_input={"customer_id": customer_id, **params}, + record=record, + ) + return record + + +def _audit_tool_call( + *, session_id: str, trace_id: str, tool_name: str, tool_input: dict, record: dict +) -> None: + """agent_tool_call 留痕（降级 warning，不阻塞对话主链路）。""" + try: + _session_repo().insert_tool_call( + session_id=session_id, + trace_id=trace_id, + tool_name=tool_name, + tool_input=json.dumps(tool_input, ensure_ascii=False, default=str), + tool_output=( + json.dumps(record["data"], ensure_ascii=False, default=str) + if record["data"] is not None + else None + ), + status=record["status"], + error_code=record["error_code"], + latency_ms=record["latency_ms"], + ) + except Exception: + logger.warning("agent_tool_call insert failed (degrade): %s", tool_name, exc_info=True) + + +# ---------- 结果摘要（LLM 上下文与降级回复共用） ---------- + + +def summarize(record: dict[str, Any]) -> str: + """Tool 结果 → 人读摘要；blocked/error 给出明确口径，供回复直接引用。""" + name = record.get("tool_name", "?") + status = record.get("status") + if status == STATUS_BLOCKED: + return f"（工具 {name} 拒绝：当前身份无权访问该客户数据，错误码 {record.get('error_code')}）" + if status == STATUS_ERROR: + return f"（工具 {name} 执行失败，查询未完成，请稍后重试）" + data = record.get("data") or {} + if name == "query_customer_profile": + if not data.get("found"): + return f"（客户档案查询：未找到客户 {data.get('customer_id', '')} 的档案）" + return ( + f"（客户档案：{data.get('display_name', '-')}（{data.get('customer_id')}），" + f"年龄 {data.get('age', '-')}，职业 {data.get('occupation', '-') or '-'}，" + f"风险测评等级 {data.get('risk_code', '-') or '-'}" + f"（测评时间 {data.get('risk_evaluated_at', '-') or '无记录'}））" + ) + if name == "query_holdings": + items = data.get("items") or [] + lines = [ + f"（持仓查询：共 {data.get('total_count', 0)} 笔，合计市值 {data.get('sum_market_value', 0)} 元）" + ] + lines += [ + f"- {it.get('product_name', it.get('product_id', '-'))}：市值 {it.get('market_value', 0)} 元" + for it in items[:5] + ] + return "\n".join(lines) + if name == "query_recent_trades": + return ( + f"（近 {data.get('days', '-')} 天交易流水：共 {data.get('total_count', 0)} 笔，" + f"申赎合计 {data.get('sum_amount', 0)} 元）" + ) + return f"（工具 {name} 完成）" + + +def context_text(records: list[dict[str, Any]]) -> str: + """全部 Tool 结果 → 注入 LLM 的上下文块（摘要 + 数据 JSON 截断）。""" + parts: list[str] = [] + for record in records: + parts.append(summarize(record)) + if record.get("data") is not None: + payload = json.dumps(record["data"], ensure_ascii=False, default=str) + if len(payload) > _RESULT_CONTEXT_MAX_CHARS: + payload = payload[:_RESULT_CONTEXT_MAX_CHARS] + "…(truncated)" + parts.append(payload) + return "[工具查询结果]\n" + "\n".join(parts) if parts else "" diff --git a/app/tool/core_tools.py b/app/tool/core_tools.py new file mode 100644 index 0000000..f937cce --- /dev/null +++ b/app/tool/core_tools.py @@ -0,0 +1,114 @@ +"""Core 模拟库只读 Tool（T-04 · FLOW §2「CoreReadOnlyRepository：持仓/流水/L0」）。 + +Tool 定义层（纯查询，无业务流程）：customer_id 由调用方（tool_service） +注入，**不接受 LLM 生成**——归属防线之一（A-01 语义，T-03 之前的止损）。 +注册表 TOOL_REGISTRY 是对话 Tool 的唯一白名单（tool_service 校验）。 + +requires_customer：Tool 是否必须绑定会话客户（True → 无归属主体即 +blocked，不触发查询）。风控分支 Tool（C1 chat_tools）注册时对台账类 +统计用 requires_customer=False，复用同一 runner。 + +返回值约定：JSON 安全 dict（Decimal→float 两位、datetime/date→isoformat）， +落库 tool_output 与 LLM 上下文共用同一结构，不做第二套序列化。 +""" + +from __future__ import annotations + +import datetime as _dt +from decimal import Decimal +from typing import Any, Callable + +from app.repository.core_ro import CoreReadOnlyRepository + +DETAIL_LIMIT = 20 # 明细条数上限（上下文与落库同限，防超长） +DEFAULT_TRADE_DAYS = 30 + + +def _jsonable(value: Any) -> Any: + """MySQL 行 → JSON 安全结构（递归；Decimal 两位小数、时间 isoformat）。""" + if isinstance(value, Decimal): + return float(round(value, 2)) + if isinstance(value, (_dt.datetime, _dt.date)): + return value.isoformat() + if isinstance(value, dict): + return {k: _jsonable(v) for k, v in value.items()} + if isinstance(value, (list, tuple)): + return [_jsonable(v) for v in value] + return value + + +def query_customer_profile( + customer_id: str, core_ro: CoreReadOnlyRepository | None = None +) -> dict[str, Any]: + """客户档案与风险测评（L0）：core_customer + core_customer_risk。""" + repo = core_ro or CoreReadOnlyRepository() + row = repo.get_customer_l0(customer_id) + if row is None: + return {"found": False, "customer_id": customer_id} + return {"found": True, **_jsonable(row)} + + +def query_holdings( + customer_id: str, core_ro: CoreReadOnlyRepository | None = None +) -> dict[str, Any]: + """持仓明细（按市值降序）+ 合计（sum_market_value）。""" + repo = core_ro or CoreReadOnlyRepository() + rows = repo.list_holdings(customer_id) + total = sum((r.get("market_value") or 0) for r in rows) + return { + "total_count": len(rows), + "sum_market_value": _jsonable(total), + "items": [_jsonable(r) for r in rows[:DETAIL_LIMIT]], + } + + +def query_recent_trades( + customer_id: str, + days: int = DEFAULT_TRADE_DAYS, + core_ro: CoreReadOnlyRepository | None = None, +) -> dict[str, Any]: + """近 N 天 confirmed 申赎流水（时间升序截断至 DETAIL_LIMIT）。""" + repo = core_ro or CoreReadOnlyRepository() + end = _dt.datetime.now() + start = end - _dt.timedelta(days=days) + rows = repo.list_trades_range(customer_id, start, end) + total = sum((r.get("amount") or 0) for r in rows) + return { + "days": days, + "total_count": len(rows), + "sum_amount": _jsonable(total), + "items": [_jsonable(r) for r in rows[:DETAIL_LIMIT]], + } + + +class ToolSpec(dict): + """注册表条目：func + 描述 + 是否必须绑定会话客户。""" + + +TOOL_REGISTRY: dict[str, ToolSpec] = { + "query_customer_profile": ToolSpec( + func=query_customer_profile, + description="查询客户档案与风险测评等级（L0）", + requires_customer=True, + ), + "query_holdings": ToolSpec( + func=query_holdings, + description="查询客户持仓明细与合计市值", + requires_customer=True, + ), + "query_recent_trades": ToolSpec( + func=query_recent_trades, + description="查询客户近期申赎交易流水", + requires_customer=True, + ), +} + + +def get_tool(name: str) -> ToolSpec | None: + """白名单查找（未知 Tool 一律 None，由 runner 拒绝）。""" + return TOOL_REGISTRY.get(name) + + +def tool_func(name: str) -> Callable[..., dict[str, Any]] | None: + spec = TOOL_REGISTRY.get(name) + return spec["func"] if spec else None diff --git a/docs/memory/FLOW.md b/docs/memory/FLOW.md index 5aa5001..b7dbe9a 100644 --- a/docs/memory/FLOW.md +++ b/docs/memory/FLOW.md @@ -48,9 +48,9 @@ RBAC 联调账号：scripts/dev/rbac-seed-reference.md Client → Gateway(JWT/RBAC) → api/chat → agent_service(LangGraph) → Tools → 存储 → 响应 + audit_log ``` -当前：**风控事件线 + Wave 0 共用底座完成**（交易网关 → 规则引擎 → 预警/AML/L3 → 4 API，`main.py` 路由挂载 + trace 中间件（X-Trace-Id/X-Request-Id + 500 兜底）+ audit 中间件（http_access）+ lifespan 完成，B1~B8；T-01 JWT 鉴权（Bearer 全环境 + dev debug 头兜底 + X-Agent-Type 准入矩阵）/ T-02 审计与 input_guard_log 双写 / T-06 chat 最小闭环（`POST /api/chat` 会话+窗口+落盘）/ T-07 LangGraph 骨架+DeepSeek（无 key 降级）均于 2026-09-07 完成并经独立 AI 评审闭环；M2 tag risk-m2 + M4 复核闭环；273 测试绿）；阶段 B 完结；**对话链路骨架已通**（chat → LangGraph → 免责声明 → 落库），Tool 节点与 RAG 待 T-04/T-21、输入防护待 T-03。 +当前：**风控事件线 + Wave 0 共用底座 + T-04 Tool 节点完成**（交易网关 → 规则引擎 → 预警/AML/L3 → 4 API，`main.py` 路由挂载 + trace 中间件（X-Trace-Id/X-Request-Id + 500 兜底）+ audit 中间件（http_access）+ lifespan 完成，B1~B8；T-01 JWT 鉴权（Bearer 全环境 + dev debug 头兜底 + X-Agent-Type 准入矩阵）/ T-02 审计与 input_guard_log 双写 / T-06 chat 最小闭环（`POST /api/chat` 会话+窗口+落盘）/ T-07 LangGraph 骨架+DeepSeek（无 key 降级）均于 2026-09-07 完成并经独立 AI 评审闭环；M2 tag risk-m2 + M4 复核闭环；T-04 Core RO Tool 节点（customer/advisor 意图触发 + 归属校验 + agent_tool_call 落库，2026-09-07）；297 测试绿）；阶段 B 完结；**对话链路已含只读查数**（chat → LangGraph(tool→llm→guard) → 免责声明 → 落库），风控四 Tool 与 RAG 待 C1/T-21、输入防护待 T-03。 -**本机已就位状态（2026-09-07 更新 · 已完成上述 ①~⑤，无需重做）：** `.env` 已配置（学习项目，`MYSQL_PASSWORD=123456`；NEO4J/DEEPSEEK 留空——DEEPSEEK_API_KEY 空时 chat 走降级回复，不阻塞演示；`.env` 无 JWT 项，签发走默认 `JWT_DEV_SECRET`，**启动时出现默认密钥 warning 属预期**，生产形态须配 `JWT_PUBLIC_KEY_PATH`）；`jinrong_core` + `jinrong_agent` 已灌库（28 客户 / AML 名单 8 条 / 演示测评已刷新 / 归属同步 28 行）；`python -m pytest` 273 用例全绿（集成测试真连本机 MySQL，未灌库的机器自动 skip 集成模块）；**演示库已重灌并清理冒烟数据（2026-09-07），全量测试可直接跑，下次演示/走查后再按 SOP §2 重灌**；本机 Redis 服务在跑（`redis://127.0.0.1:6379/0`，B9a 订阅脚本已验证）。 +**本机已就位状态（2026-09-07 更新 · 已完成上述 ①~⑤，无需重做）：** `.env` 已配置（学习项目，`MYSQL_PASSWORD=123456`；NEO4J/DEEPSEEK 留空——DEEPSEEK_API_KEY 空时 chat 走降级回复，不阻塞演示；`.env` 无 JWT 项，签发走默认 `JWT_DEV_SECRET`，**启动时出现默认密钥 warning 属预期**，生产形态须配 `JWT_PUBLIC_KEY_PATH`）；`jinrong_core` + `jinrong_agent` 已灌库（28 客户 / AML 名单 8 条 / 演示测评已刷新 / 归属同步 28 行）；`python -m pytest` 297 用例全绿（集成测试真连本机 MySQL，未灌库的机器自动 skip 集成模块）；**演示库已重灌并清理冒烟数据（2026-09-07），全量测试可直接跑，下次演示/走查后再按 SOP §2 重灌**；本机 Redis 服务在跑（`redis://127.0.0.1:6379/0`，B9a 订阅脚本已验证）。 **本机已知坑：** `mysql.exe` 不在 PATH（位于 `C:\Program Files\MySQL\MySQL Server 8.0\bin`）；`reset.ps1` 的交互式 `-p` 在自动化执行时会卡死——脚本化重灌用 `MYSQL_PWD` 环境变量传密码（交互执行不受影响，不把密码写进仓库脚本）；**`redis` 与 `python-jose[cryptography]` 包均出现过 requirements 有而环境漏装（B9a 补装 redis、Wave 0 补装 jose，2026-09-07），重装环境后先 `python -c "import jose, redis"` 自检**。 @@ -67,17 +67,17 @@ api/chat：SessionGuard；创建/续 agent_session 【T-06 已做 ↓ memory_service：Redis 读最近 N 轮；MySQL 落盘 agent_message 【T-06 已做】（窗口 TTL2h N≤20，miss 回源；同步落盘） ↓ -agent_service：LangGraph StateGraph + DeepSeek；按 Agent 类型选 Tool 节点 【T-07 骨架已做（llm→guard+免责声明）；Tool 节点待阶段 C/T-04】 +agent_service：LangGraph StateGraph + DeepSeek；tool(意图→Tool) → llm → guard 【T-04 已做（tool 节点：关键词意图仅 customer/advisor；risk/analyst 空转）；LLM intent 归 C2】 ↓ -Tool 示例： - - CoreReadOnlyRepository：持仓/流水/L0（自动注入 customer_id） 【Repository 已有，Tool 未接 T-04】 +Tool（tool_service.run_tool：白名单 → 归属校验 → 执行 → agent_tool_call 落库）： + - CoreReadOnlyRepository：持仓/流水/L0（customer_id 由会话注入，不来自 LLM） 【T-04 已做 app/tool/core_tools.py 三 Tool】 + - 风控四 Tool：alert_query/customer_context/suitability_check/aml_lookup 【待做 C1 chat_tools，复用 run_tool】 - milvus_tool：产品规则 RAG + source_refs 【待做 T-21】 - memory_service：读/写 L1/L2/L3（ProfileGuard） 【窗口读写已做；画像读写待做】 - - 风控 service 账号：R-02 适当性（无会话） 【待做 阶段 C】 ↓ 输出：assistant 消息 + has_disclaimer（customer/risk 固定免责，T-07 guard 节点） ↓ -落库：agent_message、agent_tool_call（待 C1 Tool 节点）、audit_log（http_access 已接，同 trace_id） +落库：agent_message、agent_tool_call（T-04 已接：success/blocked/error 全留痕，message_id 一期 NULL）、audit_log（http_access 已接，同 trace_id） ``` ------ diff --git a/docs/memory/FRAMEWORK.md b/docs/memory/FRAMEWORK.md index a36f209..3537c0b 100644 --- a/docs/memory/FRAMEWORK.md +++ b/docs/memory/FRAMEWORK.md @@ -39,9 +39,9 @@ | 代理人助手 Agent | L2 画像、RAG、草稿 | L1 只读、Milvus | 空壳 service（chat 骨架已通） | | 数据分析 Agent | NL→SQL→解读 | Core RO、画像只读 | 空壳 service（chat 骨架已通） | | 风控监测 Agent | 预警、L3、R-02 适当性 | 交易事件、AML 名单 | **已实现**（B1~B7：rules/alert_service/profile_l3/aml_service/engine/scoring 占位 + 4 API + suitability + main 集成与挂账①~⑦） | -| Core 只读层 | L0 事实查询 | `jinrong_core` | **core_ro.py 已实现**（含风控扩展查询） | -| 共用底座 | 会话、审计、输入防护 | MySQL 11 表 + Redis | **大部分接入（2026-09-07）**：会话（T-06 session_repository + memory_service 窗口）、审计中间件（T-02 http_access + input_guard_log 双写）已接；输入防护规则待 T-03 | -| 对话编排 | LangGraph StateGraph + DeepSeek | langgraph/langchain-openai | **骨架已实现（T-07，2026-09-07）**：llm→guard（免责声明）；Tool 节点待 T-04/阶段 C | +| Core 只读层 | L0 事实查询 | `jinrong_core` | **已实现 + 已接对话 Tool（T-04）**：core_ro 经 app/tool/core_tools.py 三只读 Tool（L0/持仓/流水）进 chat；风控扩展查询照旧 | +| 共用底座 | 会话、审计、输入防护 | MySQL 11 表 + Redis | **大部分接入（2026-09-07）**：会话（T-06 session_repository + memory_service 窗口）、审计中间件（T-02 http_access + input_guard_log 双写）、agent_tool_call Tool 留痕（T-04）已接；输入防护规则待 T-03 | +| 对话编排 | LangGraph StateGraph + DeepSeek | langgraph/langchain-openai | **已实现（T-07 骨架 + T-04 Tool 节点，2026-09-07）**：tool（关键词意图→Tool）→llm→guard（免责声明）；LLM intent 与风控四 Tool 归 C1/C2 | | 同步脚本 | 归属、Neo4j | Core → agent / 图库 | **sync_*.py 已实现** | ------ @@ -51,11 +51,13 @@ ```text api/ → risk / simulate / chat / deps / audit_middleware 已实现；knowledge、admin 空壳 【大部分】 service/ → risk/*（rules/alert/aml/engine/l3/scoring/locks/redis_gateway）+ suitability + - auth_service（T-01 JWT）+ agent_service（T-07 StateGraph）+ memory_service（T-06）已实现； + auth_service（T-01 JWT）+ agent_service（T-07 图 + T-04 tool 节点）+ memory_service（T-06）+ + tool_service（T-04 对话 Tool 编排：意图/归属校验/run_tool 落库）已实现； rag_service 空壳 【大部分】 -tool/ → document_parser、embedding_tool、milvus_tool 【空壳】 +tool/ → core_tools（T-04：L0/持仓/流水只读 Tool + TOOL_REGISTRY 白名单）已实现； + document_parser、embedding_tool、milvus_tool 【部分；RAG 三件空壳】 repository/ → core_ro（Core 只读）+ risk_repository（风控四表+审计双写）+ - session_repository（agent_session/agent_message，T-06） 【已实现】 + session_repository（agent_session/agent_message/agent_tool_call，T-06/T-04） 【已实现】 gateway/ → 模拟交易网关（trade_gateway + gateway_repository 仅 INSERT core_trade） 【已实现 B5】 model/ → schemas（Pydantic）、entities（ORM） 【占位】（风控暂用 dict，收敛待后续） config/ → settings（双库 + risk_* 阈值 + JWT 配置）、database 【settings 已实现】 diff --git a/docs/memory/MEMORY.md b/docs/memory/MEMORY.md index e8448e0..e9e81ce 100644 --- a/docs/memory/MEMORY.md +++ b/docs/memory/MEMORY.md @@ -9,7 +9,7 @@ **项目是什么：** 金融四 Agent（客户财富 / 代理人 / 数据分析 / 风控）共用数据层与合规底座；**不**互调 LLM，跨 Agent 走 L1/L2/L3 画像与预警表。 -**当前进度：** 需求与表设计已定 · **风控模块 B1~B9b 全部完成（M2 tag risk-m2），M4 复核已闭环（2026-09-07），风控阶段 B 正式完结** · **Wave 0 已完成（2026-09-07，经独立 AI 评审闭环）**：T-01 JWT 鉴权（auth_service Auth SDK + deps 工厂替换 + X-Agent-Type 准入矩阵）/ T-02 审计中间件（http_access + 独立 request_id + 4xx/500 统一错误体 + input_guard_log 双写）/ T-06 chat 最小闭环（POST /api/chat + 会话落库 + Redis 窗口）/ T-07 LangGraph StateGraph 骨架 + DeepSeek（无 key 降级），**273 测试绿**。**下一步：T-03 输入防护 / T-04 Core RO 封装 Tool / T-21 Milvus / 风控阶段 C（C0~C3，A-6 对话验收）**。**开发在分支 `feature/risk`（未合入 main）。** +**当前进度：** 需求与表设计已定 · **风控模块 B1~B9b 全部完成（M2 tag risk-m2），M4 复核已闭环（2026-09-07），风控阶段 B 正式完结** · **Wave 0 已完成（2026-09-07，经独立 AI 评审闭环）**：T-01 JWT 鉴权（auth_service Auth SDK + deps 工厂替换 + X-Agent-Type 准入矩阵）/ T-02 审计中间件（http_access + 独立 request_id + 4xx/500 统一错误体 + input_guard_log 双写）/ T-06 chat 最小闭环（POST /api/chat + 会话落库 + Redis 窗口）/ T-07 LangGraph StateGraph 骨架 + DeepSeek（无 key 降级）· **T-04 Core RO Tool 节点已完成（2026-09-07）**：app/tool/core_tools.py 三只读 Tool + tool_service（意图/归属校验/run_tool）+ 图 tool 节点 + agent_tool_call 落库，**297 测试绿**。**下一步：T-21 Milvus / T-03 输入防护 / 风控阶段 C（C0~C3，A-6 对话验收，C1 复用 run_tool）**。**开发在分支 `feature/risk`（未合入 main）。** **仓库地图：** @@ -21,8 +21,9 @@ | `app/api/knowledge.py` `admin.py` | 空壳 | 待 T-21（Milvus/RAG）与审计查询台 | | `app/service/auth_service.py` | **已实现（T-01）** | Auth SDK：JWT 验签（HS256 dev/RS256 生产）、必填 claims、jti 吊销（Redis fail-open）；签发 CLI `scripts/dev/issue_dev_token.py` | | `app/service/risk/*` + `service/suitability.py` | **已实现** | RISK-001~005 规则 / 预警聚合 / L3 写入 / AML / 引擎编排 / SUIT-001~008 / locks+redis_gateway 公共原语（B7） | -| `app/service/agent_service.py` `memory_service.py` | **已实现（T-07/T-06）** | LangGraph StateGraph（llm→guard）+ DeepSeek（无 key 降级）；Redis 会话窗口 + MySQL 回源 | -| `app/repository/session_repository.py` | **已实现（T-06）** | agent_session / agent_message 读写 | +| `app/service/agent_service.py` `memory_service.py` | **已实现（T-07/T-06/T-04）** | LangGraph StateGraph（tool→llm→guard）+ DeepSeek（无 key 降级）；T-04 tool 节点（关键词意图 customer/advisor + tool_service.run_tool）；Redis 会话窗口 + MySQL 回源 | +| `app/service/tool_service.py` `app/tool/core_tools.py` | **已实现（T-04）** | 对话 Tool 编排（白名单/归属校验/agent_tool_call 落库）+ Core RO 三只读 Tool（L0/持仓/流水）；风控四 Tool 归 C1 复用 run_tool | +| `app/repository/session_repository.py` | **已实现（T-06/T-04）** | agent_session / agent_message / agent_tool_call 读写 | | `app/gateway/` | **已实现** | 模拟交易网关（仅 gateway_repository 可 INSERT core_trade，B5） | | `app/repository/core_ro.py` | **已实现** | Core 只读 SELECT（含风控扩展 sum_trades_on_date / list_trades_range / list_active_customers） | | `app/repository/risk_repository.py` | **已实现** | risk_alert / risk_suitability_log / L3 / risk_aml_list / audit_log / input_guard_log 读写 | @@ -31,7 +32,7 @@ | `scripts/core/*.sql` + `reset.ps1` | **已实现** | Core 模拟库 DDL + 种子 | | `scripts/agent/` `scripts/demo/` `scripts/dev/` | **已实现** | AML 名单种子 + 风控演示数据 + subscribe_alerts/rebuild_alerts + issue_dev_token（JWT 签发） | | `scripts/sync/*.py` | **已实现** | 归属同步 + Neo4j 全图 | -| `tests/` | **已实现** | 21 个测试文件 273 用例（sqlite 隔离；DDL 单一事实源 `_ddl.py`；`test_integration_risk.py` 走真 MySQL + TRD-TEST- 前缀隔离） | +| `tests/` | **已实现** | 22 个测试文件 297 用例（sqlite 隔离；DDL 单一事实源 `_ddl.py`；`test_integration_risk.py` 走真 MySQL + TRD-TEST- 前缀隔离） | | `docs/需求拆解/` | 已定 | 场景 P0、矩阵、合规原文 | | `docs/PRD/PRD-风控监测Agent.md` | **已冻结** | 风控 PRD v1.0 + 规则表附录 | | `docs/项目框架设计/表设计/` | 已定 | Agent 共用 11 表 + agent 专用 SQL | @@ -49,13 +50,13 @@ 5. mysql … < scripts/agent/seed-aml-list.sql # AML 名单种子 （风控演示：scripts/demo/prepare_risk_demo.sql，reset 后重跑） 6. python scripts/sync/sync_advisor_rel.py && python scripts/sync/sync_neo4j.py -7. uvicorn app.main:app --reload → GET /health；python -m pytest（209 绿） +7. uvicorn app.main:app --reload → GET /health；python -m pytest（297 绿） ``` > **本机 ①~⑤已执行、`.env` 已配置，勿重做**；本机状态与已知坑（mysql.exe 路径 / reset.ps1 交互式 -p / `redis`、`python-jose` 包漏装已补）见 `FLOW.md` §0 尾注。 > **演示库状态（2026-09-07）**：已按演示 SOP §2 重灌并清理冒烟数据（alerts 0 / AML 8 / 会话 0），pytest 全量可直接跑；**下次演示/走查后再按 SOP §2 重灌**（核查单⑥前置断言会对走查残留显式 fail，属防护行为）。 -**下一步开发（见 TODO）：** T-03 输入防护 / T-04 Core RO 封装 Tool（agent_tool_call 随 C1）/ T-21 Milvus / 风控阶段 C（C0~C3，A-6 对话验收）。演示走查按 `docs/项目框架设计/演示SOP-风控模块.md`（debug 头通道仍有效；JWT 通道签发用 `scripts/dev/issue_dev_token.py`）。 +**下一步开发（见 TODO）：** T-21 Milvus / T-03 输入防护 / 风控阶段 C（C0~C3，A-6 对话验收；C1 风控四 Tool 复用 T-04 run_tool）。演示走查按 `docs/项目框架设计/演示SOP-风控模块.md`（debug 头通道仍有效；JWT 通道签发用 `scripts/dev/issue_dev_token.py`）。 **禁止（改代码前必记）：** Core 正式 C1~C5 不可被画像覆盖 · 审计表只 INSERT · 代理人草稿不外发 · 仅 R-02 可阻断交易 · 四 Agent 不互调 LLM。 @@ -64,8 +65,8 @@ ## 1. 项目简介 - **名称：** JinRong 金融四 Agent 智能管家 -- **当前阶段：** **Wave 0 共用底座完成（2026-09-07：JWT / 审计中间件 / chat 闭环 / LangGraph 骨架，273 绿）**；风控模块 B1~B9b 完成（M2 tag risk-m2）+ M4 复核闭环，阶段 B 完结；下一批 = T-03/T-04/T-21 + 风控阶段 C（A-6 对话线验收） -- **当前优先级：** Wave 0（T-01 JWT / T-02 审计中间件 / T-06 / T-07 LangGraph）→ 风控阶段 C（A-6）→ Wave 1 +- **当前阶段：** **T-04 Core RO Tool 节点完成（2026-09-07：tool 节点+归属校验+agent_tool_call 落库，297 绿）**；Wave 0 共用底座完成（T-01 JWT / T-02 审计中间件 / T-06 / T-07 LangGraph）；风控模块 B1~B9b 完成（M2 tag risk-m2）+ M4 复核闭环，阶段 B 完结；下一批 = T-21/T-03 + 风控阶段 C（A-6 对话线验收） +- **当前优先级：** T-21 / T-03 → 风控阶段 C（A-6）→ Wave 1 ------ @@ -128,7 +129,7 @@ Core 模拟：scripts/core/reset.ps1 · 文档 docs/项目框架设计/Core模 种子：scripts/agent/seed-aml-list.sql（AML 名单）· scripts/demo/prepare_risk_demo.sql（reset 后重跑） 依赖：requirements.txt（LangGraph + langchain-core/openai + FastAPI + SQLAlchemy） 启动：uvicorn app.main:app --reload → GET /health -测试：python -m pytest（273 用例；集成测试需本机演示数据，未灌库时自动 skip） +测试：python -m pytest（297 用例；集成测试需本机演示数据，未灌库时自动 skip） 运维/演示脚本：scripts/demo/subscribe_alerts.py（订阅推送演示）· rebuild_alerts.py TRD-xxx（引擎异常补偿重放） JWT 联调：python scripts/dev/issue_dev_token.py --sub STAFF-30001 --roles risk_officer（+ Authorization: Bearer + X-Agent-Type） 配置：.env（见 .env.example） @@ -172,6 +173,6 @@ RBAC 联调账号：scripts/dev/rbac-seed-reference.md 2. 改动属于 api / service / tool / repository 哪一层？ 3. 是否需 customer_id 归属与 JWT RBAC？ 4. Core 是模拟库只读还是 agent 库读写？ -5. 如何验证？（`python -m pytest` 全量（当前 273 绿）· uvicorn 启动 + /health · SQL / sync 脚本 · 对照 REQUIREMENTS 验收列） +5. 如何验证？（`python -m pytest` 全量（当前 297 绿）· uvicorn 启动 + /health · SQL / sync 脚本 · 对照 REQUIREMENTS 验收列） 大任务：FRAMEWORK/FLOW 与实现状态不符时先更新 memory 再编码（用户确认跳过除外）。 diff --git a/docs/memory/REQUIREMENTS.md b/docs/memory/REQUIREMENTS.md index 3329e87..40f1d94 100644 --- a/docs/memory/REQUIREMENTS.md +++ b/docs/memory/REQUIREMENTS.md @@ -20,13 +20,13 @@ | F-01 | JWT + RBAC + 数据归属 | 越权 403 + audit；JWT 手册 §13 | **已完成（T-01，2026-09-07）**：auth_service 验签/吊销 + deps 工厂（Bearer 全环境 + dev debug 头兜底）+ X-Agent-Type 准入矩阵 + 归属断言（B6 已有）；越权 403 + audit + input_guard_log 双写 | ~~T-01~~ | | F-02 | 全量审计留痕 | trace_id 可还原 | **已完成（T-02，2026-09-07）**：http_access 审计中间件 + X-Trace-Id/X-Request-Id 双贯通 + 4xx/500 统一错误体；业务判定审计 B6 已接 | ~~T-02~~ | | F-03 | 输入防护 | input_guard_log | **部分（2026-09-07）**：input_guard_log 双写通道已由 T-02 接通（鉴权拒绝类）；防护规则（prompt_injection/oversize/限流）待做 | T-03 | -| F-04 | Core 只读层 | 不改 Core 账；Repository 只 SELECT | **部分**（core_ro 已扩展风控查询；接入 api/service Tool 与归属校验待 T-04） | T-04 | +| F-04 | Core 只读层 | 不改 Core 账；Repository 只 SELECT | **已完成（T-04，2026-09-07）**：core_ro（含风控扩展）经 app/tool/core_tools.py 三只读 Tool 接入 chat（L0/持仓/流水）；Tool 层归属校验（A-01 语义：customer 本人/advisor assigned/risk_officer 全量）+ agent_tool_call 全留痕；Core 侧仍仅 SELECT | ~~T-04~~ | | R0-DB | MySQL 共用 11 表 + Redis | 01-mysql-共用底座.sql | SQL 已定；本机已灌库（B9b 走查验证） | T-05（本机已完成） | | R0-CORE | Core 模拟 + 同步 | reset.ps1 + sync 脚本 | **脚本已落地** | T-05 | **F-01 部分完成说明：** B6 已落地 `app/api/deps.py`（AuthContext Pydantic + `get_auth_context()` 工厂，dev 从 `X-Debug-Role`/`X-Debug-Actor` 头构造；T-01 就绪后仅替换工厂内部为 JWT 解析）。 -**F-04 部分完成说明：** `app/repository/core_ro.py`（含风控扩展 sum_trades_on_date / list_trades_range / list_active_customers）+ `scripts/core/*` + `settings.mysql_core_database` 已有；尚未接入对话 Tool 与归属校验。 +**F-04 完成说明（T-04，2026-09-07）：** `app/repository/core_ro.py`（含风控扩展）仅 SELECT 不变；对话接入经 `app/tool/core_tools.py`（query_customer_profile / query_holdings / query_recent_trades，TOOL_REGISTRY 白名单）+ `app/service/tool_service.py`（run_tool：归属校验→执行→agent_tool_call 落库）；agent_service 图为 START→tool→llm→guard，customer_id 由会话注入不接受 LLM 生成。风控四 Tool 归 C1 复用同一 runner。 ## Wave 1 · 内部 Agent P0 diff --git a/docs/memory/TODO.md b/docs/memory/TODO.md index 1ae77fd..2e4c060 100644 --- a/docs/memory/TODO.md +++ b/docs/memory/TODO.md @@ -9,10 +9,10 @@ ## 待办（推荐顺序） -- [ ] T-04 Core RO 封装为 Tool 节点；A-01 归属校验（chat Tool 接入 + agent_tool_call 落库） +- [x] T-04 Core RO 封装为 Tool 节点；A-01 归属校验（chat Tool 接入 + agent_tool_call 落库）—— **完成（2026-09-07）**：`app/tool/core_tools.py`（L0/持仓/流水三只读 Tool + TOOL_REGISTRY 白名单）+ `app/service/tool_service.py`（关键词意图仅 customer/advisor；归属校验对齐 deps.assert_customer_access 口径——Tool 层拒绝不抛 403 而 blocked 留痕，纵深防御；run_tool 执行+落库，落库失败降级）+ agent_service 图 tool 节点（START→tool→llm→guard；结果注入 LLM 上下文，降级回复带摘要）+ session_repository.insert_tool_call（message_id 一期 NULL）。**A-01 口径：本任务只做 Tool 层归属校验语义（customer 本人/advisor assigned/risk_officer 全量/其余拒），A-01 完整场景（代理人画像）仍归 T-20**；risk 分支四 Tool 归 C1 复用 run_tool - [ ] T-21 Milvus Lite + kb_product_rules 首批入库 - [ ] T-03 输入防护规则（F-03：prompt_injection/oversize/限流；input_guard_log 双写通道已由 T-02 接通） -- [ ] 风控阶段 C（C0 pytest-asyncio / C1 chat_tools / C2 risk 分支 StateGraph / C3 A-6 验收——依赖 T-07 已就绪） +- [ ] 风控阶段 C（C0 pytest-asyncio / C1 chat_tools / C2 risk 分支 StateGraph / C3 A-6 验收——依赖 T-07 已就绪；C1 复用 T-04 run_tool 基建） - [ ] 前端 React 多 Agent 入口（HashRouter，`web/` init） ### 风控模块（PRD v1.0 已冻结 · `docs/PRD/PRD-风控监测Agent.md`，事件驱动线不依赖 T-07 可先行） diff --git a/tests/_ddl.py b/tests/_ddl.py index c99a0ee..c9fe11a 100644 --- a/tests/_ddl.py +++ b/tests/_ddl.py @@ -32,6 +32,11 @@ SQLITE_TABLES: dict[str, str] = { product_id VARCHAR(64) PRIMARY KEY, product_name VARCHAR(128), min_risk_code VARCHAR(8), product_type VARCHAR(32)) """, + "core_holding": """ + CREATE TABLE core_holding ( + customer_id VARCHAR(64), product_id VARCHAR(64), + market_value DECIMAL, quantity DECIMAL) + """, "core_trade": """ CREATE TABLE core_trade ( trade_id VARCHAR(64) PRIMARY KEY, customer_id VARCHAR(64), product_id VARCHAR(64), @@ -69,6 +74,14 @@ SQLITE_TABLES: dict[str, str] = { content TEXT, has_disclaimer TINYINT DEFAULT 0, token_est INTEGER, created_at {_TS}) """, + "agent_tool_call": f""" + CREATE TABLE agent_tool_call ( + id INTEGER PRIMARY KEY AUTOINCREMENT, session_id VARCHAR(64), + trace_id VARCHAR(64), message_id INTEGER, tool_name VARCHAR(128), + tool_input TEXT, tool_output TEXT, status VARCHAR(16), + error_code VARCHAR(64), latency_ms INTEGER, + created_at {_TS}) + """, "risk_alert": f""" CREATE TABLE risk_alert ( alert_id VARCHAR(64) PRIMARY KEY, trace_id VARCHAR(64), customer_id VARCHAR(64), diff --git a/tests/test_chat.py b/tests/test_chat.py index 6f693ae..b7e3d29 100644 --- a/tests/test_chat.py +++ b/tests/test_chat.py @@ -24,6 +24,7 @@ from app.repository.core_ro import CoreReadOnlyRepository from app.repository.risk_repository import RiskRepository from app.repository.session_repository import SessionRepository from app.service import memory_service +from app.service import tool_service from app.service.auth_service import issue_dev_token as _issue # noqa: F401 from app.service import auth_service from app.service.risk import redis_gateway @@ -92,6 +93,9 @@ def env(monkeypatch): monkeypatch.setattr(chat_mod, "_session_repo", lambda: session_repo) monkeypatch.setattr(chat_mod, "_core_ro", lambda: core_ro) monkeypatch.setattr(memory_service, "_session_repo", lambda: session_repo) + # T-04：Tool 节点经 tool_service 落库/查库，同一 sqlite 注入 + monkeypatch.setattr(tool_service, "_session_repo", lambda: session_repo) + monkeypatch.setattr(tool_service, "_core_ro", lambda: core_ro) monkeypatch.setattr(risk_api, "_repo", lambda: repo) monkeypatch.setattr(audit_mod, "_repo", lambda: repo) monkeypatch.setattr(deps_mod, "RiskRepository", lambda: repo) @@ -224,6 +228,64 @@ def test_chat_session_not_found_and_closed(env): assert r2.status_code == 409 and r2.json()["error_code"] == "STATE_CONFLICT" +# ---------- T-04 Tool 全链路（经真路由栈触发 Tool 节点） ---------- + + +def test_chat_tool_holdings_full_link(env): + """持仓关键词触发 Tool：agent_tool_call 落库（trace/session 贯通）+ 降级回复带摘要。""" + r = env["client"].post("/api/chat", json={"message": "查一下我的持仓"}, headers=CUSTOMER) + assert r.status_code == 200 + body = r.json() + rows = _rows(env["engine"], "SELECT * FROM agent_tool_call") + assert len(rows) == 1 + row = rows[0] + assert (row["session_id"], row["trace_id"], row["tool_name"], row["status"]) == ( + body["session_id"], body["trace_id"], "query_holdings", "success" + ) + assert "LLM 未配置" in body["reply"] # env 无 key 走降级 + assert "持仓查询" in body["reply"] # 降级回复携带 Tool 摘要（env 无持仓种子 → 0 笔） + + +def test_chat_tool_no_intent_no_call(env): + r = env["client"].post("/api/chat", json={"message": "今天天气如何"}, headers=CUSTOMER) + assert r.status_code == 200 + assert _rows(env["engine"], "SELECT 1 FROM agent_tool_call") == [] + + +def test_chat_tool_advisor_profile_query(env): + """advisor（已分配）测评关键词 → L0 查询 success（A-01 归属链路）。""" + with env["engine"].begin() as conn: + conn.execute( + text( + "INSERT INTO core_customer (customer_id, display_name, age, is_active)" + " VALUES ('CUST-9527', '张三', 45, 1)" + ) + ) + conn.execute( + text( + "INSERT INTO core_customer_risk (customer_id, risk_code, evaluated_at)" + " VALUES ('CUST-9527', 'B', '2026-08-01 10:00:00')" + ) + ) + r = env["client"].post( + "/api/chat", json={"message": "客户的风险测评等级", "customer_id": "CUST-9527"}, headers=ADVISOR + ) + assert r.status_code == 200 + rows = _rows(env["engine"], "SELECT tool_name, status FROM agent_tool_call") + assert [(x["tool_name"], x["status"]) for x in rows] == [("query_customer_profile", "success")] + + +def test_chat_tool_risk_agent_no_intent_rule(env): + """risk 分支无意图规则（C1/C2 接风控 Tool），不触发 Core RO 查询。""" + r = env["client"].post( + "/api/chat", + json={"message": "查一下我的持仓", "customer_id": "CUST-9527"}, + headers={"X-Debug-Role": "risk_officer", "X-Debug-Actor": "STAFF-30001", "X-Agent-Type": "risk"}, + ) + assert r.status_code == 200 + assert _rows(env["engine"], "SELECT 1 FROM agent_tool_call") == [] + + # ---------- JWT 通道（生产主链路 · 评审 P2-3） ---------- diff --git a/tests/test_chat_tools.py b/tests/test_chat_tools.py new file mode 100644 index 0000000..9cabe19 --- /dev/null +++ b/tests/test_chat_tools.py @@ -0,0 +1,407 @@ +"""T-04 对话 Tool：意图匹配 / 归属校验 / agent_tool_call 落库 / 图节点注入。 + +单测层：run_tool 直调（sqlite 注入，success/blocked/error 三态与落库字段、 +落库降级）；图集成：FakeLLM 捕获注入的 [工具查询结果] 上下文、降级回复 +带摘要、无会话上下文空转。全链路（TestClient 真路由栈）见 test_chat.py。 +""" + +from __future__ import annotations + +import datetime as _dt + +import pytest +from langchain_core.messages import AIMessage +from sqlalchemy import text + +from _ddl import create_sqlite_engine + +from app.config.settings import settings +from app.repository.core_ro import CoreReadOnlyRepository +from app.repository.session_repository import SessionRepository +from app.service import agent_service, tool_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) + + +ACTOR_CUSTOMER = {"actor_id": "CUST-9527", "roles": ["customer"], "token_type": "customer"} +ACTOR_ADVISOR = {"actor_id": "STAFF-10086", "roles": ["advisor"], "token_type": "staff"} +ACTOR_RISK = {"actor_id": "STAFF-30001", "roles": ["risk_officer"], "token_type": "staff"} +ACTOR_ANALYST = {"actor_id": "STAFF-40001", "roles": ["analyst"], "token_type": "staff"} + + +@pytest.fixture() +def tool_env(monkeypatch): + engine = create_sqlite_engine() + session_repo = SessionRepository(engine=engine) + core_ro = CoreReadOnlyRepository(engine=engine) + with engine.begin() as conn: + conn.execute( + text( + "INSERT INTO core_customer (customer_id, display_name, age, occupation, is_active)" + " VALUES ('CUST-9527', '张三', 45, '工程师', 1)" + ) + ) + conn.execute( + text( + "INSERT INTO core_customer_risk (customer_id, risk_code, evaluated_at)" + " VALUES ('CUST-9527', 'A', '2026-08-01 10:00:00')" + ) + ) + conn.execute( + text( + "INSERT INTO core_product (product_id, product_name, min_risk_code, product_type)" + " VALUES ('P-001', '稳健一号', 'A', 'fund')" + ) + ) + conn.execute( + text( + "INSERT INTO core_holding (customer_id, product_id, market_value, quantity)" + " VALUES ('CUST-9527', 'P-001', 50000.00, 100)" + ) + ) + conn.execute( + text( + "INSERT INTO core_trade (trade_id, customer_id, product_id, trade_type, amount," + " trade_status, traded_at) VALUES ('TRD-001', 'CUST-9527', 'P-001', 'subscribe'," + " 10000.00, 'confirmed', :ts)" + ), + {"ts": _dt.datetime.now()}, + ) + conn.execute( + text( + "INSERT INTO core_customer_advisor (advisor_id, customer_id, rel_status)" + " VALUES ('STAFF-10086', 'CUST-9527', 'active')" + ) + ) + monkeypatch.setattr(tool_service, "_session_repo", lambda: session_repo) + monkeypatch.setattr(tool_service, "_core_ro", lambda: core_ro) + yield {"engine": engine, "session_repo": session_repo} + engine.dispose() + + +def _tool_rows(engine): + with engine.connect() as conn: + return [ + dict(r) + for r in conn.execute( + text("SELECT * FROM agent_tool_call ORDER BY id") + ).mappings().all() + ] + + +# ---------- 意图匹配 ---------- + + +def test_match_intent_hits(): + assert tool_service.match_intent("customer", "查一下我的持仓") == "query_holdings" + assert tool_service.match_intent("customer", "我的风险测评结果是什么") == "query_customer_profile" + assert tool_service.match_intent("advisor", "看看客户交易记录") == "query_recent_trades" + + +def test_match_intent_miss_or_agent(): + assert tool_service.match_intent("customer", "你好呀") is None + assert tool_service.match_intent("risk", "查一下我的持仓") is None # risk 分支归 C1 + assert tool_service.match_intent("analyst", "交易流水") is None + + +# ---------- run_tool：success / blocked / error 与落库 ---------- + + +def test_run_tool_holdings_success_and_audit(tool_env): + record = tool_service.run_tool( + tool_name="query_holdings", + agent_type="customer", + actor=ACTOR_CUSTOMER, + customer_id="CUST-9527", + session_id="sess-t1", + trace_id="trace-t1", + ) + assert record["status"] == "success" and record["error_code"] is None + assert record["data"]["total_count"] == 1 + assert record["data"]["sum_market_value"] == 50000.0 + assert record["data"]["items"][0]["product_name"] == "稳健一号" # JOIN 生效 + + rows = _tool_rows(tool_env["engine"]) + assert len(rows) == 1 + row = rows[0] + assert (row["session_id"], row["trace_id"], row["tool_name"], row["status"]) == ( + "sess-t1", + "trace-t1", + "query_holdings", + "success", + ) + assert row["error_code"] is None and row["latency_ms"] is not None + assert '"customer_id": "CUST-9527"' in row["tool_input"] + assert '"total_count": 1' in row["tool_output"] + + +def test_run_tool_customer_profile(tool_env): + record = tool_service.run_tool( + tool_name="query_customer_profile", + agent_type="customer", + actor=ACTOR_CUSTOMER, + customer_id="CUST-9527", + session_id="sess-t2", + ) + assert record["status"] == "success" + assert record["data"]["found"] is True + assert record["data"]["risk_code"] == "A" + + +def test_run_tool_recent_trades_params(tool_env): + record = tool_service.run_tool( + tool_name="query_recent_trades", + agent_type="customer", + actor=ACTOR_CUSTOMER, + customer_id="CUST-9527", + tool_input={"days": 7}, + session_id="sess-t3", + ) + assert record["status"] == "success" + assert record["data"]["days"] == 7 and record["data"]["total_count"] == 1 + + +def test_run_tool_blocked_not_owner(tool_env): + """customer 查他人 → blocked（AUTH_403_NOT_OWNER，与 deps 同码）+ 留痕。""" + record = tool_service.run_tool( + tool_name="query_holdings", + agent_type="customer", + actor=ACTOR_CUSTOMER, + customer_id="CUST-1001", + session_id="sess-b1", + ) + assert record["status"] == "blocked" and record["error_code"] == "AUTH_403_NOT_OWNER" + assert record["data"] is None + row = _tool_rows(tool_env["engine"])[0] + assert (row["status"], row["error_code"]) == ("blocked", "AUTH_403_NOT_OWNER") + + +def test_run_tool_blocked_advisor_not_assigned(tool_env): + record = tool_service.run_tool( + tool_name="query_holdings", + agent_type="advisor", + actor=ACTOR_ADVISOR, + customer_id="CUST-1010", + session_id="sess-b2", + ) + assert record["status"] == "blocked" and record["error_code"] == "AUTH_403_NOT_ASSIGNED" + + +def test_run_tool_blocked_scope(tool_env): + """analyst 等未授权角色 fail-closed（AUTH_403_SCOPE）。""" + record = tool_service.run_tool( + tool_name="query_holdings", + agent_type="customer", + actor=ACTOR_ANALYST, + customer_id="CUST-9527", + session_id="sess-b3", + ) + assert record["status"] == "blocked" and record["error_code"] == "AUTH_403_SCOPE" + + +def test_run_tool_risk_officer_full_access(tool_env): + """risk_officer 全量（对齐 assert_customer_access 口径）。""" + record = tool_service.run_tool( + tool_name="query_holdings", + agent_type="risk", + actor=ACTOR_RISK, + customer_id="CUST-9527", + session_id="sess-r1", + ) + assert record["status"] == "success" + + +def test_run_tool_blocked_no_customer(tool_env): + record = tool_service.run_tool( + tool_name="query_holdings", + agent_type="customer", + actor=ACTOR_CUSTOMER, + customer_id="", + session_id="sess-n1", + ) + assert record["status"] == "blocked" and record["error_code"] == "TOOL_BLOCKED_NO_CUSTOMER" + + +def test_run_tool_unknown_tool(tool_env): + record = tool_service.run_tool( + tool_name="drop_database", + agent_type="customer", + actor=ACTOR_CUSTOMER, + customer_id="CUST-9527", + session_id="sess-u1", + ) + assert record["status"] == "blocked" and record["error_code"] == "TOOL_UNKNOWN" + + +def test_run_tool_error_swallowed_and_audited(tool_env, monkeypatch): + """Tool 执行异常 → error 落痕且不向对话链路抛。""" + from app.tool import core_tools + + original = core_tools.TOOL_REGISTRY["query_holdings"]["func"] + + def boom(customer_id, core_ro): + raise RuntimeError("db exploded") + + core_tools.TOOL_REGISTRY["query_holdings"]["func"] = boom + try: + record = tool_service.run_tool( + tool_name="query_holdings", + agent_type="customer", + actor=ACTOR_CUSTOMER, + customer_id="CUST-9527", + session_id="sess-e1", + ) + finally: + core_tools.TOOL_REGISTRY["query_holdings"]["func"] = original + assert record["status"] == "error" and record["error_code"] == "TOOL_ERROR" + row = _tool_rows(tool_env["engine"])[0] + assert (row["status"], row["error_code"]) == ("error", "TOOL_ERROR") + + +def test_run_tool_audit_insert_degrades(tool_env, monkeypatch): + """留痕失败降级 warning（不阻塞对话，口径同 T-02 审计降级）。""" + def broken_repo(): + raise RuntimeError("repo down") + + monkeypatch.setattr(tool_service, "_session_repo", broken_repo) + record = tool_service.run_tool( + tool_name="query_holdings", + agent_type="customer", + actor=ACTOR_CUSTOMER, + customer_id="CUST-9527", + session_id="sess-d1", + ) + assert record["status"] == "success" # 对话链路不受留痕故障影响 + + +# ---------- 图集成：tool 节点注入与降级 ---------- + + +@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() + + +def test_graph_tool_result_injected_into_llm(fake_llm, tool_env): + out = agent_service.chat( + "customer", + [], + "查一下我的持仓", + session_id="sess-g1", + trace_id="trace-g1", + actor=ACTOR_CUSTOMER, + customer_id="CUST-9527", + ) + msgs = fake_llm.calls[0] + assert msgs[1].__class__.__name__ == "SystemMessage" + assert "[工具查询结果]" in msgs[1].content + assert "合计市值 50000" in msgs[1].content + assert '"total_count": 1' in msgs[1].content # 数据 JSON 一并注入 + assert len(_tool_rows(tool_env["engine"])) == 1 + assert out["tool_results"][0]["status"] == "success" + + +def test_graph_no_intent_skips_tool(fake_llm, tool_env): + agent_service.chat( + "customer", + [], + "你好", + session_id="sess-g2", + actor=ACTOR_CUSTOMER, + customer_id="CUST-9527", + ) + tool_msgs = [ + m + for m in fake_llm.calls[0] + if m.__class__.__name__ == "SystemMessage" and "[工具查询结果]" in m.content + ] + assert tool_msgs == [] + assert _tool_rows(tool_env["engine"]) == [] + + +def test_graph_blocked_result_visible_to_llm(fake_llm, tool_env): + """归属拒绝以 blocked 结果注入（对话内呈现，非 403）。""" + agent_service.chat( + "customer", + [], + "查一下我的持仓", + session_id="sess-g3", + actor=ACTOR_CUSTOMER, + customer_id="CUST-1001", + ) + content = fake_llm.calls[0][1].content + assert "拒绝" in content and "AUTH_403_NOT_OWNER" in content + + +def test_graph_without_session_context_no_tool(fake_llm, tool_env): + """无会话上下文（旧调用方式）Tool 空转——T-07 兼容。""" + out = agent_service.chat("customer", [], "查一下我的持仓") + assert out["tool_results"] == [] + assert "[工具查询结果]" not in fake_llm.calls[0][1].content + assert _tool_rows(tool_env["engine"]) == [] + + +def test_graph_degraded_reply_includes_summary(tool_env, monkeypatch): + """无 key 降级：回复携带 Tool 查询摘要（查询不白跑）。""" + monkeypatch.setattr(settings, "deepseek_api_key", "") + agent_service.reset_cache() + out = agent_service.chat( + "customer", + [], + "查一下我的持仓", + session_id="sess-g4", + actor=ACTOR_CUSTOMER, + customer_id="CUST-9527", + ) + assert "LLM 未配置" in out["reply"] + assert "合计市值 50000" in out["reply"] + assert out["has_disclaimer"] is True # 降级回复同样经 guard + + +# ---------- summarize / context_text ---------- + + +def test_summarize_variants(tool_env): + ok = tool_service.run_tool( + tool_name="query_customer_profile", + agent_type="customer", + actor=ACTOR_CUSTOMER, + customer_id="CUST-9527", + session_id="sess-s1", + ) + text = tool_service.summarize(ok) + assert "张三" in text and "A" in text + + blocked = tool_service.run_tool( + tool_name="query_holdings", + agent_type="customer", + actor=ACTOR_CUSTOMER, + customer_id="CUST-1001", + session_id="sess-s2", + ) + assert "无权访问" in tool_service.summarize(blocked) + + not_found = tool_service.run_tool( + tool_name="query_customer_profile", + agent_type="risk", + actor=ACTOR_RISK, + customer_id="CUST-NOPE", + session_id="sess-s3", + ) + assert "未找到" in tool_service.summarize(not_found) + + +def test_context_text_empty(): + assert tool_service.context_text([]) == ""