在构建复杂AI应用时你是否曾为如何协调多个AI智能体、管理它们之间的对话流和状态而感到头疼传统的链式调用LangChain在处理简单任务时得心应手但一旦涉及多角色协作、循环决策或复杂工作流就显得力不从心。这正是LangGraph诞生的背景——它不是一个独立的框架而是LangChain生态系统中的一个专门用于构建有状态、多智能体工作流的库。本文将彻底拆解LangGraph从核心概念到企业级多智能体架构实战手把手带你构建一个可运行的智能体系统即使你是AI应用开发的新手也能跟着一步步实现。本文适合所有对AI应用开发、智能体Agent架构感兴趣的开发者。你将学到LangGraph的核心概念图、节点、边、状态。如何构建一个基础的、有状态的对话智能体。如何设计并实现一个包含“主管”Supervisor和多个“专家”智能体的多智能体系统。如何利用LangGraph Studio进行可视化开发和调试。企业级架构中的关键考量长期记忆、错误处理与可观测性。1. 背景与核心概念为什么需要LangGraph在深入代码之前我们必须理解LangGraph要解决的根本问题。传统的LangChain通过“链”Chain将多个组件顺序连接这适用于“输入-处理-输出”的线性场景。然而许多现实任务是非线性的例如客服路由根据用户问题类型决定是转接给“技术客服”还是“订单客服”。代码审查先由“代码分析器”检查语法再由“安全扫描器”检查漏洞最后“文档生成器”输出报告这个过程可能需要循环。多专家协作一个任务被拆解后由不同的专业智能体如“数据分析师”、“文案写手”、“审核员”接力或并行完成。这些场景都需要有状态的、基于图的工作流。这就是LangGraph的用武之地。1.1 核心概念拆解图Graph 这是LangGraph的核心抽象。一个图由节点Nodes和边Edges组成。节点代表一个执行单元例如调用一次LLM或执行一个工具边定义了节点之间的流转条件。状态State 图在执行过程中维护的一个共享数据对象。所有节点都可以读取和修改这个状态。LangGraph使用Pydantic模型来严格定义状态的Schema这确保了类型安全和数据的清晰流动。常见的状态字段包括messages对话历史、next指定下一个节点等。节点Node 一个普通的Python函数或可调用对象它接收当前的State作为参数对其进行修改并返回更新后的State。这是你编写业务逻辑的地方。边Edge 决定在某个节点执行完毕后下一个应该执行哪个节点。边可以是条件边根据状态内容动态路由或固定边始终指向下一个节点。1.2 LangGraph vs. LangChain这是一个常见的困惑点。简单来说LangChain 是一个完整的框架提供了与LLM、工具、记忆、索引等交互的所有基础组件。它包含Chains、Agents、Tools等概念。LangGraph 是LangChain库中的一个子模块langgraph。它不替代LangChain而是增强它专门用于构建比简单Chain更复杂的、有状态的、基于图的工作流。你可以把LangGraph看作是LangChain这个“乐高积木套装”里用来搭建复杂机械结构如带齿轮和传动的装置的那一套特殊连接件。一句话总结用LangChain准备零件LLM、工具用LangGraph设计并组装它们的工作流水线。2. 环境准备与版本说明在开始实战前请确保你的环境已就绪。本文将使用OpenAI的GPT模型作为LLM引擎但你也可以替换为其他兼容的模型如通过Ollama运行的本地模型。操作系统 Windows/macOS/Linux 均可。Python版本 建议使用 Python 3.10 或 3.11。核心依赖库# 创建并激活虚拟环境推荐 python -m venv langgraph-env source langgraph-env/bin/activate # Linux/macOS # 或 langgraph-env\Scripts\activate # Windows # 安装依赖 pip install langgraph langchain-openai python-dotenvlanggraph: 核心库。langchain-openai: 用于调用OpenAI API的LangChain集成包。python-dotenv: 用于管理环境变量如API密钥。重要提示 你需要一个有效的OpenAI API密钥。将其保存在项目根目录的.env文件中# .env 文件 OPENAI_API_KEY你的-api-key-here项目结构my-langgraph-project/ ├── .env ├── basic_agent.py # 基础单智能体示例 ├── multi_agent.py # 多智能体系统示例 └── requirements.txt3. 核心语法与原理拆解让我们通过构建一个最简单的智能体来理解LangGraph的运作机制。这个智能体将能够进行有状态的对话。3.1 定义状态State状态是一个Pydantic模型它定义了在整个图执行过程中流动的数据结构。# basic_agent.py from typing import TypedDict, Annotated, List from langgraph.graph.message import add_messages import operator # 1. 定义状态Schema class AgentState(TypedDict): # add_messages 是一个特殊的缩减器reducer # 它能自动将新消息追加到历史消息列表中非常方便。 messages: Annotated[List, add_messages] # 你可以添加其他任意字段例如用户ID、会话标记等。 # user_id: strTypedDict 用于定义状态的类型。Annotated[List, add_messages] 这是一个关键注解。它声明messages字段是一个列表并且指定了add_messages作为“缩减器”。这意味着当多个节点并发修改messages时add_messages会确保它们被正确地追加append到列表中而不是覆盖。这是实现对话记忆的基础。3.2 创建节点Node节点是一个函数它操作状态。# basic_agent.py (续) from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage, AIMessage import os from dotenv import load_dotenv load_dotenv() # 加载 .env 文件中的环境变量 # 初始化LLM llm ChatOpenAI(modelgpt-4o-mini, api_keyos.getenv(OPENAI_API_KEY)) # 2. 定义“调用模型”节点 def call_model(state: AgentState): print(f[节点 call_model] 收到消息历史: {state[messages]}) # 将整个消息历史上下文发送给LLM response llm.invoke(state[messages]) # 返回更新后的状态将AI的回复追加到消息列表 return {messages: [response]} # 3. 定义“人工输入”节点模拟用户 def call_human(state: AgentState): # 在实际应用中这里可能连接前端或API user_input input(\n[用户] 请输入: ) human_message HumanMessage(contentuser_input) return {messages: [human_message]}call_model节点 它接收当前状态包含对话历史调用LLM获取回复并将回复作为新的AIMessage放入状态。call_human节点 模拟用户输入将输入转换为HumanMessage放入状态。3.3 构建图Graph并定义边现在我们将节点组装成图并定义它们之间的流转关系。# basic_agent.py (续) from langgraph.graph import StateGraph, END # 4. 创建图构建器 graph_builder StateGraph(AgentState) # 5. 将节点添加到图中 graph_builder.add_node(human, call_human) graph_builder.add_node(model, call_model) # 6. 设置入口点从“human”节点开始用户先说话 graph_builder.set_entry_point(human) # 7. 添加边定义执行流 # 从“human”节点执行完后总是到“model”节点 graph_builder.add_edge(human, model) # 从“model”节点执行完后总是回到“human”节点形成对话循环 graph_builder.add_edge(model, human) # 8. 编译图得到可执行对象 graph graph_builder.compile()StateGraph 图构建器传入我们定义的状态Schema。add_node 注册节点并给节点起个名字如”human”,”model”。set_entry_point 指定图的起始节点。add_edge 添加固定边。add_edge(“human”, “model”)意味着human节点执行完后无条件地执行model节点。compile() 将图定义编译成可执行对象这是最后一步。3.4 运行图# basic_agent.py (续) if __name__ __main__: # 初始化状态 initial_state AgentState(messages[]) # 运行图 print(对话开始 (输入 quit 退出)...) for event in graph.stream(initial_state, stream_modevalues): # stream 会返回一个生成器按节点执行顺序产出状态快照 if messages in event and event[messages]: last_message event[messages][-1] if last_message.type ai: print(f[AI] {last_message.content}) elif last_message.type human: # 输入时已打印这里可跳过或处理 pass运行python basic_agent.py你将看到一个简单的对话循环对话开始 (输入 quit 退出)... [用户] 请输入: 你好你是谁 [节点 call_model] 收到消息历史: [HumanMessage(content你好你是谁)] [AI] 你好我是一个AI助手由OpenAI的技术驱动。我可以帮助你回答问题、提供信息、进行对话等等。有什么我可以为你做的吗 [用户] 请输入: 请用Python写一个Hello World。 ...这个例子虽然简单但完整展示了LangGraph的核心流程定义状态 - 创建节点 - 组装成图 - 执行。接下来我们将进入更强大的多智能体世界。4. 完整实战构建多智能体协作系统Supervisor模式我们将构建一个经典的“主管-专家”模式的多智能体系统。系统包含主管Supervisor 一个LLM负责理解用户任务并将其拆解、路由给合适的专家最后汇总专家的结果。专家Specialists 多个具备特定功能的智能体。本例中我们创建两个代码专家Coder 擅长编写和解释代码。文案专家Writer 擅长润色文本、撰写文档。4.1 定义多智能体系统的状态在多智能体系统中状态需要更丰富的信息来协调工作。# multi_agent.py from typing import TypedDict, Annotated, List, Literal, Optional from langgraph.graph.message import add_messages import operator class MultiAgentState(TypedDict): # 消息历史 messages: Annotated[List, add_messages] # 下一个要执行的节点名由主管决定 next: str # 可以添加其他字段如任务描述、中间结果等 current_task: Optional[str]next字段至关重要。主管节点将根据分析结果设置next的值为”coder”、”writer”或”supervisor”表示需要继续分析或汇总从而驱动图的流转。4.2 创建专家节点每个专家节点本质上也是一个调用LLM的函数但具有特定的系统提示词System Prompt来定义其角色和能力。# multi_agent.py (续) from langchain_openai import ChatOpenAI from langchain_core.messages import SystemMessage, HumanMessage, AIMessage import os from dotenv import load_dotenv load_dotenv() llm ChatOpenAI(modelgpt-4o-mini, temperature0.7, api_keyos.getenv(OPENAI_API_KEY)) def create_specialist_node(specialist_name: str, system_prompt: str): 工厂函数创建特定专家的节点 def node_function(state: MultiAgentState): # 构建对话上下文系统提示 最新的用户问题或主管指令 # 通常我们只传递最新的几条消息给专家避免上下文过长。 # 这里简单地将所有消息传递过去在实际应用中可能需要更精细的管理。 prompt [ SystemMessage(contentsystem_prompt), *state[“messages”][-6:], # 取最近6条消息可根据需要调整 ] response llm.invoke(prompt) # 专家回复后默认将控制权交还给主管supervisor return {messages: [response], next: supervisor} return node_function # 定义专家系统提示词 CODER_SYSTEM_PROMPT 你是一位资深软件开发工程师精通多种编程语言。 你的职责是 1. 根据需求编写正确、高效、可读性强的代码。 2. 解释代码的逻辑和关键点。 3. 分析代码中的潜在问题并提供优化建议。 请专注于技术实现保持回答简洁专业。 WRITER_SYSTEM_PROMPT 你是一位专业的文案编辑和技术文档工程师。 你的职责是 1. 对给定的文本进行润色使其更流畅、专业、易懂。 2. 将技术性描述转化为用户友好的语言。 3. 撰写清晰的技术文档、API说明或项目总结。 请专注于语言表达和文档结构避免深入技术细节实现。 # 创建专家节点 coder_node create_specialist_node(“coder”, CODER_SYSTEM_PROMPT) writer_node create_specialist_node(“writer”, WRITER_SYSTEM_PROMPT)4.3 创建主管Supervisor节点主管节点是系统的大脑它需要做两件事分析 判断用户请求应该由哪个或哪些专家处理。路由 通过修改状态中的next字段将任务派发出去。# multi_agent.py (续) SUPERVISOR_SYSTEM_PROMPT 你是多智能体系统的总调度主管。你的团队中有两位专家 1. **代码专家Coder** 负责所有与编程、代码生成、代码解释、算法相关的问题。 2. **文案专家Writer** 负责所有与文本润色、文档撰写、语言优化、内容总结相关的问题。 你的工作流程 1. 分析用户的请求。 2. 决定将任务派发给“coder”还是“writer”或者用户的问题需要你直接回答。 3. 在你的回复中**必须且只能**在最后一行以格式 NEXT: 节点名 来指定下一个执行的节点。 - 例如NEXT: coder 或 NEXT: writer 或 NEXT: supervisor。 4. 如果任务需要多个专家协作请逐步调度。例如先让coder写代码再让writer为代码写文档。 请先理解任务给出简要的调度思路然后输出NEXT指令。 def supervisor_node(state: MultiAgentState): # 准备给主管LLM的上下文 prompt [ SystemMessage(contentSUPERVISOR_SYSTEM_PROMPT), *state[“messages”][-10:], # 给主管更多的上下文 ] response llm.invoke(prompt) content response.content # **关键步骤解析回复提取 NEXT 指令** next_node “supervisor” # 默认值 lines content.strip().split(‘\n’) for line in reversed(lines): # 从最后一行开始找 if line.startswith(‘NEXT:’): next_node_candidate line.split(‘NEXT:’)[1].strip().lower() if next_node_candidate in [“coder”, “writer”, “supervisor”]: next_node next_node_candidate break # 返回更新后的状态最重要的是设置 next 字段 return { “messages”: [response], “next”: next_node, “current_task”: f”主管调度至: {next_node}” }4.4 组装多智能体图现在我们将所有节点组装起来并使用条件边来实现动态路由。# multi_agent.py (续) from langgraph.graph import StateGraph, END # 1. 创建图 workflow StateGraph(MultiAgentState) # 2. 添加所有节点 workflow.add_node(“supervisor”, supervisor_node) workflow.add_node(“coder”, coder_node) workflow.add_node(“writer”, writer_node) # 3. 设置入口点用户请求先到主管 workflow.set_entry_point(“supervisor”) # 4. 定义条件边根据状态中的 next 字段决定去向 def route_after_supervisor(state: MultiAgentState): 主管执行完后根据它设置的 next 字段路由 return state[“next”] # 从主管节点出发有三条可能的路由 workflow.add_conditional_edges( “supervisor”, # 源节点 route_after_supervisor, # 路由判断函数 { “coder”: “coder”, # 如果返回”coder”则去coder节点 “writer”: “writer”, # 如果返回”writer”则去writer节点 “supervisor”: “supervisor”, # 如果返回”supervisor”则继续留在主管节点例如需要进一步分析 # 理论上还可以有 END但本例中主管负责循环调度 } ) # 5. 专家执行完后固定返回主管节点继续调度或结束 workflow.add_edge(“coder”, “supervisor”) workflow.add_edge(“writer”, “supervisor”) # 6. 编译图 app workflow.compile()4.5 运行与测试多智能体系统让我们用一个需要两者协作的任务来测试“写一个Python函数计算斐波那契数列并为其生成一段友好的使用说明。”# multi_agent.py (续) from langchain_core.messages import HumanMessage if __name__ “__main__”: # 初始化状态用户输入第一个请求 initial_state MultiAgentState( messages[ HumanMessage(content”写一个Python函数计算斐波那契数列并为其生成一段友好的使用说明。”) ], next”supervisor”, # 初始下一个节点是主管 current_taskNone ) print(“ 多智能体系统启动 \n”) final_state None # 我们限制步数防止无限循环 max_steps 10 for i, step in enumerate(app.stream(initial_state, stream_mode“values”)): node_name list(step.keys())[0] if isinstance(step, dict) else ‘unknown’ print(f”\n[步骤 {i1}] 执行节点: {node_name}“) if “messages” in step and step[“messages”]: last_msg step[“messages”][-1] print(f” 角色: {last_msg.type}“) # 打印主管和专家的思考/输出 if last_msg.type ‘ai’: # 简单处理打印前500字符 preview last_msg.content[:500] (‘…’ if len(last_msg.content) 500 else ‘’) print(f” 内容预览:\n {preview}“) if “next” in step: print(f” 下一步指向: {step[‘next’]}“) final_state step if i max_steps - 1: print(“\n达到最大步数停止。”) break print(“\n 执行结束 ) if final_state and “messages” in final_state: print(“\n完整的对话历史:”) for msg in final_state[“messages”]: print(f”[{msg.type}] {msg.content[:200]}…“ if len(msg.content) 200 else f”[{msg.type}] {msg.content}“)运行python multi_agent.py你会看到类似以下的输出清晰地展示了任务在主管、代码专家、文案专家之间的流转 多智能体系统启动 [步骤 1] 执行节点: supervisor 角色: ai 内容预览: 用户请求涉及两个部分1) 编写计算斐波那契数列的Python函数编程任务2) 生成友好的使用说明文档任务。因此我需要先调度代码专家Coder完成函数编写然后调度文案专家Writer基于代码生成说明。 NEXT: coder 下一步指向: coder [步骤 2] 执行节点: coder 角色: ai 内容预览: python def fibonacci(n): “”“计算第n个斐波那契数”“” if n 0: raise ValueError(“输入必须为正整数”) elif n 1 or n 2: return 1 a, b 1, 1 for _ in range(3, n1): a, b b, a b return b … 下一步指向: supervisor [步骤 3] 执行节点: supervisor 角色: ai 内容预览: 代码专家已提供斐波那契函数。现在需要文案专家为此函数撰写一段清晰、友好的使用说明解释功能、参数、返回值及示例。 NEXT: writer 下一步指向: writer [步骤 4] 执行节点: writer 角色: ai 内容预览: **斐波那契数列计算函数使用说明** fibonacci(n) 函数用于计算斐波那契数列中第 n 个数的值… 下一步指向: supervisor这个流程完美演示了多智能体协作主管分析任务并路由给代码专家代码专家完成代码后返回主管主管再次分析后路由给文案专家完成文档。这就是基于LangGraph构建的、可扩展的多智能体系统核心。5. 常见问题与排查思路在开发LangGraph应用时你可能会遇到以下典型问题问题现象常见原因解决思路KeyError或状态字段访问错误1. 状态Schema (TypedDict) 中未定义该字段。2. 节点返回的字典键与Schema定义不匹配。1. 检查TypedDict类确保所有用到的字段都已明确定义。2. 确保每个节点函数返回的字典中的键都是状态Schema中存在的字段。图陷入无限循环1. 边Edges的逻辑有误形成了环且没有退出条件。2. 条件边add_conditional_edges的路由函数始终返回同一个节点。1. 使用langgraph的graph.get_graph().draw_mermaid()输出图结构可视化检查循环。2. 在条件边路由函数中添加日志打印其返回值。3. 确保有一个节点能将next指向END来终止流程。LLM不遵循指令格式如不输出NEXT1. 系统提示词System Prompt不够清晰或强制。2. LLM的temperature参数过高导致输出随机。1. 强化提示词工程在提示词中明确要求输出格式并给出多个示例。2. 降低temperature如设为0以获得更确定性的输出。3. 在节点代码中增加更健壮的解析逻辑包括格式错误的回退机制。消息历史messages过长导致Token超限或性能下降默认的add_messages缩减器会无限制地追加消息。1. 在状态Schema中定义新的字段来存储摘要或关键信息而非完整历史。2. 在节点函数中有选择地传递最近N条消息给LLM而不是全部。3. 实现一个“总结”节点定期将长对话历史压缩成摘要。无法安装langgraph或依赖冲突1. Python版本不兼容。2. 与现有项目的其他包存在版本冲突。1. 确认使用Python 3.10。2. 在全新的虚拟环境中安装python -m venv env source env/bin/activate。3. 使用pip install langgraph0.0.xx指定一个已知稳定的版本。6. 进阶主题与最佳实践掌握了基础和多智能体构建后以下是提升应用鲁棒性和工程化水平的关键点。6.1 长期记忆Persistent Memory上述示例的状态仅在单次运行期间存在于内存中。在生产环境中你需要将状态持久化到数据库如PostgreSQL, Redis或向量库中以实现跨会话的记忆。LangGraph的设计使得这很容易实现。核心思路 自定义一个Checkpointer来存储和加载State。你可以基于任何后端实现BaseCheckpointSaver接口。# 伪代码示例 from langgraph.checkpoint.base import BaseCheckpointSaver import pickle # 假设有一个简单的KV存储 class MyRedisCheckpointer(BaseCheckpointSaver): def __init__(self, redis_client): self.client redis_client def put(self, config, checkpoint): # 将checkpoint序列化后存入Rediskey为config[“configurable”][“thread_id”] serialized pickle.dumps(checkpoint) self.client.set(f”langgraph:{config[‘configurable’][‘thread_id’]}“, serialized) def get(self, config): # 从Redis读取并反序列化 serialized self.client.get(f”langgraph:{config[‘configurable’][‘thread_id’]}“) return pickle.loads(serialized) if serialized else None # 在编译图时传入checkpointer graph workflow.compile(checkpointerMyRedisCheckpointer(redis_client))6.2 错误处理与回退节点尤其是调用外部API的LLM节点可能失败。LangGraph提供了Node包装器来增加重试、超时和回退逻辑。from langgraph.graph import Node from tenacity import retry, stop_after_attempt, wait_exponential retry(stopstop_after_attempt(3), waitwait_exponential(multiplier1, min4, max10)) def reliable_llm_call(state): # 你的LLM调用逻辑 response llm.invoke(state[“messages”]) return {“messages”: [response]} # 将普通的函数包装成具有容错能力的Node reliable_node Node(reliable_llm_call, name”reliable_llm”) workflow.add_node(“reliable_llm”, reliable_node)6.3 可观测性与调试LangGraph StudioLangGraph提供了一个强大的可视化调试工具——LangGraph Studio。它是一个本地Web应用可以让你可视化图结构 直观看到节点和边。跟踪执行过程 逐步执行查看每个节点输入/输出的状态快照。编辑和测试 直接修改提示词或代码并重新运行。安装与运行pip install langgraph-cli langgraph studio然后在浏览器中打开http://localhost:5678导入或创建你的图即可进行交互式调试。这对于复杂工作流的开发和问题排查不可或缺。6.4 架构设计建议状态设计最小化 只将真正需要在节点间共享的数据放入State。过度庞大的状态会影响性能和清晰度。节点职责单一 每个节点应只做一件事如“调用LLM”、“查询数据库”、“格式化输出”。这提高了可测试性和可复用性。善用子图Subgraph 对于复杂的、可复用的逻辑序列可以将其封装成一个子图。主图通过一个节点调用子图这有助于管理复杂度。为生产环境配置超时和限流 在调用外部服务LLM API、数据库的节点上务必设置超时。考虑对整个图的执行时间设置上限。日志与监控 在每个节点的入口和出口添加详细的日志记录状态变化、耗时和错误。这对接入APM如Prometheus, Datadog至关重要。从简单的对话循环到复杂的企业级多智能体调度系统LangGraph通过“图”这一抽象为我们提供了一种强大而优雅的方式来构建有状态的AI工作流。它解决了传统链式结构在处理分支、循环和持久化状态时的短板。回顾本文你应当掌握核心概念 图、状态、节点、边是构建一切的基石。开发流程 定义状态Schema - 编写节点函数 - 组装图并定义边 - 编译运行。多智能体模式 通过“主管”节点和条件边可以实现灵活的智能体路由与协作。工程化要素 长期记忆、错误处理、可视化调试是走向生产环境的必经之路。LangGraph的学习曲线初看可能较陡但一旦理解其范式你会发现它极大地提升了构建复杂AI应用的效率和可控性。建议你从本文的示例出发尝试修改状态结构、增加新的专家节点如“搜索专家”、“数据库专家”或引入工具Tools调用来打造属于你自己的智能体系统。
