feat(agent): add LangGraph scaffold for assignment 1

- src/agent/state.py: minimal five-field AgentState
- src/agent/graph.py: START -> normalize_input -> reply -> END,
  compiled with InMemorySaver; reply emits <state>1002</state> prefix
- examples/langgraph_smoke.py: offline smoke, turns 1/2/1 verified
- test/agent/test_scaffold.py: turn accumulation, thread isolation,
  fixed-template assertions (3 passed)
- requirements.txt: pin langgraph==1.2.9

Known limit: InMemorySaver state disappears on process exit.
This commit is contained in:
Eric Wang
2026-10-19 13:32:43 +08:00
parent a9a1393f64
commit 5020903791
6 changed files with 211 additions and 0 deletions
+38
View File
@@ -0,0 +1,38 @@
"""LangGraph 脚手架冒烟脚本(作业 1)。
同一 thread 输入两次、另一个 thread 输入一次,打印轮数和文本。
预期输出轮数依次为 1、2、1。
本脚本不导入 FastGPT、不读取其密钥;离线即可运行:
python examples/langgraph_smoke.py
"""
import asyncio
from src.agent.graph import build_graph
async def main() -> None:
graph = build_graph()
async def ask(thread_id: str, text: str) -> None:
# 只传本轮变化的字段;会话字段(turn_count 等)由 checkpoint 保管。
# 注意:对已有 checkpoint 的 thread 传入完整初始状态,
# 其中的 turn_count=0 会覆盖已累计的值。
result = await graph.ainvoke(
{"session_id": thread_id, "input_text": text},
config={"configurable": {"thread_id": thread_id}},
)
print(
f"thread={thread_id} turn_count={result['turn_count']} "
f"stage_code={result['stage_code']} text={result['response_text']}"
)
await ask("thread-a", "你好")
await ask("thread-a", "发生了交通事故")
await ask("thread-b", "你好")
if __name__ == "__main__":
asyncio.run(main())
+1
View File
@@ -21,3 +21,4 @@ requests
sqlalchemy
pymysql
PyYAML>=6.0
langgraph==1.2.9
+1
View File
@@ -0,0 +1 @@
"""LangGraph accident-flow agent (作业 1 起逐步交付)。"""
+93
View File
@@ -0,0 +1,93 @@
"""LangGraph 最小脚手架(作业 1)。
图结构:START → normalize_input → reply → END
骨架已装配好:状态、节点注册、边和编译(默认 InMemorySaver)。
两个节点的具体逻辑留待实现(见函数体内的 TODO)。
离线运行示例:
python examples/langgraph_smoke.py
对应测试:test/agent/test_scaffold.py
"""
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, StateGraph
from .state import AgentState
# 脚手架阶段固定返回的阶段码;作业 3 起由转换表驱动。
DEFAULT_STAGE_CODE = "1002"
def normalize_input(state: AgentState) -> dict:
"""归一化本轮输入,为回复节点准备上下文。
TODO(作业 1): 实现该节点。建议职责:
- 校验/清理 ``state["input_text"]``(例如去除首尾空白);
- 返回需要写入状态的字段增量,例如
``{"input_text": cleaned}``;
- 保持纯函数:不读全局状态、不做 IO。
"""
cleaned = state["input_text"].strip()
return {"input_text": cleaned}
def reply(state: AgentState) -> dict:
"""生成固定模板回复,并把轮数加一。
TODO(作业 1): 实现该节点。验收要求:
- 每次调用使 ``turn_count`` 恰好 +1;首轮会话没有历史
checkpoint,用 ``state.get("turn_count") or 0`` 起步;
- ``response_text`` 使用固定模板,**包含本轮输入文本**
(``state["input_text"]``);
- 响应文本需带公共协议前缀,例如
``<state>1002</state>...``(``stage_code`` 用
``DEFAULT_STAGE_CODE``,API 层负责解析该前缀);
- ``stage_code`` 写入 ``DEFAULT_STAGE_CODE``。
参考实现::
turn_count = (state.get("turn_count") or 0) + 1
response_text = (
f"<state>{DEFAULT_STAGE_CODE}</state>"
f"已收到:{state['input_text']}"
)
return {
"turn_count": turn_count,
"response_text": response_text,
"stage_code": DEFAULT_STAGE_CODE,
}
"""
turn_count = (state.get("turn_count") or 0) + 1
response_text = (
f"<state>{DEFAULT_STAGE_CODE}</state>"
f"已收到:{state['input_text']}"
)
return {
"turn_count": turn_count,
"response_text": response_text,
"stage_code": DEFAULT_STAGE_CODE,
}
def build_graph(checkpointer=None):
"""装配并编译最小图。
参数:
checkpointer: 自定义 checkpointer;缺省使用 ``InMemorySaver``。
作业 7 会替换为 ``AsyncPostgresSaver``。
返回:
编译后的图。调用方通过
``config={"configurable": {"thread_id": <会话 ID>}}``
区分会话。
"""
builder = StateGraph(AgentState)
builder.add_node("normalize_input", normalize_input)
builder.add_node("reply", reply)
builder.add_edge(START, "normalize_input")
builder.add_edge("normalize_input", "reply")
builder.add_edge("reply", END)
return builder.compile(checkpointer=checkpointer or InMemorySaver())
+22
View File
@@ -0,0 +1,22 @@
"""图状态定义(作业 1)。
最小状态只包含脚手架所需的五个字段;作业 3 会扩充领域状态
(阶段、事故字段、人员字段、澄清计数、本轮 patch 等)。
所有字段必须可序列化,不得保存 HTTP client、数据库连接等资源。
"""
from typing import TypedDict
class AgentState(TypedDict):
"""一次对话轮次在图中流动的状态。
``turn_count`` 由同一 ``thread_id`` 的 checkpoint 承载,
节点每轮只负责 +1,不负责跨会话记忆。
"""
session_id: str
turn_count: int
input_text: str
response_text: str
stage_code: str
+56
View File
@@ -0,0 +1,56 @@
"""作业 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"]