本文深入解析LangGraph的底层执行引擎通过状态机设计实现AI的智能控制流。文章详细介绍了状态机三要素状态、节点、边在LangGraph中的应用以及Reducer的核心机制如何实现状态更新。此外还探讨了条件边、并行执行和编译产物的相关知识帮助读者全面理解LangGraph的工作原理为构建复杂的AI应用打下坚实基础。上一篇我们把 LangChain Memory 模块拆了个底朝天搞清楚了截断、总结、检索三种记忆方案各自的适用场景。这一篇我们进入 LangGraph 的核心地带——它的底层执行引擎。你有没有想过为什么同样是调用 LLM用了 LangGraph 之后AI 就能「想一步、停一步、判断一步」就能在工具调用和 LLM 推理之间来回切换还不会乱套答案就藏在它的状态机设计里。01 先说说 LangChain 的老问题LangChain 的早期架构是线性的——Chain。用户输入 → PromptTemplate → LLM → OutputParser → 输出这个模型干净利落但有个致命弱点控制流是固定的。一旦你想实现「如果 LLM 觉得需要查数据库就查不需要就跳过」或者「出错了重试三次」你会发现原来的 Chain 根本放不下这些逻辑。要么把判断逻辑硬编码在节点里要么用 Python 的 if/else 在外层手动管流程——乱成一锅粥。现实中的 Agent 场景根本不是线性的调工具 → 看结果 → 决定调下一个还是直接回答生成草稿 → 自我审查 → 不行就重写多轮对话 → 根据意图走不同分支这些都需要带状态的循环控制流。这正是 LangGraph 解决的核心问题。02 状态机是什么用最简单的例子解释状态机State Machine不是什么新概念。你手机的蓝牙就是个状态机[关闭] --打开-- [扫描中] --发现设备-- [配对中] --成功-- [已连接] | --失败-- [扫描中]三个核心要素┌─────────────────────────────────────────────┐ │ 状态机三要素 │ ├──────────┬──────────────────────────────────┤ │ State │ 当前的数据快照系统处于什么情况│ │ Node │ 执行动作并更新 State │ │ Edge │ 决定下一步去哪个 Node │ └──────────┴──────────────────────────────────┘LangGraph 就是把这套机制套在了 LLM 上State 对话历史 工具结果 中间变量Node LLM 调用 / 工具执行 / 业务逻辑Edge 普通跳转 / 条件分支Conditional Edge03 StateGraph 执行引擎它的心脏长什么样打开 LangGraph 的源码Python 版langgraph/graph/state.pyJS 版同理你会看到StateGraph类维护了几个核心数据结构┌──────────────────────────────────────────────────┐ │ StateGraph 内部 │ ├─────────────────┬────────────────────────────────┤ │ nodes │ Map名称, 函数 │ │ edges │ Mapfrom, to[] │ │ conditional_ │ Mapfrom, (state)节点名 │ │ edges │ │ │ channels/ │ 每个状态字段的 Reducer │ │ schema │ │ └─────────────────┴────────────────────────────────┘当你调用graph.compile()时它做了什么compile() 阶段 │ ├─ 验证图结构有没有孤立节点有没有到 END 的路径 ├─ 构建邻接表预计算每个节点的后继 ├─ 初始化 Channels每个状态字段注册 Reducer └─ 返回 CompiledGraph可以 invoke/streamcompile()本质上是把你「描述的图」变成「可执行的调度器」。04 一次完整执行从 invoke 到节点运行的全流程来看最简单的例子一个单节点 LLM Agentimport { StateGraph, START, END } from langchain/langgraph; import { Annotation } from langchain/langgraph; import { ChatOpenAI } from langchain/openai; import { HumanMessage } from langchain/core/messages; // 1. 定义状态结构 const AgentState Annotation.Root({ messages: AnnotationHumanMessage[]({ reducer: (prev, next) [...prev, ...next], default: () [], }), }); // 2. 初始化 LLM const llm new ChatOpenAI({ model: gpt-4o-mini }); // 3. 定义节点函数 async function callLLM(state: typeof AgentState.State) { const response await llm.invoke(state.messages); return { messages: [response] }; } // 4. 构建图 const graph new StateGraph(AgentState) .addNode(llm, callLLM) .addEdge(START, llm) .addEdge(llm, END) .compile(); // 5. 运行 const result await graph.invoke({ messages: [new HumanMessage(你好介绍一下自己)], });执行流程逐步拆解graph.invoke({ messages: [...] }) │ ▼ 1. 初始化 State messages [HumanMessage(你好...)] │ ▼ 2. 调度器从 START 出发 → 找到边START → llm │ ▼ 3. 执行节点 llm callLLM(state) 被调用 → llm.invoke(messages) → 返回 { messages: [AIMessage(我是...)] } │ ▼ 4. 合并状态Reducer 新 messages [...旧, AIMessage(我是...)] │ ▼ 5. 调度器检查下一步 → llm 的边指向 END → 执行结束 │ ▼ 6. 返回最终 State05 Reducer状态更新的核心机制这是很多人没搞清楚的地方。节点函数返回的不是「新 State」而是「State 的更新片段」。LangGraph 用 Reducer 把旧状态和更新片段合并。// 三种常见 Reducer 写法 // 方式1用内置 messagesStateReducer追加消息 const State1 Annotation.Root({ messages: AnnotationBaseMessage[]({ reducer: (prev, next) [...prev, ...next], }), }); // 方式2覆盖最新值覆盖旧值 const State2 Annotation.Root({ step: Annotationnumber({ reducer: (_, next) next, // 直接替换 }), }); // 方式3累加计数 const State3 Annotation.Root({ callCount: Annotationnumber({ reducer: (prev, next) prev next, default: () 0, }), });Reducer 的执行时机节点返回 { key: value } │ ▼ 对每个 key找到对应 Reducer newState[key] reducer(oldState[key], value) │ ▼ 生成新 State 快照 旧 State 不变不可变数据结构关键点State 是不可变的。 每次节点执行都产生一个新的 State 快照旧快照被保留这是 Checkpoint 能实现的基础后面讲。06 图的调度器它怎么决定下一步去哪这是 LangGraph 最有意思的部分——调度器Scheduler。调度器本质上是一个事件循环while 当前节点 ! END: 1. 执行当前节点 node(state) → partial_update 2. 用 Reducer 合并状态 → new_state 3. 查询当前节点的出边 - 普通边直接去下一个节点 - 条件边调用路由函数 router(new_state) → 返回节点名 4. 把下一个节点加入执行队列 5. 取队列头 → 重复条件边Conditional Edge长这样import { StateGraph, START, END } from langchain/langgraph; // 路由函数根据 state 返回下一个节点名 function routeAfterLLM(state: typeof AgentState.State): string { const lastMessage state.messages[state.messages.length - 1]; // 如果 LLM 想调工具 if (tool_calls in lastMessage lastMessage.tool_calls?.length) { return tools; // → 去工具节点 } return END; // → 直接结束 } const graph new StateGraph(AgentState) .addNode(llm, callLLM) .addNode(tools, callTools) .addEdge(START, llm) .addConditionalEdges(llm, routeAfterLLM, { tools: tools, // 映射路由函数返回值 → 节点名 [END]: END, }) .addEdge(tools, llm) // 工具执行完回到 LLM .compile();执行路径示意START → llm → [判断] → 需要工具? Yes → tools → llm → [判断] → 不需要了? Yes → END No → END这就是 ReAct Agent 的本质一个带条件边的循环图。07 并行执行Fan-out / Fan-in 模式LangGraph 支持一个节点连到多个后继节点实现并行执行。// Fan-out一个节点触发多个并行节点 const parallelGraph new StateGraph(AgentState) .addNode(start, startNode) .addNode(search_web, searchWeb) // 并行 .addNode(search_db, searchDB) // 并行 .addNode(merge, mergeResults) // 合并 .addEdge(START, start) .addEdge(start, search_web) // 同时出发 .addEdge(start, search_db) .addEdge(search_web, merge) // 都完成后汇聚 .addEdge(search_db, merge) .addEdge(merge, END) .compile();调度器处理并行的方式start 节点执行完毕 │ ├──→ search_web加入执行队列 └──→ search_db加入执行队列 执行队列[search_web, search_db] 同时调度异步并发执行 两者都完成后 merge 节点的所有前驱都就绪 → merge 入队 │ ▼ merge 执行 → END这个 Fan-out/Fan-in 模式是并行搜索、多 Agent 协作的基础。08 编译产物CompiledGraph 里藏了什么compile()返回的CompiledGraph暴露了四个主要接口interface CompiledGraph { // 同步执行返回最终 State invoke(input: State, config?: RunnableConfig): PromiseState; // 流式执行每个节点完成后 yield 一次 stream( input: State, config?: RunnableConfig ): AsyncGeneratorRecordstring, State; // 可视化图结构调试用 getGraph(): DrawableGraph; // 获取当前状态需要 Checkpointer getState(config: RunnableConfig): PromiseStateSnapshot; }stream()的输出格式for await (const chunk of graph.stream(input)) { // chunk 是一个对象{ 节点名: 该节点产生的状态更新 } // 例如 // { llm: { messages: [AIMessage(...)] } } // { tools: { messages: [ToolMessage(...)] } } console.log(chunk); }这就是为什么前端能实现「打字机效果」——每个节点完成前端就能收到一次更新。总结这篇我们从底层拆解了 LangGraph 把 LLM 变成状态机的核心机制StateGraph 三要素State 存数据、Node 执行动作、Edge 决定走向缺一不可Reducer 是关键节点返回的是「更新片段」而非「完整新状态」Reducer 负责合并State 始终不可变调度器是心脏本质是一个事件循环条件边让它能根据运行时状态动态决策compile() 的意义把声明式的图描述变成可执行的调度引擎做结构验证和邻接表预计算并行靠 Fan-out多条出边同时触发Fan-in 等待所有前驱完成调度器自动处理同步stream() 是流式基础每个节点完成即 yield这是打字机效果和实时反馈的技术支撑最后如果说程序员已经是高薪职业那么干AI的程序员就是高薪中的高薪。现在的市场已经用数据给程序员指明了方向学AI大模型就是冲刺高薪的最优解看着身边越来越多的同行转型大模型、拿到高薪offer很多人心里都动了心但真正的难题来了零基础小白不知道从哪入门有基础的程序员找不到系统学习路径实战项目练手无门面试不知道考什么别慌今天就给大家整理了一份【2026年最新版】AI大模型免费学习资源包覆盖从入门到实战、从理论到面试、从基础到进阶的全流程所有资料均已整理归档无冗余、无套路免费分享给每一位想抓住AI风口的程序员和小白扫码免费领取全部内容1、大模型系统化学习路线2、大模型学习书籍文档3、AI大模型最新行业报告4、大模型项目实战配套源码5、大模型大厂面试真题四阶段精细化学习规划附时间节点可直接照做结合上述资源给大家整理了一份可直接落地的四阶段学习规划总时长约2个月小白可循序渐进程序员可根据自身基础调整节奏高效掌握大模型核心能力快速实现从“入门”到“能落地、能面试”的跨越。第一阶段10天初阶应用该阶段让大家对大模型 AI有一个最前沿的认识对大模型 AI 的理解超过 95% 的人可以在相关讨论时发表高级、不跟风、又接地气的见解别人只会和 AI 聊天而你能调教 AI并能用代码将大模型和业务衔接。大模型 AI 能干什么大模型是怎样获得「智能」的用好 AI 的核心心法大模型应用业务架构大模型应用技术架构代码示例向 GPT-3.5 灌入新知识提示工程的意义和核心思想Prompt 典型构成指令调优方法论思维链和思维树Prompt 攻击和防范…第二阶段30天高阶应用该阶段我们正式进入大模型 AI 进阶实战学习学会构造私有知识库扩展 AI 的能力。快速开发一个完整的基于 agent 对话机器人。掌握功能最强的大模型开发框架抓住最新的技术进展适合 Python 和 JavaScript 程序员。为什么要做 RAG搭建一个简单的 ChatPDF检索的基础概念什么是向量表示Embeddings向量数据库与向量检索基于向量检索的 RAG搭建 RAG 系统的扩展知识混合检索与 RAG-Fusion 简介向量模型本地部署…第三阶段30天模型训练恭喜你如果学到这里你基本可以找到一份大模型 AI相关的工作自己也能训练 GPT 了通过微调训练自己的垂直大模型能独立训练开源多模态大模型掌握更多技术方案。到此为止大概2个月的时间。你已经成为了一名“AI小子”。那么你还想往下探索吗为什么要做 RAG什么是模型什么是模型训练求解器 损失函数简介小实验2手写一个简单的神经网络并训练它什么是训练/预训练/微调/轻量化微调Transformer结构简介轻量化微调实验数据集的构建…第四阶段20天商业闭环对全球大模型从性能、吞吐量、成本等方面有一定的认知可以在云端和本地等多种环境下部署大模型找到适合自己的项目/创业方向做一名被 AI 武装的产品经理。硬件选型带你了解全球大模型使用国产大模型服务搭建 OpenAI 代理热身基于阿里云 PAI 部署 Stable Diffusion在本地计算机运行大模型大模型的私有化部署基于 vLLM 部署大模型案例如何优雅地在阿里云私有部署开源大模型部署一套开源 LLM 项目内容安全互联网信息服务算法备案…扫码免费领取全部内容6、这些资料真的有用吗这份资料由我和鲁为民博士(北京清华大学学士和美国加州理工学院博士)共同整理现任上海殷泊信息科技CEO其创立的MoPaaS云平台获Forrester全球’强劲表现者’认证服务航天科工、国家电网等1000企业以第一作者在IEEE Transactions发表论文50篇获NASA JPL火星探测系统强化学习专利等35项中美专利。本套AI大模型课程由清华大学-加州理工双料博士、吴文俊人工智能奖得主鲁为民教授领衔研发。资料内容涵盖了从入门到进阶的各类视频教程和实战项目无论你是小白还是有些技术基础的技术人员这份资料都绝对能帮助你提升薪资待遇转行大模型岗位。这份完整版的大模型 AI 学习资料已经上传CSDN朋友们如果需要可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费】