从接触 Agent 开发到现在我踩过不少坑状态管理完全靠手写、节点之间的数据流转不透明、并行调度和人工介入复杂到想放弃。直到用上 LangGraph才真正感受到“给 Agent 建流程图”是一件这么直观的事。如果你也想构建一个可以落地的企业级 Agent或者正在学习 LangGraph 但不知道如何从零搭起这篇文章会给你一条清晰的进阶路径。我尽量把概念讲明白把代码写全把容易踩的坑标注出来。文章会从 LangGraph 是什么开始讲到核心组件再到单 Agent、带记忆的 Agent、工具调用、多智能体架构最后给出一批可复制的实战代码和生产环境建议。零基础可以跟着跑通有经验的开发者可以重点看第 5、6、8 节。1. LangGraph 是什么为什么它适合构建企业级 Agent1.1 LangGraph 的基本定位LangGraph 是一个面向 LLM 应用的可编排框架核心目标是让开发者把 Agent 的流程“图”出来。你可以把 Agent 的每一步看作一个节点Node节点之间用边Edge连接复杂流程中还有条件边Conditional Edge帮你动态决定下一步走哪条分支。传统开发中写一个 Agent 往往靠while True循环 一堆if else控制逻辑简单的对话还能应付一旦涉及多个大模型调用工具调用链用户反馈后重新推理多个子 Agent 并行 / 分层协作对话记忆持久化。代码很快就会失控。LangGraph 的方式更像是在画一张有向图节点是业务动作边是流转关系状态对象则负责在节点之间传递数据。这种结构天然适合业务复杂、需要维护和扩展的企业级 Agent。1.2 LangGraph 与 LangChain 的关系和区别很多人会把 LangChain 和 LangGraph 混淆简单理解LangChain 是一套面向 LLM 应用的开发工具包提供了模型封装、提示词模板、文档加载、向量存储、链式调用等能力。LangGraph 是一个更底层的编排框架强调用图结构管理 Agent 的状态流转和执行流程。用 LangChain 也能写一个链式调用比如“取数据 → 写提示词 → 调模型 → 解析结果”但流程一旦变成动态的“工具循环”“多分支决策”“人工确认”LangChain 的传统 Chain 模式就比较吃力了。LangGraph 不是要替代 LangChain而是和 LangChain 互补。实际项目中常用 LangChain 的模型封装、消息格式和工具规范同时用 LangGraph 来承载整个 Agent 运行逻辑。1.3 典型应用场景企业知识库助手检索文档 → 判断是否需要追问 → 生成回答 → 提供引用来源。自动化客服工单处理识别意图 → 查询订单系统 → 调用 API 创建工单 → 向用户确认。数据分析 Agent根据用户问题编写 SQL → 执行查询 → 分析结果 → 输出报告。多智能体协作系统一个主管 Agent 负责拆解任务多个子 Agent 分别负责搜索、计算、写作最后汇总生成结果。这些场景共同的特点是流程不是单线而是存在分支、循环、并行以及可能的人工介入。LangGraph 恰好为这类流程提供了结构化方案。2. 环境准备与核心概念2.1 环境准备本文以 Python 环境为主使用常见的 LangGraph、LangChain 组件。安装命令如下pip install langgraph langchain-openai python-dotenv如果你本地有 Ollama想用开源模型体验可以添加上依赖pip install langgraph langchain-ollama python-dotenv安装完成后建议创建一个独立项目目录langgraph-demo/ ├── main.py ├── requirements.txt └── .envrequirements.txt内容建议根据你的实际依赖编写至少要包含langgraph和对应模型接入包。也可以直接先用 pip 安装不写文件。如果你使用 OpenAI 接口需要在.env中配置OPENAI_API_KEY你的密钥 OPENAI_BASE_URLhttps://api.openai.com/v1注意不要把密钥写死在代码里更不要提交到 Git 仓库。企业项目一般通过密钥管理平台注入环境变量。2.2 核心组件速览在使用 LangGraph 之前必须先理解下面几个名词组件作用类比State图执行过程中共享的数据结构各个节点都能读写的“黑板”Node图中的执行单元可以是一个函数、一个工具调用或一个子图流水线上的工位Edge连接节点之间的路径流水线轨道Conditional Edge根据状态条件动态选择去哪个节点岔路口的分流器Checkpointer保存每次运行的快照实现记忆和断点续跑游戏存档Send API动态创建多个并行分支用于 Map-Reduce 场景工单派发员关键点是每个节点都接收当前 State 作为输入处理后返回一个字典LangGraph 会把返回的字典合并进全局 State。因此节点函数不用关心分支和流程问题只负责“处理自己这一步”。3. 第一个 LangGraph 程序从 StateGraph 开始3.1 定义 State先看一个最简单的 StateGraph 示例它包含两个节点顺序执行。from typing import TypedDict, Annotated # 定义一个全局状态 class DemoState(TypedDict): user_name: str result: str这里的DemoState继承自TypedDict表示图中的全局状态。user_name和result是所有节点都可以访问的字段。3.2 编写节点函数节点就是一个普通函数接收状态字典返回更新后的字段。from langgraph.graph import StateGraph, START, END def collect_node(state: DemoState): # 模拟业务处理 return { result: f用户 {state[user_name]} 已进入流程 } def finish_node(state: DemoState): print(最终输出:, state[result]) return {}collect_node从输入状态中取出user_name写入resultfinish_node读取result并打印。两个节点之间靠 State 传递数据不需要额外传参。3.3 构建图并运行def build_graph(): graph StateGraph(DemoState) # 添加节点 graph.add_node(collect, collect_node) graph.add_node(finish, finish_node) # 添加边 graph.add_edge(START, collect) graph.add_edge(collect, finish) graph.add_edge(finish, END) # 编译 return graph.compile() app build_graph() result app.invoke({ user_name: 张三 }) print(图执行结果:, result)预期输出大致如下最终输出: 用户 张三 已进入流程 图执行结果: {user_name: 张三, result: 用户 张三 已进入流程}这段代码展示了 LangGraph 最基础但又最重要的思想节点只关注数据转换图的运行过程被显式表达出来。3.4 条件边让流程“活”起来真实业务中很少会一条路走到尾。比如用户输入“查询天气”和“我要订机票”系统需要走不同分支。条件边就是用来解决这个问题的。def check_input(state: DemoState): user_input state[user_name] if 天气 in user_input: return weather return default builder StateGraph(DemoState) builder.add_node(parse, check_input) builder.add_node(weather, lambda state: {result: 今天晴温度22℃}) builder.add_node(default, lambda state: {result: 你说的是 state[user_name]}) builder.add_edge(START, parse) builder.add_conditional_edges( parse, check_input, { weather: weather, default: default, } ) builder.add_edge(weather, END) builder.add_edge(default, END)注意这里的check_input同时被当作节点和条件判断函数使用。作为节点时它接收并返回 State作为条件函数时返回字符串映射到后续节点。4. 给 Agent 加记忆Checkpointer 与 thread_id4.1 为什么需要记忆如果没有记忆每次invoke都是全新状态模型无法记住用户上一轮说了什么。对话型 Agent 必须保存历史消息。LangGraph 的 Checkpointer 机制可以让图在某次执行结束时保存状态下次执行时从“某个时间点”继续。4.2 使用 MemorySaver下面用MemorySaver做一个带记忆的简单对话流程。为了更接近真实场景先定义消息状态。from typing import TypedDict, Annotated from langgraph.graph import StateGraph, START, END from langgraph.graph.message import add_messages from langchain_core.messages import AIMessage, HumanMessage from langgraph.checkpoint.memory import MemorySaver from langchain_openai import ChatOpenAI class AgentState(TypedDict): messages: Annotated[list, add_messages] def chat_node(state: AgentState): model ChatOpenAI(modelgpt-4o-mini, temperature0) response model.invoke(state[messages]) return {messages: [response]} graph_builder StateGraph(AgentState) graph_builder.add_node(chat, chat_node) graph_builder.add_edge(START, chat) graph_builder.add_edge(chat, END) # compile 时传入 checkpointer checkpointer MemorySaver() agent graph_builder.compile(checkpointercheckpointer) # 通过 thread_id 区分不同会话 config {configurable: {thread_id: conversation-001}} print(agent.invoke({messages: [HumanMessage(content我叫小明)]}, config)) print(agent.invoke({messages: [HumanMessage(content我刚才说我叫什么)]}, config))关键点add_messages这一定义表示 messages 字段在节点返回时会自动“追加”而不是覆盖。thread_id是对话会话的唯一标识。同一个thread_id会共享历史不同thread_id相互隔离。生产环境下可以使用持久化存储比如 PostgreSQL、Redis 等MemorySaver只适合本地测试和演示。5. 工具调用让 Agent 能动手执行任务5.1 为什么需要工具大模型本身不能实时查询天气、不能查数据库、更不能执行外部 API 操作。工具调用本质上是模型根据用户需求生成一个结构化调用请求LangGraph 帮助执行该请求再把结果返回给模型。5.2 定义一个工具使用 LangChain 的装饰器定义一个简单工具from langchain_core.tools import tool tool def add(a: int, b: int) - int: 计算两个整数的和。 return a b tool def multiply(a: int, b: int) - int: 计算两个整数的乘积。 return a * b这里的 docstring 非常重要模型会根据函数名和 docstring 判断何时调用工具。5.3 构建 Agent 并支持工具LangGraph 生态中ToolNode和相关辅助函数能简化工具调用流程。from langgraph.prebuilt import ToolNode, tools_condition from langchain_openai import ChatOpenAI from langgraph.graph import StateGraph, START, END from langgraph.graph.message import add_messages from typing import TypedDict, Annotated from langchain_core.messages import AIMessage class ToolAgentState(TypedDict): messages: Annotated[list, add_messages] tools [add, multiply] def agent_node(state: ToolAgentState): model ChatOpenAI(modelgpt-4o-mini, temperature0) model_with_tools model.bind_tools(tools) response model_with_tools.invoke(state[messages]) return {messages: [response]} builder StateGraph(ToolAgentState) builder.add_node(agent, agent_node) # 工具节点执行模型请求的工具 builder.add_node(tools, ToolNode(tools)) # 入口 builder.add_edge(START, agent) # 条件边模型认为需要调用工具就进入 tools否则结束 builder.add_conditional_edges( agent, tools_condition, ) # 工具执行完后回到 agent继续让模型判断结果 builder.add_edge(tools, agent) builder.add_edge(agent, END) app builder.compile() res app.invoke({messages: [(user, 计算 1234 乘以 56 等于多少)]}) for message in res[messages]: message.pretty_print()执行过程中LangGraph 会自动完成“模型生成 tool call → ToolNode 执行工具 → 返回工具结果 → 模型生成最终回答”的循环。tools_condition是 prebuilt 工具模块提供的条件函数作用是判断模型返回的消息里是否包含tool_calls如果有就进入工具节点否则结束。需要注意的是模型返回的工具调用结果不会自动保存到持久层如果要实现多轮上下文理解仍然要配合第 4 节的 Checkpointer 使用。6. 多智能体架构实战从单 Agent 到团队协作理解了单 Agent 后再来看企业级应用里更常见的多智能体架构。多智能体并不是“把多个模型函数放在一个图里”更关键的是怎样分工、怎样决策、怎样传递结果。6.1 两种常见多智能体模式网络模式Network多个 Agent 直接互相对话消息自由传递。灵活但难控制适合研究原型。主管-员工模式Supervisor一个主管 Agent 负责理解用户请求、选择调用哪个子 Agent、汇总结果。这种模式责任清晰更适合企业落地。下面用一个简化版 Supervisor 模式演示。我们创建两个 Worker一个擅长做加法一个擅长做乘法由主管模型决定走哪个分支。from typing import Literal from langchain_openai import ChatOpenAI from langgraph.graph import StateGraph, START, END from langgraph.graph.message import add_messages from langgraph.prebuilt import ToolNode, tools_condition from typing import TypedDict, Annotated from langchain_core.messages import AIMessage class TeamState(TypedDict): messages: Annotated[list, add_messages] # 子 Agent 1负责处理加法 def add_agent_node(state: TeamState): model ChatOpenAI(modelgpt-4o-mini, temperature0) model_tools model.bind_tools([add]) return {messages: [model_tools.invoke(state[messages])]} # 子 Agent 2负责处理乘法 def multiply_agent_node(state: TeamState): model ChatOpenAI(modelgpt-4o-mini, temperature0) model_tools model.bind_tools([multiply]) return {messages: [model_tools.invoke(state[messages])]} # 主管节点根据用户请求分派任务 def supervisor_node(state: TeamState): system_prompt ( 你是团队主管。根据用户需求选择合适节点。 涉及加法时回复 add_agent涉及乘法时回复 multiply_agent 如果都不涉及直接回复 END。 ) messages [{role: system, content: system_prompt}] state[messages] model ChatOpenAI(modelgpt-4o-mini, temperature0) response model.invoke(messages) return {messages: [response]}这里主管节点返回的 AIMessage 内容可以直接作为路由判断依据。在生产项目中更稳妥的方式是让模型输出结构化结果而不是解析自然语言。def route_supervisor(state: TeamState): last_message state[messages][-1].content if multiply in last_message: return multiply_agent if add in last_message: return add_agent return END构建图的逻辑如下graph StateGraph(TeamState) graph.add_node(supervisor, supervisor_node) graph.add_node(add_agent, add_agent_node) graph.add_node(multiply_agent, multiply_agent_node) graph.add_edge(START, supervisor) graph.add_conditional_edges( supervisor, route_supervisor, { add_agent: add_agent, multiply_agent: multiply_agent, END: END, } ) graph.add_edge(add_agent, supervisor) graph.add_edge(multiply_agent, supervisor) app graph.compile()这个简化示例展示的核心思想是子 Agent 是节点主管 Agent 是路由器全局 State 负责传消息。实际项目中每个子 Agent 内部完全可以又是一个完整的 LangGraph 子图这样就能形成“图里有图”的层次化架构。6.2 使用 Send API 做并行分派有时候需要一次性处理大量相似任务比如批量分析多篇文章、批量审核多条工单。如果一个个循环处理速度会很慢。LangGraph 的SendAPI 可以从当前节点动态创建多个分支并行执行。from langgraph.types import Send from typing import TypedDict, List class BatchState(TypedDict): articles: List[str] results: List[str] def start_branch(state: BatchState): # 为每一篇文章创建一个任务 return [ Send(analyze_one, {article: article}) for article in state[articles] ] def analyze_one(state): article state[article] # 这里可以调用相关模型或处理函数 return {results: [f分析完成{article[:20]}...]} builder StateGraph(BatchState) builder.add_node(start, start_branch) builder.add_node(analyze_one, analyze_one) builder.add_edge(START, start) builder.add_edge(analyze_one, END) app builder.compile() res app.invoke({articles: [文章1内容……, 文章2内容……, 文章3内容……]}) print(res[results])Send的本质是“动态建图”。它适合 Map-Reduce 场景先分裂出多个任务最后再汇总。需要注意的是并行分支共享同一份全局 State在分支中更新需要小心字段是否会被覆盖。实际项目里通常会为每个分支设计独立的状态字段或者用不同的状态 key 避免竞争。7. 常见问题与排查思路LangGraph 开发中下面这些问题出现频率很高我这里整理成一个速查表。问题现象常见原因解决思路节点返回字段不在 State 中State 类型没定义该字段在TypedDict中补充字段定义工具节点报“Unknown tool”工具没有传给ToolNode检查ToolNode(tools)中的工具列表对话时消息被覆盖而不是追加缺少Annotated[list, add_messages]注解在 State 的 messages 字段加add_messages图执行直接结束没有调用工具模型未返回tool_calls检查是否调用了bind_tools工具也许与问题不相关thread_id相同但记忆不生效编译后没有传入 checkpointer确认compile(checkpointer...)并且invoke时传了config出现InvalidUpdateError节点返回的字段状态更新冲突检查是否有两个并行节点同时更新同一字段执行一直重复进入同一个节点条件边判断有误或模型反复输出工具调用在条件函数中打印日志人工验证路由几个排查建议先简化把一个多节点图拆成最小复现确认哪个节点出问题。使用print观察节点输入输出理解 State 传递是否正确。查看官方文档中的 API 变更。LangGraph 社区更新很快部分接口名在不同版本有差异一定要参考当前安装版本的文档。在调用外部 API 时做好超时和重试避免 Agent 卡死。8. 企业级 Agent 的最佳实践建议8.1 State 设计明确边界避免大而全State 是各个节点通信的“黑板”字段越多越容易混乱。建议把数据分成几类用户输入与输出消息临时中间数据外部系统返回的原始数据终态结果。不同类别的字段要命名清晰必要时使用嵌套结构。字段更新时要明确是“覆盖”还是“追加”尽量使用不可变数据或显式更新。8.2 工具设计安全第一工具是 Agent 的“手”也是风险入口。企业级项目尤其要注意不要给 Agent 开放删除数据库、直接 drop 表的权限外部 API 调用必须有超时、鉴权和审计日志对 Agent 发出的命令要做白名单校验涉及真实业务数据变更时加上人工确认节点Human-in-the-loop。LangGraph 支持interrupt_before和interrupt_after参数可以在关键节点暂停图等待用户确认后再继续。8.3 可观测性记录每一步生产环境下任何 Agent 都可能出现不可预期的行为。建议至少记录每轮用户输入和模型输出每次工具调用参数和返回结果路由决策结果异常栈信息和耗时。如果使用 LangGraph 官方生态可以接入 LangSmith 做链路追踪。即使不接入也要在自己项目中打结构化日志。8.4 测试与灰度Agent 的行为具有随机性不可以只做“能跑通”测试。建议建立一套回归测试集核心功能用例边界输入用例攻击性输入用例工具调用准确率用例长时间运行的稳定性用例。上线前先在低流量环境灰度观察工具调用失败率和用户满意度后再全量放开。8.5 成本和性能控制每个 Agent 节点都可能调用一次模型多智能体架构下成本会成倍增加。可以做的优化包括简单问题走规则匹配避免无意义的大模型调用使用缓存复用结果子 Agent 间只传必要信息不要把整段历史重复传给每个节点给模型设置max_tokens、超时时间避免无限生成。9. 总结与下一步学习路线写到这儿你已经走完了从“零基础理解 LangGraph”到“能构建带工具、带记忆、带多智能体协作 Agent”的完整路径。你现在掌握的关键点包括LangGraph 与 LangChain 的区别State、Node、Edge、Conditional Edge 的核心用法Checkpointer 如何给 Agent 增加记忆ToolNode 如何让模型调用外部工具Supervisor 模式多智能体如何协作Send API 如何做并行任务分派企业落地时需要关注的安全、可观测性和成本问题。接下来你可以尝试做一个综合小项目把今天学的都串起来比如设计一个“智能工单助手”。用户提出“请帮我查询订单 Z123 的状态并生成一份简要报告”。一个 Agent 负责查询 API一个 Agent 负责写报告主管 Agent 负责调度。给整个流程加上线程记忆和人工确认节点。一边写一遍回头查官方文档很多源码层的疑问会自然解开。尤其建议关注langgraph官方仓库中的示例代码理解别人如何设计复杂状态这对你后续写企业级 Agent 会非常有帮助。如果你在实践中遇到问题欢迎留言交流我会持续更新常见问题清单。动手写一个自己的第一个多智能体项目永远是最好的学习方式。