57 lines
1.7 KiB
Python
57 lines
1.7 KiB
Python
|
"""作业 1 验收测试：LangGraph 最小图。
|
||
|
|
|
||
|
|
通过标准（对应迁移计划作业 1）：
|
||
|
|
- 同一会话（thread）轮数累计：1 → 2；
|
||
|
|
- 不同会话相互隔离：新 thread 的轮数从 1 开始；
|
||
|
|
- 不依赖 FastGPT 与真实模型，离线可重复执行。
|
||
|
|
|
||
|
|
注意：节点实现（src/agent/graph.py 中的 TODO）完成前，
|
||
|
|
本文件的测试会失败，属于预期状态。
|
||
|
|
"""
|
||
|
|
|
||
|
|
import pytest
|
||
|
|
|
||
|
|
from src.agent.graph import build_graph
|
||
|
|
|
||
|
|
|
||
|
|
async def ask(graph, thread_id: str, text: str) -> dict:
|
||
|
|
# 只传本轮变化的字段；对已有 checkpoint 的 thread 传完整初始状态，
|
||
|
|
# 其中的 turn_count=0 会覆盖已累计的值（LangGraph 输入按通道覆盖）。
|
||
|
|
return await graph.ainvoke(
|
||
|
|
{"session_id": thread_id, "input_text": text},
|
||
|
|
config={"configurable": {"thread_id": thread_id}},
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
@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_reply_uses_fixed_template_with_stage_prefix() -> None:
|
||
|
|
graph = build_graph()
|
||
|
|
|
||
|
|
result = await ask(graph, "thread-a", "你好")
|
||
|
|
|
||
|
|
assert result["stage_code"] == "1002"
|
||
|
|
assert result["response_text"].startswith("<state>1002</state>")
|
||
|
|
assert "你好" in result["response_text"]
|