diff --git a/examples/langgraph_smoke.py b/examples/langgraph_smoke.py
new file mode 100644
index 0000000..143d0ce
--- /dev/null
+++ b/examples/langgraph_smoke.py
@@ -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())
diff --git a/requirements.txt b/requirements.txt
index 62fbfeb..64f244a 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -21,3 +21,4 @@ requests
 sqlalchemy
 pymysql
 PyYAML>=6.0
+langgraph==1.2.9
diff --git a/src/agent/__init__.py b/src/agent/__init__.py
new file mode 100644
index 0000000..0bf7c46
--- /dev/null
+++ b/src/agent/__init__.py
@@ -0,0 +1 @@
+"""LangGraph accident-flow agent (作业 1 起逐步交付)。"""
diff --git a/src/agent/graph.py b/src/agent/graph.py
new file mode 100644
index 0000000..8ca0152
--- /dev/null
+++ b/src/agent/graph.py
@@ -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())
diff --git a/src/agent/state.py b/src/agent/state.py
new file mode 100644
index 0000000..89b58eb
--- /dev/null
+++ b/src/agent/state.py
@@ -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
diff --git a/test/agent/test_scaffold.py b/test/agent/test_scaffold.py
new file mode 100644
index 0000000..dba661f
--- /dev/null
+++ b/test/agent/test_scaffold.py
@@ -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"]