LangGraph实战：从单智能体到多智能体协作的工程化指南

30款热门AI模型一站整合DeepSeek/GLM/Qwen 随心用限时 5 折。 点击领海量免费额度最近在尝试把一些零散任务自动化时我遇到了一个典型困境单个大模型调用能解决简单问题但面对稍微复杂、需要多步骤协作的场景就显得力不从心。要么是上下文太长导致模型“失忆”要么是任务拆解不清晰结果来回打补丁代码越写越乱。这让我重新审视了“AI智能体”这个概念。它远不止是让模型调用个工具那么简单。真正的价值在于它把一次性的、脆弱的提示工程变成了一个可定义、可编排、可复用的工作流系统。而LangChain和LangGraph正是构建这类系统的核心框架。很多人一上来就纠结于“LangChain和LangGraph有什么区别”、“OpenClaw怎么装”。这些问题当然重要但如果只停留在工具安装和API调用很容易陷入“调参工程师”的困境只见树木不见森林。这篇文章我想和你聊点不一样的。我们不只讲“怎么搭”更要讲清楚“为什么这么搭”以及从单智能体到多智能体协作整个设计思路的演进。我会用一个从简单到复杂的案例手把手带你理解如何用LangGraph搭建一个具备感知、决策、执行能力的智能体并最终将其工程化部署。你会发现核心难点往往不在代码本身而在于如何清晰地定义状态、划分职责和管理通信。1. 重新理解“智能体”从单次调用到可编排的工作流在开始写代码之前我们必须先统一认知当我们谈论“AI智能体”时到底在谈论什么一个常见的误解是智能体就是一个能调用工具的LLM。这个定义只对了一半。一个孤立的、能调用搜索工具的ChatGPT可以算作一个“工具调用者”但未必是一个合格的“智能体”。智能体的核心特征在于持续的自主性和基于状态的目标驱动行为。想象一下人类处理复杂任务的过程接到一个任务状态输入分析并制定计划决策执行计划中的某个步骤行动观察结果状态更新根据新状态决定下一步是继续执行、调整计划还是求助他人循环决策。这个过程是循环的、有状态的、可调整的。LangChain早期提供的AgentExecutor其实已经实现了这个循环的简化版LLM思考Think- 决定调用工具Act- 观察工具结果Observe- 继续思考。但它的问题在于这个循环是黑盒的状态管理尤其是复杂的多轮记忆和流程控制如条件分支、并行执行的定制能力较弱。而LangGraph的诞生正是为了将这种“智能体工作流”显式化、可视化、可编程化。它用“图”Graph的概念来建模智能体节点Node代表一个处理单元可以是LLM调用、工具执行、条件判断边Edge代表控制流和数据流。这样一来智能体的推理逻辑不再是隐藏在提示词中的“魔法”而是一个清晰可见、可调试、可扩展的工程结构。所以LangChain和LangGraph的关系可以这样理解LangChain提供了构建智能体所需的基础组件如模型封装、提示模板、记忆模块、工具链以及最基础的Agent执行器。它像是一个丰富的“零件库”。LangGraph则提供了组装这些零件、构建复杂工作流引擎的能力。它定义了零件之间如何连接、数据如何流动、流程如何控制。它关注的是系统的“骨架”和“神经系统”。对于智能体开发我的建议是用LangChain准备“材料”用LangGraph设计“蓝图”并搭建“房子”。2. 搭建你的第一个智能体从“状态”定义开始让我们暂时忘掉那些复杂的多智能体架构先从构建一个最简单的、具有思考-行动循环的单一智能体开始。这个智能体能根据用户问题决定是直接回答还是调用一个搜索工具。在LangGraph中构建任何工作流的第一步永远是定义状态State。状态是整个图运行时共享的数据容器它决定了节点之间能传递什么信息。from typing import TypedDict, Annotated, List from langgraph.graph.message import add_messages import operator class AgentState(TypedDict): # 消息历史这是智能体与用户、工具交互的核心记录 messages: Annotated[List, add_messages] # 用户的最新问题我们可以选择将其从历史中分离出来便于节点访问 user_input: str # 其他任何你想在流程中传递的上下文比如已执行的步骤、中间结果等 # intermediate_results: List[str]这里我们定义了一个简单的状态它主要包含消息历史。Annotated和add_messages是LangGraph提供的语法糖用于自动合并消息列表非常方便。接下来我们定义这个智能体的两个核心“器官”大脑LLM节点负责思考决定下一步做什么。手工具节点负责执行具体的动作比如搜索。我们先准备“手”——一个简单的搜索工具这里用模拟函数代替真实网络请求from langchain_core.tools import tool tool def search_web(query: str): 在互联网上搜索信息。当用户的问题需要最新或外部知识时使用此工具。 # 模拟搜索返回 print(f[工具调用] 正在搜索: {query}) # 这里应该接入真实的搜索API如Serper、Tavily等 simulated_result f根据搜索{query}获得的结果是AI智能体是一种能感知环境、制定决策并执行行动的程序。 return simulated_result然后我们构建“大脑”。这个节点需要做三件事接收当前状态主要是对话历史让LLM思考并根据LLM的决定来指导下一步流程。from langchain_openai import ChatOpenAI from langgraph.prebuilt import ToolNode from langgraph.graph import END # 初始化LLM model ChatOpenAI(modelgpt-4o-mini, temperature0) # 将工具包装成列表供LLM调用 tools [search_web] model_with_tools model.bind_tools(tools) def llm_node(state: AgentState): 智能体的‘大脑’。 1. 获取最新的用户消息。 2. 让LLM根据对话历史和工具描述进行思考。 3. 根据LLM的输出是直接回复还是调用工具来决定下一步。 print(f[LLM节点] 当前消息数: {len(state[messages])}) # 调用LLM response model_with_tools.invoke(state[messages]) # 关键LLM的响应可能是一个普通的AIMessage也可能是一个带有工具调用的AIMessage if response.tool_calls: print(f[LLM节点] 决定调用工具: {[tc[name] for tc in response.tool_calls]}) # 如果有工具调用我们将包含此响应的消息添加到历史中然后路由到‘工具执行节点’ return {messages: [response]} # 状态更新添加LLM的工具调用请求 else: print([LLM节点] 决定直接回复用户。) # 如果LLM直接回复我们将回复添加到历史中并路由到流程终点END return {messages: [response]} # 状态更新添加LLM的最终回复现在我们需要告诉LangGraph当llm_node执行完后下一步该去哪。这就是边Edge的逻辑。我们根据LLM是否有工具调用来决定路由。def route_after_llm(state: AgentState): 路由函数检查LLM节点的输出决定下一步是执行工具还是结束。 last_message state[messages][-1] # 如果最后一条消息是AIMessage且包含工具调用则去执行工具 if hasattr(last_message, tool_calls) and last_message.tool_calls: return tool_node # 否则流程结束 return END最后我们把所有零件组装起来形成一张“图”。from langgraph.graph import StateGraph, START # 创建图构建器并指定我们定义的状态类型 builder StateGraph(AgentState) # 添加节点 builder.add_node(llm_node, llm_node) # “大脑”节点 # 使用预建的ToolNode作为“手”节点它会自动执行LLM请求调用的工具 builder.add_node(tool_node, ToolNode(tools)) # “手”节点 # 设置起点 builder.add_edge(START, llm_node) # 设置条件边llm_node之后根据路由函数的结果走 builder.add_conditional_edges( llm_node, route_after_llm, # 路由判断函数 { tool_node: tool_node, # 如果返回“tool_node”则前往tool_node END: END # 如果返回END则结束 } ) # 设置固定边tool_node执行完后必须回到llm_node进行下一步思考 builder.add_edge(tool_node, llm_node) # 编译图得到可执行的工作流 agent_graph builder.compile()现在让我们运行这个简单的智能体# 初始化状态 initial_state AgentState(messages[(user, 什么是AI智能体)], user_input什么是AI智能体) # 执行图 final_state agent_graph.invoke(initial_state) # 查看最终的消息历史 for msg in final_state[messages]: print(f{msg.type}: {msg.content})这个智能体会先思考发现可以直接回答于是输出答案。如果你问“今天北京的天气怎么样”LLM可能会决定调用search_web工具图就会在llm_node-tool_node-llm_node之间循环直到LLM给出最终答案。至此你已经完成了一个具备基础ReActReasoning-Acting能力的智能体。它的工作流清晰可见开始 - 思考 - (执行工具 - 思考)循环 - 结束。3. 升级为多智能体协作架构选择与通信设计单一智能体处理简单任务尚可但面对“分析一篇技术文章总结其要点并评估其市场前景”这类复合任务时就会显得笨拙。我们需要专业化分工。多智能体系统的核心优势在于模块化和专业化。你可以为不同子任务设计专家智能体如“分析员”、“总结员”、“评估员”。但随之而来的问题是它们如何协作谁来决定下一个谁工作LangGraph文档中提到了几种经典的多智能体架构我们需要根据业务场景来选择架构核心思想适用场景在LangGraph中的关键实现主管Supervisor一个中央“主管”智能体负责接收任务并决定调用哪个下属智能体或何时结束。任务有明确的阶段划分且需要一个中心节点做调度。如客服系统主管先理解问题再分派给“技术客服”或“订单客服”。主管节点返回Command(goto“某智能体”)。下属智能体完成后返回Command(goto“supervisor”)。网络Network每个智能体都可以直接与其他智能体通信共同协商完成任务。去中心化、协作性强的场景如多角色辩论、头脑风暴。每个智能体节点都可以根据自身逻辑返回Command跳转到任意其他节点。层级式Hierarchical多层主管结构。顶层主管管理多个团队主管每个团队主管管理自己的专家智能体。超大型、任务高度结构化的系统。如一个公司CEO - 部门总监 - 员工。将子团队本身也编译成子图Subgraph作为父图的一个节点。工具调用式Supervisor with Tool Calling将下属智能体“包装”成工具主管智能体像使用普通工具一样调用它们。希望复用现有单智能体模式且交互逻辑简单的场景。主管无需理解复杂的状态传递。使用create_react_agent创建主管将子智能体函数作为tools传入。让我们以实现一个“主管架构”为例构建一个包含“分析员”、“总结员”、“评估员”的智能体团队。首先定义团队状态和主管from typing import Literal from langgraph.types import Command class TeamState(TypedDict): messages: Annotated[List, add_messages] # 主管需要知道该让谁工作 next: Literal[analyst, summarizer, evaluator, __end__] # 各专家智能体的私有工作区可选用于复杂状态隔离 # analyst_output: str # summarizer_output: str # 主管智能体决定任务分派 def supervisor_node(state: TeamState) - Command[Literal[analyst, summarizer, evaluator, __end__]]: 主管节点。分析当前对话和任务决定下一步由哪个专家执行。 # 这里可以基于简单规则也可以让一个LLM来做决策 last_msg state[messages][-1].content if state[messages] else if 分析 in last_msg or not state.get(analyst_done, False): print([主管] 分派给分析员。) return Command(gotoanalyst) elif 总结 in last_msg or not state.get(summarizer_done, False): print([主管] 分派给总结员。) return Command(gotosummarizer) elif 评估 in last_msg or not state.get(evaluator_done, False): print([主管] 分派给评估员。) return Command(gotoevaluator) else: print([主管] 任务完成结束。) return Command(goto__end__)然后定义三个专家智能体。它们完成任务后都需要将控制权交还给主管。# 分析员智能体 def analyst_node(state: TeamState) - Command[Literal[supervisor]]: print([分析员] 开始分析文章...) # 模拟分析工作 analysis_result 文章核心观点是LangGraph通过图计算模型让多智能体协作变得可编程。 new_message (assistant, f分析结果{analysis_result}) # 更新状态并告知主管我完成了 return Command( gotosupervisor, update{ messages: [new_message], analyst_done: True, # 在状态中标记本步骤完成 analysis: analysis_result # 传递结果给后续节点 } ) # 总结员智能体 def summarizer_node(state: TeamState) - Command[Literal[supervisor]]: print([总结员] 开始总结分析结果...) analysis state.get(analysis, ) summary f基于分析‘{analysis}’总结要点如下1. 核心是图计算2. 关键在可编程性。 new_message (assistant, f总结结果{summary}) return Command( gotosupervisor, update{ messages: [new_message], summarizer_done: True, summary: summary } ) # 评估员智能体 def evaluator_node(state: TeamState) - Command[Literal[supervisor]]: print([评估员] 开始评估市场前景...) summary state.get(summary, ) evaluation f基于总结‘{summary}’评估其市场前景在自动化流程领域有较高应用价值。 new_message (assistant, f评估结果{evaluation}) return Command( gotosupervisor, update{ messages: [new_message], evaluator_done: True } )最后组装这个多智能体图builder StateGraph(TeamState) builder.add_node(supervisor, supervisor_node) builder.add_node(analyst, analyst_node) builder.add_node(summarizer, summarizer_node) builder.add_node(evaluator, evaluator_node) # 设置起点为主管 builder.add_edge(START, supervisor) # 主管根据Command.goto动态路由到各个专家节点 # 专家节点完成后固定返回主管节点 builder.add_edge(analyst, supervisor) builder.add_edge(summarizer, supervisor) builder.add_edge(evaluator, supervisor) team_graph builder.compile()运行这个团队initial_team_state TeamState( messages[(user, 请分析、总结并评估这篇关于LangGraph的文章。)], nextsupervisor, analyst_doneFalse, summarizer_doneFalse, evaluator_doneFalse ) # 使用流式接口可以看清执行过程 for step in team_graph.stream(initial_team_state, subgraphsTrue): print(f当前活跃节点: {list(step.keys())})这个例子展示了多智能体协作的基本模式中心调度专业分工状态共享。主管根据任务进度通过状态中的_done标志或消息内容来协调专家们的工作顺序。4. 工程化与部署从原型到可用的服务搭建好一个能在笔记本里跑通的智能体图只是万里长征第一步。要让它成为一个真正可用的服务我们必须考虑工程化问题状态持久化、错误处理、可观测性、以及部署。4.1 状态持久化与长期记忆智能体的对话历史状态是其“记忆”。对于一次会话内的多轮交互我们需要将会话状态保存下来以便下次能接着聊。LangGraph内置了检查点Checkpoint机制。from langgraph.checkpoint import MemorySaver # 创建一个内存检查点存储器生产环境应使用数据库如Redis、PostgreSQL memory MemorySaver() # 在编译图时启用检查点 persistent_graph builder.compile(checkpointermemory) # 调用时需要指定一个线程IDthread_id来标识会话 config {configurable: {thread_id: user_123_session_1}} initial_state TeamState(...) # 第一次调用状态会被保存 result1 persistent_graph.invoke(initial_state, configconfig) # 第二次调用会从上次的检查点恢复状态 result2 persistent_graph.invoke({messages: [(user, 能再详细说说分析部分吗)]}, configconfig)这样智能体就能记住之前的对话上下文。生产环境中你需要将MemorySaver替换为RedisSaver或PostgresSaver等持久化存储。4.2 错误处理与鲁棒性智能体在调用外部工具、模型时可能失败。一个健壮的智能体需要容错机制。工具调用超时/失败可以在ToolNode外层包裹错误处理逻辑或在自定义工具函数中使用try-catch返回明确的错误信息给LLM。LLM输出格式异常使用LangChain的with_structured_output或Pydantic来强制LLM返回结构化数据避免解析失败。图执行中断利用LangGraph的interrupt机制在特定节点如需要人工审核时暂停执行等待外部输入后再继续。4.3 可观测性与调试当智能体工作流变得复杂时调试是个挑战。LangGraph提供了强大的可视化和追踪Tracing功能。# 方法1打印图结构非常直观 from IPython.display import Image, display try: display(Image(persistent_graph.get_graph().draw_mermaid_png())) except: print(如需生成图像请安装pygraphviz或pydot。) # 方法2使用流式输出实时观察执行路径 for step in persistent_graph.stream(initial_state, configconfig, subgraphsTrue): node_name list(step.keys())[0] print(f 进入节点: {node_name}) # 可以在这里记录日志或发送到监控系统 # 方法3集成LangSmithLangChain的官方监控平台可以查看每次调用的详细链路、耗时、输入输出。4.4 部署为API服务最终我们需要将智能体图暴露为API。可以使用FastAPI等框架轻松包装。from fastapi import FastAPI, HTTPException from pydantic import BaseModel import uvicorn app FastAPI(titleAI智能体团队API) class ChatRequest(BaseModel): thread_id: str # 会话ID message: str # 用户输入 class ChatResponse(BaseModel): response: str # 智能体回复 thread_id: str app.post(/chat, response_modelChatResponse) async def chat_with_agent(req: ChatRequest): config {configurable: {thread_id: req.thread_id}} try: # 调用持久化的图 final_state persistent_graph.invoke( {messages: [(user, req.message)]}, configconfig ) # 提取最后一条AI消息作为回复 last_msg final_state[messages][-1] if last_msg.type ! ai: raise HTTPException(status_code500, detailAgent did not return an AI message.) return ChatResponse(responselast_msg.content, thread_idreq.thread_id) except Exception as e: raise HTTPException(status_code500, detailfAgent execution failed: {str(e)}) if __name__ __main__: uvicorn.run(app, host0.0.0.0, port8000)现在你的多智能体系统就变成了一个标准的Web服务可以被前端或其他系统调用。关于OpenClaw在搜索材料中频繁出现的“OpenClaw”通常是特定厂商如火山引擎提供的AI智能体开发平台或框架。它的定位可以类比为云原生的、企业级的LangChain/LangGraph发行版。它可能提供了托管服务无需自己部署模型和向量数据库。可视化编排通过拖拽搭建智能体工作流。预置组件集成了多种模型、工具和数据连接器。运维监控企业级的权限、审计和性能监控。如果你的团队追求快速上线、不想管理底层基础设施且业务在特定云平台上那么类似OpenClaw的平台是一个不错的选择。但如果你需要深度定制、对架构有完全控制权、或希望技术栈保持灵活那么从开源的LangGraph开始自建是更扎实的路径。5. 避坑指南与进阶思考在实战中有几个关键点极易被忽略却直接影响智能体的稳定性和效果。1. 状态设计是重中之重状态是智能体系统的“血液”。设计不佳的状态会导致信息传递混乱。建议最小化共享状态只在不同节点间必需传递的数据才放入全局状态。专家智能体的中间过程可放在私有状态。使用强类型如TypedDict便于IDE提示和错误检查。考虑状态版本化当升级智能体逻辑时旧检查点可能不兼容需要设计迁移策略。2. 工具描述的质量决定智能体能力上限LLM通过工具描述来理解何时、如何使用工具。模糊的描述会导致误调用。清晰search_web(query: str)就比search(query: str)好。具体在描述中说明工具的精确用途、输入格式、输出示例。适度不要一次性给智能体太多工具如超过10个容易造成选择困难。可以动态加载工具集。3. 控制流复杂度与可维护性的权衡LangGraph非常灵活可以构建极其复杂的流程图。但越复杂调试越难。遵循“单向流”尽量让控制流清晰单向避免形成难以理解的循环。多用子图Subgraph封装将功能内聚的多个节点封装成一个子图作为父图的一个节点简化顶层视图。文档化你的图用注释或文档说明每个节点的职责和边的条件。4. 成本与延迟优化多智能体意味着多次LLM调用成本和时间激增。缓存对确定性高的工具调用或LLM响应进行缓存。轻量模型分工让主管或路由节点使用轻量、快速的模型如GPT-4o-mini专家节点再用更强的模型。异步执行对于可以并行的任务利用LangGraph的StateGraph的并发能力。5. 测试策略智能体系统是非确定性的测试挑战大。单元测试节点函数确保每个单独的LLM调用或工具函数逻辑正确。集成测试工作流用典型输入测试整个图断言最终的输出状态或消息符合预期。“金丝雀”测试用小流量真实请求对比新旧版本智能体的输出。构建AI智能体尤其是多智能体系统是一个典型的系统设计问题。它考验的不是你对某个API的熟悉程度而是你如何将一个模糊的业务目标分解成清晰的状态、节点和边并让它们可靠、高效地协作。从这个角度看LangGraph不仅仅是一个库它更是一种关于复杂AI系统编排的思维方式。当你开始用“图”来思考任务流时很多原本纠缠不清的问题会突然变得结构清晰、可被解决。这才是智能体技术带给开发者最深层的价值。 30款热门AI模型一站整合DeepSeek/GLM/Qwen 随心用限时 5 折。 点击领海量免费额度