在实际 AI 应用开发中构建能够协同工作的多智能体系统正成为解决复杂任务的关键路径。LangGraph 作为 LangChain 生态中用于构建有状态、多参与者工作流的核心库它并非简单地串联多个 Agent而是提供了一套基于图Graph的编程模型允许开发者清晰地定义智能体之间的控制流、状态共享和循环逻辑。理解 LangGraph 的架构、核心组件并通过代码将其落地是进阶 AI 应用开发的必修课。本文将以一个模拟的“技术文章评审与发布”多智能体工作流为例带你从零开始深入 LangGraph 的内部机制完成一个可运行、可调试的多智能体系统实战。1. 理解 LangGraph从状态机到智能体协作图在深入代码之前必须厘清 LangGraph 要解决的核心问题以及其背后的设计哲学。传统的单智能体或简单链式调用在处理需要多步骤决策、状态持久化或循环反馈的任务时显得力不从心。1.1 为什么需要 LangGraph单智能体与工作流的局限一个功能强大的大语言模型智能体Agent可以理解指令、调用工具并给出回答。然而当任务复杂度上升时例如“分析一篇技术博客的初稿给出修改建议并根据建议重写最后生成发布摘要”单个智能体会面临几个挑战职责过载一个智能体需要同时具备分析、评审、写作、总结等多种能力Prompt 会变得极其复杂且难以维护。状态管理困难任务涉及多个阶段分析、修改、总结每个阶段的输入和输出即状态需要被有效地传递和更新。在简单的链式调用中状态传递容易出错或丢失。缺乏结构化流程任务可能包含条件分支如果文章质量高则直接发布否则需要修改或循环根据评审意见多次修改直至达标。用代码硬编码这些逻辑会使系统僵化而用自然语言让智能体自行决定则不可靠。LangGraph 的提出正是为了将工作流的控制逻辑与单个智能体的推理能力解耦。它允许你将一个复杂任务分解为多个专门的智能体或节点并通过一个有向图来定义它们之间的执行顺序和条件跳转。1.2 LangGraph 核心模型图、状态与边LangGraph 的核心抽象非常简单却极其强大图Graph代表整个工作流由节点Nodes和边Edges构成。节点Node工作流中的一个步骤。它通常是一个函数接收当前工作流状态执行操作如调用 LLM、运行工具并返回一个更新后的状态。一个节点可以对应一个智能体、一个工具调用或任何自定义逻辑。状态State一个共享的、类型化的字典在工作流执行过程中在所有节点间传递和修改。它记录了任务的当前进展、中间结果和最终输出。边Edge定义了节点之间的流转条件。分为两种起始边Start Edge指定工作流从哪个节点开始。条件边Conditional Edge根据当前状态的值动态决定下一个要执行的节点。这是实现分支和循环的关键。这种基于图的模型使得复杂的工作流变得可视化、可维护且易于调试。你可以清晰地看到任务从Analyzer节点开始然后根据分析结果决定是跳转到Rewriter还是Summarizer。1.3 LangGraph 与 LangChain 的关系LangGraph 是 LangChain 框架的一个子库但它解决的是不同层次的问题。一个常见的混淆在于认为它们是并列的选择。LangChain提供了构建基于大语言模型应用的基础模块如模型封装LLMs、提示模板PromptTemplates、记忆Memory、文档加载器Document Loaders以及最基础的链Chains和智能体Agents。它关注的是“如何与 LLM 交互”。LangGraph建立在 LangChain 组件之上专注于“如何编排多个与 LLM 交互的步骤”。它使用 LangChain 的Runnable接口来定义节点逻辑但引入了图结构来管理这些Runnable之间的复杂关系。简而言之你用 LangChain 来构造单个智能体的“零件”然后用 LangGraph 把这些“零件”组装成一条能处理复杂任务的“自动化流水线”。2. 环境准备与项目初始化我们将构建一个模拟的“技术文章评审与发布”多智能体系统。这个系统包含三个智能体分析员、重写员和总结员它们将协作处理一篇输入的文章。2.1 环境与依赖配置首先确保你的 Python 环境版本在 3.8 或以上。然后安装必要的依赖包。我们使用 OpenAI 的 GPT 模型作为底层 LLM因此需要其 SDK。同时安装 LangChain 和 LangGraph。# 创建并激活虚拟环境可选但推荐 python -m venv venv source venv/bin/activate # Linux/Mac # venv\Scripts\activate # Windows # 安装核心依赖 pip install langchain langgraph openai注意本文示例使用 OpenAI API你需要准备一个有效的OPENAI_API_KEY环境变量。你也可以替换为其他 LangChain 支持的模型如 Anthropic、Groq 或本地模型通过 Ollama但初始化方式会有所不同。2.2 项目结构规划一个清晰的项目结构有助于管理复杂的多智能体工作流。建议按以下方式组织tech_article_agent_system/ ├── agents/ # 存放各个智能体的定义 │ ├── __init__.py │ ├── analyzer.py │ ├── rewriter.py │ └── summarizer.py ├── graph/ # 存放 LangGraph 定义 │ ├── __init__.py │ └── workflow.py # 主图定义文件 ├── state.py # 定义共享状态 Schema ├── config.py # 配置如模型、API Key ├── main.py # 主入口执行工作流 └── requirements.txt我们先从定义共享状态开始这是整个工作流的“数据总线”。3. 定义多智能体系统的共享状态在state.py中我们使用TypedDict来定义状态的类型这能提供良好的代码提示和类型检查。# state.py from typing import TypedDict, List, Optional class ArticleState(TypedDict): 多智能体文章处理工作流的共享状态。 所有节点都读取和修改这个字典的某些字段。 # 输入 original_article: str # 用户输入的原始文章 # 中间产物 analysis_result: Optional[str] # 分析员给出的分析结果 rewritten_article: Optional[str] # 重写员生成的重写文章 # 输出 final_summary: Optional[str] # 总结员生成的最终摘要 # 控制流 needs_rewrite: Optional[bool] # 决定工作流走向的标志这个ArticleState定义了工作流中流动的所有数据。每个节点智能体的职责就是消费某些字段并更新另一些字段。例如analyzer节点会读取original_article生成analysis_result并判断needs_rewrite。4. 实现核心智能体节点我们将三个智能体分别实现为三个函数它们都遵循LangGraph节点的签名接收一个状态字典返回一个更新后的状态字典。4.1 分析员智能体Analyzer分析员负责评估文章质量并决定是否需要重写。在agents/analyzer.py中# agents/analyzer.py from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser import re # 初始化 LLM llm ChatOpenAI(modelgpt-4o-mini, temperature0.1) def analyze_article(state: dict) - dict: 分析文章节点。 输入state[original_article] 输出更新 state[analysis_result], state[needs_rewrite] article state[original_article] # 1. 构建分析提示词 analysis_prompt ChatPromptTemplate.from_messages([ (system, 你是一位资深技术编辑。请严格评估以下技术文章草稿的质量。), (user, 请分析这篇技术文章 --- {article} --- 请按以下格式回复 ## 分析结果 [这里写你对文章逻辑、结构、技术准确性和可读性的分析] ## 是否需要重写 只需要回答“是”或“否”。如果文章存在明显逻辑问题、关键信息缺失或结构混乱则回答“是”。 ) ]) # 2. 构造链并调用 analysis_chain analysis_prompt | llm | StrOutputParser() analysis_response analysis_chain.invoke({article: article}) # 3. 解析响应更新状态 # 简单通过正则分离分析结果和决策 parts re.split(r## 是否需要重写\, analysis_response) analysis_text parts[0].replace(## 分析结果, ).strip() decision_text parts[1].strip().lower() if len(parts) 1 else needs_rewrite 是 in decision_text or yes in decision_text # 返回更新后的状态片段 return { analysis_result: analysis_text, needs_rewrite: needs_rewrite }关键点解释节点函数签名函数接收一个state字典返回一个只包含需要更新字段的字典。LangGraph 会自动将其与原有状态合并。Prompt 设计我们要求 LLM 以特定格式回复便于后续程序化解析。这是构建可靠智能体的关键技巧。输出解析使用StrOutputParser获取文本响应然后用正则表达式提取结构化信息。对于更复杂的场景可以使用JsonOutputParser。4.2 重写员智能体Rewriter重写员在分析员认为需要修改时被触发。在agents/rewriter.py中# agents/rewriter.py from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser llm ChatOpenAI(modelgpt-4o-mini, temperature0.3) # 重写可以更有创造性 def rewrite_article(state: dict) - dict: 重写文章节点。 输入state[original_article], state[analysis_result] 输出更新 state[rewritten_article] article state[original_article] analysis state.get(analysis_result, ) rewrite_prompt ChatPromptTemplate.from_messages([ (system, 你是一位技术写作专家擅长根据修改建议优化文章。), (user, 原始文章 --- {article} --- 编辑分析建议 --- {analysis} --- 请根据以上分析对原始文章进行重写和优化。保持核心技术内容不变但改进逻辑、结构和表达。 直接输出优化后的完整文章不要额外解释。 ) ]) rewrite_chain rewrite_prompt | llm | StrOutputParser() rewritten rewrite_chain.invoke({article: article, analysis: analysis}) return {rewritten_article: rewritten}4.3 总结员智能体Summarizer总结员负责生成最终版本的摘要。在agents/summarizer.py中# agents/summarizer.py from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser llm ChatOpenAI(modelgpt-4o-mini, temperature0.1) def summarize_article(state: dict) - dict: 总结文章节点。 输入state[rewritten_article] 或 state[original_article] 输出更新 state[final_summary] # 优先使用重写后的文章如果没有则使用原文 article_to_summarize state.get(rewritten_article) or state[original_article] summarize_prompt ChatPromptTemplate.from_messages([ (system, 你是一位技术内容总结者。), (user, 请为以下技术文章生成一个简洁的发布摘要约150字突出其核心观点和技术价值。 文章 --- {article} --- 摘要 ) ]) summarize_chain summarize_prompt | llm | StrOutputParser() summary summarize_chain.invoke({article: article_to_summarize}) return {final_summary: summary}至此三个智能体节点都已定义完毕。它们都是独立的、可测试的函数。接下来我们需要用 LangGraph 将它们“编织”成一个协同工作流。5. 使用 LangGraph 构建并编排工作流这是 LangGraph 的核心部分。我们在graph/workflow.py中创建图。5.1 构建图结构# graph/workflow.py from langgraph.graph import StateGraph, END from typing import Literal import agents.analyzer as analyzer import agents.rewriter as rewriter import agents.summarizer as summarizer from state import ArticleState # 1. 初始化一个状态图并指定状态 Schema workflow StateGraph(ArticleState) # 2. 添加节点 # 节点名与函数名可以不同这里为了清晰保持一致 workflow.add_node(analyzer, analyzer.analyze_article) workflow.add_node(rewriter, rewriter.rewrite_article) workflow.add_node(summarizer, summarizer.summarize_article) # 3. 设置起始节点工作流总是从分析员开始 workflow.set_entry_point(analyzer) # 4. 定义条件边分析完成后根据 needs_rewrite 决定下一步 def route_after_analysis(state: ArticleState) - Literal[rewriter, summarizer]: 路由函数。根据状态中的 needs_rewrite 布尔值决定下一个节点。 if state.get(needs_rewrite): return rewriter # 需要重写则跳转到重写员 else: return summarizer # 不需要重写则直接跳转到总结员 # 将分析员节点连接到这个条件边 workflow.add_conditional_edges( analyzer, route_after_analysis, { rewriter: rewriter, summarizer: summarizer } ) # 5. 定义普通边重写员完成后必然进入总结员 workflow.add_edge(rewriter, summarizer) # 总结员完成后工作流结束 workflow.add_edge(summarizer, END) # 6. 编译图得到一个可执行对象 app workflow.compile()代码详解StateGraph(ArticleState)创建图时传入状态类型有助于类型检查和 IDE 提示。add_node将之前定义的函数注册为图节点。set_entry_point指定工作流的起点。add_conditional_edges这是实现分支逻辑的关键。它连接源节点“analyzer”到一个路由函数route_after_analysis。该函数检查当前状态返回下一个节点的名称。下面的映射字典将这个名称映射到实际的节点对象。add_edge定义无条件跳转。“rewriter” - “summarizer”表示重写员完成后一定进入总结员。“summarizer” - END表示总结员是工作流的终点。compile()将图定义编译成一个可调用的app。这个app的invoke方法就是工作流的执行入口。5.2 可视化工作流可选但强烈推荐LangGraph 提供了将图导出为 PNG 图像的功能这能极大帮助理解和调试。# 在 graph/workflow.py 末尾添加 try: # 显示图结构 from IPython.display import Image, display # 只有在 Jupyter 环境中才尝试显示 display(Image(app.get_graph().draw_mermaid_png())) except: # 非 Jupyter 环境可以保存到文件 print(非交互环境尝试保存图表到文件...) graph_image app.get_graph().draw_mermaid_png() with open(article_workflow.png, wb) as f: f.write(graph_image) print(图表已保存为 article_workflow.png)生成的图会清晰地显示analyzer开始根据条件指向rewriter或summarizerrewriter指向summarizer最后结束。6. 运行与验证多智能体工作流现在我们创建一个主入口文件main.py来运行整个系统。# main.py import os from graph.workflow import app from state import ArticleState # 设置 OpenAI API Key (如果未在环境变量中设置) # os.environ[OPENAI_API_KEY] your-api-key-here def main(): # 1. 准备输入文章这里用一段简短的示例 sample_article LangGraph是一个库。它用来做多智能体。很多人觉得它难。 它基于图。有节点和边。节点是智能体。边是流转。 我们可以用它构建工作流。工作流可以很复杂。 # 2. 构造初始状态 initial_state: ArticleState { original_article: sample_article, analysis_result: None, rewritten_article: None, final_summary: None, needs_rewrite: None, # 初始为 None由 analyzer 决定 } print( * 50) print(初始文章) print(sample_article) print( * 50) print(\n开始执行多智能体工作流...\n) # 3. 执行编译好的图应用 final_state app.invoke(initial_state) # 4. 打印最终结果 print( * 50) print(工作流执行完成) print( * 50) print(\n【分析结果】) print(final_state.get(analysis_result, N/A)) print(f\n【是否需要重写】: {final_state.get(needs_rewrite)}) if final_state.get(rewritten_article): print(\n【重写后的文章】) print(final_state[rewritten_article]) else: print(\n【文章未重写】) print(\n【最终发布摘要】) print(final_state.get(final_summary, N/A)) # 5. 打印完整状态调试用 print(\n *50) print(完整最终状态调试信息:) for key, value in final_state.items(): print(f{key}: {value[:100] if isinstance(value, str) and len(value) 100 else value}) if __name__ __main__: main()运行这个程序python main.py你应该能看到类似以下的输出清晰地展示了工作流的执行过程和每个智能体的产出 初始文章 LangGraph是一个库。它用来做多智能体。很多人觉得它难。 它基于图。有节点和边。节点是智能体。边是流转。 我们可以用它构建工作流。工作流可以很复杂。 开始执行多智能体工作流... 工作流执行完成 【分析结果】 文章内容过于简略缺乏具体解释和实例。句子之间连贯性不强更像是零散的要点罗列而非一篇完整的文章。技术准确性方面虽然提到了基本概念但未能深入说明LangGraph的核心价值、工作原理或与其他工具如LangChain的区别。可读性较差对于不了解背景的读者难以形成清晰认知。 【是否需要重写】: True 【重写后的文章】 LangGraph 是 LangChain 生态系统中的一个关键库专门用于构建复杂、有状态的多智能体工作流。它通过“图”这一抽象模型将复杂的AI任务分解为多个相互协作的智能体节点并通过定义节点之间的流转规则边来编排整个执行流程。 与传统的单智能体或简单链式调用相比LangGraph 解决了职责过载、状态管理困难和流程缺乏结构化等挑战。例如在一个技术文章评审场景中你可以创建“分析”、“重写”、“总结”三个独立的智能体节点。分析节点评估文章质量并决定是否需要修改如果需要则流转到重写节点进行优化最后无论是否经过重写都会进入总结节点生成最终摘要。这种基于图的设计使得工作流清晰、可维护且易于调试。 【最终发布摘要】 本文介绍了LangGraph库它是构建多智能体系统的核心工具。通过图结构节点和边来编排智能体工作流能有效管理复杂任务的状态和流程。文章以技术文章评审为例说明了如何将分析、重写、总结等步骤分解为独立的智能体并通过条件逻辑串联从而提升AI应用的处理能力和可维护性。这个输出验证了我们的工作流分析员认为示例文章质量不高needs_rewrite: True于是流程走向重写员重写员生成了一篇结构更清晰、内容更丰富的文章最后总结员基于重写后的文章生成了摘要。7. 核心机制剖析与高级特性7.1 状态State的深层次理解在上面的例子中我们使用了简单的字典合并来更新状态。LangGraph 的状态管理实际上更强大。它支持对状态进行细粒度的读写控制这是通过StateGraph的state_schema和节点的input/output关键字参数实现的。# 高级状态管理示例显式声明节点读写字段 from langgraph.graph import StateGraph, START from langgraph.graph.message import add_messages # 定义一个更复杂的状态包含消息历史 class ChatState(TypedDict): messages: list # 对话历史 summary: str # 当前对话摘要 graph_builder StateGraph(ChatState) # 添加节点时可以指定它修改状态的哪个字段 # 这有助于图的优化和避免意外覆盖 def ai_node(state: ChatState): # 这个节点只读取 messages 并更新 messages last_msg state[“messages”][-1] response llm.invoke(last_msg.content) return {“messages”: add_messages(state[“messages”], response)} # 在复杂工作流中明确读写字段能提升可维护性 # graph_builder.add_node(“ai”, ai_node, input“messages”, output“messages”)7.2 人工干预Human-in-the-Loop机制LangGraph 内置了对人工干预的支持这是构建可靠生产系统的重要特性。你可以在工作流的任何节点暂停等待外部输入如人工审核然后再继续。from langgraph.graph import StateGraph from langgraph.checkpoint import MemorySaver from langgraph.prebuilt import ToolNode, tools_condition # 1. 使用检查点Checkpointer持久化状态 memory MemorySaver() app_with_checkpoints workflow.compile(checkpointermemory) # 2. 在某个节点后中断等待人工输入 # 假设我们在 analyzer 后需要人工确认 def human_review_node(state): # 这里可以是将任务发送到审核队列的逻辑 # 或者直接抛出中断异常由外部系统接管 print(f”分析完成请人工审核分析结果: {state[‘analysis_result’]}”) # 在实际中这里会暂停图执行直到收到外部信号如API回调 # 为了示例我们模拟人工确认需要重写 return {“human_approved_rewrite”: True} workflow.add_node(“human_review”, human_review_node) # 修改边让 analyzer 之后进入 human_review workflow.add_edge(“analyzer”, “human_review”) # 然后 human_review 根据人工输入决定下一步 def after_human_review(state): if state.get(“human_approved_rewrite”): return “rewriter” else: return “summarizer” workflow.add_conditional_edges(“human_review”, after_human_review, …)7.3 多智能体协作模式我们的例子是线性条件流。LangGraph 支持更复杂的协作模式广播Broadcast一个节点的输出同时发送给多个后续节点。聚合Aggregate多个节点的输出汇聚到一个节点进行处理。循环Loop节点可以指向之前的节点形成循环直到满足某个条件退出例如重写-评审循环直到质量达标。这些模式通过巧妙地组合add_edge,add_conditional_edges以及节点函数内的逻辑来实现。8. 常见问题排查与调试指南在开发 LangGraph 多智能体应用时你可能会遇到以下典型问题。8.1 状态更新不生效问题现象可能原因检查与解决节点修改了状态但后续节点读取到的仍是旧值。1. 节点函数返回的字典键名与状态 Schema 中定义的字段名不匹配。2. 使用了错误的状态合并逻辑如直接赋值而非返回更新字典。1. 确保节点返回的字典键名与TypedDict中定义的完全一致。2. LangGraph 节点函数应返回一个字典其中包含需要更新的字段及其新值。不要修改传入的state对象本身。状态字段值为None。1. 字段在初始状态中未提供且节点未成功写入。2. 条件边路由函数访问了尚未被赋值的字段。1. 在TypedDict中使用Optional[...]声明可为空的字段。2. 在路由函数中访问状态字段时使用state.get(‘field_name’)并提供默认值。8.2 条件边路由错误问题现象可能原因检查与解决工作流抛出KeyError或进入意外节点。1. 路由函数返回的字符串与add_conditional_edges中映射的键名不匹配。2. 路由函数的返回值不是Literal类型中声明的值。1. 仔细核对add_conditional_edges的第三个参数映射字典的键名是否与路由函数所有可能的返回值完全一致。2. 使用typing.Literal明确限定路由函数的返回类型有助于 IDE 检查和避免拼写错误。8.3 LLM 调用失败或超时问题现象可能原因检查与解决节点执行卡住或报错APIError。1. API Key 未设置或无效。2. 网络问题或模型服务不可用。3. Prompt 设计不当导致 LLM 输出格式不符合解析预期。1. 确认OPENAI_API_KEY环境变量已设置且有效。2. 在节点函数中添加重试机制和超时设置。例如使用tenacity库进行重试。3. 在调用 LLM 链的invoke前后添加日志打印输入 Prompt 和原始输出确保输出能被后续解析逻辑正确处理。8.4 调试与日志记录为每个节点添加详细的日志是调试多智能体工作流的最有效方法。import logging logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) def analyze_article_with_log(state: dict) - dict: logger.info(f”分析员节点开始执行。输入文章长度: {len(state[‘original_article’])}”) # … 原有逻辑 … logger.info(f”分析完成。needs_rewrite 决策为: {needs_rewrite}”) return {“analysis_result”: analysis_text, “needs_rewrite”: needs_rewrite}此外LangGraph 的app.invoke()返回的最终状态包含了所有中间状态。通过打印或记录这个完整状态你可以追溯每个节点的输入和输出。9. 生产环境最佳实践与扩展方向将 LangGraph 多智能体系统用于生产需要考虑以下几个方面。9.1 配置与安全性密钥管理永远不要将 API Key 硬编码在代码中。使用环境变量、密钥管理服务如 AWS Secrets Manager或配置文件。配置外置将模型类型、温度参数、超时时间等配置项放在配置文件如config.yaml或环境变量中。输入验证与清理在第一个处理用户输入的节点对输入进行验证、长度限制和清理防止 Prompt 注入或资源耗尽攻击。9.2 可观测性与监控结构化日志为每个节点的开始、结束、关键决策点记录结构化日志如 JSON 格式便于集中收集和分析。链路追踪为每次工作流执行生成一个唯一trace_id并贯穿所有节点调用和外部服务如 LLM API便于问题追踪。性能指标监控每个节点的执行耗时、LLM 调用的 Token 消耗、工作流整体成功率等指标。9.3 性能与成本优化异步执行如果节点之间没有严格的先后依赖可以考虑使用 LangGraph 的异步特性或将其中的某些节点改为异步任务。缓存对于纯函数式、输入相同则输出必然相同的节点如某些格式化或计算节点可以考虑引入缓存机制避免重复计算或重复调用 LLM。LLM 调用优化合理设置temperature参数对于非创造性任务使用更小、更快的模型如gpt-4o-mini而非gpt-4考虑对输出长度进行限制。9.4 扩展方向集成外部工具让智能体能够调用外部 API、数据库查询、代码执行器等。这可以通过 LangChain 的Tool接口轻松实现并将其封装成节点。动态图构建根据运行时条件动态添加或移除节点。这需要更高级的图操作但 LangGraph 的 API 支持在编译前动态修改图结构。与 Web 框架集成将编译好的app封装成 FastAPI 或 Flask 的一个端点提供 HTTP API 服务。持久化检查点使用数据库或 Redis 作为Checkpointer实现工作流的长时运行、暂停和恢复这对于需要人工审核或等待外部事件的任务至关重要。通过本教程你不仅学会了如何使用 LangGraph 构建一个可运行的多智能体系统更重要的是理解了其基于图编排的核心思想。从明确的状态定义、职责清晰的节点函数、到有条件分支的工作流设计这套方法论可以扩展到客服对话、数据分析、代码生成等众多复杂场景。下一步尝试为你手头的某个复杂任务设计一个图并思考如何将其分解为多个协同的智能体节点。