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用于消息列表的可累加更新。你也可以定义自己的 reducerfrom 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 最核心的编程模型。
