最简单的工具 Agent用一个 while 循环就能写出来while True: ai_message model.invoke(messages) messages.append(ai_message) if not ai_message.tool_calls: return messages for tool_call in ai_message.tool_calls: result run_tool(tool_call) messages.append(result)这段代码没有错。模型决定是否调用工具工具结果放回消息再让模型继续判断ReAct 的最小闭环已经成立。但只要需求再向前一步循环就会迅速膨胀两个工具要并行执行状态怎样合并middleware 要在 model 前后插入节点顺序怎样表达某个工具要人工审批进程重启后怎样继续streaming、checkpoint、cache 和 tracing 应该挂在哪里子 Agent 如何拥有独立作用域又能回到父执行链某一步失败后是重试节点、回到模型还是终止整次运行LangChain 没有继续给 while 循环叠加更多 if而是让create_agent()变成一个小型编译器它把模型、工具、middleware、状态协议和路由规则编译成StateGraph再交给 LangGraph 的 Pregel 运行时执行。这不是为了把流程画得更好看而是把状态、并行、暂停恢复和可观察性变成运行时的一等能力。图 1create_agent 如何把配置编译成可执行 StateGraph一、create_agent()返回的不是 AgentExecutor从函数签名看create_agent()的返回类型已经给出答案CompiledStateGraph[ AgentState, ContextT, InputAgentState, OutputAgentState, ]它最终仍然实现 Runnable 接口所以调用方式看起来很熟悉agent.invoke(inputs) agent.stream(inputs) await agent.ainvoke(inputs)但执行内核已经不是“一个函数调用另一个函数”而是StateGraph builder - 校验节点与边 - 把 state 字段转换成 channels - 把节点转换成 Pregel actors - 把边转换成触发关系 - 生成 CompiledStateGraphStateGraph只是构建器不能直接运行只有compile()之后才得到支持 invoke、stream、checkpoint、interrupt 和状态查询的执行对象。二、图的第一块基石不是节点而是状态 schema普通循环通常直接修改一个 dict 或消息列表。图运行时不能接受这种隐式共享因为多个节点可能在同一步并行写入同一个字段。StateGraph要求先声明状态 schema。它把每个字段转换成一个 channel并决定这个字段存什么类型 节点提交的是完整值还是增量 多个更新如何合并 该字段能否出现在输入和输出中LangChain 的默认AgentState很精炼class AgentState(TypedDict): messages: Annotated[list[AnyMessage], add_messages] jump_to: Annotated[JumpTo | None, EphemeralValue, PrivateStateAttr] structured_response: Annotated[ResponseT, OmitFromInput]三个字段对应三种完全不同的 channel 语义。三、messages为什么必须带 reducermessages不是普通 LastValue而是使用add_messagesreducer。模型节点可能提交{messages: [AIMessage(...)]}工具节点可能在下一步并行提交{messages: [ToolMessage(...)]}middleware 还可能通过额外Command追加或替换一条带相同 id 的消息。如果每个更新都直接覆盖列表最后只会剩下某个节点的结果如果所有节点都原地 append共享写入又会引入顺序和并发问题。reducer 把规则写进状态类型节点只提交自己的增量屏障阶段统一执行add_messages(old, update)。这样消息历史怎样合并不再由每个节点临时决定。这也是图运行时能检测冲突的原因。测试中两个Command在同一个 super-step 同时写入没有 reducer 的structured_response会触发InvalidUpdateError而同时写入messages可以按 reducer 合并。四、jump_to为什么是临时且私有的jump_to同时带有EphemeralValue PrivateStateAttr它不是业务记忆而是 middleware 给下一条条件边的临时控制信号{jump_to: end}EphemeralValue避免旧路由指令长期留在状态中影响后续步骤PrivateStateAttr则让它同时从公开输入和输出 schema 中消失。调用者不应该在最终结果里看到一个内部跳转标记也不应该把上一次运行遗留的jump_to当成新输入。状态 schema 在这里同时承担了类型协议和安全边界。五、输入、内部状态与输出并不是同一张表create_agent()会分别生成StateSchema 完整内部状态 InputSchema 调用者允许传入的字段 OutputSchema 调用者最终能看到的字段OmitFromInput、OmitFromOutput和PrivateStateAttr决定字段在哪些边界出现。例如structured_response可以出现在最终输出却不能让调用者在输入时伪造middleware 的私有计数器可以参与节点执行但不进入输入或结果。多个 middleware 还能各自贡献state_schema。源码按照注册顺序合并字段最后放入调用者显式传入的state_schema因此冲突时后者拥有最终决定权。这意味着 middleware 不只是“回调函数集合”还可以给整张图增加受 reducer 管理的共享数据通道。六、节点只做一件事读取快照返回局部更新StateGraph的节点契约可以概括为State - PartialState节点不需要复制完整状态也不应该直接修改全局对象。它读取这一轮可见的快照只返回自己要提交的字段。create_agent()至少会加入model构造ModelRequest、执行模型、解析结构化输出tools由ToolNode执行客户端工具middleware.before_agent整次运行前执行middleware.before_model每轮模型调用前执行middleware.after_model每轮模型调用后执行middleware.after_agent整次运行结束前执行。同步和异步实现会被包装进同一个RunnableCallable节点。运行时根据调用入口选择对应执行路径图结构本身不需要维护两套版本。七、model 节点为什么返回Command列表模型调用完成后源码不会直接修改 state而是先把ModelResponse转成Command( update{ messages: model_response.result, structured_response: ..., } )如果多层wrap_model_callmiddleware 还返回了额外状态更新这些Command会按“模型结果在前、middleware 内层到外层在后”的顺序一起返回。这样每一份更新都会经过 channel reducer而不是在 wrapper 调用栈里悄悄修改 dict。这里还有一条有意保留的边界wrap model middleware 返回的Command目前只允许更新状态goto、resume和跨图graph参数会明确报错。路由应使用生命周期 hook 的jump_to恢复应走图运行时的 interrupt/resume 协议。换句话说Command虽然是图级控制对象但不同扩展点只开放自己能够安全承诺的子集。八、边把“接下来做什么”从节点代码中拿了出来普通循环里模型执行、工具执行和路由判断常常挤在同一个函数中。图中则分成节点这一步产生什么更新 边更新提交后下一步去哪里create_agent()先计算四个位置entry_node 整次运行入口可能包含 before_agent loop_entry_node 每轮入口可能包含 before_model loop_exit_node 每轮出口可能包含 after_model exit_node 整次出口可能包含 after_agent然后再添加静态边和条件边。model 出口的条件边会依次判断是否存在 middleware 的jump_to是否还有尚未回应的普通 tool call是否已经获得structured_response是否需要回到模型重试否则结束。tools 出口则判断return_direct、结构化工具结果或回到下一轮模型。流程控制因此成为可检查、可绘制、可插入 interrupt 的图结构而不是埋在某个巨大函数里的分支。九、Send解决的不是跳转而是动态并行一条普通条件边通常返回一个节点名return tools但一条 AIMessage 可能同时包含多个普通工具调用。源码会为每个 pending call 生成return [ Send(tools, [tool_call_a]), Send(tools, [tool_call_b]), ]这里不是静态创建tools_a、tools_b两个节点而是在运行时向同一个tools节点扇出多份任务。每个任务只携带自己的 tool call不再复制整份 Agent state。ToolNode 在执行时通过 channel read 给ToolRuntime.state补齐状态快照。这避免工具数量增加时把完整 state 重复写入每个 task 的开销。Send因此同时解决了两个问题图结构保持稳定调用数量可以动态变化每个并行任务拥有独立输入和 tool call id又能读取同一轮状态快照。图 2两个并行工具如何跨三个 super-step 合并回 Agent 状态十、Pregel 为什么要把执行分成 super-stepCompiledStateGraph的运行内核基于 Pregel也就是 Bulk Synchronous Parallel 模型。每个 super-step 分成三阶段Plan 根据上一轮更新确定本轮要运行哪些节点或任务 Execution 并行执行已选任务 Update 等任务完成后通过 channel/reducer 统一提交写入最关键的一条规则是同一个 super-step 中产生的 channel 更新在该 step 完成前对其他并行任务不可见。假设模型一次调用工具 A 和工具 Bstep N model 产生两个 tool calls 屏障提交 AIMessage step N1 tool A 与 tool B 读取同一份已提交 state 两者并行执行 任何一方都看不到兄弟工具尚未提交的更新 屏障用 reducer 合并两个结果 step N2 model 同时看到 A、B 的 ToolMessage这正是上一篇提到的“并行工具快照语义”。它不是 ToolNode 自己加锁实现的而是图运行时的 step 边界天然保证。十一、屏障让并行结果可预测也让错误更明确如果多个并行节点都更新同一个字段channel 类型决定结果channel 语义同一步多次更新add_messages等 reducer按规则聚合BinaryOperatorAggregate用二元操作连续合并LastValue通常只允许一个更新冲突会报错EphemeralValue只服务临时 step 信号这比“最后完成的线程覆盖前一个线程”可靠得多。图运行时不会假装所有字段都能安全合并。你必须在 schema 中明确表达聚合规则没有规则却发生并发写入时失败比静默丢数据更正确。十二、checkpoint 保存的不是聊天记录而是图的执行位置消息列表只能回答“聊了什么”不能完整回答哪个节点已经完成 下一步准备运行哪些任务 哪次 interrupt 正在等待恢复 各个 channel 的版本是什么compile(checkpointer...)把版本化 checkpoint 接入图运行时。调用时使用thread_id标识一条独立执行线config {configurable: {thread_id: conversation-42}} result agent.invoke(inputs, config)同一个 thread 可以继续累积状态也可以从暂停点恢复或读取历史快照。这就是为什么人工审批不能只用input()阻塞进程。interrupt()会把待审批数据写入可持久化执行状态外部系统稍后用 resume 命令继续原进程不需要一直存活。十三、checkpointer、store 和 cache 不是同一种存储create_agent()在 compile 阶段同时接受三类后端能力作用checkpointer保存单个 thread 的状态版本与执行位置支持暂停、恢复、回放store跨 thread 的长期数据例如用户记忆或共享资料cache复用节点执行结果减少重复计算三者不能互相替代。把用户偏好放进 checkpoint会被限制在某个 thread把待恢复的节点位置只写入 store运行时又不知道从哪一步继续cache 更不能承担业务真相。图编译把这些能力安装到统一运行时但数据生命周期仍然通过不同抽象保持分离。十四、interrupt、debug、streaming 为什么都适合挂在图上compile()还接收interrupt_before / interrupt_after debug name transformers它们都依赖明确的节点或事件边界interrupt 要知道停在谁之前或之后debug 要展示每一步选中了哪些节点、写入了什么streaming 要区分 messages、tools、values、tasks 等通道transformer 要按 graph namespace 创建独立实例子 Agent 要把嵌套图的生命周期挂回父 scope。如果执行只是一段 while 代码这些能力只能靠侵入业务函数实现。图把边界显式化后它们可以由运行时统一提供。十五、静态图并不意味着 Agent 失去动态性很多人看到 graph 会担心模型行为明明不可预测为什么能提前画图因为静态的是能力边界动态的是每次运行的选择静态有哪些节点、合法目标、状态字段、reducer、interrupt 位置 动态模型生成几个 tool calls、条件边选哪个目标、Send 扇出几个任务middleware 的can_jump_to在编译期声明可能跳到 model、tools 或 endjump_to在运行时选择本次目标ToolNode 只有一个静态节点Send可以动态创建任意数量的调用任务。图不是预先写死每一步结果而是先定义合法执行空间再让运行时数据在空间中选择路径。十六、compile()是从声明到运行时的真正分界线StateGraph.compile()会完成这些工作校验节点、边、branch 与 interrupt 目标 确定 input / output / stream channels 把 state 字段实例化为 channel 把节点附着为 Pregel actors 把边与条件分支附着为触发关系 安装 checkpointer / store / cache 生成 CompiledStateGraphcreate_agent()随后还会绑定默认配置较高的 recursion limit、LangChain Agent integration metadata以及可选的 Agent name。从这一步开始Agent 才同时拥有 Runnable 的调用表面和图运行时的状态能力。十七、Agent 变成图真正换掉的是执行模型把整条源码主线压缩起来模型、工具、middleware、response format - 合并 state / input / output schema - 创建 model、tools 与 hook 节点 - 编译静态边、条件边和 Send 分支 - reducer 定义状态合并 - Pregel super-step 执行与屏障提交 - checkpoint / interrupt / stream 接管生命周期普通 while 循环以调用栈为中心代码运行到哪里执行状态就在哪里。StateGraph 以版本化状态和任务为中心节点可以结束进程可以退出之后仍能根据 checkpoint 知道已经发生什么、下一步应该做什么。这就是 LangChain Agent 选择 LangGraph 的根本原因。图不是 ReAct 循环的装饰层而是让并行、恢复、状态契约和扩展边界同时成立的执行基础。学AI大模型的正确顺序千万不要搞错了2026年AI风口已来各行各业的AI渗透肉眼可见超多公司要么转型做AI相关产品要么高薪挖AI技术人才机遇直接摆在眼前有往AI方向发展或者本身有后端编程基础的朋友直接冲AI大模型应用开发转岗超合适就算暂时不打算转岗了解大模型、RAG、Prompt、Agent这些热门概念能上手做简单项目也绝对是求职加分王给大家整理了超全最新的AI大模型应用开发学习清单和资料手把手帮你快速入门学习路线:✅大模型基础认知—大模型核心原理、发展历程、主流模型GPT、文心一言等特点解析✅核心技术模块—RAG检索增强生成、Prompt工程实战、Agent智能体开发逻辑✅开发基础能力—Python进阶、API接口调用、大模型开发框架LangChain等实操✅应用场景开发—智能问答系统、企业知识库、AIGC内容生成工具、行业定制化大模型应用✅项目落地流程—需求拆解、技术选型、模型调优、测试上线、运维迭代✅面试求职冲刺—岗位JD解析、简历AI项目包装、高频面试题汇总、模拟面经以上6大模块看似清晰好上手实则每个部分都有扎实的核心内容需要吃透我把大模型的学习全流程已经整理好了抓住AI时代风口轻松解锁职业新可能希望大家都能把握机遇实现薪资/职业跃迁这份完整版的大模型 AI 学习资料已经上传CSDN朋友们如果需要可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费】