LangGraph实战：用状态图编排智能体工作流与工具调用

说实话第一次看到LangGraph这个项目时我的第一反应是这不就是给LangChain加了个循环吗但当我自己动手写了几个稍复杂的Agent并尝试用它来管理多步骤、有状态、需要动态决策的任务流之后我才意识到这玩意儿解决的痛点有多大。如果你已经厌倦了让大模型在单轮或多轮对话里反复猜你的意图却始终不具备真正的执行力如果你在构建AI应用时不仅希望它能回答怎么做更希望它能自己拆解任务、调用工具、根据中间结果修正下一步行动——那这篇基于LangGraph的智能体开发实操总结大概率正是你需要的。这篇文章我不会去复述官方文档里的那些术语定义而是从为什么要用图来编排Agent这个最根本的问题讲起然后带着你一步步搭建一个能真实落地、具备工具调用与条件路由能力的智能体工作流。全程配有可直接参考的代码、设计思路和避坑经验无论你是刚开始接触智能体开发的新手还是在LangChain和LangGraph之间犹豫不决的开发者相信都能从中找到有价值的参考。1. 从问答机器到智能体缺的不只是记忆1.1 传统对话链路的本质局限当我们用OpenAI的函数调用、或者LangChain的单个Chain去构建应用时典型流程是这样的用户输入 - 大模型生成回复 - 返回结果。这个流程在纯问答场景下没有任何问题但对稍微复杂一点的业务需求比如帮我调研一下某行业的头部公司并生成一份对比报告整个链路就断掉了。因为模型需要先检索资料、调用搜索工具拿到结果再基于结果做分析然后可能还要调用内部API查数据最后根据查到的数据生成文档。这个过程不是一次的生成而是多次感知-决策-行动的循环。没有状态管理的情况下代码会变得极其混乱——你需要手动维护每一步的中间变量把上下文一次次拼接进提示词还要写一大堆if-else来处理模型可能出现的各种分支走向。我有一次在一个金融问答项目里硬写了超过600行的分支判断就是为了处理用户问的是一只股票还是一只基金、是否需要换算单位、是否需要给出历史净值这类逻辑。结果就是代码极其脆弱换一个模型、调一个prompt整个逻辑全崩。1.2 智能体需要的三个核心能力后来我逐步总结出一个真正的智能体至少需要具备三个核心能力第一状态持久化——每一步执行完中间结果要能被后续步骤访问不能每次重头再来。第二动态路由——下一步执行什么不应该由代码写死而应该由当前的状态和模型判断出来的意图共同决定。第三工具调用闭环——模型不仅要知道该调工具还要能读懂工具返回的结果并决定是继续调用、修正参数、还是直接给出最终答案。LangGraph之所以能成为解决这三个问题的利器本质原因是它把智能体的执行过程抽象成了一张有向图。节点是具体的执行动作比如调用一次搜索、运行一段计算边是执行顺序的约束而全局状态则像一张黑板所有节点都可以读取和写入。更关键的是它允许图中存在条件分支和循环——模型可以自主决定从A节点跑到B节点还是回到A节点重新执行。用图来描述智能体的思考过程比用Stack或队列来描述表达力强了不止一个量级。1.3 LangGraph到底是LangChain的何方神圣对于刚接触的朋友可以先把它理解成LangChain生态里的编排引擎。LangChain本身提供了模型封装ChatOpenAI等、工具封装TavilySearch等、以及诸如Chain的基本调用链但Chain的拓扑结构是静态的。LangGraph则打破了链的直线结构允许构建环状和条件跳转。它们的关系不是替代而是互补LangChain负责提供积木LangGraph负责决定积木怎么搭、搭完之后怎么跑。在实际开发中我通常用LangChain的create_react_agent快速搭建原型当原型复杂度上升、需要精细控制流程和持久化状态时再切换到LangGraph进行重构。2. 上手LangGraph核心概念与最小工作流拆解2.1 五个你必须想清楚的概念如果你去看官方文档会被State、Node、Edge、Conditional Edges、StateGraph这几个词砸晕。我用自己理解的方式给它们做个翻译State状态整个工作流的共享内存。通常定义为一个TypedDict或Pydantic模型里面放着消息历史、当前轮次、中间数据、最终答案等字段。要注意的是LangGraph的State更新不是覆盖而是默认合并——只要你的状态定义里用了Annotated[list, operator.add]这种操作符多个节点就可以往同一个列表字段里追加内容这个特性在保存多轮Agent推理过程时非常实用。Node节点就是一个普通的Python函数。输入是State字典输出是一个字典表示要对State做哪些更新。你可以把任何逻辑写进节点里调用模型、执行代码、查数据库什么都行。Edge边节点之间的连接。一条边表示上一个节点执行完立即进入下一个节点。Conditional Edges条件边节点的出口有多个执行哪个由函数返回值决定。这是LangGraph实现模型自主决策下一步动作的命脉。StateGraph状态图把它们全部组装起来的容器。最后调用compile()编译成可执行对象这一点类似于把它当做一个新的工具接入LangChain。2.2 手写一个最简问答工作流我通常建议新手先别急着追最新API用最原始的写法感受一下图执行的过程。下面这段代码是一个只有两个节点的极简工作流import operator from typing import Annotated, TypedDict from langgraph.graph import StateGraph, START, END # 1. 定义全局状态 class MyState(TypedDict): query: str answer: str reasoning_steps: Annotated[list, operator.add] # 追加模式保留所有中间推理 # 2. 定义节点函数 async def analyze_query(state: MyState) - dict: query state[query] # 模拟一次意图分析 reasoning f已收到用户问题{query}准备进入检索环节 return {reasoning_steps: [reasoning]} async def generate_answer(state: MyState) - dict: reasoning state[reasoning_steps] answer f针对你的问题{state[query]}这是最终的答复。 return {answer: answer, reasoning_steps: [已完成答案生成]} # 3. 构建图 builder StateGraph(MyState) builder.add_node(analyze, analyze_query) builder.add_node(generate, generate_answer) # 4. 连接START - analyze - generate - END builder.add_edge(START, analyze) builder.add_edge(analyze, generate) builder.add_edge(generate, END) # 5. 编译并运行 graph builder.compile() result await graph.ainvoke({query: 今天北京天气怎么样}) print(result[answer]) print(result[reasoning_steps])你可以看到整个过程跟搭乐高一样先定状态再写节点然后连线最后运行。节点之间的调用完全由框架调度你只需要关心每个节点输入什么、输出什么。2.3 为什么State必须是共享内存而不是传参初次接触的人容易犯一个误区把上一节点的返回值手动传给下一个节点。这在简单的线性流程里可行但一旦出现循环或者两个分支并行执行再汇聚手工传参就会变成一场灾难。你无法确定哪个分支先结束、哪个结果应该被覆盖。LangGraph的做法是把State当作全局黑板每个节点只需声明我需要读哪些字段、写哪些字段。并行节点往同一个列表字段追加内容也会被正确合并且不丢失任何数据。这个设计背后的思想是数据流与执行流解耦和我们在分布式系统里常见的高可用的架构思路很一致。理解了这一点你再去看官方文档里那些复杂的图结构思路就清晰了。3. 让智能体真正干活工具调用与条件路由的完整实现3.1 从模型生成文本到模型调度工具图结构搭好了真正让它从问答机器蜕变为智能体的关键一步是让模型可以自主发起工具调用。OpenAI已经在函数调用接口层面支持了这个能力LangGraph要做的就是把模型的工具调用请求解析出来执行工具然后把结果以消息形式反馈给模型——这个循环可以持续多轮直到模型认为不再需要工具为止。我在项目里常用的是一个研究助理Agent用户输入一个行业关键词智能体判断是否需要联网搜索如果需要就调用Tavily搜索API把搜索结果整理后继续追问或者给出最终报告。这个Agent的核心代码如下采用的是LangGraph官方推荐的ReAct模式封装from langchain_openai import ChatOpenAI from langchain_community.tools.tavily_search import TavilySearchResults from langgraph.prebuilt import create_react_agent # 1. 准备大模型与工具 llm ChatOpenAI(modelgpt-4o, temperature0) search_tool TavilySearchResults(max_results5) # 2. 使用 prebuilt 快速创建 ReAct Agent agent create_react_agent( modelllm, tools[search_tool], prompt你是一名严谨的行业研究助理必须基于检索结果回答用户问题。 ) # 3. 直接对话自动完成 工具调用-总结 循环 result agent.invoke({messages: [{role: user, content: 帮我调研2025年国内AI Agent赛道的主要玩家和融资情况}]}) for message in result[messages]: print(f{message.type.upper()}: {message.content})这段代码的简洁程度可能让人有点意外但它内部做的工作量远比你想象的大。create_react_agent内部构建了一个预定义的图模型节点 - 工具节点 - 条件判断是否继续调用工具- 循环直到输出最终答案。它会自动把工具返回的结果打包成ToolMessage重新注入模型上下文还会对部分模型的特殊函数调用格式做适配。3.2 手动构建工具调用图搞懂底层的每一环create_react_agent用起来确实爽但如果你要定制路由逻辑或者想插入自己的判断节点就得手动搭图了。这里我拆解一个自己实际用过的通用工具调用图它比prebuilt版本更灵活也更能帮助你理解LangGraph的执行机制import json from typing import Annotated, TypedDict, Literal from langchain_openai import ChatOpenAI from langchain_community.tools.tavily_search import TavilySearchResults from langchain_core.messages import HumanMessage, AIMessage, ToolMessage from langgraph.graph import StateGraph, START, END from langgraph.prebuilt import ToolNode from langgraph.graph.message import add_messages # 1. 定义状态 class AgentState(TypedDict): messages: Annotated[list, add_messages] # 2. 准备模型与工具 llm ChatOpenAI(modelgpt-4o) tools [TavilySearchResults(max_results5)] llm_with_tools llm.bind_tools(tools) # 3. 节点A让模型决定下一步 async def call_model(state: AgentState) - dict: response await llm_with_tools.ainvoke(state[messages]) return {messages: [response]}注意这里的关键点是bind_tools(tools)。它会让模型的输出结构带上tool_calls字段。如果模型认为需要搜索就会返回一个包含搜索参数的工具调用指令而不是普通文本。# 4. 节点B执行工具 tool_node ToolNode(tools) # 5. 定义条件路由模型是否想调用工具 def should_continue(state: AgentState) - Literal[tools, continue]: last_message state[messages][-1] if getattr(last_message, tool_calls, None): return tools return continue # 6. 构建图 builder StateGraph(AgentState) builder.add_node(agent, call_model) builder.add_node(tools, tool_node) builder.add_node(final, lambda state: {messages: [AIMessage(content流程结束以上为最终答复。)]}) builder.add_edge(START, agent) builder.add_conditional_edges( agent, should_continue, { tools: tools, # 有工具调用 - 进入工具节点 continue: final, # 没有工具调用 - 直接输出结果 }, ) builder.add_edge(tools, agent) # 工具执行完回到模型继续判断 builder.add_edge(final, END) graph builder.compile()执行过程很好理解模型节点拿到对话历史如果觉得需要搜索返回携带tool_calls的消息should_continue检测到这一字段把执行权交给tools节点工具执行后结果包装成ToolMessage回写到状态中然后边把执行权传回agent节点模型再次根据新的、包含搜索结果的完整上下文做决策。这个模型 - 工具 - 模型的循环正是智能体区别于普通问答系统的核心特征。3.3 用HumanMessage模拟一次完整的多轮工具调用为了验证图是否正常我会用以下方式模拟真实用户请求# 模拟用户提问 user_query 2025年AI Agent赛道有哪些值得关注的初创公司请给出5个并说明理由。 # 执行智能体多轮自动驾驶 result await graph.ainvoke({ messages: [HumanMessage(contentuser_query)] }) # 打印完整执行过程 for msg in result[messages]: if isinstance(msg, AIMessage) and msg.tool_calls: print(f[模型发起工具调用] {msg.tool_calls[0][name]}({msg.tool_calls[0][args]})) elif isinstance(msg, ToolMessage): print(f[工具返回结果] 长度: {len(msg.content)} 字符) else: print(f[文本消息] {msg.content[:200]})实操中你会发现模型可能会连续发起三次搜索第一次搜AI Agent 初创公司 融资第二次搜AI Agent 创投 2025第三次搜某家公司的具体新闻。这三轮搜索之间模型会根据前一轮的搜索结果来修正下一轮的搜索词。这个动态调整的过程是传统基于规则的爬虫或脚本完全做不到的也正是智能体与自动化程序的分水岭。4. 从Demo到生产状态持久化、流式输出与人工介入4.1 断点续跑让Agent具备短期记忆一个很常见的需求是用户和智能体的对话跨越多个会话每次调用之间对话历史不能丢。LangGraph为此提供了checkpointer机制。本质上它在每次图执行前把状态快照保存到持久化存储中下次执行时把快照加载回来。官方支持内存、SQLite、PostgreSQL等后端本地调试用MemorySaver即可from langgraph.checkpoint.memory import MemorySaver checkpointer MemorySaver() graph_with_memory builder.compile(checkpointercheckpointer) # 用 thread_id 区分不同会话 config {configurable: {thread_id: user-abc-001}} result1 await graph_with_memory.ainvoke( {messages: [HumanMessage(content我叫张三想了解AI Agent)]}, config ) result2 await graph_with_memory.ainvoke( {messages: [HumanMessage(content刚才聊的那个帮我深入讲讲技术架构)]}, config )当第二次调用时LangGraph会自动定位到thread_id对应的Conversation历史模型可以引用第一次对话的内容。这个机制对生产级应用非常重要。我踩过的一个坑是如果Checkpointer实现了状态合并但你的状态里某个字段是普通字符串而不是列表第二次调用读到的可能还是旧值。解决办法是对所有需要跨轮累积的字段都使用Annotated[list, operator.add]或者在节点里手动覆盖。4.2 流式输出让用户看到思考过程如果你直接ainvoke一个多步骤Agent用户可能要等上十几秒才能收到完整回复这个体验很差。LangGraph原生支持流式返回事件async for event in graph.astream_events({messages: [HumanMessage(contentuser_query)]}, versionv1): kind event[event] if kind on_chat_model_stream: chunk event[data].get(chunk) if chunk and chunk.content: print(chunk.content, end, flushTrue) elif kind on_tool_start: print(f\n[正在调用工具] {event.get(name)}...)这样就能实现模型吐字和工具执行进度交替出现的效果。用户体验上有一个质的提升。但要注意astream_events在LangChain和LangGraph的历史版本里返回结构有一些调整——如果你升级之后发现取不到chunk直接去官方GitHub的Release Notes里确认on_chat_model_stream事件的最新字段路径不要盲目相信旧博客的代码。4.3 人工介入让老板或审核者插一脚另一个容易被忽视的杀手级功能是interrupt_before。允许你在某个关键节点执行前暂停图等人工确认后继续。比如生成对外发布的投资分析报告之前加了人工审核节点graph_with_human builder.compile( checkpointercheckpointer, interrupt_before[final], # 在 final 节点前暂停 ) # 第一次调用会在 final 前停下 await graph_with_human.ainvoke({messages: [HumanMessage(content生成报告)]}, config) # 从快照恢复传入人工审核结果后继续 await graph_with_human.ainvoke( {messages: [AIMessage(content审核已通过继续执行)]}, config )这种机制在很多企业级场景下非常实用AI负责把所有前期工作做完人在关键决策节点只做一次确认既保留了自动化效率又满足了风控要求。5. 我踩过的坑与2026年智能体开发的前瞻5.1 五个典型问题与排查思路在LangGraph上写坏过很多个图之后我总结了一份高频问题速查表面试和实战中都很常见现象常见原因解决办法图运行卡死不结束条件边写错让循环永远为真在should_continue里增加最大迭代轮次判断用state字段记录步数工具调用的参数总是不对模型不知道工具的完整参数格式给工具增加详细的description尤其是参数说明必要时用pydantic定义入参结构体多个节点并行执行后数据丢失State定义里用了普通字典没有定义追加操作符检查Annotated[list, operator.add] 是否应用在需要累积的字段上恢复断点后状态错乱使用了带副作用的节点让Node保持纯函数所有对外部世界的读取和写入都在节点内部处理不要依赖全局变量模型在工具循环中反复调用同一个工具工具返回结果不明确或者温度太高提高工具的返回信息质量把temperature降到0在工具返回内容中增加若未查找到有效信息请明确说明之类的提示5.2 面试中关于LangGraph的高频考点在我接触到的多个智能体开发岗位的面试中LangGraph相关的问题几乎都会出现这里整理了几个核心考点供大家参考LangChain和LangGraph有什么区别基础但高频。LangChain提供模型封装、工具集成和Chain抽象LangGraph是一个图编排框架支持循环、分支、状态管理和人工介入。Chain是静态拓扑Graph是动态拓扑。LangGraph中StateGraph和Graph有什么区别StateGraph是带状态类型的图节点函数接收TypedDict并返回部分更新Graph是不带状态的简单图应用场景有限现在大部分智能体需求都可以采用StateGraph实现。ToolNode的默认行为是什么它接收AIMessage中的tool_calls依次调用对应工具返回ToolMessage列表。需要注意并行工具调用和异步调用的资源限制。Checkpointer的作用持久化每步状态实现断点续跑、具备短期记忆、支持interrupt_before/after人工介入。生产环境推荐用PostgreSQL版的checkpointer避免MemorySaver在重启后丢失状态。如何控制在多Agent协作中各个子Agent的并发数量LangGraph支持send() API实现动态Fan-out配合Semaphore或手动限制并发工具数量。5.3 工业智能体2026年真正的分水岭最近行业内有个判断我认为很中肯2026年是工业智能体从概念演示走向工程化落地的分水岭。去参加WAIC这类展会也会明显感觉到智能体不再只是能聊天的Demo而是必须解决真实业务问题自动生成专利辅助材料、销售线索自动跟进、考公知识点梳理、测试用例自动生成……这些场景的共同特征是流程固定、工具明确、结果可验收。LangGraph这类框架恰好提供了把大模型嵌入固定流程、同时保留灵活性空间的能力——这比单纯追求模型有多聪明更接近商业价值。对我来说LangGraph最大的价值不在于代码技巧而在于它让我们用工程化的思维去构建Agent。AI不只是生成一个答案而是完成一个流程不只是聪明的单次推理而是稳健的多步执行。如果你正在做的项目还停留在调用大模型-输出文本的阶段真心建议花一个下午把LangGraph的图执行逻辑跑一遍。从问答机器到智能体中间差的往往不是模型能力而是一张合理的状态流转图。