LangGraph实战：构建有状态多智能体系统的工程指南

在实际企业级 AI 应用开发中构建一个能处理复杂、多步骤任务的智能体Agent往往比训练一个单一的大语言模型更具挑战性。开发者常常面临状态管理混乱、工具调用顺序失控、多角色协作困难等问题。LangGraph 作为 LangChain 生态中用于构建有状态、多智能体工作流的核心框架通过将智能体行为建模为图Graph清晰地定义了状态流转和节点执行逻辑为复杂 Agent 架构提供了工程化的解决方案。本文将从 LangGraph 的核心概念入手逐步构建一个包含监督者Supervisor和长期记忆State的多智能体系统并最终部署一个可运行的本地 AI 智能体。无论你是希望理解 Agent 架构设计还是需要将 LangGraph 应用于实际项目本文都将提供一条从入门到实战的清晰路径。1. 理解 LangGraph为什么图是构建智能体的最佳抽象在深入代码之前必须理解 LangGraph 解决的核心问题。传统的链式调用Chain在处理线性任务时表现良好但面对需要循环、分支、回溯或并行执行的复杂场景时其表达能力就显得捉襟见肘。例如一个客服机器人可能需要根据用户意图决定调用知识库查询、订单状态检查或人工坐席转接等多个工具并且这些调用可能不是一次性的而是根据中间结果动态调整的。1.1 图Graph与状态State模型LangGraph 将整个智能体的工作流抽象为一个有向图。图中的节点Node代表一个可执行单元例如调用一个大语言模型、执行一个工具Tool或者进行逻辑判断。边Edge则定义了节点之间的流转条件决定了执行完一个节点后下一步应该走向哪里。这种抽象天然适合描述包含判断、循环和并发的业务流程。与图紧密相关的是状态State模型。在 LangGraph 中一个类型化的State对象贯穿整个图的执行过程。每个节点都可以读取和修改这个共享状态。这解决了传统链式调用中状态传递隐式、易丢失的问题。常见的状态类型是MessageState它专门用于管理对话历史但你可以定义任何符合业务需求的TypedDict作为状态。1.2 LangGraph 与 LangChain 的关系与区别这是一个常见的困惑点。LangChain 是一个更广泛的框架提供了与各种大语言模型、向量数据库、工具等集成的组件。你可以把 LangChain 看作是一个“工具箱”和“连接器”。而 LangGraph 是 LangChain 生态系统中的一个专门用于构建有状态、多步骤工作流的库。它依赖于 LangChain 的核心组件如 LLM、 Tools但提供了更强大的流程控制能力。简单来说LangChain擅长“做什么”——集成模型、调用工具、检索文档。LangGraph擅长“按什么顺序、在什么条件下做”——编排复杂的、有状态的执行流程。当你需要构建一个能进行多轮交互、根据历史决策的智能体时LangGraph 是你的首选。1.3 核心概念节点Node、边Edge与检查点Checkpoint节点Node一个接收状态、执行操作、返回新状态的函数。它可以是调用 LLM、运行工具或者一个简单的逻辑函数。边Edge连接节点的路径。分为条件边Conditional Edge和普通边。条件边允许根据当前状态的值动态决定下一个节点这是实现分支和循环的关键。检查点CheckpointLangGraph 支持在执行的特定点保存状态快照。这使得工作流可以暂停、恢复甚至实现类似“长期记忆”的机制对于构建复杂的、可中断的对话系统至关重要。理解了这些我们就知道 LangGraph 不是替代 LLM而是为 LLM 驱动的智能体提供了一个可靠、可调试的执行引擎。2. 环境准备与核心依赖配置在开始构建智能体之前需要搭建一个稳定的 Python 开发环境。本文将使用开源模型通过 Ollama 运行和 LangChain/LangGraph 的最新稳定版本来演示确保所有步骤都可以在本地复现。2.1 创建虚拟环境与安装依赖首先创建一个独立的 Python 环境以避免包冲突。# 创建并激活虚拟环境以 conda 为例也可使用 venv conda create -n langgraph-demo python3.10 -y conda activate langgraph-demo # 安装核心框架 pip install langgraph langchain langchain-community # 安装用于本地模型交互的库 pip install ollama # 可选用于可视化图的库 pip install pygraphviz注意pygraphviz的安装可能需要系统级的 Graphviz 开发库。在 Ubuntu 上可以运行sudo apt-get install graphviz graphviz-dev在 macOS 上可以运行brew install graphviz。2.2 启动本地模型服务Ollama我们将使用 Ollama 在本地运行开源大语言模型这比调用远程 API 更快速、私密且无成本。前往 Ollama 官网 下载并安装对应操作系统的客户端。安装完成后在终端拉取一个轻量级模型例如llama3.2或qwen2.5。# 拉取模型 ollama pull llama3.2:3b # 或 ollama pull qwen2.5:3b确保 Ollama 服务在后台运行。安装后通常会自动启动服务。2.3 验证环境创建一个简单的 Python 脚本测试 LangChain 能否成功调用本地 Ollama 模型。# test_env.py from langchain_community.llms import Ollama # 初始化本地 LLM llm Ollama(modelllama3.2:3b) # 进行一次简单调用 response llm.invoke(请用中文回答什么是人工智能) print(response)运行此脚本python test_env.py如果能看到模型返回的连贯中文回答说明环境配置成功。3. 构建你的第一个 LangGraph 智能体单节点工作流我们从最简单的图开始一个只包含一个节点调用 LLM的工作流。目标是理解如何定义状态、创建图和运行它。3.1 定义状态State状态是一个TypedDict它规定了图中可以流转哪些信息。我们从最简单的对话状态开始。# simple_agent.py from typing import TypedDict, List from langgraph.graph import StateGraph, END # 1. 定义状态包含消息列表 class AgentState(TypedDict): messages: List[str] # 存储对话历史 # 2. 定义节点函数 def call_model(state: AgentState) - AgentState: 节点调用LLM生成回复 from langchain_community.llms import Ollama llm Ollama(modelllama3.2:3b) # 获取最新的用户消息这里简单处理取最后一条 user_input state[messages][-1] if state[messages] else 你好 prompt f用户说{user_input}\n请给出友好、简洁的回复 # 调用模型 response llm.invoke(prompt) # 更新状态将AI回复加入消息列表 new_messages state[messages] [fAI: {response}] return {messages: new_messages} # 3. 构建图 graph_builder StateGraph(AgentState) graph_builder.add_node(assistant, call_model) # 添加名为“assistant”的节点 graph_builder.set_entry_point(assistant) # 设置入口节点 graph_builder.add_edge(assistant, END) # 设置出口执行完即结束 # 编译图 graph graph_builder.compile() # 4. 运行图 initial_state {messages: [用户: 今天的天气怎么样]} result graph.invoke(initial_state) print(最终状态中的消息, result[messages])这个例子中我们定义了一个状态AgentState它只有一个字段messages。图只有一个节点assistant该节点读取状态中的最后一条消息调用 LLM 生成回复并将回复追加到messages中然后流程结束。3.2 理解图的编译与执行graph.compile()是关键一步它将我们定义的节点和边编译成一个可执行的对象。graph.invoke(initial_state)则是以初始状态启动图的执行。图会从入口节点开始按照边的定义依次执行节点直到到达END。执行上述代码你会看到类似[用户: 今天的天气怎么样, AI: 今天天气晴朗适合外出。]的输出。虽然简单但这已经是一个完整的有状态工作流。4. 实现多智能体与监督者Supervisor架构单智能体能力有限。现实任务往往需要多个专家智能体协作并由一个监督者Supervisor来协调。例如一个任务可能先由“分类器”判断类型再路由给“翻译器”或“总结器”处理。4.1 设计多智能体系统我们将构建一个包含三个智能体的系统翻译智能体Translator负责将输入翻译成英文。总结智能体Summarizer负责总结文本内容。问答智能体QA负责回答基于文本的问题。监督者Supervisor根据用户输入的意图决定将任务派发给哪个智能体。4.2 定义包含意图识别的状态我们需要扩展状态以包含路由决策。# multi_agent.py from typing import TypedDict, List, Literal, Optional from langgraph.graph import StateGraph, END from langchain_community.llms import Ollama # 定义更丰富的状态 class MultiAgentState(TypedDict): messages: List[str] current_agent: Optional[str] # 当前执行的智能体名称 next_agent: Optional[str] # 监督者决定的下一个智能体 final_output: Optional[str] # 最终输出 # 初始化一个共享的LLM避免重复创建 llm Ollama(modelqwen2.5:3b) def supervisor_node(state: MultiAgentState) - MultiAgentState: 监督者节点分析用户意图路由到对应智能体 user_input state[messages][-1] prompt f 请分析用户意图并只返回以下三个选项之一 - translator: 如果用户要求翻译或内容涉及多语言。 - summarizer: 如果用户要求总结、概括或提炼要点。 - qa: 如果用户提出了一个具体问题需要回答。 用户输入{user_input} 意图 intent llm.invoke(prompt).strip().lower() # 简单清理LLM输出确保是三个选项之一 if translator in intent: next_agent translator elif summarizer in intent: next_agent summarizer elif qa in intent: next_agent qa else: next_agent translator # 默认路由 return {next_agent: next_agent} def translator_node(state: MultiAgentState) - MultiAgentState: 翻译智能体节点 user_input state[messages][-1] prompt f将以下中文文本翻译成英文{user_input} translation llm.invoke(prompt) return {final_output: f翻译结果{translation}, current_agent: translator} def summarizer_node(state: MultiAgentState) - MultiAgentState: 总结智能体节点 user_input state[messages][-1] prompt f用中文总结以下文本的核心内容{user_input} summary llm.invoke(prompt) return {final_output: f总结结果{summary}, current_agent: summarizer} def qa_node(state: MultiAgentState) - MultiAgentState: 问答智能体节点 user_input state[messages][-1] prompt f请基于你的知识回答以下问题{user_input} answer llm.invoke(prompt) return {final_output: f答案{answer}, current_agent: qa}4.3 构建带有条件路由的图关键步骤在于监督者节点执行后需要根据其输出的next_agent值动态选择下一个节点。这需要使用add_conditional_edges。# 继续 multi_agent.py # 构建图 builder StateGraph(MultiAgentState) # 添加节点 builder.add_node(supervisor, supervisor_node) builder.add_node(translator, translator_node) builder.add_node(summarizer, summarizer_node) builder.add_node(qa, qa_node) # 设置入口点为监督者 builder.set_entry_point(supervisor) # 定义条件路由函数 def route_after_supervisor(state: MultiAgentState) - str: 根据监督者决定的 next_agent 返回下一个节点名 return state.get(next_agent, translator) # 添加条件边从supervisor出发根据条件路由到三个工作节点之一 builder.add_conditional_edges( supervisor, route_after_supervisor, { translator: translator, summarizer: summarizer, qa: qa, } ) # 为每个工作节点添加指向 END 的边执行完即结束 builder.add_edge(translator, END) builder.add_edge(summarizer, END) builder.add_edge(qa, END) # 编译图 graph builder.compile() # 运行测试 print( 测试多智能体系统 ) test_inputs [ 将‘你好世界’翻译成英文, 概括一下《红楼梦》的主要情节, 珠穆朗玛峰的高度是多少 ] for inp in test_inputs: print(f\n用户输入{inp}) initial_state {messages: [inp], current_agent: None, next_agent: None, final_output: None} result graph.invoke(initial_state) print(f执行代理{result[current_agent]}) print(f最终输出{result[final_output]})运行此脚本你会看到对于不同的输入监督者成功地将任务路由到了不同的智能体并得到了相应的输出。这便是一个最基本的多智能体协作系统。5. 集成长期记忆与复杂状态管理上述示例中状态在单次执行后即丢弃。为了实现多轮对话和上下文感知我们需要引入长期记忆机制。LangGraph 的Checkpointer和更复杂的状态设计是实现这一目标的关键。5.1 使用 MessageState 管理对话历史LangGraph 预定义了MessageState它专门用于处理对话场景内部使用list[BaseMessage]来存储消息。我们改造之前的例子使用MessageState并加入记忆。# memory_agent.py from typing import Literal from langgraph.graph import StateGraph, END from langgraph.checkpoint.memory import MemorySaver # 内存检查点用于演示 from langchain_core.messages import HumanMessage, AIMessage from langchain_community.llms import Ollama # 1. 导入预定义的MessageState from langgraph.graph import MessageState llm Ollama(modelllama3.2:3b) # 2. 定义节点函数现在接收和返回的是MessageState def call_llm(state: MessageState): 节点调用LLM基于完整对话历史生成回复 # state[“messages”] 是一个 BaseMessage 列表 messages state[messages] # 调用模型传入整个历史 response llm.invoke(messages) # 将AI回复作为AIMessage加入状态 return {messages: [AIMessage(contentresponse)]} # 3. 构建图并配置检查点Checkpointer memory MemorySaver() # 使用内存存储检查点生产环境可换为数据库 builder StateGraph(state_schemaMessageState) builder.add_node(assistant, call_llm) builder.set_entry_point(assistant) builder.add_edge(assistant, END) # 编译图时传入检查点管理器 graph builder.compile(checkpointermemory) # 4. 运行图并保存线程ThreadID以实现多轮对话 config {configurable: {thread_id: user_123}} # 唯一线程ID # 第一轮对话 initial_state {messages: [HumanMessage(content我叫小明。)]} result1 graph.invoke(initial_state, configconfig) print(第一轮回复, result1[messages][-1].content) # 第二轮对话图会从检查点恢复状态包含历史消息 result2 graph.invoke({messages: [HumanMessage(content我刚才说我叫什么名字)]}, configconfig) print(第二轮回复有记忆, result2[messages][-1].content)通过使用MemorySaver和唯一的thread_id我们为对话创建了持久化的线程。每次调用invoke时如果传入相同的thread_idLangGraph 会从检查点加载之前的状态从而实现跨轮次的记忆。MessageState自动处理了消息的累加。5.2 设计自定义的长期记忆模块对于更复杂的场景如需要从大量历史中筛选相关记忆可以结合向量数据库。思路是将对话摘要或关键信息存入向量库在需要时进行检索。# 伪代码示例结合向量数据库的记忆模块 class LongTermMemory: def __init__(self, vector_store): self.store vector_store def remember(self, query: str, k: int3): 检索相关记忆 return self.store.similarity_search(query, kk) def memorize(self, text: str, metadata: dict): 存储新的记忆 self.store.add_texts([text], metadatas[metadata]) # 在智能体节点中可以先检索记忆再将记忆作为上下文注入给LLM这超出了本文基础范围但指出了扩展方向将 LangGraph 的状态管理与外部的知识库相结合可以构建出能力强大的、有长期记忆的智能体。6. 运行、调试与可视化构建复杂的图之后如何调试和验证其执行流程至关重要。6.1 使用 LangGraph Studio 进行可视化可选LangGraph 提供了 Studio 工具可以可视化图结构并逐步调试。安装后可以通过编写一个简单的描述文件来启动。# 安装 langgraph-cli pip install langgraph-cli # 在项目目录下创建 langgraph.json 描述文件 # 然后运行 langgraph dev这会在本地启动一个 Web 服务允许你上传图定义并可视化执行步骤。对于复杂工作流这是一个强大的调试助手。6.2 在代码中跟踪执行状态更直接的方式是在invoke时启用流式输出观察执行路径。# 使用 stream 模式运行图观察节点执行顺序 for event in graph.stream(initial_state, configconfig, stream_modevalues): node_name list(event.keys())[0] print(f执行节点: {node_name}) # 可以进一步打印 state 的变化 # print(f状态: {event[node_name]})此外在每个节点函数内部添加详细的日志打印是定位问题最有效的方法。7. 常见问题排查与优化实践在实际开发中你会遇到各种问题。下表列出了一些典型问题及其排查思路问题现象可能原因检查与解决思路图编译失败状态State类型定义错误节点函数签名与状态不匹配。1. 检查TypedDict的字段名和类型是否与节点函数中读写的一致。2. 确保节点函数返回一个字典其键是状态的子集。节点未被调用边Edge设置错误入口点Entry Point设置错误。1. 使用graph.get_graph().draw_mermaid()输出图结构检查节点连接是否正确。2. 确认set_entry_point设置的是已添加的节点名。状态更新不生效节点函数修改了局部变量但未正确返回更新字典多个节点并发写入冲突需用send更新。1. 节点函数必须返回{“field_name”: new_value}。2. 对于复杂并发研究StateGraph的send方法。条件路由Conditional Edge不工作路由函数返回的值不在预设的映射中路由函数逻辑错误。1. 在路由函数内打印state和返回值确保返回值是add_conditional_edges中定义的键之一。2. 检查LLM在监督者节点中的输出是否被正确解析。内存检查点未保存未在compile时传入checkpointer每次调用使用了不同的thread_id。1. 确认builder.compile(checkpointerMemorySaver())。2. 确保多轮对话中config里的thread_id保持不变。执行速度慢节点中的 LLM 调用是同步阻塞的图结构存在不必要的串行。1. 考虑使用异步节点async def和ainvoke。2. 分析图将无依赖的节点设置为并发执行add_edge时指定多个目标。7.1 最佳实践建议状态设计最小化只将需要在节点间传递的数据放入 State。避免将整个应用上下文塞进去。节点职责单一每个节点应只完成一件明确的事情。这有利于测试、复用和调试。善用检查点对于耗时长的流程或需要暂停/恢复的场景检查点是必备功能。错误处理在关键节点尤其是调用外部 API 或工具时添加try...except并考虑将错误信息写入状态由专门的错误处理节点来响应。测试驱动为每个节点函数编写单元测试为整个图编写集成测试模拟各种输入和状态。生产环境部署考虑使用更持久化的检查点存储如 PostgreSQL并为图执行添加超时、重试和监控机制。8. 从演示到生产架构扩展思考本文的示例均在单机、单进程内运行。要将其发展为可服务大量用户的生产级系统需要考虑以下扩展分布式执行对于计算密集或 IO 密集的节点可以将其部署为独立的微服务LangGraph 通过远程调用RPC来协调。Ray 是一个值得考虑的分布式执行框架可以与 LangGraph 结合。高可用检查点将MemorySaver替换为基于数据库如 Redis, PostgreSQL的检查点存储实现确保状态持久化和多实例共享。可观测性在图执行过程中注入追踪Tracing信息收集每个节点的耗时、输入输出和错误集成到如 LangSmith 或 OpenTelemetry 等可观测性平台。版本化管理图的定义节点和边会随着业务迭代而变化。需要建立图的版本管理机制实现灰度发布和回滚。LangGraph 提供的是一种强大的编排范式它将智能体的“决策逻辑”图定义与“执行环境”节点实现解耦。掌握这种范式后你可以根据业务复杂度灵活地选择从简单的内存图到复杂的分布式工作流引擎构建出真正可靠、可维护的企业级 Agent 架构。