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
71 lines
2.4 KiB
Python
71 lines
2.4 KiB
Python
"""作业 1 验收测试：LangGraph 最小图。
|
||
|
||
通过标准（对应迁移计划作业 1）：
|
||
- 同一会话（thread）轮数累计：1 → 2；
|
||
- 不同会话相互隔离：新 thread 的轮数从 1 开始；
|
||
- 不依赖 FastGPT 与真实模型，离线可重复执行。
|
||
|
||
注意：节点实现（src/agent/graph.py 中的 TODO）完成前，
|
||
本文件的测试会失败，属于预期状态。
|
||
"""
|
||
|
||
import pytest
|
||
|
||
from src.agent.graph import GraphDeps, build_graph, workflow_fake_llm
|
||
|
||
|
||
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()),
|
||
)
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_same_thread_accumulates_turns() -> None:
|
||
graph = build_graph()
|
||
|
||
first = await ask(graph, "thread-a", "你好")
|
||
second = await ask(graph, "thread-a", "发生了交通事故")
|
||
|
||
assert first["turn_count"] == 1
|
||
assert second["turn_count"] == 2
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_different_threads_are_isolated() -> None:
|
||
graph = build_graph()
|
||
|
||
await ask(graph, "thread-a", "你好")
|
||
await ask(graph, "thread-a", "发生了交通事故")
|
||
other = await ask(graph, "thread-b", "你好")
|
||
|
||
assert other["turn_count"] == 1
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_first_turn_is_welcome_stage() -> None:
|
||
# 新会话首条消息：welcome 节点固定问候，返回 1001（对照工作流
|
||
# work_start is_first_message → Assigned Reply#3），不调用模型
|
||
graph = build_graph()
|
||
result = await ask(graph, "thread-a", "你好")
|
||
|
||
assert result["stage_code"] == "1001"
|
||
assert result["response_text"].startswith("<state>1001</state>")
|
||
assert result["turn_count"] == 1
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_second_turn_keeps_stage_without_event() -> None:
|
||
# 问候后的普通输入：脚手架版无事件则保持 1001，
|
||
# 【继续办理】等事件经白名单进入 1002（见 test_transitions.py）
|
||
graph = build_graph()
|
||
await ask(graph, "thread-a", "你好")
|
||
result = await ask(graph, "thread-a", "我出车祸了")
|
||
|
||
assert result["stage_code"] == "1001"
|
||
assert "我出车祸了" in result["response_text"]
|