diff --git a/examples/langgraph_smoke.py b/examples/langgraph_smoke.py index 143d0ce..fa9ee89 100644 --- a/examples/langgraph_smoke.py +++ b/examples/langgraph_smoke.py @@ -1,28 +1,31 @@ -"""LangGraph 脚手架冒烟脚本（作业 1）。 +"""LangGraph 脚手架冒烟脚本。 同一 thread 输入两次、另一个 thread 输入一次，打印轮数和文本。 -预期输出轮数依次为 1、2、1。 +预期轮数依次为 1、2、1；首轮为 welcome（1001 固定问候，不调模型）， +第二轮起由注入的 workflow fake LLM 按提示词语义回复。 本脚本不导入 FastGPT、不读取其密钥；离线即可运行： - python examples/langgraph_smoke.py + python -m examples.langgraph_smoke """ import asyncio -from src.agent.graph import build_graph +from src.agent.graph import GraphDeps, build_graph, workflow_fake_llm async def main() -> None: graph = build_graph() + deps = GraphDeps(llm=workflow_fake_llm()) async def ask(thread_id: str, text: str) -> None: - # 只传本轮变化的字段；会话字段（turn_count 等）由 checkpoint 保管。 - # 注意：对已有 checkpoint 的 thread 传入完整初始状态， - # 其中的 turn_count=0 会覆盖已累计的值。 + # 只传本轮变化的字段；会话字段（turn_count/stage_code 等） + # 由 checkpoint 保管。注意：对已有 checkpoint 的 thread 传 + # 完整初始状态会覆盖已累计的值。 result = await graph.ainvoke( {"session_id": thread_id, "input_text": text}, config={"configurable": {"thread_id": thread_id}}, + context=deps, ) print( f"thread={thread_id} turn_count={result['turn_count']} " diff --git a/src/agent/confirm.py b/src/agent/confirm.py new file mode 100644 index 0000000..52643b3 --- /dev/null +++ b/src/agent/confirm.py @@ -0,0 +1,154 @@ +"""信息核实节点（作业 5，LLM 决策版）。 + +设计：提示词即程序——核实链的顺序、话术、阈值、全局中断全部 +在 ``prompts/single_confirm.md`` / ``dual_confirm.md`` 中（与工作流 +chatNode 提示词一致），代码不维护状态链。 + +节点职责（薄运行时）： +1. 把上下文注入提示词：已核实字段（person_fields，模型据此跳过 + 已确认项）、当前人员序号（person_index，双车两方隔离）、 + 对话历史、无回复计数； +2. 一次 LLM 调用产出 ```` 前缀回复；前缀缺失/非法提议 + 受控修复一次（复用 collect 的语义），仍失败抛 CollectReplyError； +3. 字段提交走**独立提取调用**（计划作业 4-4 允许，需记录次数）： + 从本轮输入提取语义明确的新字段，validate_candidate 校验后 + 按 person_index 隔离写入 person_fields / turn_patch； +4. 阶段码一律经白名单校验（3001/3002 → 0000/0001/0002/0003/0004）。 + +模型决定“问什么、去哪一步”；代码决定“能不能去、字段能不能写”。 +""" + +import json + +from .collect import ( + _bookkeeping, + context_prompt, + conversation_history, + validated_reply, +) +from .fields import FieldCandidate, FieldStatus, validate_candidate +from .llm import LLMService +from .prompts import DUAL_CONFIRM, SINGLE_CONFIRM, render_prompt +from .state import AgentState + +# 提取调用的系统提示标记（测试与分流依据；不以 开头， +# 故不会被误当对话回复解析） +_EXTRACTION_MARKER = "【字段提取】" + +_EXTRACTION_PROMPT = ( + _EXTRACTION_MARKER + + "从【用户输入】中提取本次新提供且语义明确的当事人信息，" + "只输出 JSON 对象（键限于给定字段表，值用原文；无新字段输出 {{}}）。" + "模糊、破碎、待确认的输入不提取。reserved 键 person_index（1 或 2）" + "仅在对话明确切换到另一方当事人时给出。\n" + "字段表（当前人员）：{keys}\n" + "【用户输入】{user_input}\n" + "【模型回复】{reply}" +) + +# 各方人员字段（来源：endpoints.get_info 分组；与 fields.KNOWN_FIELD_KEYS 对齐） +_PERSON1_KEYS = ("xm1", "sfzmhm1", "sfzmwh1", "sjhm1", "sjwh1", "csbw1") +_PERSON2_KEYS = ("xm2", "sfzmhm2", "sfzmwh2", "sjhm2", "sjwh2", "csbw2") + + +def _person_keys(person_index: int) -> tuple[str, ...]: + return _PERSON1_KEYS if person_index != 2 else _PERSON2_KEYS + + +async def _extract_turn_fields( + state: AgentState, llm: LLMService, reply: str +) -> tuple[list[FieldCandidate], int | None]: + """独立提取调用：返回 (候选列表, 模型声明的 person_index 或 None)。 + + 调用次数由调用方计入（计划作业 4-4：记录原因/次数/延迟）。 + """ + person_index = state.get("person_index") or 1 + raw = await llm.complete( + _EXTRACTION_PROMPT.format( + keys="、".join(_person_keys(person_index)), + user_input=state.get("input_text", ""), + reply=reply, + ), + "", + ) + try: + data = json.loads(raw.strip()) + if not isinstance(data, dict): + return [], None + except json.JSONDecodeError: + return [], None + next_person = data.pop("person_index", None) + if next_person not in (1, 2): + next_person = None + candidates = [ + FieldCandidate(key=key, value=value, source_turn=state.get("turn_count") or 0) + for key, value in data.items() + if key in _person_keys(person_index) + ] + return candidates, next_person + + +async def _run_confirm(state: AgentState, llm: LLMService, prompt_name: str) -> dict: + """核实节点共享运行时（单车/双车仅提示词不同）。""" + system_prompt = render_prompt(prompt_name, state) + context_prompt(state) + user_message = state.get("input_text", "") + current = state.get("stage_code") or "3001" + raw, stage = await validated_reply( + llm, system_prompt, user_message, current, history=conversation_history(state) + ) + + + # 字段提取与按人员隔离提交（提取调用本身可能失败，不阻塞对话） + person_fields = dict(state.get("person_fields") or {}) + confirm_status = dict(state.get("confirm_status") or {}) + patch: dict = {} + new_person_index = state.get("person_index") or 1 + try: + candidates, declared = await _extract_turn_fields(state, llm, raw) + except Exception: + candidates, declared = [], None + for cand in candidates: + validated = validate_candidate(cand) + confirm_status[validated.key] = validated.status + if validated.status is FieldStatus.CONFIRMED: + person_fields[validated.key] = validated.value + patch[validated.key] = validated.value + if declared is not None: + new_person_index = declared + + no_reply_count, _clarify = _bookkeeping(state) + # clarify_count 的“计入”是模型按提示词（连续 4 次）判断的语义， + # 代码侧只做簿记透传，不用 collect 的近似增量 + clarify_count = state.get("clarify_count") or 0 + dialogue = list(state.get("dialogue") or []) + dialogue.append({"role": "user", "content": state.get("input_text", "")}) + dialogue.append({"role": "assistant", "content": raw}) + + return { + "turn_count": (state.get("turn_count") or 0) + 1, + "no_reply_count": no_reply_count, + "clarify_count": clarify_count, + "response_text": raw, + "stage_code": stage, + "raw_status_code": stage, + "person_fields": person_fields, + "confirm_status": confirm_status, + "turn_patch": patch, + "person_index": new_person_index, + "dialogue": dialogue, + } + + +async def run_single_confirm(state: AgentState, llm: LLMService) -> dict: + """单车信息确认节点（提示词：prompts/single_confirm.md）。""" + return await _run_confirm(state, llm, SINGLE_CONFIRM) + + +async def run_dual_confirm(state: AgentState, llm: LLMService) -> dict: + """双车信息确认节点（提示词：prompts/dual_confirm.md）。 + + 与单车共用运行时；两方的区分靠注入的 person_fields 与 + person_index，提取调用的字段表按当前人员隔离，模型声明切换 + 到另一方（person_index）时由代码应用。 + """ + return await _run_confirm(state, llm, DUAL_CONFIRM) diff --git a/src/agent/graph.py b/src/agent/graph.py index 2db8e20..2742c09 100644 --- a/src/agent/graph.py +++ b/src/agent/graph.py @@ -1,19 +1,19 @@ """LangGraph 事故流程图。 -架构（对照 ``workflow/20261016/事故信息采集2025118.json``，作业 3 修订）： +架构（对照 ``workflow/20261020/事故信息采集2025118.json``）： -- 每个主要阶段是独立节点，与工作流 chatNode 一一对应： - ``collect``（事故信息收集）、``single_photo``（单车拍照）、 - ``dual_photo``（双车拍照）、``single_confirm`` / ``dual_confirm`` - （信息确认，作业 5 实现）； -- 节点间的大切换由 ``route_by_stage`` 条件边按 ``stage_code`` 路由 - （对照工作流的 Condition#2/#7/#9 等）；LLM 接入后由节点内的 - “解析状态”代码写出 ``stage_code``，条件边只做分发，不含业务规则； -- 节点内部的阶段推进（拍照步骤、失败/无回复处理、分流）由节点内 - LLM 提议目标 ``stage_code``（脚手架期由 ``_fake_llm_propose`` - 模拟），代码按 ``transitions`` 白名单校验、非法提议不写状态； - 代码只负责事件解析与计数等上下文簿记，不代替模型决定； -- LLM 接入前（作业 4 之前），节点共用固定话术模板。 +- 每个主要阶段一个节点，与工作流 chatNode 一一对应： + ``collect``（事故信息收集）、``single_photo`` / ``dual_photo`` + （拍照）、``single_confirm`` / ``dual_confirm``（信息确认）； +- 节点间大切换由 ``route_by_stage`` 条件边按 ``stage_code`` 分发 + （对照 Condition#2/#7/#9 等）；一轮 HTTP 请求 = 一次图执行， + 跨轮记忆由 checkpoint 承载； +- 节点内阶段码一律由 LLM 提议（提示词即程序），代码只校验 + （前缀格式、白名单、clientMode 出口）与落库； +- 依赖注入走 LangGraph 原生机制：``context_schema=GraphDeps``， + 节点签名 ``(state, runtime: Runtime[GraphDeps])``，调用时 + ``context=GraphDeps(llm=...)``。LLM client 属于运行时依赖， + 不进 state、不进 checkpoint（计划作业 3-2）。 图结构:: @@ -21,40 +21,43 @@ ├→ collect ───────────→ END ├→ single_photo ──────→ END ├→ dual_photo ────────→ END - ├→ single_confirm ────→ END (作业 5) - ├→ dual_confirm ──────→ END (作业 5) + ├→ single_confirm ────→ END + ├→ dual_confirm ──────→ END └→ END (终态 0000—0005) -首轮语义（对照工作流 work_start 的 is_first_message → Assigned Reply#3）： -新会话第一条消息由 ``welcome`` 节点以固定话术返回 ``1001`` -（未准备好通话），不调用模型；之后用户说【继续办理】或继续对话， -由 collect 节点的 LLM 提议 ``1001 → 1002``（白名单已含该转换）。 - -离线运行示例： +离线运行示例（默认注入 workflow fake LLM）： python -m examples.langgraph_smoke 对应测试：test/agent/test_scaffold.py、test/agent/test_events.py、 -test/agent/test_transitions.py +test/agent/test_transitions.py、test/agent/test_photo.py、 +test/agent/test_confirm.py """ +from dataclasses import dataclass + from langgraph.checkpoint.memory import InMemorySaver from langgraph.graph import END, START, StateGraph +from langgraph.runtime import Runtime -from .events import ClientEvent, parse_client_event +from .collect import run_collect +from .confirm import run_dual_confirm, run_single_confirm +from .events import parse_client_event +from .llm import LLMService, workflow_fake_llm +from .photo import run_photo from .state import AgentState -from .transitions import TERMINAL_STAGES, is_legal_transition +from .transitions import TERMINAL_STAGES -# 缺省阶段（首轮 welcome 之后的兜底）；后续由规则/模型驱动。 -DEFAULT_STAGE_CODE = "1002" -# 首轮固定问候（对照工作流 Assigned Reply#3 + Variable Update#6）： -# 新会话首条消息返回 1001，不调用模型。 -WELCOME_STAGE = "1001" -WELCOME_TEXT = ( - "1001您好，请问您是来办理交通事故信息采集的吗？" - "请说“继续办理”开始。" -) +@dataclass +class GraphDeps: + """图节点的运行时依赖。 + + 通过 ``context_schema`` 注入（``ainvoke(..., context=GraphDeps(...))``）， + 只存在于调用期，不序列化、不进入 checkpoint。 + """ + + llm: LLMService def normalize_input(state: AgentState) -> dict: @@ -99,157 +102,79 @@ def route_by_stage(state: AgentState) -> str: return "collect" -def welcome(state: AgentState) -> dict: - """首轮问候节点：新会话第一条消息的固定话术（对照工作流 - work_start is_first_message → Assigned Reply#3）。 +# 首轮固定问候（对照工作流 Assigned Reply#3）：新会话首条消息返回 +# 1001，不调用模型；话术与 Variable Update#6 后的记录一致。 +WELCOME_STAGE = "1001" +WELCOME_TEXT = ( + "1001您好，这里是无锡交警，我将为您远程处理交通事故。" + "请将人员撤离至路侧安全区域，开启危险报警双闪灯、放置三角警告牌、" + "做好安全防护，谨防二次事故伤害。若您已经准备好了，请点击继续办理，" + "如需人工服务，请说转人工。" +) - 确定性节点，不调用模型；返回 1001（未准备好通话）， - 用户确认后由 collect 节点的 LLM 经白名单守卫进入 1002。 + +def welcome(state: AgentState) -> dict: + """首轮问候节点（确定性，不调用模型）。 + + 返回 1001（未准备好通话），并把问候写入 dialogue， + 供后续节点作为聊天历史注入（对齐 FastGPT 的 chatId 历史）。 """ return { "turn_count": (state.get("turn_count") or 0) + 1, "response_text": WELCOME_TEXT, "stage_code": WELCOME_STAGE, "raw_status_code": WELCOME_STAGE, + "dialogue": [ + {"role": "user", "content": state.get("input_text", "")}, + {"role": "assistant", "content": WELCOME_TEXT}, + ], } -def _fake_llm_propose(state: AgentState, event: ClientEvent | None) -> str | None: - """假 LLM：提出下一步阶段码（作业 4 起由真实 LLM 取代）。 - - 模拟工作流 chatNode 提示词的判断（来源见各分支注释）： - - 会话标记且当前 1001 → 1002（work_start is_first_message）； - - 【拍摄完成】拍照链前进一步（单车/双车拍照提示词）； - - 拍照失败标记 → 0005（客户端已统计三次，提示词要求直接转人工）； - - 连续第二次【用户无回复】 → 0004（no_reply_count 由代码簿记， - 本函数读取的是本轮计数前的历史值）。 - - 提议是否生效由 ``is_legal_transition`` 校验。 - """ - current = state.get("stage_code") or "" - if event in (ClientEvent.SESSION_START, ClientEvent.CONTINUE): - if current == "1001": - return "1002" - if event is ClientEvent.PHOTO_DONE: - chain = { - "2000": "2001", "2001": "2002", "2002": "2003", - "2003": "2005", - "2010": "2011", "2011": "2012", "2012": "2013", - "2013": "2014", "2014": "2015", - } - return chain.get(current) - if event is ClientEvent.PHOTO_FAILED_3X: - return "0005" - if event is ClientEvent.NO_REPLY and (state.get("no_reply_count") or 0) >= 1: - return "0004" - return None +async def collect_node(state: AgentState, runtime: Runtime[GraphDeps]) -> dict: + """事故信息收集节点（工作流「事故信息收集」chatNode）。""" + return await run_collect(state, runtime.context.llm) -def _event_reply(state: AgentState) -> dict: - """事件簿记 + 阶段提议/校验 + 固定话术（作业 3 脚手架版）。 - - - 代码只维护上下文：``no_reply_count`` 累计（NO_REPLY）与 - 重置（其他输入）； - - 阶段码（含失败/无回复处理）一律由 ``_fake_llm_propose`` 提议 - （作业 4 换真实 LLM），经白名单校验，非法提议保持当前阶段； - - 轮数 +1，响应带 ```` 前缀（API 层解析）。 - - TODO(作业 4): 真实 LLM 接入后，话术与阶段提议移入各节点； - TODO(作业 4-10): 对非法提议做一次受控修复重试（当前直接拒绝）。 - """ - raw_event = state.get("client_event") or "" - event = ClientEvent(raw_event) if raw_event else None - - no_reply_count = state.get("no_reply_count") or 0 - if event is ClientEvent.NO_REPLY: - no_reply_count += 1 - elif (state.get("input_text") or "").strip(): - no_reply_count = 0 - - current = state.get("stage_code") or DEFAULT_STAGE_CODE - proposed = _fake_llm_propose(state, event) - if proposed and is_legal_transition(current, proposed): - stage_code = proposed - else: - stage_code = current - - response_text = ( - f"{stage_code}" - f"已收到：{state['input_text']}" - ) - return { - "turn_count": (state.get("turn_count") or 0) + 1, - "no_reply_count": no_reply_count, - "response_text": response_text, - "stage_code": stage_code, - } +async def single_photo_node(state: AgentState, runtime: Runtime[GraphDeps]) -> dict: + """单车拍照节点（工作流「单车拍照」/「单车拍照#6」chatNode）。""" + return await run_photo(state, runtime.context.llm, kind="single") -def collect(state: AgentState) -> dict: - """事故信息收集节点（工作流“事故信息收集” chatNode）。 - - TODO(作业 4): 接入 LLM——多轮采集、跳过已知信息、一次补齐分流； - 模型回复中的状态前缀由代码校验（解析状态，对照工作流 - “解析状态” code 节点），不由模型自由写阶段。 - """ - return _event_reply(state) +async def dual_photo_node(state: AgentState, runtime: Runtime[GraphDeps]) -> dict: + """双车拍照节点（工作流「双车拍照」/「双车拍照#7」chatNode）。""" + return await run_photo(state, runtime.context.llm, kind="dual") -def single_photo(state: AgentState) -> dict: - """单车拍照节点（工作流“单车拍照” chatNode）。 - - TODO(作业 5): 按 2000→2001→2002→2003→2005 生成对应拍照话术； - 步骤推进的确定性部分已在 ``transitions`` 表覆盖。 - """ - return _event_reply(state) +async def single_confirm_node(state: AgentState, runtime: Runtime[GraphDeps]) -> dict: + """单车信息确认节点（工作流「单车信息确认」chatNode）。""" + return await run_single_confirm(state, runtime.context.llm) -def dual_photo(state: AgentState) -> dict: - """双车拍照节点（工作流“双车拍照” chatNode）。 - - TODO(作业 5): 按 2010→…→2015 生成对应话术；direct 模式末次 - 拍摄完成进入 3002（同轮询问第一方身份）。 - """ - return _event_reply(state) +async def dual_confirm_node(state: AgentState, runtime: Runtime[GraphDeps]) -> dict: + """双车信息确认节点（工作流「双车信息确认」chatNode）。""" + return await run_dual_confirm(state, runtime.context.llm) -def single_confirm(state: AgentState) -> dict: - """单车信息确认节点（工作流“单车信息确认” chatNode，作业 5）。""" - raise NotImplementedError("TODO(作业 5): 实现单车信息确认节点") - - -def dual_confirm(state: AgentState) -> dict: - """双车信息确认节点（工作流“双车信息确认” chatNode，作业 5）。""" - raise NotImplementedError("TODO(作业 5): 实现双车信息确认节点") - - -def build_graph(checkpointer=None, llm=None): +def build_graph(checkpointer=None): """装配并编译图。 参数: checkpointer: 自定义 checkpointer；缺省 ``InMemorySaver``， 作业 7 替换为 ``AsyncPostgresSaver``。 - llm: LLM 服务（``llm.LLMService``）；提供后 collect 节点 - 走 ``collect.run_collect``（作业 4），否则保持离线 - 脚手架行为（``_event_reply``，作业 1—3 的测试依赖）。 - 调用方通过 ``config={"configurable": {"thread_id": <会话>}}`` 区分会话。 + LLM 依赖不在此处绑定——调用时通过 + ``config={"configurable": {"thread_id": <会话>}}`` 区分会话， + ``context=GraphDeps(llm=)`` 注入运行时依赖。 """ - builder = StateGraph(AgentState) + builder = StateGraph(AgentState, context_schema=GraphDeps) builder.add_node("normalize_input", normalize_input) builder.add_node("welcome", welcome) - if llm is None: - collect_node = collect - else: - from .collect import run_collect - - async def collect_node(state): # noqa: F811 - return await run_collect(state, llm) builder.add_node("collect", collect_node) - builder.add_node("single_photo", single_photo) - builder.add_node("dual_photo", dual_photo) - builder.add_node("single_confirm", single_confirm) - builder.add_node("dual_confirm", dual_confirm) + builder.add_node("single_photo", single_photo_node) + builder.add_node("dual_photo", dual_photo_node) + builder.add_node("single_confirm", single_confirm_node) + builder.add_node("dual_confirm", dual_confirm_node) builder.add_edge(START, "normalize_input") builder.add_conditional_edges( @@ -265,10 +190,17 @@ def build_graph(checkpointer=None, llm=None): END: END, }, ) - builder.add_edge("welcome", END) - builder.add_edge("collect", END) - builder.add_edge("single_photo", END) - builder.add_edge("dual_photo", END) - builder.add_edge("single_confirm", END) - builder.add_edge("dual_confirm", END) + for name in ( + "welcome", "collect", "single_photo", "dual_photo", + "single_confirm", "dual_confirm", + ): + builder.add_edge(name, END) return builder.compile(checkpointer=checkpointer or InMemorySaver()) + + +__all__ = [ + "AgentState", + "GraphDeps", + "build_graph", + "workflow_fake_llm", +] diff --git a/src/agent/photo.py b/src/agent/photo.py new file mode 100644 index 0000000..65c8a13 --- /dev/null +++ b/src/agent/photo.py @@ -0,0 +1,83 @@ +"""拍照节点：按工作流分支选提示词，由 LLM 同时决定下一阶段和话术。 + +来源：workflow/20261020/事故信息采集2025118.json 的四个拍照 chatNode。 +【拍摄完成】、拍照失败、无回复均作为模型输入；代码校验前缀、转换 +白名单和 clientMode 出口，不覆盖模型的判断或正文。 +""" + +from .collect import ( + _bookkeeping, + _build_user_message, + context_prompt, + conversation_history, + validated_reply, +) +from .llm import LLMService +from .prompts import ( + DUAL_PHOTO, + DUAL_PHOTO_ADDON, + SINGLE_PHOTO, + SINGLE_PHOTO_ADDON, + render_prompt, +) +from .state import AgentState +from .transitions import is_legal_transition + +PHOTO_PROMPTS = { + ("single", "direct"): SINGLE_PHOTO, + ("single", "browser_addon"): SINGLE_PHOTO_ADDON, + ("dual", "direct"): DUAL_PHOTO, + ("dual", "browser_addon"): DUAL_PHOTO_ADDON, +} + + +def is_legal_photo_transition(source: str, target: str, *, mode: str) -> bool: + """按工作流 Condition#11/#12 校验模式出口；不替模型改写目标。 + + direct 拍照结束进入 3001/3002；addon 拍照结束进入 0000。 + 两种模式均允许提示词定义的转人工码和保持当前阶段。 + """ + if not is_legal_transition(source, target): + return False + if mode == "browser_addon" and target in {"3001", "3002"}: + return False + if mode == "direct" and target == "0000": + return False + return True + + +async def run_photo(state: AgentState, llm: LLMService, *, kind: str) -> dict: + """调用对应拍照模型一次；输出不合规时最多修复一次。 + + 历史包含上一轮已经提出的拍照/身份问题，随 checkpoint 传到核实 + 节点，避免切换节点后丢失上下文。车损位置等自然语言保留原样。 + """ + mode = state.get("client_mode") or "direct" + prompt_name = PHOTO_PROMPTS[(kind, mode)] + current = state["stage_code"] + raw, stage = await validated_reply( + llm, + render_prompt(prompt_name, state) + context_prompt(state), + _build_user_message(state), + current, + history=conversation_history(state), + validator=lambda source, target: is_legal_photo_transition( + source, target, mode=mode + ), + ) + no_reply_count, clarify_count = _bookkeeping(state) + dialogue = list(state.get("dialogue") or []) + dialogue.extend([ + {"role": "user", "content": state.get("input_text", "")}, + {"role": "assistant", "content": raw}, + ]) + return { + "turn_count": (state.get("turn_count") or 0) + 1, + "response_text": raw, + "stage_code": stage, + "raw_status_code": stage, + "no_reply_count": no_reply_count, + "clarify_count": clarify_count, + "dialogue": dialogue, + "turn_patch": {}, + } diff --git a/src/backends/langgraph.py b/src/backends/langgraph.py index 15c9e9a..65201f4 100644 --- a/src/backends/langgraph.py +++ b/src/backends/langgraph.py @@ -14,18 +14,21 @@ from collections.abc import AsyncIterator -from ..agent.graph import build_graph +from ..agent.graph import GraphDeps, build_graph, workflow_fake_llm from .chat import ChatInput, ChatResult, ChatStreamEvent, ChatBackend, TextDelta class LangGraphBackend(ChatBackend): """Adapt a compiled LangGraph to the neutral chat contract.""" - def __init__(self, graph=None): + def __init__(self, graph=None, llm=None): """参数: - graph: 编译后的图；缺省调用 ``build_graph()``（内存 checkpoint）。 + graph: 编译后的图；缺省调用 ``build_graph()``。 + llm: LLM 服务（运行时依赖）；缺省为离线 workflow fake。 + 生产环境由 lifespan 注入 ``build_llm_service()`` 的实例。 """ self._graph = graph if graph is not None else build_graph() + self._llm = llm if llm is not None else workflow_fake_llm() async def complete(self, chat_input: ChatInput) -> ChatResult: """非流式：执行一轮图并翻译结果。 @@ -40,8 +43,13 @@ class LangGraphBackend(ChatBackend): （作业 6 起实现）。 """ result = await self._graph.ainvoke( - {"session_id": chat_input.session_id, "input_text": chat_input.text}, + { + "session_id": chat_input.session_id, + "input_text": chat_input.text, + "client_mode": chat_input.client_mode, + }, config={"configurable": {"thread_id": chat_input.session_id}}, + context=GraphDeps(llm=self._llm), ) return ChatResult( content = result["response_text"], diff --git a/src/core/fastgpt_client.py b/src/core/fastgpt_client.py index 567eb53..fdd307a 100644 --- a/src/core/fastgpt_client.py +++ b/src/core/fastgpt_client.py @@ -26,8 +26,8 @@ async def lifespan(app: FastAPI): from ..agent.llm import build_llm_service global _langgraph_graph - _langgraph_graph = build_graph(llm=build_llm_service()) - _chat_backend = LangGraphBackend(_langgraph_graph) + _langgraph_graph = build_graph() + _chat_backend = LangGraphBackend(_langgraph_graph, llm=build_llm_service()) else: _fastgpt_client = AsyncChatClient( api_key=Config.FASTGPT_API_KEY, diff --git a/test/agent/test_collect_llm.py b/test/agent/test_collect_llm.py index d0120ff..d610eab 100644 --- a/test/agent/test_collect_llm.py +++ b/test/agent/test_collect_llm.py @@ -26,7 +26,8 @@ from src.agent.prompts import COLLECTION, load_prompt def collect_state(**overrides) -> dict: state = { "session_id": "s1", - "input_text": "你好", + # 使用完整句而非短促词，避免触发分流守卫（“你好”<=2 字） + "input_text": "发生了交通事故，两辆车追尾", "stage_code": "1002", "turn_count": 1, "no_reply_count": 0, @@ -138,13 +139,16 @@ async def test_collect_illegal_stage_proposal_not_committed(): @pytest.mark.asyncio async def test_graph_collect_node_with_llm(): - # 图级：build_graph(llm=...) 时首轮走 welcome（不调模型）， - # 第二轮起 collect 节点走 run_collect + # 图级：首轮走 welcome（确定性，不调模型）；第二轮起 collect + # 节点经 Runtime 注入的 llm 走 run_collect + from src.agent.graph import GraphDeps + llm = FakeLLMService(lambda s, u: "1002收到。") - graph = build_graph(llm=llm) + graph = build_graph() first = await graph.ainvoke( {"session_id": "s1", "input_text": "追尾了"}, config={"configurable": {"thread_id": "s1"}}, + context=GraphDeps(llm=llm), ) assert len(llm.calls) == 0 # welcome 是确定性节点 assert first["stage_code"] == "1001" @@ -152,6 +156,7 @@ async def test_graph_collect_node_with_llm(): result = await graph.ainvoke( {"session_id": "s1", "input_text": "继续办理"}, config={"configurable": {"thread_id": "s1"}}, + context=GraphDeps(llm=llm), ) assert len(llm.calls) == 1 assert result["response_text"].startswith("1002") diff --git a/test/agent/test_confirm.py b/test/agent/test_confirm.py new file mode 100644 index 0000000..72444a5 --- /dev/null +++ b/test/agent/test_confirm.py @@ -0,0 +1,193 @@ +"""作业 5 验收测试：信息核实节点（LLM 决策版）。 + +测试用脚本化 FakeLLM 模拟“行为良好的模型”：按用户最新输入 +返回对应的带前缀回复；字段提取调用按提示词返回 JSON。 +断言仍是行为级的：stage 推进、person_fields 落库、person_index +切换、模糊不计入失败。 +""" + +import json + +import pytest + +from src.agent.graph import GraphDeps, build_graph, workflow_fake_llm +from src.agent.llm import FakeLLMService + + +def latest_input(message: str) -> str: + return message.split("【用户最新输入】")[-1].strip() + + +def make_llm(dialogue_script, extraction_script=None): + """dialogue_script(latest_input) -> 带前缀回复； + extraction_script(extraction_system_prompt) -> JSON 字符串。""" + + def responder(system, user): + if system.startswith("【字段提取】"): + if extraction_script: + return extraction_script(system) + return "{}" + return dialogue_script(latest_input(user)) + + return FakeLLMService(responder) + + +async def invoke(graph, thread_id, text, llm=None, **initial): + payload = {"session_id": thread_id, "input_text": text} + payload.update(initial) + return await graph.ainvoke( + payload, + config={"configurable": {"thread_id": thread_id}}, + context=GraphDeps(llm=llm or workflow_fake_llm()), + ) + + +# ---------- 单车核实 ---------- + + +@pytest.mark.asyncio +async def test_entry_consumes_pending_identity_question(): + # 用户确认车主身份 → 模型直接问姓名（身份问题的提问发生在 + # 拍照节点进入 3001 的同一回复里，这里不得重复） + llm = make_llm( + lambda q: "3001好的，请问您的姓名是什么？" + ) + graph = build_graph() + result = await invoke( + graph, "t-entry", "是的，我是车主", llm=llm, + stage_code="3001", client_mode="direct", + ) + + assert result["stage_code"] == "3001" + assert "姓名" in result["response_text"] + assert "车主" not in result["response_text"] + + +@pytest.mark.asyncio +async def test_single_confirm_full_chain_to_0000(): + def dialogue(q): + if q == "是的": + return "3001好的，请问您的姓名是什么？" + if q == "张三": + return "3001张三您好，请问您的身份证号码后四位是否正确？" + if q == "是对的": + return "3001请问您的手机号码后四位是否正确？" + return "0000好的，信息已全部确认无误。感谢您的配合。" + + def extraction(system): + if "【用户输入】张三" in system: + return json.dumps({"xm1": "张三"}, ensure_ascii=False) + return "{}" + + llm = make_llm(dialogue, extraction) + graph = build_graph() + + r1 = await invoke(graph, "t-chain", "是的", llm=llm, stage_code="3001") + assert "姓名" in r1["response_text"] + r2 = await invoke(graph, "t-chain", "张三", llm=llm) + assert "身份证" in r2["response_text"] + assert r2["person_fields"]["xm1"] == "张三" + r3 = await invoke(graph, "t-chain", "是对的", llm=llm) + assert "手机" in r3["response_text"] + r4 = await invoke(graph, "t-chain", "没错", llm=llm) + assert r4["stage_code"] == "0000" + assert r4["response_text"].startswith("0000") + + +@pytest.mark.asyncio +async def test_deny_collects_correct_value(): + llm = make_llm( + lambda q: "3001抱歉，请告诉我您正确的身份证号码后四位。", + lambda system: "{}", + ) + graph = build_graph() + result = await invoke( + graph, "t-deny", "不对", llm=llm, + stage_code="3001", person_fields={"xm1": "张三"}, client_mode="direct", + ) + + assert result["stage_code"] == "3001" + assert "身份证" in result["response_text"] + assert result["person_fields"].get("sfzmwh1") in (None, "") + + +@pytest.mark.asyncio +async def test_vague_answer_stays_without_counting_failure(): + llm = make_llm( + lambda q: "3001请问您是否是这辆车的车主？请回答“是”或“不是”。", + lambda system: "{}", + ) + graph = build_graph() + result = await invoke( + graph, "t-vague", "嗯", llm=llm, + stage_code="3001", clarify_count=0, client_mode="direct", + ) + + assert result["stage_code"] == "3001" + assert "车主" in result["response_text"] + assert (result.get("clarify_count") or 0) == 0 + + +# ---------- 双车核实 ---------- + + +@pytest.mark.asyncio +async def test_dual_person_index_switches_after_handoff(): + llm = make_llm( + lambda q: "3002您好。请问这辆车是由您驾驶的吗？", + lambda system: json.dumps({"person_index": 2}), + ) + graph = build_graph() + result = await invoke( + graph, "t-dual", "移交完毕", llm=llm, + stage_code="3002", person_index=1, + person_fields={"xm1": "张三"}, client_mode="direct", + ) + + assert result["stage_code"] == "3002" + assert result["person_index"] == 2 + assert "驾驶" in result["response_text"] + + +@pytest.mark.asyncio +async def test_dual_second_person_fields_isolated(): + def dialogue(q): + if q == "是的": + return "3002好的，驾驶员您好。请问您的姓名是什么？" + return "3002请问您的身份证号码后四位是否正确？" + + def extraction(system): + if "【用户输入】李四" in system: + return json.dumps({"xm2": "李四"}, ensure_ascii=False) + return "{}" + + llm = make_llm(dialogue, extraction) + graph = build_graph() + + await invoke( + graph, "t-dual2", "是的", llm=llm, + stage_code="3002", person_index=2, + person_fields={"xm1": "张三", "sjhm1": "13800001111"}, + client_mode="direct", + ) + result = await invoke(graph, "t-dual2", "李四", llm=llm) + + assert result["person_fields"]["xm2"] == "李四" + assert result["person_fields"]["xm1"] == "张三" + + +@pytest.mark.asyncio +async def test_dual_completion_goes_to_0000(): + llm = make_llm( + lambda q: "0000好的，两位驾驶员的信息均已核实完毕。感谢配合。", + lambda system: "{}", + ) + graph = build_graph() + result = await invoke( + graph, "t-dual-done", "没错", llm=llm, + stage_code="3002", person_index=2, + person_fields={"xm1": "张三", "xm2": "李四"}, + client_mode="direct", + ) + + assert result["stage_code"] == "0000" diff --git a/test/agent/test_photo.py b/test/agent/test_photo.py new file mode 100644 index 0000000..e72eb06 --- /dev/null +++ b/test/agent/test_photo.py @@ -0,0 +1,116 @@ +"""拍照节点回归：LLM 提议阶段与正文，运行时校验并保存上下文。 + +fake 输出仅验证接线与校验，不代替真实模型对提示词服从性的评测。 +""" + +import pytest + +from src.agent.collect import CollectReplyError +from src.agent.graph import GraphDeps, build_graph, workflow_fake_llm +from src.agent.llm import FakeLLMService + + +async def invoke(graph, thread_id, text, llm=None, **initial): + return await graph.ainvoke( + {"session_id": thread_id, "input_text": text, **initial}, + config={"configurable": {"thread_id": thread_id}}, + context=GraphDeps(llm=llm or workflow_fake_llm()), + ) + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("source", "target"), + [("2000", "2001"), ("2001", "2002"), ("2002", "2003"), ("2003", "2005"), + ("2010", "2011"), ("2011", "2012"), ("2012", "2013"), + ("2013", "2014"), ("2014", "2015")], +) +async def test_photo_applies_model_proposal(source, target): + raw = f"{target}模型生成的本轮拍照引导。" + llm = FakeLLMService(lambda s, u: raw) + result = await invoke(build_graph(), "photo", "【拍摄完成】", llm=llm, + stage_code=source, client_mode="direct") + assert result["stage_code"] == target + assert result["response_text"] == raw + assert len(llm.calls) == 1 + assert "【拍摄完成】" in llm.calls[0]["user_message"] + + +@pytest.mark.asyncio +@pytest.mark.parametrize("text", ["【拍摄完成】", "【客户端连续3次拍摄识别失败：图片过于模糊】", "【用户无回复】"]) +async def test_event_alone_does_not_force_stage(text): + # 模型保持阶段时，运行时不会仅凭事件强制跳转。 + llm = FakeLLMService(lambda s, u: "2012请确认当前操作。") + result = await invoke(build_graph(), "hold", text, llm=llm, + stage_code="2012", client_mode="direct") + assert result["stage_code"] == "2012" + assert len(llm.calls) == 1 + + +@pytest.mark.asyncio +async def test_failure_stage_also_comes_from_model(): + llm = FakeLLMService(lambda s, u: "0005正在转接人工。") + result = await invoke(build_graph(), "fail", "【客户端连续3次拍摄识别失败：图片过于模糊】", llm=llm, + stage_code="2012", client_mode="direct") + assert result["stage_code"] == "0005" + assert len(llm.calls) == 1 + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("source", "mode", "target", "node"), + [("2005", "direct", "3001", "单车拍照"), + ("2005", "browser_addon", "0000", "单车拍照#6"), + ("2015", "direct", "3002", "双车拍照"), + ("2015", "browser_addon", "0000", "双车拍照#7")], +) +async def test_mode_selects_workflow_prompt(source, mode, target, node): + from src.agent.prompts import load_prompt + from src.agent.photo import PHOTO_PROMPTS + + kind = "single" if source == "2005" else "dual" + expected_prompt = load_prompt(PHOTO_PROMPTS[(kind, mode)]) + expected_body = expected_prompt.split("--- 原文开始 ---", 1)[1].split("--- 原文结束 ---", 1)[0].strip() + raw = f"{target}对应模式的结束话术。" + llm = FakeLLMService(lambda s, u: raw) + text = "车子左后方" if source == "2005" else "【拍摄完成】" + result = await invoke(build_graph(), "end", text, llm=llm, + stage_code=source, client_mode=mode) + assert result["stage_code"] == target + assert len(llm.calls) == 1 + assert f"节点「{node}」" in expected_prompt + # 模型收到原文，预填变量已经渲染，来源头部不进入模型上下文。 + assert "{{$" not in llm.calls[0]["system_prompt"] + assert expected_body.splitlines()[0] in llm.calls[0]["system_prompt"] + + +@pytest.mark.asyncio +async def test_photo_preserves_guidance_and_handoff_history(): + replies = iter([ + "2015您可以点击手机界面中间的拍照按钮。", + "3002请问第一辆车是由您驾驶的吗？", + ]) + llm = FakeLLMService(lambda s, u: next(replies)) + graph = build_graph() + first = await invoke(graph, "history", "拍照按钮在哪里？", llm=llm, + stage_code="2015", client_mode="direct") + second = await invoke(graph, "history", "【拍摄完成】", llm=llm) + assert "手机界面中间" in first["response_text"] + assert llm.calls[1]["history"][0] == {"role": "user", "content": "拍照按钮在哪里？"} + assert llm.calls[1]["user_message"] == "【拍摄完成】" + assert second["dialogue"][-1]["content"] == second["response_text"] + assert "驾驶" in second["dialogue"][-1]["content"] + + +@pytest.mark.asyncio +@pytest.mark.parametrize(("source", "mode", "target"), [ + ("2000", "direct", "2003"), + ("2015", "browser_addon", "3002"), + ("2005", "direct", "0000"), +]) +async def test_illegal_proposal_fails_after_one_repair(source, mode, target): + llm = FakeLLMService(lambda s, u: f"{target}非法提议") + with pytest.raises(CollectReplyError): + await invoke(build_graph(), "illegal", "继续", llm=llm, + stage_code=source, client_mode=mode) + assert len(llm.calls) == 2 diff --git a/test/agent/test_scaffold.py b/test/agent/test_scaffold.py index f482eb6..b5021ec 100644 --- a/test/agent/test_scaffold.py +++ b/test/agent/test_scaffold.py @@ -11,15 +11,16 @@ import pytest -from src.agent.graph import build_graph +from src.agent.graph import GraphDeps, build_graph, workflow_fake_llm -async def ask(graph, thread_id: str, text: str) -> dict: +async def ask(graph, thread_id: str, text: str, llm=None) -> dict: # 只传本轮变化的字段；对已有 checkpoint 的 thread 传完整初始状态， # 其中的 turn_count=0 会覆盖已累计的值（LangGraph 输入按通道覆盖）。 return await graph.ainvoke( {"session_id": thread_id, "input_text": text}, config={"configurable": {"thread_id": thread_id}}, + context=GraphDeps(llm=llm or workflow_fake_llm()), ) diff --git a/test/agent/test_transitions.py b/test/agent/test_transitions.py index a137035..2242410 100644 --- a/test/agent/test_transitions.py +++ b/test/agent/test_transitions.py @@ -12,8 +12,9 @@ import pytest -import src.agent.graph as graph_mod -from src.agent.graph import build_graph +from src.agent.llm import FakeLLMService + +from src.agent.graph import GraphDeps, build_graph, workflow_fake_llm from src.agent.transitions import ( TERMINAL_STAGES, TransitionSpec, @@ -21,12 +22,13 @@ from src.agent.transitions import ( ) -def invoke(graph, thread_id, text, **initial): +def invoke(graph, thread_id, text, llm=None, **initial): payload = {"session_id": thread_id, "input_text": text} payload.update(initial) return graph.ainvoke( payload, config={"configurable": {"thread_id": thread_id}}, + context=GraphDeps(llm=llm or workflow_fake_llm()), ) @@ -82,12 +84,21 @@ async def test_valid_reply_resets_no_reply_count(): @pytest.mark.asyncio -async def test_illegal_proposal_keeps_stage(monkeypatch): - # LLM 提议非法跳步（2001 → 2003）：校验拒绝，状态不被污染 +async def test_illegal_proposal_raises_and_state_untouched(): + # LLM 坚持非法跳步（2001 → 2003）：白名单拒绝 + 一次修复后 + # 仍非法 → CollectReplyError，非法阶段不写状态 + from src.agent.collect import CollectReplyError + + llm = FakeLLMService(lambda s, u: "2003跳步") graph = build_graph() - monkeypatch.setattr(graph_mod, "_fake_llm_propose", lambda s, e: "2003") - result = await invoke(graph, "t-illegal", "【拍摄完成】", stage_code="2001") - assert result["stage_code"] == "2001" + with pytest.raises(CollectReplyError): + await invoke(graph, "t-illegal", "【拍摄完成】", llm=llm, stage_code="2001") + assert len(llm.calls) == 2 + + # 会话未被污染：下一轮用守规矩的 fake 继续，阶段仍是 2001 + good = FakeLLMService(lambda s, u: "2002已收到。") + result = await invoke(graph, "t-illegal", "【拍摄完成】", llm=good, stage_code="2001") + assert result["stage_code"] == "2002" @pytest.mark.asyncio diff --git a/test/agent/test_workflow_prompts.py b/test/agent/test_workflow_prompts.py new file mode 100644 index 0000000..b56719c --- /dev/null +++ b/test/agent/test_workflow_prompts.py @@ -0,0 +1,101 @@ +"""20261020 工作流的提示词移植与模型决策回归。""" + +from datetime import datetime, timezone +import json +from pathlib import Path + +import pytest + +from src.agent.collect import CollectReplyError, run_collect +from src.agent.llm import FakeLLMService +from src.agent.prompts import COLLECTION, load_prompt, render_prompt + +WORKFLOW = Path(__file__).resolve().parents[2] / "workflow/20261020/事故信息采集2025118.json" +NODES = { + "collection": "事故信息收集", + "single_photo": "单车拍照", + "dual_photo": "双车拍照", + "single_photo_addon": "单车拍照#6", + "dual_photo_addon": "双车拍照#7", + "single_confirm": "单车信息确认", + "dual_confirm": "双车信息确认", +} + + +@pytest.mark.parametrize(("prompt", "node_name"), NODES.items()) +def test_archived_prompt_matches_workflow(prompt, node_name): + workflow = json.loads(WORKFLOW.read_text(encoding="utf-8-sig")) + node = next(n for n in workflow["nodes"] if n["name"] == node_name) + original = next(i["value"] for i in node["inputs"] if i["key"] == "systemPrompt") + archived = load_prompt(prompt).split("--- 原文开始 ---", 1)[1].split("--- 原文结束 ---", 1)[0] + assert archived.strip() == original.strip() + + +def test_render_replaces_time_and_person_variables(): + now = datetime(2026, 10, 20, 12, 0, tzinfo=timezone.utc) + rendered = render_prompt(COLLECTION, {}, now=now) + assert "2026-10-20 12:00:00 +0000" in rendered + rendered = render_prompt("dual_confirm", { + "person_fields": {"hphm1": "测试A", "hphm2": "测试B", "sfzmwh1": "1234"} + }) + assert "测试A" in rendered and "测试B" in rendered + assert "1234" in rendered + assert "{{$" not in rendered + assert "未提供" in rendered + + +@pytest.mark.asyncio +async def test_short_answer_does_not_override_model_routing(): + # “在”可能是明确回答；运行时不得按字数一律拒绝分流。 + llm = FakeLLMService(lambda s, u: "2000信息已齐，请拍摄车辆前方。") + result = await run_collect({ + "stage_code": "1002", "input_text": "在", "clarify_count": 0, + "dialogue": [{"role": "assistant", "content": "您还在事故现场吗？"}], + }, llm) + assert result["stage_code"] == "2000" + assert result["clarify_count"] == 0 + assert len(llm.calls) == 1 + + +@pytest.mark.asyncio +async def test_format_and_stage_errors_share_one_repair_budget(): + replies = iter(["缺少前缀", "9999非法阶段"]) + llm = FakeLLMService(lambda s, u: next(replies)) + with pytest.raises(CollectReplyError): + await run_collect({"stage_code": "1002", "input_text": "追尾了"}, llm) + assert len(llm.calls) == 2 + + +@pytest.mark.asyncio +async def test_collect_sends_history_as_messages(): + history = [ + {"role": "user", "content": "没有人受伤"}, + {"role": "assistant", "content": "1002请问有几辆车？"}, + ] + llm = FakeLLMService(lambda s, u: "1002请问您还在现场吗？") + await run_collect({ + "stage_code": "1002", "input_text": "两辆", "dialogue": history, + }, llm) + assert llm.calls[0]["history"] == history + assert llm.calls[0]["user_message"] == "两辆" + assert "没有人受伤" not in llm.calls[0]["user_message"] + + +@pytest.mark.asyncio +async def test_openai_adapter_preserves_message_roles(): + from types import SimpleNamespace + from unittest.mock import AsyncMock + from src.agent.llm import OpenAILLMService + + create = AsyncMock(return_value=SimpleNamespace(choices=[ + SimpleNamespace(message=SimpleNamespace(content="1002收到。")) + ])) + llm = OpenAILLMService(model="test", api_key="test", max_retries=0) + llm._client = SimpleNamespace(chat=SimpleNamespace(completions=SimpleNamespace(create=create))) + history = [{"role": "assistant", "content": "之前的问题"}] + await llm.complete("系统提示词", "本轮输入", history=history) + assert create.call_args.kwargs["messages"] == [ + {"role": "system", "content": "系统提示词"}, + {"role": "assistant", "content": "之前的问题"}, + {"role": "user", "content": "本轮输入"}, + ]