refactor(agent): inject deps via context_schema, LLM-driven photo/confirm nodes

- graph.py: LangGraph-native DI - StateGraph(context_schema=GraphDeps),
  nodes take (state, runtime: Runtime[GraphDeps]); build_graph() decoupled
  from LLM; deleted scaffold _event_reply/_fake_llm_propose/stub nodes
- photo.py: model proposes stage+body via per-mode prompts (direct/addon);
  validator enforces clientMode exits; deterministic event forcing removed
- confirm.py: single runtime for single/dual confirm, extraction call
  isolated per person_index
- llm.py: workflow_fake_llm() (offline workflow-faithful responder,
  history-aware) for smoke/tests
- backend: LangGraphBackend(graph, llm) passes context per call
- tests: invoke helpers inject GraphDeps; illegal-proposal test asserts
  error + state untouched via second well-behaved fake

Verified: 143 passed; smoke turns 1/2/1 with original welcome text
This commit is contained in:
Eric Wang
2026-10-21 15:11:16 +08:00
parent a5407e03b2
commit 6d0a2169c8
12 changed files with 794 additions and 187 deletions
+10 -7
View File
@@ -1,28 +1,31 @@
"""LangGraph 脚手架冒烟脚本(作业 1)。 """LangGraph 脚手架冒烟脚本。
同一 thread 输入两次、另一个 thread 输入一次,打印轮数和文本。 同一 thread 输入两次、另一个 thread 输入一次,打印轮数和文本。
预期输出轮数依次为 1、2、1。 预期轮数依次为 1、2、1;首轮为 welcome(1001 固定问候,不调模型),
第二轮起由注入的 workflow fake LLM 按提示词语义回复。
本脚本不导入 FastGPT、不读取其密钥;离线即可运行: 本脚本不导入 FastGPT、不读取其密钥;离线即可运行:
python examples/langgraph_smoke.py python -m examples.langgraph_smoke
""" """
import asyncio import asyncio
from src.agent.graph import build_graph from src.agent.graph import GraphDeps, build_graph, workflow_fake_llm
async def main() -> None: async def main() -> None:
graph = build_graph() graph = build_graph()
deps = GraphDeps(llm=workflow_fake_llm())
async def ask(thread_id: str, text: str) -> None: async def ask(thread_id: str, text: str) -> None:
# 只传本轮变化的字段;会话字段(turn_count 等)由 checkpoint 保管。 # 只传本轮变化的字段;会话字段(turn_count/stage_code 等)
# 注意:对已有 checkpoint 的 thread 传入完整初始状态, # 由 checkpoint 保管。注意:对已有 checkpoint 的 thread 传
# 其中的 turn_count=0 会覆盖已累计的值。 # 完整初始状态会覆盖已累计的值。
result = await graph.ainvoke( result = await graph.ainvoke(
{"session_id": thread_id, "input_text": text}, {"session_id": thread_id, "input_text": text},
config={"configurable": {"thread_id": thread_id}}, config={"configurable": {"thread_id": thread_id}},
context=deps,
) )
print( print(
f"thread={thread_id} turn_count={result['turn_count']} " f"thread={thread_id} turn_count={result['turn_count']} "
+154
View File
@@ -0,0 +1,154 @@
"""信息核实节点(作业 5,LLM 决策版)。
设计:提示词即程序——核实链的顺序、话术、阈值、全局中断全部
在 ``prompts/single_confirm.md`` / ``dual_confirm.md`` 中(与工作流
chatNode 提示词一致),代码不维护状态链。
节点职责(薄运行时):
1. 把上下文注入提示词:已核实字段(person_fields,模型据此跳过
已确认项)、当前人员序号(person_index,双车两方隔离)、
对话历史、无回复计数;
2. 一次 LLM 调用产出 ``<state>`` 前缀回复;前缀缺失/非法提议
受控修复一次(复用 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
# 提取调用的系统提示标记(测试与分流依据;不以 <state> 开头,
# 故不会被误当对话回复解析)
_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)
+92 -160
View File
@@ -1,19 +1,19 @@
"""LangGraph 事故流程图。 """LangGraph 事故流程图。
架构(对照 ``workflow/20261016/事故信息采集2025118.json``,作业 3 修订): 架构(对照 ``workflow/20261020/事故信息采集2025118.json``):
- 每个主要阶段是独立节点,与工作流 chatNode 一一对应: - 每个主要阶段一个节点,与工作流 chatNode 一一对应:
``collect``(事故信息收集)、``single_photo``(单车拍照)、 ``collect``(事故信息收集)、``single_photo`` / ``dual_photo``
``dual_photo``(双车拍照)、``single_confirm`` / ``dual_confirm`` (拍照)、``single_confirm`` / ``dual_confirm``(信息确认);
(信息确认,作业 5 实现); - 节点间大切换由 ``route_by_stage`` 条件边按 ``stage_code`` 分发
- 节点间的大切换由 ``route_by_stage`` 条件边按 ``stage_code`` 路由 (对照 Condition#2/#7/#9 等);一轮 HTTP 请求 = 一次图执行,
(对照工作流的 Condition#2/#7/#9 等);LLM 接入后由节点内的 跨轮记忆由 checkpoint 承载;
“解析状态”代码写出 ``stage_code``,条件边只做分发,不含业务规则; - 节点内阶段码一律由 LLM 提议(提示词即程序),代码只校验
- 节点内部的阶段推进(拍照步骤、失败/无回复处理、分流)由节点内 (前缀格式、白名单、clientMode 出口)与落库;
LLM 提议目标 ``stage_code``(脚手架期由 ``_fake_llm_propose`` - 依赖注入走 LangGraph 原生机制:``context_schema=GraphDeps``,
模拟),代码按 ``transitions`` 白名单校验、非法提议不写状态; 节点签名 ``(state, runtime: Runtime[GraphDeps])``,调用时
代码只负责事件解析与计数等上下文簿记,不代替模型决定; ``context=GraphDeps(llm=...)``。LLM client 属于运行时依赖,
- LLM 接入前(作业 4 之前),节点共用固定话术模板。 不进 state、不进 checkpoint(计划作业 3-2)。
图结构:: 图结构::
@@ -21,40 +21,43 @@
├→ collect ───────────→ END ├→ collect ───────────→ END
├→ single_photo ──────→ END ├→ single_photo ──────→ END
├→ dual_photo ────────→ END ├→ dual_photo ────────→ END
├→ single_confirm ────→ END (作业 5) ├→ single_confirm ────→ END
├→ dual_confirm ──────→ END (作业 5) ├→ dual_confirm ──────→ END
└→ END (终态 0000—0005) └→ END (终态 0000—0005)
首轮语义(对照工作流 work_start 的 is_first_message → Assigned Reply#3): 离线运行示例(默认注入 workflow fake LLM):
新会话第一条消息由 ``welcome`` 节点以固定话术返回 ``1001``
(未准备好通话),不调用模型;之后用户说【继续办理】或继续对话,
由 collect 节点的 LLM 提议 ``1001 → 1002``(白名单已含该转换)。
离线运行示例:
python -m examples.langgraph_smoke python -m examples.langgraph_smoke
对应测试:test/agent/test_scaffold.py、test/agent/test_events.py、 对应测试: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.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, StateGraph 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 .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): @dataclass
# 新会话首条消息返回 1001,不调用模型。 class GraphDeps:
WELCOME_STAGE = "1001" """图节点的运行时依赖。
WELCOME_TEXT = (
"<state>1001</state>您好,请问您是来办理交通事故信息采集的吗?" 通过 ``context_schema`` 注入(``ainvoke(..., context=GraphDeps(...))``),
"请说“继续办理”开始。" 只存在于调用期,不序列化、不进入 checkpoint。
) """
llm: LLMService
def normalize_input(state: AgentState) -> dict: def normalize_input(state: AgentState) -> dict:
@@ -99,157 +102,79 @@ def route_by_stage(state: AgentState) -> str:
return "collect" return "collect"
def welcome(state: AgentState) -> dict: # 首轮固定问候(对照工作流 Assigned Reply#3):新会话首条消息返回
"""首轮问候节点:新会话第一条消息的固定话术(对照工作流 # 1001,不调用模型;话术与 Variable Update#6 后的记录一致。
work_start is_first_message → Assigned Reply#3)。 WELCOME_STAGE = "1001"
WELCOME_TEXT = (
"<state>1001</state>您好,这里是无锡交警,我将为您远程处理交通事故。"
"请将人员撤离至路侧安全区域,开启危险报警双闪灯、放置三角警告牌、"
"做好安全防护,谨防二次事故伤害。若您已经准备好了,请点击继续办理,"
"如需人工服务,请说转人工。"
)
确定性节点,不调用模型;返回 1001(未准备好通话),
用户确认后由 collect 节点的 LLM 经白名单守卫进入 1002。 def welcome(state: AgentState) -> dict:
"""首轮问候节点(确定性,不调用模型)。
返回 1001(未准备好通话),并把问候写入 dialogue,
供后续节点作为聊天历史注入(对齐 FastGPT 的 chatId 历史)。
""" """
return { return {
"turn_count": (state.get("turn_count") or 0) + 1, "turn_count": (state.get("turn_count") or 0) + 1,
"response_text": WELCOME_TEXT, "response_text": WELCOME_TEXT,
"stage_code": WELCOME_STAGE, "stage_code": WELCOME_STAGE,
"raw_status_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: async def collect_node(state: AgentState, runtime: Runtime[GraphDeps]) -> dict:
"""假 LLM:提出下一步阶段码(作业 4 起由真实 LLM 取代)。 """事故信息收集节点(工作流「事故信息收集」chatNode)。"""
return await run_collect(state, runtime.context.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
def _event_reply(state: AgentState) -> dict: async def single_photo_node(state: AgentState, runtime: Runtime[GraphDeps]) -> dict:
"""事件簿记 + 阶段提议/校验 + 固定话术(作业 3 脚手架版)。 """单车拍照节点(工作流「单车拍照」/「单车拍照#6」chatNode)。"""
return await run_photo(state, runtime.context.llm, kind="single")
- 代码只维护上下文:``no_reply_count`` 累计(NO_REPLY)与
重置(其他输入);
- 阶段码(含失败/无回复处理)一律由 ``_fake_llm_propose`` 提议
(作业 4 换真实 LLM),经白名单校验,非法提议保持当前阶段;
- 轮数 +1,响应带 ``<state>`` 前缀(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"<state>{stage_code}</state>"
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,
}
def collect(state: AgentState) -> dict: async def dual_photo_node(state: AgentState, runtime: Runtime[GraphDeps]) -> dict:
"""事故信息收集节点(工作流“事故信息收集” chatNode)。 """双车拍照节点(工作流「双车拍照」/「双车拍照#7」chatNode)。"""
return await run_photo(state, runtime.context.llm, kind="dual")
TODO(作业 4): 接入 LLM——多轮采集、跳过已知信息、一次补齐分流;
模型回复中的状态前缀由代码校验(解析状态,对照工作流
“解析状态” code 节点),不由模型自由写阶段。
"""
return _event_reply(state)
def single_photo(state: AgentState) -> dict: async def single_confirm_node(state: AgentState, runtime: Runtime[GraphDeps]) -> dict:
"""单车拍照节点(工作流“单车拍照” chatNode)。 """单车信息确认节点(工作流「单车信息确认」chatNode)。"""
return await run_single_confirm(state, runtime.context.llm)
TODO(作业 5): 按 2000→2001→2002→2003→2005 生成对应拍照话术;
步骤推进的确定性部分已在 ``transitions`` 表覆盖。
"""
return _event_reply(state)
def dual_photo(state: AgentState) -> dict: async def dual_confirm_node(state: AgentState, runtime: Runtime[GraphDeps]) -> dict:
"""双车拍照节点(工作流“双车拍照” chatNode)。 """双车信息确认节点(工作流「双车信息确认」chatNode)。"""
return await run_dual_confirm(state, runtime.context.llm)
TODO(作业 5): 按 2010→…→2015 生成对应话术;direct 模式末次
拍摄完成进入 3002(同轮询问第一方身份)。
"""
return _event_reply(state)
def single_confirm(state: AgentState) -> dict: def build_graph(checkpointer=None):
"""单车信息确认节点(工作流“单车信息确认” 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):
"""装配并编译图。 """装配并编译图。
参数: 参数:
checkpointer: 自定义 checkpointer;缺省 ``InMemorySaver``, checkpointer: 自定义 checkpointer;缺省 ``InMemorySaver``,
作业 7 替换为 ``AsyncPostgresSaver``。 作业 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=<LLMService>)`` 注入运行时依赖。
""" """
builder = StateGraph(AgentState) builder = StateGraph(AgentState, context_schema=GraphDeps)
builder.add_node("normalize_input", normalize_input) builder.add_node("normalize_input", normalize_input)
builder.add_node("welcome", welcome) 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("collect", collect_node)
builder.add_node("single_photo", single_photo) builder.add_node("single_photo", single_photo_node)
builder.add_node("dual_photo", dual_photo) builder.add_node("dual_photo", dual_photo_node)
builder.add_node("single_confirm", single_confirm) builder.add_node("single_confirm", single_confirm_node)
builder.add_node("dual_confirm", dual_confirm) builder.add_node("dual_confirm", dual_confirm_node)
builder.add_edge(START, "normalize_input") builder.add_edge(START, "normalize_input")
builder.add_conditional_edges( builder.add_conditional_edges(
@@ -265,10 +190,17 @@ def build_graph(checkpointer=None, llm=None):
END: END, END: END,
}, },
) )
builder.add_edge("welcome", END) for name in (
builder.add_edge("collect", END) "welcome", "collect", "single_photo", "dual_photo",
builder.add_edge("single_photo", END) "single_confirm", "dual_confirm",
builder.add_edge("dual_photo", END) ):
builder.add_edge("single_confirm", END) builder.add_edge(name, END)
builder.add_edge("dual_confirm", END)
return builder.compile(checkpointer=checkpointer or InMemorySaver()) return builder.compile(checkpointer=checkpointer or InMemorySaver())
__all__ = [
"AgentState",
"GraphDeps",
"build_graph",
"workflow_fake_llm",
]
+83
View File
@@ -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": {},
}
+12 -4
View File
@@ -14,18 +14,21 @@
from collections.abc import AsyncIterator 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 from .chat import ChatInput, ChatResult, ChatStreamEvent, ChatBackend, TextDelta
class LangGraphBackend(ChatBackend): class LangGraphBackend(ChatBackend):
"""Adapt a compiled LangGraph to the neutral chat contract.""" """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._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: async def complete(self, chat_input: ChatInput) -> ChatResult:
"""非流式:执行一轮图并翻译结果。 """非流式:执行一轮图并翻译结果。
@@ -40,8 +43,13 @@ class LangGraphBackend(ChatBackend):
(作业 6 起实现)。 (作业 6 起实现)。
""" """
result = await self._graph.ainvoke( 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}}, config={"configurable": {"thread_id": chat_input.session_id}},
context=GraphDeps(llm=self._llm),
) )
return ChatResult( return ChatResult(
content = result["response_text"], content = result["response_text"],
+2 -2
View File
@@ -26,8 +26,8 @@ async def lifespan(app: FastAPI):
from ..agent.llm import build_llm_service from ..agent.llm import build_llm_service
global _langgraph_graph global _langgraph_graph
_langgraph_graph = build_graph(llm=build_llm_service()) _langgraph_graph = build_graph()
_chat_backend = LangGraphBackend(_langgraph_graph) _chat_backend = LangGraphBackend(_langgraph_graph, llm=build_llm_service())
else: else:
_fastgpt_client = AsyncChatClient( _fastgpt_client = AsyncChatClient(
api_key=Config.FASTGPT_API_KEY, api_key=Config.FASTGPT_API_KEY,
+9 -4
View File
@@ -26,7 +26,8 @@ from src.agent.prompts import COLLECTION, load_prompt
def collect_state(**overrides) -> dict: def collect_state(**overrides) -> dict:
state = { state = {
"session_id": "s1", "session_id": "s1",
"input_text": "你好", # 使用完整句而非短促词,避免触发分流守卫(“你好”<=2 字)
"input_text": "发生了交通事故,两辆车追尾",
"stage_code": "1002", "stage_code": "1002",
"turn_count": 1, "turn_count": 1,
"no_reply_count": 0, "no_reply_count": 0,
@@ -138,13 +139,16 @@ async def test_collect_illegal_stage_proposal_not_committed():
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_graph_collect_node_with_llm(): async def test_graph_collect_node_with_llm():
# 图级:build_graph(llm=...) 时首轮走 welcome(不调模型), # 图级:首轮走 welcome(确定性,不调模型);第二轮起 collect
# 第二轮起 collect 节点走 run_collect # 节点经 Runtime 注入的 llm 走 run_collect
from src.agent.graph import GraphDeps
llm = FakeLLMService(lambda s, u: "<state>1002</state>收到。") llm = FakeLLMService(lambda s, u: "<state>1002</state>收到。")
graph = build_graph(llm=llm) graph = build_graph()
first = await graph.ainvoke( first = await graph.ainvoke(
{"session_id": "s1", "input_text": "追尾了"}, {"session_id": "s1", "input_text": "追尾了"},
config={"configurable": {"thread_id": "s1"}}, config={"configurable": {"thread_id": "s1"}},
context=GraphDeps(llm=llm),
) )
assert len(llm.calls) == 0 # welcome 是确定性节点 assert len(llm.calls) == 0 # welcome 是确定性节点
assert first["stage_code"] == "1001" assert first["stage_code"] == "1001"
@@ -152,6 +156,7 @@ async def test_graph_collect_node_with_llm():
result = await graph.ainvoke( result = await graph.ainvoke(
{"session_id": "s1", "input_text": "继续办理"}, {"session_id": "s1", "input_text": "继续办理"},
config={"configurable": {"thread_id": "s1"}}, config={"configurable": {"thread_id": "s1"}},
context=GraphDeps(llm=llm),
) )
assert len(llm.calls) == 1 assert len(llm.calls) == 1
assert result["response_text"].startswith("<state>1002</state>") assert result["response_text"].startswith("<state>1002</state>")
+193
View File
@@ -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: "<state>3001</state>好的,请问您的姓名是什么?"
)
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 "<state>3001</state>好的,请问您的姓名是什么?"
if q == "张三":
return "<state>3001</state>张三您好,请问您的身份证号码后四位是否正确?"
if q == "是对的":
return "<state>3001</state>请问您的手机号码后四位是否正确?"
return "<state>0000</state>好的,信息已全部确认无误。感谢您的配合。"
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("<state>0000</state>")
@pytest.mark.asyncio
async def test_deny_collects_correct_value():
llm = make_llm(
lambda q: "<state>3001</state>抱歉,请告诉我您正确的身份证号码后四位。",
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: "<state>3001</state>请问您是否是这辆车的车主?请回答“是”或“不是”。",
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: "<state>3002</state>您好。请问这辆车是由您驾驶的吗?",
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 "<state>3002</state>好的,驾驶员您好。请问您的姓名是什么?"
return "<state>3002</state>请问您的身份证号码后四位是否正确?"
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: "<state>0000</state>好的,两位驾驶员的信息均已核实完毕。感谢配合。",
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"
+116
View File
@@ -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"<state>{target}</state>模型生成的本轮拍照引导。"
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: "<state>2012</state>请确认当前操作。")
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: "<state>0005</state>正在转接人工。")
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"<state>{target}</state>对应模式的结束话术。"
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([
"<state>2015</state>您可以点击手机界面中间的拍照按钮。",
"<state>3002</state>请问第一辆车是由您驾驶的吗?",
])
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"<state>{target}</state>非法提议")
with pytest.raises(CollectReplyError):
await invoke(build_graph(), "illegal", "继续", llm=llm,
stage_code=source, client_mode=mode)
assert len(llm.calls) == 2
+3 -2
View File
@@ -11,15 +11,16 @@
import pytest 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 传完整初始状态, # 只传本轮变化的字段;对已有 checkpoint 的 thread 传完整初始状态,
# 其中的 turn_count=0 会覆盖已累计的值(LangGraph 输入按通道覆盖)。 # 其中的 turn_count=0 会覆盖已累计的值(LangGraph 输入按通道覆盖)。
return await graph.ainvoke( return await graph.ainvoke(
{"session_id": thread_id, "input_text": text}, {"session_id": thread_id, "input_text": text},
config={"configurable": {"thread_id": thread_id}}, config={"configurable": {"thread_id": thread_id}},
context=GraphDeps(llm=llm or workflow_fake_llm()),
) )
+19 -8
View File
@@ -12,8 +12,9 @@
import pytest import pytest
import src.agent.graph as graph_mod from src.agent.llm import FakeLLMService
from src.agent.graph import build_graph
from src.agent.graph import GraphDeps, build_graph, workflow_fake_llm
from src.agent.transitions import ( from src.agent.transitions import (
TERMINAL_STAGES, TERMINAL_STAGES,
TransitionSpec, 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 = {"session_id": thread_id, "input_text": text}
payload.update(initial) payload.update(initial)
return graph.ainvoke( return graph.ainvoke(
payload, payload,
config={"configurable": {"thread_id": thread_id}}, 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 @pytest.mark.asyncio
async def test_illegal_proposal_keeps_stage(monkeypatch): async def test_illegal_proposal_raises_and_state_untouched():
# LLM 提议非法跳步(2001 → 2003):校验拒绝,状态不被污染 # LLM 坚持非法跳步(2001 → 2003):白名单拒绝 + 一次修复后
# 仍非法 → CollectReplyError,非法阶段不写状态
from src.agent.collect import CollectReplyError
llm = FakeLLMService(lambda s, u: "<state>2003</state>跳步")
graph = build_graph() graph = build_graph()
monkeypatch.setattr(graph_mod, "_fake_llm_propose", lambda s, e: "2003") with pytest.raises(CollectReplyError):
result = await invoke(graph, "t-illegal", "【拍摄完成】", stage_code="2001") await invoke(graph, "t-illegal", "【拍摄完成】", llm=llm, stage_code="2001")
assert result["stage_code"] == "2001" assert len(llm.calls) == 2
# 会话未被污染:下一轮用守规矩的 fake 继续,阶段仍是 2001
good = FakeLLMService(lambda s, u: "<state>2002</state>已收到。")
result = await invoke(graph, "t-illegal", "【拍摄完成】", llm=good, stage_code="2001")
assert result["stage_code"] == "2002"
@pytest.mark.asyncio @pytest.mark.asyncio
+101
View File
@@ -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: "<state>2000</state>信息已齐,请拍摄车辆前方。")
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(["缺少前缀", "<state>9999</state>非法阶段"])
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": "<state>1002</state>请问有几辆车?"},
]
llm = FakeLLMService(lambda s, u: "<state>1002</state>请问您还在现场吗?")
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="<state>1002</state>收到。"))
]))
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": "本轮输入"},
]