Eric Wang
6d0a2169c8
- 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
117 lines
5.1 KiB
Python
117 lines
5.1 KiB
Python
"""拍照节点回归：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
|