LangGraph企业级Agent实战：状态管理、记忆与多智能体协作

从接触 Agent 开发到现在，我踩过不少坑：状态管理完全靠手写、节点之间的数据流转不透明、并行调度和人工介入复杂到想放弃。直到用上 LangGraph，才真正感受到“给 Agent 建流程图”是一件这么直观的事。如果你也想构建一个可以落地的企业级 Agent，或者正在学习 LangGraph 但不知道如何从零搭起，这篇文章会给你一条清晰的进阶路径。

我尽量把概念讲明白，把代码写全，把容易踩的坑标注出来。文章会从 LangGraph 是什么开始，讲到核心组件，再到单 Agent、带记忆的 Agent、工具调用、多智能体架构，最后给出一批可复制的实战代码和生产环境建议。零基础可以跟着跑通，有经验的开发者可以重点看第 5、6、8 节。

1. LangGraph 是什么？为什么它适合构建企业级 Agent

1.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 └── .env

requirements.txt内容建议根据你的实际依赖编写，至少要包含langgraph和对应模型接入包。也可以直接先用 pip 安装，不写文件。

如果你使用 OpenAI 接口，需要在.env中配置：

OPENAI_API_KEY=你的密钥 OPENAI_BASE_URL=https://api.openai.com/v1

注意：不要把密钥写死在代码里，更不要提交到 Git 仓库。企业项目一般通过密钥管理平台注入环境变量。

2.2 核心组件速览

在使用 LangGraph 之前，必须先理解下面几个名词：

关键点是：每个节点都接收当前 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，写入result；finish_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_id

4.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(model="gpt-4o-mini", temperature=0) 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(checkpointer=checkpointer) # 通过 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(model="gpt-4o-mini", temperature=0) 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(model="gpt-4o-mini", temperature=0) model_tools = model.bind_tools([add]) return {"messages": [model_tools.invoke(state["messages"])]} # 子 Agent 2：负责处理乘法 def multiply_agent_node(state: TeamState): model = ChatOpenAI(model="gpt-4o-mini", temperature=0) 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(model="gpt-4o-mini", temperature=0) 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 开发中，下面这些问题出现频率很高，我这里整理成一个速查表。

几个排查建议：

• 先简化：把一个多节点图拆成最小复现，确认哪个节点出问题。
• 使用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 如何做并行任务分派；
• 企业落地时需要关注的安全、可观测性和成本问题。

接下来，你可以尝试做一个综合小项目，把今天学的都串起来，比如：

1. 设计一个“智能工单助手”。
2. 用户提出“请帮我查询订单 Z123 的状态，并生成一份简要报告”。
3. 一个 Agent 负责查询 API，一个 Agent 负责写报告，主管 Agent 负责调度。
4. 给整个流程加上线程记忆和人工确认节点。

一边写一遍回头查官方文档，很多源码层的疑问会自然解开。尤其建议关注langgraph官方仓库中的示例代码，理解别人如何设计复杂状态，这对你后续写企业级 Agent 会非常有帮助。

如果你在实践中遇到问题，欢迎留言交流，我会持续更新常见问题清单。动手写一个自己的第一个多智能体项目，永远是最好的学习方式。