18 LangGraph — 有状态 Agent 运行时 从入门到手写 Agent

学习理念：LangGraph 是 2026 年生产级 Agent 的事实标准。LangChain 是"链式思维"（直线或 DAG），LangGraph 是图式思维（节点 + 边 + 状态 + 循环）。它是 Ch16 LangChain 的升级方向——官方已明确：LangChain 1.0 的 create_agent 底层跑的就是 LangGraph。学完本章，你将理解 Agent 的底层机制，并能手写一个生产级的 Agent 运行时。

海外对标：LangGraph ↔ Temporal（分布式持久化执行）/ OpenAI Assistant API（有状态 Agent）/ Vercel AI SDK（流式 Agent）。LangGraph 是目前开源生态中唯一同时具备有状态图执行、持久化检查点、Human-in-the-loop 原生支持的 Agent 运行时。

本节 AI 替代率：~60% | 人工干预率：~40%

角色	能力范围
🤖 AI 擅长	生成 LangGraph 代码骨架、解释 State/Node/Edge 概念、写 Reducer 函数
👤 人类需理解	图式思维模型（状态机 + 图遍历）、Reducer 的选择策略、interrupt 的幂等性陷阱、条件边的路由逻辑

技术栈健康度速览（OPC 7.4）

技术	健康度	说明
LangGraph	🔥 巅峰	生产级有状态 Agent 运行时，2026 年增长最快的 LLM 框架
Checkpointer (InMemory/SQLite/Postgres)	🔥 巅峰	持久化检查点，Durable Execution 基石
StateGraph + Reducer	🔥 巅峰	核心设计模式，理解后端状态管理的人 1 天上手
interrupt / Human-in-the-loop	🔥 巅峰	LangGraph 相对于竞品的最大差异化能力
Node Stream + Event Stream	🟢 稳定	三层流式（Token / State / Event）
LangChain Agent (AgentExecutor)	🟡 已过峰值	已被 LangGraph 替代，本章就是教你手写替代品

---

§0 ⚡ 开篇：从链式思维到图式思维

为什么你需要 LangGraph

LangGraph vs LangChain vs 裸 SDK

维度	裸 SDK	LangChain	LangGraph
思维模型	函数调用	线性管道 (LCEL)	有向图 (StateGraph)
状态管理	无——自己维护	无——靠外部 Memory	一等公民——TypedDict + Reducer
循环能力	手写 while 循环	AgentExecutor 有缺陷	一等公民——Node → Condition → Node
Human-in-the-loop	手写中断逻辑	无	原生——interrupt() + Command(resume=True)
持久化	手写	Checkpointer（LangGraph 借用）	Checkpointer（核心内置）
流式输出	SDK 原生	Token 流	Token + State + Event 三层流
多 Agent	手写编排	尴尬	Subgraph + Swarm + RemoteGraph
生产就绪	✅ 最高	🟡 抽象层有摩擦	✅ 为 Durable Execution 设计

从 Chain 到 Graph 的思维转变

Chain 思维（Ch16）:                    Graph 思维（Ch18）:
Prompt → LLM → Output                  State ↗ Node → Node → Node ↘
                                            ↑                    ↓
                                      Condition ← ← ← ←  Condition

给程序员的比喻：

• Chain = 一个顺序执行的函数列表（[func1, func2, func3]）
• Graph = 一个有限状态机（State Machine）——有 State、有 State 更新规则（Reducer）、有状态转移条件（Conditional Edge）
• 如果你写过 Redux（前端）/ Spring State Machine（后端）/ BPMN 工作流引擎，LangGraph 的思维模型你 1 天上手

---

一、🟢 快速入门：5 行搭一个图（Day01 前半）

1.1 三节点并联示例

from typing import TypedDict
from langgraph.graph import StateGraph
from langgraph.constants import START, END

# 1. 定义状态
class MyState(TypedDict):
    query: str
    rag_result: str
    web_result: str
    answer: str

# 2. 定义节点
def rag_node(state: MyState):
    return {"rag_result": f"知识库检索: {state['query']}"}

def web_node(state: MyState):
    return {"web_result": f"网络检索: {state['query']}"}

def llm_node(state: MyState):
    return {"answer": f"综合RAG和网络结果生成回答"}

# 3. 构建图
builder = StateGraph(MyState)
builder.add_node(rag_node)
builder.add_node(web_node)
builder.add_node(llm_node)

# 4. 连接边（rag 和 web 并联 → llm）
builder.add_edge(START, "rag_node")
builder.add_edge(START, "web_node")
builder.add_edge("rag_node", "llm_node")
builder.add_edge("web_node", "llm_node")
builder.add_edge("llm_node", END)

# 5. 编译并执行
graph = builder.compile()
result = graph.invoke({"query": "什么是LangGraph"})
print(result["answer"])

🔑 与 LCEL 的对应关系：

• StateGraph ≈ RunnableParallel（多个源合并到一个目标）
• add_node ≈ | 管道中的每个步骤
• add_edge ≈ | 管道符的连接
• LangGraph 多了 State，这是 LCEL 没有的

1.2 核心概念速览

概念	English	本质	比喻
状态	State	图的全局数据，TypedDict 定义	全局变量（但有 Reducer 保护）
节点	Node	一个处理函数：State → State 的部分更新	函数：读 state，写回部分字段
边	Edge	节点之间的连接	函数调用关系
条件边	Conditional Edge	根据 State 决定下一个节点	if-else 分支
Reducer	Reducer Function	定义"多个写入冲突时怎么合并"	数据库的 Merge 策略
Checkpointer	Checkpoint Saver	持久化每次超步后的 State	Git commit
Interrupt	interrupt()	暂停图执行，等人类审批	代码中的断点

1.3 🤔 并联节点是怎么执行的？（Fan-in 行为）

§1.1 示例中，rag_node 和 web_node 并联后都指向 llm_node。初学者常问：rag 和 web 是同时执行吗？llm_node 什么时候触发？

答案是：LangGraph 按 SuperStep 分层执行，每层的所有节点并行执行**，全部完成后才进入下一层。**

关键规则：

• 同一 SuperStep 内的节点并行执行：rag_node 和 web_node 同时运行，互不等待
• 所有前驱节点全部完成 → 合并 State → 触发下游节点：llm_node 必须等 rag 和 web 都返回后才触发
• Reducer 在每层结束时调用一次：rag 返回 {"rag_result":"..."}，web 返回 {"web_result":"..."}，Reducer 将它们合并到全局 State 中

🤔 如果 rag 快、web 慢，llm_node 会执行两次吗？ 不会。LangGraph 的 BFS 调度会等当前 SuperStep 所有节点完成后再推进到下一层。同一个 State 字段被多个节点写，由 Reducer 决定合并策略。

---

二、🔥 状态系统：LangGraph 的核心设计（Day01 后半）

2.1 State：图的"全局变量"

LangGraph 中，State 是图中所有节点共享的唯一数据源。每个节点读 State → 处理 → 返回部分更新。

from typing import TypedDict

class MyState(TypedDict):
    query: str              # 无 Reducer = 覆盖策略
    messages: list          # 有 Reducer = 合并策略

2.2 Reducer：解决"写入冲突"

当多个节点写入同一个 State 字段时（如并联节点），Reducer 决定如何合并：

三种 Reducer 策略

Reducer 类型	定义方式	行为	适用场景
默认（override）	不加 Annotated	new_value 直接覆盖旧值	单一来源的字段（如 query）
追加（operator.add）	Annotated[list, add]	列表拼接 list1 + list2	messages 列表
自定义	Annotated[type, func]	自定义合并逻辑	需要特殊合并策略的数据

from typing import TypedDict, Annotated, List
from operator import add
from langchain.messages import AnyMessage, HumanMessage, AIMessage

# 1. 默认 Reducer（覆盖）
class MyState(TypedDict):
    query: str          # 多次写入 = 最后一次覆盖之前

# 2. operator.add（追加——最常用）
class AgentState(TypedDict):
    messages: Annotated[List[AnyMessage], add]  # 多次写入 = 列表拼接

# 3. 自定义 Reducer
def my_message_reducer(left: List[AnyMessage], right: List[AnyMessage]) -> List[AnyMessage]:
    """去重后再追加——防止同一个消息被多个来源重复添加"""
    existing_ids = {m.id for m in left}
    new_messages = [m for m in right if m.id not in existing_ids]
    return left + new_messages

class CustomState(TypedDict):
    messages: Annotated[List[AnyMessage], my_message_reducer]

💡 Reducer 的调用时机：LangGraph 在每个 SuperStep（一批可并行的节点执行完后）调用 Reducer，将各节点的增量更新合并到全局 State 中。Reducer 是幂等的——同一个 State 多次经过 Reducer 结果一致。

2.5 ⚡ 深入：SuperStep 执行算法

LangGraph 的底层执行算法是 BFS（广度优先）分层调度，而不是简单的顺序执行。理解它就能解释很多"奇怪"的行为。

SuperStep 的核心规则：

规则	说明	为什么重要
同一层并行	当前 SuperStep 的所有节点同时执行	不关心执行顺序，只关心最终合并结果
全部完成才推进	所有节点都返回后，Reducer 合并，再进入下一层	不会出现"部分执行就触发下游"的情况
执行编号 + Checkpoint	每次 SuperStep 是一个编号，每个 Checkpoint 保存的是该编号结束后的 State	时间旅行 = 回溯到某个执行编号
条件边 = 动态决定下一层	条件边函数在当前 SuperStep 所有节点完成后执行，决定下一层包含哪些节点	分支逻辑是在"整层结束后"才判断

🤔 常见困惑：条件边是在每个节点后判断，还是整层后判断？ 条件边是在当前 SuperStep 所有节点都完成后执行的。所以一个节点返回后不会立即跳转——要等整层全部跑完。这避免了"一个节点跑得快就提前跳走"的问题。

2.3 Input / Output Schema

控制图的输入和输出——输入只接受需要的字段，输出只暴露需要的字段：

class AgentInput(TypedDict):
    query: str

class AgentOutput(TypedDict):
    answer: str

class AgentState(TypedDict):
    query: str
    rag_result: str
    web_result: str
    answer: str

builder = StateGraph(
    state_schema=AgentState,    # 内部完整状态
    input_schema=AgentInput,    # 外部输入只接受 query
    output_schema=AgentOutput,  # 外部输出只暴露 answer
)

2.4 Checkpointer：持久化与时间旅行

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.checkpoint.sqlite import SqliteSaver
# from langgraph.checkpoint.postgres import PostgresSaver  # 生产级

# 选择 Checkpointer
# checkpointer = InMemorySaver()              # 内存（开发/测试）
checkpointer = SqliteSaver.from_conn_string("agent.db")  # SQLite（单机生产）
# checkpointer = PostgresSaver(...)           # Postgres（分布式生产）

graph = builder.compile(checkpointer=checkpointer)

# 每次 invoke 需要传入 thread_id（会话 ID）
result = graph.invoke(
    {"query": "什么是LangGraph"},
    config={"configurable": {"thread_id": "session_001"}}
)

# ✅ 同一 thread_id 恢复对话上下文
result = graph.invoke(
    {"query": "我刚才问了什么？"},
    config={"configurable": {"thread_id": "session_001"}}
)

Checkpointer 的核心能力

能力	方法	说明
状态持久化	invoke(config)	每个 SuperStep 后自动保存 State
状态历史	graph.get_state_history(config)	查看所有历史状态（时间旅行）
状态恢复	graph.get_state(config, checkpoint_id)	回到任意历史状态
分支执行	graph.invoke(Command(resume=...))	从分支点继续执行

Checkpointer 类型	持久化	适合
InMemorySaver	❌ 内存	开发测试
SqliteSaver	✅ 本地文件	单机生产
PostgresSaver	✅ 远程 DB	分布式生产

---

三、🔥 节点与边（Day02 前半）

3.1 节点的能力

# 1. 节点输入：接收完整 State，只读取自己需要的字段
def my_node(state: MyState):
    query = state["query"]
    return {"result": f"处理: {query}"}

# 2. 节点输出：返回部分 State 更新（Reducer 会合并它）
#    不能直接修改 state 并 return state——必须返回 {"field": value}

# 3. 节点缓存（幂等性保证）
builder.add_node(my_node)  # 默认不缓存
# Node Cache + Checkpointer 配合实现幂等重试

# 4. 节点重试（失败自动重试）
from langgraph.graph import add_node
builder.add_node(
    "risky_node", 
    risky_function,
    retry_policy={
        "max_attempts": 3,
        "retry_on": [ConnectionError, TimeoutError],
        "backoff_factor": 2.0,  # 指数退避
    }
)

3.2 条件边（Conditional Edge）

def route_decision(state: MyState) -> str:
    """根据状态决定下一个节点"""
    if state["requires_search"]:
        return "web_search_node"
    elif state["has_rag_result"]:
        return "llm_node"
    else:
        return "__end__"

# 注册条件边
builder.add_conditional_edges(
    source="router_node",
    path=route_decision,  # 返回目标节点名称
)

给程序员的比喻：条件边 = 后端路由——根据请求参数（State）分发到不同的 Handler（Node）。

3.3 Node 的 runtime 参数

在 ToolNode 的定义中有一个隐藏参数 runtime：

class ToolNode:
    def __call__(self, state, runtime):  # ← runtime 是什么？
        ...

runtime 是 LangGraph 自动注入的运行时上下文对象，包含：

• 配置信息：当前执行的 config（含 thread_id、recursion_limit 等）
• 内存存储：Store 对象（用于跨会话的长期记忆）
• 执行状态：当前 SuperStep 编号、Checkpoint ID 等

初学者不需要手动使用 runtime——它是给 Middleware 和 Checkpointer 等基础设施用的。你只需要知道：函数签名写上 state, runtime 两个参数，LangGraph 会自动传值。

参数	谁传的	你什么时候需要手动用
state	LangGraph 自动注入当前 State	每次都需要——这是你的业务数据
runtime	LangGraph 自动注入运行时配置	基本不需要——只有写底层框架时需要

3.4 递归限制（防止无限循环）

# LangGraph 默认 recursion_limit=25
# 如果图中有循环边，超过限制会抛出异常

graph = builder.compile()
graph.invoke(
    {"query": "循环测试"},
    {"recursion_limit": 50}  # 自定义限制
)

3.4 流式输出

# 1. 节点级别的流式输出（Node.stream）
# 在节点的回调中 yield 中间结果

# 2. 图级别的流式输出（三种模式）
for event in graph.stream(
    {"query": "你好"},
    stream_mode="updates"   # 每步返回节点更新
):
    print(event)

# stream_mode 可选：
# "values"    → 每次 SuperStep 后的完整 State
# "updates"   → 每次 SuperStep 后的增量更新 ← 最常用
# "messages"  → Token 级别的消息流（类似 LLM 流式）

---

四、🔥 Human-in-the-loop（Day02 中段）

4.1 interrupt()：暂停图执行

from langgraph.types import interrupt, Command

def approval_node(state: MyState):
    """需要人类审批的节点"""
    
    # 暂停执行，等待人类输入
    user_input = interrupt({
        "query": state["query"],
        "msg": "是否确认进行知识库查询？"
    })
    
    if user_input:
        return {"rag_result": "知识库查询结果"}
    else:
        return {"rag_result": "用户取消查询"}

4.2 完整 HIL 流程

4.3 ⚠️ interrupt 的幂等性陷阱

def dangerous_node(state: MyState):
    """不幂等的节点——interrupt 恢复时会再次执行"""
    
    # ❌ 危险操作：每次执行都会插入一条记录
    db.insert_log("开始查询")  # <- interrupt 恢复后再次执行! 重复插入!
    
    user_input = interrupt({"确认信息": state["data"]})
    result = db.search(user_input)
    return {"result": result}

解决方案：把非幂等操作放在 interrupt 之前独立的节点中，或确保操作是幂等的（如 INSERT IF NOT EXISTS）。

---

五、🔥 手写 create_agent（Day02 后半）

这是本章的"毕业设计"——用 LangGraph 从零手写一个与 LangChain 兼容的 Agent 运行时。

5.1 架构设计

5.2 核心代码

from langchain_openai import ChatOpenAI
from langchain.tools import tool
from langgraph.graph import StateGraph
from langgraph.checkpoint.base import BaseCheckpointSaver
from typing import TypedDict, Annotated, List
from langchain.messages import AnyMessage, AIMessage, ToolMessage
from operator import add
from concurrent.futures import ThreadPoolExecutor

# ── 1. 状态定义 ──
class AgentState(TypedDict):
    messages: Annotated[List[AnyMessage], add]

# ── 2. ToolNode（解析 tool_call → 并行执行 → 返回 ToolMessage）──
class ToolNode:
    def __init__(self, tools: list):
        self.name_to_tool = {t.name: t for t in tools}
        
    def __call__(self, state, runtime):
        last_message = state["messages"][-1]  # AIMessage with tool_calls
        with ThreadPoolExecutor(max_workers=3) as executor:
            results = executor.map(self._run_one, last_message.tool_calls)
        return {"messages": list(results)}
    
    def _run_one(self, tool_call: dict) -> ToolMessage:
        try:
            result = self.name_to_tool[tool_call["name"]].invoke(tool_call["args"])
            return ToolMessage(content=result, tool_call_id=tool_call["id"])
        except Exception as e:
            return ToolMessage(
                content=f"工具调用失败: {e}",
                tool_call_id=tool_call["id"]
            )

# ── 3. Middleware 基类 ──
class BaseMiddleware:
    def before_agent(self, state) -> dict: return {}
    def before_model(self, state) -> dict: return {}

# ── 4. 构建 Agent 图 ──
def build_agent(model, tools, system_prompt, middleware_list, checkpointer):
    tool_node = ToolNode(tools)
    model = model.bind_tools(tools)
    
    def model_node(state: AgentState):
        ai_msg = model.invoke(state["messages"])
        return {"messages": [ai_msg]}
    
    # 条件边：有 tool_calls → tool_node，否则结束
    def should_continue(state: AgentState) -> str:
        last_msg = state["messages"][-1]
        if isinstance(last_msg, AIMessage) and last_msg.tool_calls:
            return "tool_node"
        return "__end__"
    
    # 构建图
    builder = StateGraph(AgentState)
    builder.add_node(model_node)
    builder.add_node("tool_node", tool_node)
    
    # 如果有 middleware，添加 middleware 节点
    if middleware_list:
        def before_agent_node(state):
            update = {}
            for mw in middleware_list:
                update.update(mw.before_agent(state))
            return update
        
        def before_model_node(state):
            update = {}
            for mw in middleware_list:
                update.update(mw.before_model(state))
            return update
        
        builder.add_node(before_agent_node)
        builder.add_node(before_model_node)
        builder.add_edge("__start__", "before_agent_node")
        builder.add_edge("before_agent_node", "before_model_node")
        builder.add_edge("before_model_node", "model_node")
    else:
        builder.add_edge("__start__", "model_node")
    
    builder.add_conditional_edges("model_node", should_continue)
    builder.add_edge("tool_node", "model_node")  # 循环：tool → model
    
    return builder.compile(checkpointer=checkpointer)

# ── 5. 使用 ──
@tool
def get_weather(city: str) -> str:
    """获取城市天气"""
    return f"{city}天气晴朗"

agent = build_agent(
    model=ChatOpenAI(model="gpt-4o-mini"),
    tools=[get_weather],
    system_prompt="你是餐厅助手",
    middleware_list=[CurrentDateMiddleware()],
    checkpointer=InMemorySaver()
)

result = agent.invoke(
    {"messages": [{"role": "user", "content": "北京天气如何？"}]},
    config={"configurable": {"thread_id": "1"}}
)
print(result["messages"][-1].content)

5.3 这个 Agent 与 LangChain create_agent 的对应关系

LangChain create_agent	我们的手写实现	说明
内部图结构 StateGraph	StateGraph(AgentState)	完全一致
model.bind_tools()	model = model.bind_tools(tools)	完全一致
工具调用执行器	ToolNode（线程池并行）	更轻量，无额外依赖
middleware=[...]	before_agent_node + before_model_node	概念一致
condition edge	should_continue()	完全一致
Checkpointer	checkpointer	完全一致

🔑 学完这个手写实现，你就理解了 LangChain create_agent 的底层原理——它就是一个 StateGraph + ToolNode + ConditionalEdge。没有魔法，全是 StateGraph。

---

技术栈深度评估

编号	技术	健康度	说明
T1	LangGraph StateGraph	🔥 巅峰	生产级 Agent 运行时事实标准
T2	Reducer (operator.add)	🔥 巅峰	LangGraph 最核心的设计模式
T3	Checkpointer (InMemory/SQLite)	🔥 巅峰	持久化基石
T4	interrupt + Command	🔥 巅峰	差异化能力——最吸引生产团队的 Feature
T5	Conditional Edge	🔥 巅峰	图式思维的核心表达
T6	ToolNode（手写）	🟢 稳定	理解原理即可，生产直接用框架的
T7	Middleware（手写）	🟢 稳定	LangChain 1.0 已原生支持
T8	Postgres Checkpointer	⏳ 成长期	分布式生产级持久化

海外对标

企业/项目	应用场景	技术方案
OpenAI	有状态 Agent	assistants.run() + Threads（闭源）
Temporal	分布式工作流	Durable Execution（Go/Java，非 LLM 专用）
Vercel AI SDK	流式 Agent	轻量级 Agent SDK（TypeScript）
CrewAI	多 Agent 编排	角色化 Agent 抽象（Python）
Microsoft AutoGen	多 Agent 对话	Agent 聊天模式（Python）

企业痛点映射

痛点	传统方案	AI Agent 方案	效率提升
多步 Agent 中途失败需重跑	从头执行，浪费资源	Checkpointer 从中断处恢复	节省 80% 重跑成本
敏感操作需要人工确认	全部暂停→审批→恢复，手写逻辑	interrupt() 原生支持	开发成本降低 90%
复杂业务需要条件路由	if-else 硬编码	Conditional Edge 状态驱动	可维护性提升 10x
Agent 行为难以调试	print 大法	State History + Snapshot 时间旅行	调试效率提升 5x

---

学习路径

优先级	内容	时间	说明
🔥	§0 开篇（LangGraph vs LCEL 对比）	5 min	先读——理解为什么需要图式思维
🔥	§一 快速入门（三节点示例）	5 min	搭起来看看图长什么样
🔥	§二 状态系统（State + Reducer + Checkpointer）	15 min	核心——60% 的价值在这里
🔥	§五 手写 create_agent	15 min	毕业设计——把概念组装成完整 Agent
🟢	§三 节点与边	10 min	细节都在这
🟢	§四 Human-in-the-loop	10 min	最吸引生产团队的 Feature

---

AI 协作指南

本文档看完后，以下问题直接问 AI 插件：

Q: "这个手写 create_agent 怎么添加 Stream+Event 流式输出？"
Q: "LangGraph 的 PostgresCheckpointer 怎么配置？"
Q: "我的场景需要多 Agent 协作，LangGraph 的 Swarm 模式怎么用？"
Q: "Reducer 中 operator.add 和自定义 Reducer 怎么选？"

AI 能做的：
  - 把本文章 5 手写 Agent 扩展到支持流式输出
  - 生成 PostgresCheckpointer 的 Docker Compose 配置
  - 用 LangGraph Subgraph 实现多 Agent 编排
  - 解释 LangGraph 的 SuperStep 执行算法细节

---

🛠️ 从"看代码"到"跑起来"

1️⃣ 初学者常见错误排查

错误信息	原因	解决方案
KeyError: 'field_x'	节点试图读取 State 中不存在的字段	检查 TypedDict 定义是否包含该字段，或拼写是否一致
RecursionLimitError	图中存在无限循环（条件边始终返回同一个节点）	检查条件边函数是否有终止路径，或调大 recursion_limit
CheckpointerError: thread_id required	使用 Checkpointer 后未传入 thread_id	在 config={"configurable": {"thread_id": "..."}} 中传入
InterruptException	中断后恢复时使用了错误的 thread_id	恢复时必须使用与中断时相同的 thread_id
ValueError: State field conflict	并联节点返回了同一个字段，且该字段没有 Reducer	给该字段加 Annotated[type, add] 或自定义 Reducer
恢复执行后工具被调用了两次	interrupt 节点不幂等——恢复时会从头执行	将非幂等操作移到 interrupt 之前独立的节点中
graph.get_state_history() 返回空	Checkpointer 未正确配置，或使用了 InMemorySaver（进程重启后丢失）	改用 SqliteSaver 持久化到文件

2️⃣ 环境检测

我的环境是 [Windows/Python 3.12]，
这个项目依赖 langgraph, langchain-openai, pydantic,
帮我生成 requirements.txt。

2️⃣ 快速验证

# 按顺序跑，5 分钟从入门到手写 Agent
python 02_get_started/get_started_demo.py          # §一 快速入门
python 03_states/03_reducer_demo.py               # §二 Reducer
python 04_nodes/06_node_human_in_the_loop.py      # §四 HIL
python 06_create_agent_demo/agent.py              # §五 手写 Agent

3️⃣ 代码审查

帮我审查 agent.py 中的 build_agent 函数：
1. ToolNode 的线程池并行执行是否正确？
2. 条件边 should_continue 的逻辑有没有边界情况？
3. Middleware 的状态合并逻辑是否完整？
4. Checkpointer 的 thread_id 管理是否合理？

---

附录：原始资料处理说明

原始文件	处理方式
尚硅谷大模型技术之LangGraphV1.1.0.docx	内容已整合到本文档
3代码/LangGraphDemo/02_get_started/	快速入门代码已提取到 §一
3代码/LangGraphDemo/03_states/（10 个文件）	Reducer / Checkpointer 核心代码已提取到 §二
3代码/LangGraphDemo/04_nodes/（6 个文件）	节点能力已整合到 §三
3代码/LangGraphDemo/05_edges/（2 个文件）	条件边已整合到 §三
3代码/LangGraphDemo/06_create_agent_demo/（4 个文件）	完整手写 Agent 已提取到 §五
视频（day01~day02，共 ~50 个）	跳过

---

修复情况：

• ✅ 技术栈健康度标签（OPC 7.4）
• ✅ 🟢🟡🟠🔴 优先级颜色
• ✅ Mermaid 图（决策树 / 状态更新流程 / HIL 时序 / create_agent 类图 / 对比图）
• ✅ LangGraph vs LangChain 逐项对比表（§0 + §五）
• ✅ 手写 create_agent 完整代码 ~80 行（§五）
• ✅ interrupt 幂等性陷阱标注（§四）
• ✅ 三种 Reducer 策略 + 自定义 Reducer 示例（§二）
• ✅ 海外对标 + 企业痛点映射
• ✅ 🛠️ 实操指引
• ✅ AI 协作指南
• ✅ 学习路径表（最短 60 分钟）