这类教程最值得先看的不是标题里“吊打付费”这种说法而是它到底能不能帮你把 LangGraph 和多智能体Multi-Agent的概念从一堆抽象术语变成能跑起来的代码。很多人学 LangGraph 卡在第一步知道它是个构建多智能体工作流的框架但一上手就分不清 Agent、State、Node、Edge 这些核心组件怎么拼起来更别说处理循环、分支和长期记忆了。如果你正在找一套能跟着敲、能理解每一步为什么这么写、并且能处理真实任务比如让多个 AI 角色协作分析、决策、调用工具的实战指南那这篇梳理会直接切入要害。我们不谈空泛的“架构优势”直接从环境搭建、核心组件拆解、到写出第一个能处理复杂逻辑的工作流最后再聊怎么把它变得更健壮、更适合生产环境。整个过程我会把那些容易让新手困惑的“坑点”——比如状态State设计、循环控制、工具调用异常——都提前标出来。1. 先拆解 LangGraph 的核心它到底解决了多智能体开发的什么痛点在直接动手写代码之前我们需要先达成一个共识LangGraph 不是一个全新的 AI 模型而是一个编排框架。它的核心价值是帮你把 LangChain 里那些零散的 Agent、Tool、Chain 组织成一个有明确流程的“图”Graph。1.1 多智能体系统的常见混乱场景假设你想做一个智能客服系统里面包含查询理解 Agent分析用户问题意图。知识检索 Agent根据意图去数据库或文档找答案。答案润色 Agent把检索到的信息整理成通顺回复。安全检查 Agent过滤掉不合适的内容。如果用最基础的 LangChain 硬拼你可能会写一堆if...else或者写一个很长的顺序链。问题很快会出现流程僵化所有问题都走固定流程但有些简单问题可能不需要“知识检索”。状态管理困难如何把“查询理解”的结果传递给“知识检索”如果“安全检查”没通过是重试还是结束循环与分支用户说“不对再查查”系统如何回到“查询理解”或“知识检索”节点调试地狱一个任务流经多个 Agent出错了很难定位是哪个环节、什么输入导致的。LangGraph 就是为了标准化地解决这些问题而设计的。它把整个协作流程抽象成一张“图”节点Node是处理单元Agent、函数、工具边Edge是流转逻辑。你可以清晰地定义什么条件下从哪个节点跳转到哪个节点。1.2 LangGraph 的核心心智模型状态机这是理解 LangGraph 最关键的一步。你可以把它想象成一个共享白板State和一群专家Nodes。共享白板State这是一个所有专家都能看到和修改的字典dict。它记录了当前任务的所有信息比如用户的原始问题input、理解后的意图intent、检索到的文档documents、生成的草稿draft、最终回复response等。这个 State 会在专家们之间传递。专家Nodes每个专家Node负责一项具体工作。比如“理解专家”只负责读白板上的input然后写出intent。“检索专家”只根据intent去查资料然后把结果documents写回白板。调度员Edges Conditional Edges决定下一个该谁看白板。可以是固定的顺序理解专家完了一定是检索专家也可以根据白板上的内容决定如果intent是“问候”就直接跳给回复专家跳过检索。这种设计带来的直接好处是流程可视化整个系统怎么跑的一目了然。状态集中管理所有中间数据都在一个地方调试时打印 State 就行。灵活的路由可以轻松实现“循环”让某个专家反复工作和“分支”根据内容走不同路径。2. 环境准备与核心组件初识别急着安装先想清楚你要什么很多人一上来就pip install langgraph然后对着报错不知所措。我的建议是先根据你的目标明确需要安装哪些包。2.1 依赖选择LangChain 是基础模型服务是引擎LangGraph 是 LangChain 生态系统的一部分。通常你需要一个“大脑”LLM 模型和 LangGraph 框架本身。基础组合推荐新手起步# 核心框架 pip install langgraph langchain # 选择一个模型提供商例如 OpenAI需要API Key pip install openai # 或者使用本地模型例如通过 Ollama免费本地运行 # pip install ollamaOpenAI最稳定文档最全但需要付费 API Key。Ollama免费在本地运行 Llama、Mistral 等开源模型适合实验和内部工具。注意本地模型的推理速度和效果取决于你的电脑配置尤其是 GPU 和内存。生产环境考虑你可能还需要langchain-community更多社区工具和集成、tavily-python联网搜索、langsmith用于链路追踪和监控等。2.2 理解四大核心组件State, Node, Edge, Graph安装好后别急着写复杂图。先用代码感受一下这四个核心对象。下面是一个“超浓缩”的认知代码不运行只看结构from typing import TypedDict, Annotated from langgraph.graph import StateGraph, END # 1. 定义 State我们的“共享白板”长什么样 class AgentState(TypedDict): # Annotated 是 LangGraph 的语法用于添加描述。add 表示这个字段的值是累加的列表。 messages: Annotated[list, 对话消息列表] # 其他你需要传递的信息比如 # user_query: str # search_results: list # 2. 定义 Node一个“专家”的工作函数 def call_llm(state: AgentState): 模拟调用大模型处理消息 # 从白板state上读取信息 latest_message state[“messages”][-1] # 这里应该真实调用 LLM例如response chat_model.invoke(latest_message) # 为了示例我们模拟一个回复 simulated_response “这是根据你的问题生成的模拟回复。” # 把新的回复写回白板的 messages 列表 return {“messages”: state[“messages”] [{“role”: “assistant”, “content”: simulated_response}]} # 3. 构建图 workflow StateGraph(AgentState) # 创建图并指定白板的结构 workflow.add_node(“assistant”, call_llm) # 添加一个名为“assistant”的节点它对应 call_llm 函数 workflow.set_entry_point(“assistant”) # 设置入口节点 workflow.add_edge(“assistant”, END) # 设置从“assistant”节点到结束的边 app workflow.compile() # 编译成可执行的应用 # 4. 运行图 initial_state {“messages”: [{“role”: “user”, “content”: “你好”}]} result app.invoke(initial_state) print(result[“messages”][-1])这段代码虽然简单但包含了所有要素AgentState定义了白板的结构。Annotated[list, add]是关键它告诉 LangGraph 这个messages字段是列表每次节点返回的messages都会追加到原有列表后面而不是覆盖。这是实现对话记忆的基础。call_llm函数这就是一个Node。它接收state处理然后返回一个字典来更新state。返回的字典键必须与AgentState中定义的字段对应。StateGraph图的容器。add_node注册节点。add_edge注册边。END是一个特殊的终止节点。compile()把蓝图变成可执行的app。invoke()输入初始状态运行图。第一个避坑点State的设计是重中之重。设计得太简单信息不够用设计得太复杂节点函数写起来很乱。一开始尽量只放必要字段。3. 实战构建一个具备分支能力的多智能体工作流现在我们来构建一个真实可用的多智能体系统。场景是一个智能助手能根据用户问题类型决定是直接回答简单问候、搜索网络事实性问题还是进行创意写作。3.1 定义更完善的状态与节点我们需要更丰富的 State以及三个不同的 Agent 节点。from typing import TypedDict, Annotated, Literal from langgraph.graph import StateGraph, END from langchain_openai import ChatOpenAI # 使用 OpenAI 模型 from langchain_community.tools.tavily_search import TavilySearchResults # 使用搜索工具 import os # 0. 设置环境变量你的 OpenAI API Key 和 Tavily API Key os.environ[“OPENAI_API_KEY”] “your-openai-key” os.environ[“TAVILY_API_KEY”] “your-tavily-key” # 1. 定义状态 class AssistantState(TypedDict): messages: Annotated[list, “对话消息列表”] next: Literal[“respond”, “search”, “write”, END] # 下一个要执行的节点名 # 2. 初始化模型和工具 llm ChatOpenAI(model“gpt-3.5-turbo”) search_tool TavilySearchResults(max_results2) # 限制搜索结果为2条 # 3. 定义节点函数 def router_node(state: AssistantState): 路由节点分析最新用户消息决定下一步做什么 last_msg state[“messages”][-1][“content”] # 这里用 LLM 判断意图。实际应用中可以用更简单的规则如关键词匹配。 # 为简化我们使用模拟逻辑 if “你好” in last_msg or “hi” in last_msg: next_step “respond” # 简单问候直接回复 elif “谁” in last_msg or “哪里” in last_msg or “何时” in last_msg: next_step “search” # 事实性问题需要搜索 else: next_step “write” # 其他问题当作创意请求 # 关键返回的字典中next 字段用于控制流程 return {“next”: next_step} def general_respond_node(state: AssistantState): 通用回复节点处理简单问候 response “你好我是你的智能助手。” return {“messages”: state[“messages”] [{“role”: “assistant”, “content”: response}], “next”: END} def search_agent_node(state: AssistantState): 搜索智能体节点执行搜索并总结 query state[“messages”][-1][“content”] # 调用搜索工具 search_results search_tool.invoke(query) # 请 LLM 总结搜索结果 summary_prompt f“”” 用户问题{query} 搜索到的信息{search_results} 请根据以上信息生成一个简洁、准确的回答。 “”” ai_msg llm.invoke(summary_prompt) return {“messages”: state[“messages”] [{“role”: “assistant”, “content”: ai_msg.content}], “next”: END} def creative_write_node(state: AssistantState): 创意写作节点 request state[“messages”][-1][“content”] prompt f“你是一个创意写作助手。请根据以下请求进行创作{request}” ai_msg llm.invoke(prompt) return {“messages”: state[“messages”] [{“role”: “assistant”, “content”: ai_msg.content}], “next”: END}3.2 组装图并设置条件边这是 LangGraph 最精彩的部分根据State中的next字段值动态决定下一步。# 4. 构建图 workflow StateGraph(AssistantState) # 添加节点 workflow.add_node(“router”, router_node) workflow.add_node(“responder”, general_respond_node) workflow.add_node(“searcher”, search_agent_node) workflow.add_node(“writer”, creative_write_node) # 设置入口 workflow.set_entry_point(“router”) # 添加条件边router 节点执行完后根据其返回的 state[‘next’] 值决定去哪个节点 workflow.add_conditional_edges( “router”, # 源节点 # 这是一个判断函数它接收最新的 state返回下一个节点的名字 lambda state: state[“next”], # 映射关系router 节点返回的 next 值 - 对应的下一个节点名 { “respond”: “responder”, “search”: “searcher”, “write”: “writer”, END: END } ) # 为其他节点添加固定边指向结束 workflow.add_edge(“responder”, END) workflow.add_edge(“searcher”, END) workflow.add_edge(“writer”, END) # 编译应用 app workflow.compile()3.3 运行与调试现在我们可以像调用函数一样调用这个多智能体工作流了。# 测试1简单问候 print(“ 测试1问候 ) result1 app.invoke({“messages”: [{“role”: “user”, “content”: “你好啊”}], “next”: None}) print(result1[“messages”][-1][“content”]) # 测试2事实性问题 print(“\n 测试2事实查询 ) result2 app.invoke({“messages”: [{“role”: “user”, “content”: “爱因斯坦什么时候出生的”}], “next”: None}) print(result2[“messages”][-1][“content”]) # 测试3创意请求 print(“\n 测试3创意写作 ) result3 app.invoke({“messages”: [{“role”: “user”, “content”: “写一首关于春天的短诗”}], “next”: None}) print(result3[“messages”][-1][“content”])运行这个流程你会看到用户输入“你好啊”router判断next“respond”跳转到responder节点直接回复问候语然后结束。用户输入“爱因斯坦...”router判断next“search”跳转到searcher节点该节点调用搜索工具用 LLM 总结返回答案然后结束。用户输入“写一首诗...”router判断next“write”跳转到writer节点进行创意生成。第二个避坑点add_conditional_edges是实现分支路由的核心。它的判断函数lambda state: state[“next”]必须返回一个字符串这个字符串要在你提供的映射字典里。确保你的节点函数如router_node返回的next值一定是其中之一。4. 进阶引入循环、人工干预与持久化一个基础的多智能体工作流跑通后你会面临更实际的需求任务可能需要多轮循环比如反复提炼问题有时需要人工介入审核并且希望工作流状态能保存下来。4.1 实现循环让智能体“反复思考”假设我们增加一个“审阅”节点如果对答案不满意就循环回“搜索”或“写作”节点重做。# 在 State 中增加一个字段记录审阅结果 class RefinementState(TypedDict): messages: Annotated[list, “对话消息列表”] draft_answer: str # 草稿答案 review_feedback: Literal[“approve”, “needs_search”, “needs_rewrite”] # 审阅意见 next: Literal[“generate”, “review”, “search”, “rewrite”, END] def generate_draft_node(state: RefinementState): 生成草稿节点初始生成 query state[“messages”][-1][“content”] prompt f“请回答以下问题{query}” ai_msg llm.invoke(prompt) return {“draft_answer”: ai_msg.content, “next”: “review”} # 下一步进入审阅 def review_node(state: RefinementState): 模拟审阅节点。实际可以是另一个LLM或人工判断 draft state[“draft_answer”] # 模拟一个简单的审阅逻辑如果答案很短就要求重写如果包含‘不确定’就要求搜索 if len(draft) 50: feedback “needs_rewrite” elif “不确定” in draft: feedback “needs_search” else: feedback “approve” return {“review_feedback”: feedback, “next”: feedback} # 将反馈直接作为下一步的节点名 def rewrite_node(state: RefinementState): 重写节点 query state[“messages”][-1][“content”] feedback “请提供更详细、准确的回答。” prompt f“问题{query}\n审阅意见{feedback}\n请重新回答” ai_msg llm.invoke(prompt) return {“draft_answer”: ai_msg.content, “next”: “review”} # 再次进入审阅形成循环 # 构建带循环的图 workflow StateGraph(RefinementState) workflow.add_node(“generate”, generate_draft_node) workflow.add_node(“review”, review_node) workflow.add_node(“rewrite”, rewrite_node) workflow.add_node(“search”, search_agent_node) # 复用之前的搜索节点 workflow.set_entry_point(“generate”) workflow.add_edge(“generate”, “review”) workflow.add_conditional_edges( “review”, lambda state: state[“review_feedback”], { “approve”: END, “needs_rewrite”: “rewrite”, “needs_search”: “search”, } ) workflow.add_edge(“rewrite”, “review”) # 关键重写后回到审阅形成循环 workflow.add_edge(“search”, “review”) # 搜索后也回到审阅 app workflow.compile()第三个避坑点循环图一定要有终止条件否则会无限循环。在上面的例子中review_node的feedback最终必须有可能返回“approve”从而走向END。在实际应用中你可能还需要设置最大循环次数可以在State里加一个loop_count字段在review_node里检查。4.2 模拟人工干预Human-in-the-Loop有时关键决策需要真人确认。LangGraph 提供了interrupt机制。思路是在特定的节点如review后不自动执行下一条边而是暂停等待外部输入。from langgraph.graph import MessagesState from langgraph.checkpoint import MemorySaver from langgraph.prebuilt import ToolNode, tools_condition # 使用 MemorySaver 实现状态持久化和中断 memory MemorySaver() workflow StateGraph(MessagesState, config_schema…) # … 添加节点和边 … workflow.add_node(“human_review”, …) # 这是一个等待人工输入的节点 # 在需要中断的边之后配置 checkpoint app workflow.compile(checkpointermemory) # 运行直到遇到配置了中断的节点 config {“configurable”: {“thread_id”: “thread_123”}} initial_state {“messages”: [{“role”: “user”, “content”: “重要问题需要人工审核”}]} # 第一次 invoke可能会停在 human_review 节点 result app.invoke(initial_state, config) # 此时app 的状态被保存在 memory 中线程 thread_123 处于暂停状态。 # 模拟人工操作获取当前状态提供输入继续执行 from langgraph.types import Command # 人工给出指令例如批准或修改 human_command Command(resume{“messages”: [{“role”: “user”, “content”: “我批准继续。”}]}) updated_state app.update_state(config, human_command) # 继续执行后续节点 final_result app.invoke(None, config) # 传入 None 表示从当前 checkpoint 继续核心概念Checkpointer如MemorySaver负责保存工作流的快照。当流程执行到预设的中断点通常在conditional_edges的判断中实现它会暂停并保存当前状态。你可以通过app.update_state注入人工指令然后invoke继续执行。这非常适合需要人工审核、确认或修正的严肃流程。4.3 状态持久化与可视化对于生产系统你肯定不希望每次重启服务所有运行中的工作流都丢失。MemorySaver在内存中重启就没了。你需要一个持久化的Checkpointer比如基于数据库的。可视化LangGraph 内置了可视化方法对于调试和理解复杂流程极其有用。# 将图导出为 PNG 图片 from langgraph.graph import StateGraph # … 构建你的 workflow … workflow.get_graph().draw_mermaid_png(output_file_path“my_workflow.png”)生成一张 Mermaid 流程图能清晰看到所有节点和边的连接关系尤其是条件分支和循环。5. 从 Demo 到生产你必须考虑的工程化问题能让一个多智能体流程在 Jupyter Notebook 里跑起来只是第一步。要把它变成一个可靠的服务还有很长的路要走。5.1 错误处理与重试节点函数尤其是调用外部 API、工具可能会失败。LangGraph 本身不自动重试你需要自己实现。策略一在节点函数内部包装 try-catchdef robust_search_node(state: AssistantState): max_retries 3 for i in range(max_retries): try: result search_tool.invoke(state[“query”]) return {“results”: result} except Exception as e: if i max_retries - 1: # 最后一次重试也失败返回一个兜底结果或明确错误 return {“error”: f“搜索失败{str(e)}“, “results”: []} time.sleep(1) # 简单等待后重试策略二使用 LangGraph 的ToolNode和tools_condition对工具调用更友好ToolNode能自动处理工具调用的输入输出格式结合tools_condition可以更好地管理工具执行后的流程。5.2 超时与资源限制超时对于每个节点或整个app.invoke()设置超时。可以使用asyncio.wait_for或threading模块包装。资源限制限制并发运行的图实例数量避免对 LLM API 或下游服务造成洪水攻击。可以使用信号量asyncio.Semaphore或任务队列如 Celery来控制。5.3 可观测性与监控当你的智能体在线上处理成千上万的任务时出了问题不能只靠print。日志在每个节点的入口和出口记录详细的日志包括state的关键内容、耗时、成功/失败状态。使用结构化的日志库如structlog。链路追踪集成LangSmith。这是 LangChain 官方的可观测性平台能自动记录每次invoke的完整流程、每个节点的输入输出、耗时和 Token 消耗。对于调试复杂的工作流和优化成本至关重要。os.environ[“LANGSMITH_TRACING”] “true” os.environ[“LANGSMITH_API_KEY”] “your-langsmith-key” # 之后所有通过 LangChain/LangGraph 的调用都会被自动记录到 LangSmith 项目指标收集业务指标如不同分支路径的占比、平均处理时间、失败率等用于评估和优化你的智能体系统。5.4 测试策略多智能体工作流的测试比单函数复杂。单元测试节点函数单独测试每个node function模拟输入state断言输出state。集成测试工作流针对几种典型的用户输入运行完整的app.invoke()断言最终的state或输出消息符合预期。模拟外部依赖在测试中把llm.invoke和tool.invoke用 Mock 对象替换确保测试快速、稳定且不消耗 API 费用。测试边界条件专门测试路由判断的边界、循环退出的条件、错误处理逻辑。5.5 与现有系统集成你的 LangGraph 工作流很可能只是大系统中的一个组件。考虑如何将它暴露出去。作为 API 端点使用 FastAPI、Flask 等框架将app.invoke()包装成一个 HTTP 端点。接收用户输入返回执行结果。注意处理并发和状态隔离为每个请求使用唯一的thread_id。作为异步任务将工作流执行丢到 Celery、Dramatiq 或 RQ 等任务队列中实现异步处理、重试和结果回调。状态存储使用持久化的Checkpointer如基于 Redis 或 PostgreSQL确保长时间运行或中断的工作流状态不丢失。6. 总结LangGraph 实战的核心心法回过头看把 LangGraph 和多智能体系统用起来关键不在于记住所有 API而在于掌握几个核心心法第一状态State设计先行。在画图之前先用TypedDict把“共享白板”上需要流通的所有信息定义清楚。这是整个系统的数据骨架。字段宁少勿多优先使用Annotated[list, add]来管理列表的追加这是对话记忆的基石。第二把复杂流程拆解成单一职责的节点Node。每个节点最好只做一件事并且这件事对应一个明确的“专家”角色。节点函数要纯净输出只依赖于输入的 State。这会让你的图更清晰也更容易测试和复用。第三用条件边Conditional Edges实现智能路由。这是 LangGraph 的灵魂。add_conditional_edges让你能根据 State 的内容动态决定流程走向从而实现分支、循环等复杂逻辑。设计好路由判断逻辑那个lambda函数是让智能体“智能”起来的关键。第四从简单到复杂逐步验证。不要一开始就设计一个包含 10 个节点、5 个循环的巨图。先做一个两三个节点的最小可行图比如“路由 - 处理 - 结束”跑通整个流程。然后逐步添加新节点和新分支每加一步都充分测试。第五生产环境可观测性重于功能性。一个能跑但看不清内部状况的多智能体系统是危险的。务必在早期就集成 LangSmith 或类似的追踪工具并建立完善的日志和监控。当用户说“答案不对”时你能快速回溯是哪个节点、基于什么输入、给出了什么输出这才是工程化的体现。最后记住 LangGraph 是一个强大的编排工具但它不解决 AI 能力本身的问题比如模型不够聪明、工具不好用。它的价值在于让你能像搭积木一样把已有的 AI 能力模型、工具、链组织成稳定、可控、可维护的自动化流程。先从一个小而具体的场景开始把它跑通、跑稳你自然会知道如何用它去构建更复杂的智能体生态。
