1. 为什么先理解状态与节点
LangGraph 是 LangChain 生态中用于构建有状态、可循环、可控制流程的 Agent 框架。和普通的大模型单次调用不同，它把一次复杂任务拆成一张「图」：图里有多个节点，节点之间通过边连接，数据则存放在共享的状态中。很多初学者直接用现成脚手架跑通 Demo，一旦遇到多步推理、工具调用、人工确认、循环重试等场景就会失控，根本原因往往是没有真正理解两件事：状态 State 如何流转，以及节点 Node 如何读写和更新状态。
本教程会从最小可运行示例出发，逐步拆解状态定义、节点写法、更新机制、条件路由和循环控制，并给出完整可运行的 Python 代码。读完本文后，你应当能独立设计一张结构清晰、行为可控的 LangGraph 工作流图。
2. LangGraph 的三要素：状态、节点、边
一张 LangGraph 图由三个核心部分组成：
- State（状态）：整个图共享的数据容器，节点读取它、返回更新字段，LangGraph 负责把更新合并回状态。它决定了「数据在节点之间如何传递」。
- Node（节点）：一个可执行函数，接收当前状态并返回需要更新的字段。它代表一个具体的处理步骤，例如调用大模型、执行工具、做判断。
- Edge（边）：节点之间的连接关系，决定执行顺序。普通边固定指向下一个节点，条件边则根据状态动态选择目标节点。
可以把状态理解为「共享内存」，节点理解为「处理单元」，边理解为「流转规则」。其中状态与节点是最需要花时间掌握的，下面逐步拆解。
3. 快速上手：最小可运行示例
先安装依赖：
pip install langgraph langchain-core下面是只有一个节点的最小示例，它的作用是演示状态如何被写入、读取和合并：
from typing import TypedDict from langgraph.graph import StateGraph, START, END class State(TypedDict): message: str count: int def node_a(state: State) -> dict: print("节点 A 开始执行") return {"message": "来自节点 A 的消息", "count": state["count"] + 1} builder = StateGraph(State) builder.add_node("a", node_a) builder.add_edge(START, "a") builder.add_edge("a", END) graph = builder.compile() result = graph.invoke({"message": "初始消息", "count": 0}) print(result)运行后输出大致如下：
节点 A 开始执行 {'message': '来自节点 A 的消息', 'count': 1}需要关注三点：第一，状态通过TypedDict声明了字段和类型；第二，节点函数只返回需要更新的字段，LangGraph 会把它与旧状态合并，未返回的字段保持不变；第三，START和END是系统提供的虚拟节点，分别表示图的入口和出口。
4. 状态 State 详解
4.1 什么是状态
状态是贯穿整张图的数据结构。LangGraph 在每一步调用节点前，都会把当前完整状态传给节点；节点执行结束后，LangGraph 读取节点返回的字典，并把其中的字段合并回状态。状态不要求每个节点都完整返回全部字段，只需要返回发生了变化的字段即可。
LangGraph 支持多种方式定义状态，最常用的是TypedDict和 Pydantic 模型。
4.2 使用 TypedDict 定义状态
TypedDict是轻量级方案，适合明确列出少量字段的中小项目。定义时可以直接给每个字段标注类型：
from typing import TypedDict, Annotated, Literal import operator from langgraph.graph import StateGraph, START, END class ChatState(TypedDict): question: str answer: str messages: Annotated[list, operator.add]这里question和answer是普通字符串字段，默认采用「后写覆盖」的更新方式：如果多个节点都返回了answer，最后写入的值会覆盖前面的值。messages则使用了Annotated[list, operator.add]，表示这个字段的更新方式是「累加」，多个节点追加的消息不会互相覆盖，而会拼接成一条长列表。这个机制被称为Reducer。
4.3 使用 Pydantic 模型定义状态
当状态字段较多、需要默认值和运行时校验时，Pydantic 模型比TypedDict更合适：
from pydantic import BaseModel from langgraph.graph import StateGraph, START, END class ChatState(BaseModel): question: str = "" answer: str = "" retry_count: int = 0 def answer_node(state: ChatState) -> dict: # 这里可以接入真实大模型，示例中先返回固定内容 return {"answer": f"已收到问题：{state.question}", "retry_count": state.retry_count + 1} builder = StateGraph(ChatState) builder.add_node("answer", answer_node) builder.add_edge(START, "answer") builder.add_edge("answer", END) graph = builder.compile() result = graph.invoke({"question": "什么是 LangGraph？"}) print(result)Pydantic 状态的好处是：可以声明默认值，避免启动时缺少字段报错；同时在节点返回数据时会按字段类型做校验。需要注意的是，Pydantic 状态默认同样采用「按字段浅合并」的更新方式，未返回的字段保持原值。
4.4 状态更新规则与自定义 Reducer
默认情况下，节点返回的新值会覆盖状态中的同名字段；如果字段使用Annotated[类型, reducer]声明，更新时就调用该 reducer 合并新旧值。最常用的内置 reducer 是operator.add，用于消息列表的可累加更新。你也可以定义自己的 reducer：
from typing import Annotated, TypedDict def keep_latest(existing, new): # 如果新值非空就覆盖，否则保留旧值 return new if new else existing class WorkState(TypedDict): messages: Annotated[list, keep_latest] score: int使用自定义 reducer 可以精确控制字段在节点之间如何合并，例如只保留最后一条消息、去重、取最大最小值等。下面的表格总结了常见的状态更新方式：
| 状态定义 | 更新行为 | 典型场景 |
|---|---|---|
TypedDict普通字段 | 后写覆盖 | 单一结果字段，如最终答案 |
Annotated[list, operator.add] | 列表累加 | 多轮对话消息 |
| 自定义 reducer | 按自定义逻辑合并 | 去重、取最新、聚合 |
| Pydantic 模型 | 浅合并并校验 | 字段多且有默认值 |
5. 节点 Node 详解
5.1 节点的基本形态
节点本质上是一个接收状态并返回更新字典的函数。同步函数和异步函数都可以作为节点：
import asyncio from langgraph.graph import StateGraph, START, END 同步节点 def sync_node(state: dict) -> dict: return {"step": "sync"} 异步节点 async def async_node(state: dict) -> dict: await asyncio.sleep(0.1) return {"step": "async"} builder = StateGraph(dict) builder.add_node("first", sync_node) builder.add_node("second", async_node) builder.add_edge(START, "first") builder.add_edge("first", "second") builder.add_edge("second", END) graph = builder.compile()LangGraph 在编译时会识别节点函数是同步还是异步，并统一处理执行方式。同步节点默认在线程池中运行，异步节点则直接等待协程完成。对于需要调用网络接口、大模型或工具的节点，推荐使用异步函数。
5.2 节点的输入与输出
每个节点函数的第一个位置参数是当前状态。返回值必须是一个字典，字典里的键对应需要更新的状态字段。节点可以只返回部分字段：
def step_one(state: ChatState) -> dict: # 只更新 question，不触碰其他字段 return {"question": state["question"].strip()}如果节点不需要修改状态，只想执行副作用，可以返回空字典。不过更推荐通过合理划分节点，让每个节点职责单一：要么处理数据，要么调用外部服务，要么做流程判断。
5.3 条件路由函数
条件路由函数在形式上也类似节点，但它的返回值不是状态更新，而是「下一个节点的名称」。它通过add_conditional_edges绑定到某个节点之后，决定后续路径：
from typing import Literal def route_after_classify(state: ChatState) -> Literal["code_node", "general_node"]: if state["category"] == "code": return "code_node" return "general_node"条件路由函数接收当前状态，返回一个字符串，该字符串必须与add_conditional_edges中映射表的键对应。这是实现「根据状态走不同分支」的核心手段。
6. 边 Edge 与流程流转
边决定节点执行的先后顺序。LangGraph 提供三种常用边：
- 普通边：
add_edge("a", "b")，表示从节点 a 固定流转到节点 b。 - 条件边：
add_conditional_edges("a", router, mapping)，根据路由函数返回值选择目标。 - 起点和终点：
START表示图入口，END表示图出口。
下图展示了一个带分类路由的流程结构：
flowchart TD START((START)) --> classify[classify 分类节点] classify -->|category = code| code_node[code_node 代码节点] classify -->|category = general| general_node[general_node 通用节点] code_node --> END((END)) general_node --> END((END))理解状态、节点和边的关系后，就能构造出任意复杂的图。下面通过完整实战把它们串起来。
7. 完整实战：带分类路由的智能问答流程
这个示例构建一个简易的小助手：先对问题做分类，再根据类别走不同的回答节点。它完整演示了状态定义、节点读写、条件路由和结果合并：
from typing import TypedDict, Literal, Annotated import operator from langgraph.graph import StateGraph, START, END class AssistState(TypedDict): query: str category: str answer: str messages: Annotated[list, operator.add] def classify_node(state: AssistState) -> dict: query = state["query"] if "代码" in query or "bug" in query.lower(): return {"category": "code"} return {"category": "general"} def code_node(state: AssistState) -> dict: return { "answer": "这是代码类问题。建议先贴出报错信息和关键代码片段，我们可以逐步定位。", "messages": [{"role": "assistant", "content": "已进入代码解答分支"}], } def general_node(state: AssistState) -> dict: return { "answer": "这是通用问题。我会结合上下文给出尽量清晰的解答。", "messages": [{"role": "assistant", "content": "已进入通用解答分支"}], } def route_by_category(state: AssistState) -> Literal["code_node", "general_node"]: if state["category"] == "code": return "code_node" return "general_node" builder = StateGraph(AssistState) builder.add_node("classify", classify_node) builder.add_node("code_node", code_node) builder.add_node("general_node", general_node) builder.add_edge(START, "classify") builder.add_conditional_edges( "classify", route_by_category, {"code_node": "code_node", "general_node": "general_node"}, ) builder.add_edge("code_node", END) builder.add_edge("general_node", END) graph = builder.compile() result = graph.invoke({"query": "我的代码报错了，帮我看看", "category": "", "answer": "", "messages": []}) print(result["category"]) print(result["answer"]) print(result["messages"])执行流程如下：classify节点先根据问题文本更新category字段；随后条件路由函数读取该字段，决定进入code_node还是general_node；最终节点写入answer，并通过operator.add把消息追加到messages列表。这个结构已经具备一个简易 Agent 的基础形态。
8. 进阶实战：工具调用与循环
真实 Agent 通常需要在「思考」和「执行工具」之间循环，直到满足停止条件。下面的示例演示如何让工具节点执行后回到助手节点，并通过条件边控制循环次数：
from typing import TypedDict, Annotated, Literal import operator from langgraph.graph import StateGraph, START, END class AgentState(TypedDict): messages: Annotated[list, operator.add] def assistant_node(state: AgentState) -> dict: # 模拟大模型生成下一步：这里固定请求调用工具两次 return {"messages": [{"role": "assistant", "content": "我需要调用工具获取天气信息"}]} def tool_node(state: AgentState) -> dict: # 模拟工具执行结果 return {"messages": [{"role": "tool", "content": "天气结果：晴，22 摄氏度"}]} def should_continue(state: AgentState) -> Literal["tool", "end"]: # 当消息数量小于 3 时继续循环，否则结束 if len(state.get("messages", [])) < 3: return "tool" return "end" builder = StateGraph(AgentState) builder.add_node("assistant", assistant_node) builder.add_node("tool", tool_node) builder.add_edge(START, "assistant") builder.add_conditional_edges( "assistant", should_continue, {"tool": "tool", "end": END}, ) builder.add_edge("tool", "assistant") graph = builder.compile() result = graph.invoke({"messages": []}) print(result["messages"])这段代码的关键在于tool节点执行完后通过普通边回到assistant节点，形成循环。循环由should_continue条件函数控制：当消息数量达到阈值时返回end，路由到END结束。把工具调用从固定数据换成真实的搜索、代码执行、数据库查询接口，就能升级为可用的 Agent。
9. 常见问题与最佳实践
状态字段不是越多越好。字段过多会让节点之间耦合加重，建议按「输入数据、中间过程、最终结果」三个层次组织。
下面是在实际开发中最常遇到的几个问题：
- 节点返回了不存在的字段：如果返回字典中含有未在状态中声明的键，LangGraph 会忽略或报错，务必保证字段一致。
- 条件路由返回了映射表之外的键：路由函数的返回值必须在映射表中存在，否则运行时会报 KeyError。
- 误用 reducer 导致消息丢失：多轮对话场景一定要使用
Annotated[list, operator.add]，否则后写入的消息会覆盖前面的消息。 - 循环缺少终止条件：只要图中有环，就必须在条件边中提供清晰、可达的退出条件，否则图会无限运行。
开发时的最佳实践建议：先用文字画出流程图，再编写代码；每个节点只做一件事；状态字段保持稳定命名；循环逻辑单独写成条件函数并加上注释。这样当流程变复杂时，图仍然易于理解和维护。
10. 总结
本文从状态、节点、边三个核心概念出发，完整讲解了 LangGraph 的基础工作机制，并给出了最小示例、分类路由实战和工具调用循环三个可运行代码。掌握之后，你可以继续延伸学习以下方向：
- 使用
SendAPI 实现并行分支； - 引入 checkpointer 实现断点续跑和多轮会话持久化；
- 结合 LangChain Tool 定义真实工具并接入大模型完成自主 Agent。
状态决定数据如何流转，节点决定每一步做什么，边决定流程如何走。把这三者理解透彻，就掌握了 LangGraph 最核心的编程模型。
