LangGraph 与 LangChain 对比：多智能体开发的框架选择指南

关键词

LangChain, LangGraph, 多智能体开发, 大语言模型(LLM)应用框架, 对话式代理, 状态机编程, Agentic Workflow

摘要

随着大语言模型（LLM）如 GPT-4o、Claude 3.5 Sonnet、Qwen3 的普及，Agentic Workflow（智能体工作流）正在成为构建复杂 AI 应用的主流范式。而在 Agentic 开发领域，LangChain 生态无疑是最成熟、社区最活跃的代表之一——从 2022 年底的单智能体链式调用工具包，到 2024 年推出的LangGraph这一专注于状态机驱动的多智能体/复杂工作流框架，LangChain 的产品迭代恰好反映了 LLM 应用从“简单任务自动化”到“复杂协作决策系统”的发展历程。

本文将以 “一步步思考” 的方式，从问题背景、核心概念、技术原理、代码实现、实际应用、最佳实践、行业趋势等多个维度，对 LangChain（核心是 LangChain Core 的 Chain/Agent）与 LangGraph 进行系统性、可落地、有深度的对比。我们会用生活化的例子（如“单人外卖员配送”vs“外卖配送调度团队协作网”）解释两者的本质区别；通过数学模型（状态转移矩阵、概率决策树）量化分析两者的适用场景；给出完整的 Python 代码示例（分别用 LangChain Agent、LangGraph 构建“客户投诉处理智能体协作系统”）；结合真实项目案例（电商客服、金融风控审核、代码审查自动化）总结最佳实践；最后梳理两者的发展历史，并展望 Agentic 开发领域的未来趋势。

无论你是 LLM 应用开发的初学者（想快速上手构建简单代理），还是资深工程师（需要构建可解释、可调试、高并发的复杂多智能体系统），本文都能为你提供清晰的框架选择指南，避免踩坑，提高开发效率。

正文

1. 背景介绍：为什么需要 Agentic 框架？LangChain 和 LangGraph 是怎么来的？

1.1 问题背景：LLM 应用的三次浪潮与 Agentic 范式的兴起

在进入正式对比之前，我们得先搞清楚：为什么会有 LangChain 这样的工具包？为什么又要推出 LangGraph？这背后其实是 LLM 应用开发的三次技术浪潮。

1.1.1 第一次浪潮：“直接调用 LLM API”的零代码/低代码时代（2022年Q1-Q3）

LLM 刚火的时候，开发一个 AI 应用特别简单：只需要写几行代码（甚至不用代码，用 OpenAI Playground、HuggingFace Spaces），输入一段提示词（Prompt），就能得到输出。比如：

  • 用 GPT-3.5 写邮件：输入“帮我写一封给导师的请假邮件，理由是外婆生病住院，需要回去照顾3天”
  • 用 Stable Diffusion 画图：输入“一只戴着帽子的柴犬坐在海边看日落，风格是宫崎骏的动画”

这一阶段的应用核心是**“提示词工程”**，几乎不需要框架，直接调用 API 即可。但很快，开发者就发现了问题：

  1. 功能太单一：只能做“输入-输出”的单步任务，没法做多步任务（比如：先搜索“外婆家到医院的路线”，再写请假邮件，再查机票价格，最后生成行程表）
  2. 提示词太长太乱：多步任务的提示词需要包含所有步骤的规则、上下文信息，容易超过 LLM 的 Token 窗口限制（2022年 GPT-3.5 的 Token 窗口只有 4k），输出也不稳定
  3. 没有记忆功能：每次调用 API 都是独立的，没法记住之前的对话历史或任务状态（比如：用户问“昨天你帮我查的柴犬价格是多少？”，直接调用 API 根本不知道“昨天查过柴犬价格”这件事）
  4. 没法调用外部工具：LLM 的知识是静态的（截止到训练数据的截止日期），没法获取实时信息（比如天气、股票价格），也没法执行具体的操作（比如订机票、发邮件、查询数据库）
1.1.2 第二次浪潮：“链式调用+工具增强”的单智能体时代（2022年Q4-2023年Q4）

为了解决第一次浪潮的问题，LangChain于 2022 年 10 月由 Harrison Chase 创立——它的核心思想是：把 LLM 应用拆分成一个个“组件”（Component），然后用“链条”（Chain）把组件串联起来。

这些组件包括：

  • LLM/Chat Model：大语言模型/对话模型（比如 OpenAI 的 gpt-4o、Anthropic 的 claude-3.5-sonnet、HuggingFace 的 qwen3-7b-chat）
  • Prompt Template：提示词模板（用来动态生成提示词，避免重复写代码）
  • Memory：记忆组件（用来存储对话历史、任务状态，支持 LLM 记住上下文）
  • Tool：工具组件（用来调用外部 API、执行 Python 代码、查询数据库等）
  • Chain：链条组件（把上面的组件串联起来，形成一个可执行的工作流）
  • Agent：代理组件（比 Chain 更高级，能根据 LLM 的推理结果自主选择工具、自主调整步骤，而不是硬编码的串联）

举个“单人外卖员配送”的生活化例子：

  • 没有框架的直接调用 API：相当于外卖员每次只接一单，而且不知道怎么规划路线，也不知道怎么联系商家和用户，完全靠用户的一句话（提示词）指导——效率极低，还经常出错
  • LangChain 的 Chain：相当于给外卖员配了一张“硬编码的路线图”（比如：先取商家 A 的餐，再送用户 B，再取商家 C 的餐，再送用户 D）——外卖员必须严格按照路线图走，不能调整（比如路上堵车了也不能绕路，用户 B 临时改地址了也不能处理）
  • LangChain 的 Agent：相当于给外卖员配了一个“智能手机导航+语音助手”——导航会根据实时路况自主调整路线，语音助手会根据用户和商家的消息自主决定下一步操作（比如：用户 B 临时改地址，语音助手会先查新地址的位置，再重新规划路线，再联系用户确认）

这一阶段的应用核心是**“链式调用+工具增强+自主决策”**，LangChain 几乎垄断了单智能体应用开发的市场——截至 2024 年 6 月，LangChain 在 GitHub 上的 Star 数已经超过了 100k，拥有超过 1000 个官方/社区贡献的组件，覆盖了几乎所有主流的 LLM、工具、数据库。

但随着应用场景越来越复杂，开发者又发现了 LangChain（特别是 Chain/Agent）的新问题：

  1. Chain 的逻辑太僵化：硬编码的串联没法处理分支、循环、并行、状态回溯等复杂逻辑（比如：外卖员取餐发现商家没做好，需要等待30分钟，这时候 Chain 就没法处理——因为 Chain 是线性的，不能循环）
  2. Agent 的逻辑太黑盒：自主决策的过程不可解释、不可调试——你不知道 Agent 为什么选择这个工具，为什么生成这个步骤，如果出错了，很难定位问题（比如：外卖员突然绕了很远的路，你根本不知道是导航坏了，还是语音助手理解错了用户的意思）
  3. 多智能体协作难实现：Chain/Agent 都是单智能体的设计，没法支持多个智能体之间的协作、竞争、通信、同步（比如：外卖平台需要多个外卖员协作配送，需要调度员智能体分配订单，需要导航员智能体规划路线，需要客服智能体处理用户投诉——这在 LangChain 的单智能体框架下很难实现）
  4. 状态管理混乱：Memory 组件虽然能存储一些信息，但没有统一的状态管理机制——每个 Chain/Agent 的状态都是独立的，状态的更新、同步、持久化都很麻烦（比如：用户 B 临时改地址的状态，调度员智能体、导航员智能体、客服智能体都需要知道，但用 LangChain 的 Memory 很难实现这种多智能体之间的状态同步）
  5. 可扩展性差：随着应用复杂度的增加，Chain/Agent 的代码会变得越来越臃肿，很难维护和扩展（比如：一个简单的客户投诉处理智能体，可能需要几十个组件，代码行数超过 1000 行）
1.1.3 第三次浪潮：“状态机驱动+多智能体协作”的复杂系统时代（2024年Q1至今）

为了解决第二次浪潮的问题，LangChain 团队于 2024 年 1 月推出了 LangGraph 的第一个 Beta 版本，并于 2024 年 6 月推出了 LangGraph 1.0 正式版。

LangGraph 的核心思想是：把 LLM 应用（特别是复杂的多智能体应用）建模成一个有限状态机（FSM）或有向无环图（DAG）**（实际上，LangGraph 支持有环的图，也就是循环状态机），其中：

  • 节点（Node）：代表一个“动作”（Action）——可以是调用 LLM、调用工具、执行自定义代码、更新状态等
  • 边（Edge）：代表“动作之间的转移条件”（Transition Condition）——可以是无条件转移、基于状态值的条件转移、基于 LLM 推理结果的条件转移等
  • 状态（State）：代表“整个系统的全局状态”——是一个统一的、结构化的对象，所有节点都可以读取和更新状态
  • 智能体（Agent）：在 LangGraph 中，智能体本质上就是“一个状态机的子图”——多个智能体可以通过共享状态或发送消息的方式进行协作

还是举“外卖配送调度团队协作网”的生活化例子：

  • LangGraph 的状态机：相当于整个外卖平台的“调度系统”——有一个全局的“调度状态”（包括所有待分配的订单、所有外卖员的位置和状态、所有商家的备餐状态等）
  • LangGraph 的节点：相当于调度系统的“各个部门”——比如：
    • 订单接收节点：接收用户的订单请求，更新调度状态
    • 外卖员分配节点：根据调度状态（订单的位置、外卖员的位置和状态）自主选择合适的外卖员
    • 路线规划节点：为外卖员规划最优路线，更新调度状态
    • 备餐监控节点：监控商家的备餐状态，更新调度状态
    • 配送监控节点：监控外卖员的配送状态，更新调度状态
    • 投诉处理节点：如果用户或商家有投诉，触发投诉处理子图（也就是另一个智能体）
  • LangGraph 的边：相当于调度系统的“工作流程规则”——比如：
    • 订单接收节点 → 外卖员分配节点（无条件转移）
    • 外卖员分配节点 → 路线规划节点（如果成功分配了外卖员）
    • 外卖员分配节点 → 等待节点（如果没有合适的外卖员）
    • 等待节点 → 外卖员分配节点（等待5分钟后无条件转移）
    • 备餐监控节点 → 配送监控节点（如果商家已经备餐完成）
    • 备餐监控节点 → 等待节点（如果商家还没备餐完成）
    • 配送监控节点 → 结束节点（如果订单已经配送完成）
    • 配送监控节点 → 投诉处理节点（如果用户或商家有投诉）
  • LangGraph 的多智能体协作：相当于调度系统的“各个部门之间的协作”——所有部门都共享同一个“调度状态”，任何一个部门更新了状态，其他部门都能立即看到；如果某个部门遇到了复杂问题（比如投诉处理），可以触发另一个专门的部门（投诉处理子图/智能体）来处理

这一阶段的应用核心是**“状态机驱动+多智能体协作+可解释可调试+统一状态管理”**，LangGraph 完美解决了 LangChain（特别是 Chain/Agent）的所有问题——截至 2024 年 9 月，LangGraph 在 GitHub 上的 Star 数已经超过了 25k，拥有超过 100 个官方/社区贡献的节点和工具，覆盖了几乎所有主流的多智能体协作场景。

1.2 目标读者

本文的目标读者包括：

  1. LLM 应用开发的初学者：想了解 LangChain 和 LangGraph 的基本概念，快速上手构建简单的单智能体或多智能体应用
  2. LangChain 的老用户：想了解 LangGraph 相对于 LangChain 的优势，知道什么时候该继续用 LangChain，什么时候该切换到 LangGraph
  3. 资深的 AI/软件工程师：想深入了解 LangChain 和 LangGraph 的技术原理（包括状态机、图论、提示词工程、工具调用机制等），构建可解释、可调试、高并发、可扩展的复杂多智能体系统
  4. AI 产品经理：想了解 Agentic 开发领域的技术现状和未来趋势，为产品规划提供技术参考
  5. AI 研究者：想了解 LangChain 和 LangGraph 的实现细节，为自己的研究工作提供参考
1.3 核心问题或挑战

在本文中，我们将重点解决以下几个核心问题或挑战：

  1. LangChain 和 LangGraph 的核心概念分别是什么？它们的本质区别是什么？
  2. LangChain 的 Chain/Agent 和 LangGraph 的 StateGraph/CompiledGraph 分别是怎么工作的？
  3. 什么时候该用 LangChain？什么时候该用 LangGraph？
  4. 如何分别用 LangChain 和 LangGraph 构建一个完整的、可落地的应用？
  5. 使用 LangChain 和 LangGraph 时，有哪些最佳实践和常见问题？
  6. LangChain 和 LangGraph 的发展历史是什么？它们的未来趋势是什么？

2. 核心概念解析：从“单人外卖员”到“外卖配送调度团队协作网”

在这一章节中，我们将用生活化的例子（如“单人外卖员配送”vs“外卖配送调度团队协作网”）解释 LangChain 和 LangGraph 的核心概念，并通过概念核心属性维度对比表格、ER 实体关系图、交互关系图来梳理它们之间的关系。

2.1 核心概念：LangChain 生态的核心组件

首先，我们来回顾一下 LangChain 生态的核心组件——注意，LangChain 生态现在已经分成了多个独立的包：

  • LangChain Core：LangChain 生态的核心库，包含了所有基础组件的定义（如 BaseLLM、BaseChatModel、BasePromptTemplate、BaseMemory、BaseTool、BaseChain、BaseAgent、AgentExecutor 等）
  • LangChain Community：包含了社区贡献的组件（如各种 LLM 的封装、各种工具的封装、各种数据库的封装等）
  • LangChain OpenAI：包含了 OpenAI 官方组件的封装（如 ChatOpenAI、OpenAIEmbeddings 等）
  • LangChain Anthropic：包含了 Anthropic 官方组件的封装（如 ChatAnthropic 等）
  • LangChain Google：包含了 Google 官方组件的封装（如 ChatGoogleGenerativeAI 等）
  • LangChain Hub：包含了 LangChain 官方维护的提示词模板库（可以直接拉取使用）
  • LangSmith：LangChain 官方维护的 LLM 应用开发平台（提供了调试、监控、评估、部署等功能）

在本文中，我们主要关注 LangChain Core 的核心组件——因为 LangGraph 也是基于 LangChain Core 构建的。

2.1.1 组件 1：LLM/Chat Model（大语言模型/对话模型）

生活化比喻：LLM/Chat Model 就像是“外卖员的大脑”——负责思考、推理、决策、生成文本。

核心定义：

  • LLM：文本生成模型（Text Generation Model），输入是一段纯文本，输出是一段纯文本（比如 OpenAI 的 text-davinci-003、HuggingFace 的 qwen3-7b）
  • Chat Model：对话生成模型（Chat Generation Model），输入是一个“消息列表”（List of Messages），每个消息有一个角色（Role）——可以是 system（系统提示词）、user（用户输入）、assistant（助手输出）、tool（工具输出），输出是一个“助手消息”（Assistant Message）（比如 OpenAI 的 gpt-4o、Anthropic 的 claude-3.5-sonnet、HuggingFace 的 qwen3-7b-chat）

文本示意图：

纯文本输入 → LLM → 纯文本输出
消息列表输入 → Chat Model → 助手消息输出

示例代码：

# 安装依赖
# pip install langchain-openai python-dotenv

from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage

# 加载环境变量（需要在 .env 文件中设置 OPENAI_API_KEY）
load_dotenv()

# 初始化 Chat Model
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)

# 构造消息列表
messages = [
    SystemMessage(content="你是一只戴着帽子的、会说话的柴犬，说话要可爱、幽默，用‘汪汪’结尾。"),
    HumanMessage(content="你今天想去哪里玩呀？"),
]

# 调用 Chat Model
response = llm.invoke(messages)
print(response.content)
# 输出示例：我今天想去海边玩！可以捡贝壳、追海鸥，还能躺在沙滩上晒太阳呢！汪汪～
2.1.2 组件 2：Prompt Template（提示词模板）

生活化比喻：Prompt Template 就像是“外卖员的工作手册模板”——外卖员只需要填入一些动态信息（比如用户的地址、商家的地址），就能生成一份完整的工作手册。

核心定义：Prompt Template 是一个“带有占位符的提示词”——占位符通常用 {variable_name} 表示，可以在运行时动态替换成具体的值。Prompt Template 可以大大提高提示词的复用性和可维护性，避免重复写代码。

文本示意图：

提示词模板（带有占位符） + 动态变量值 → 生成完整的提示词

示例代码：

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

# 初始化 Chat Prompt Template（支持消息列表，适合 Chat Model）
prompt_template = ChatPromptTemplate.from_messages([
    SystemMessage(content="你是一只戴着帽子的、会说话的柴犬，说话要可爱、幽默，用‘汪汪’结尾。"),
    MessagesPlaceholder(variable_name="chat_history"),  # 占位符：对话历史
    HumanMessage(content="{user_input}"),  # 占位符：用户输入
])

# 构造动态变量值
variables = {
    "chat_history": [
        HumanMessage(content="你叫什么名字呀？"),
        SystemMessage(content="我叫柴小帽！汪汪～"),
    ],
    "user_input": "你今天想去哪里玩呀？",
}

# 生成完整的提示词
prompt = prompt_template.invoke(variables)
print(prompt)
# 输出：
# ChatPromptValue(messages=[
#   SystemMessage(content='你是一只戴着帽子的、会说话的柴犬，说话要可爱、幽默，用‘汪汪’结尾。'),
#   HumanMessage(content='你叫什么名字呀？'),
#   SystemMessage(content='我叫柴小帽！汪汪～'),
#   HumanMessage(content='你今天想去哪里玩呀？')
# ])

# 调用 Chat Model
response = llm.invoke(prompt)
print(response.content)
# 输出示例：我今天想去公园玩！可以和别的小狗一起跑步、玩飞盘，还能吃主人给的小零食呢！汪汪～
2.1.3 组件 3：Memory（记忆组件）

生活化比喻：Memory 就像是“外卖员的手机备忘录”——用来存储之前的对话历史、任务状态，支持 LLM 记住上下文。

核心定义：Memory 是一个“用来存储和检索信息的组件”——常见的 Memory 类型包括：

  • ConversationBufferMemory：最简单的记忆组件，直接存储所有的对话历史（适合 Token 窗口大的 LLM）
  • ConversationSummaryMemory：总结对话历史的记忆组件，只存储对话历史的总结（适合 Token 窗口小的 LLM）
  • ConversationBufferWindowMemory：滑动窗口记忆组件，只存储最近的 k 条对话历史（适合 Token 窗口小的 LLM）
  • VectorStoreRetrieverMemory：向量检索记忆组件，用向量数据库存储对话历史，根据用户输入的语义检索相关的对话历史（适合 Token 窗口小、需要长时记忆的 LLM）

文本示意图：

用户输入 → Memory 检索相关信息 → 生成包含上下文的提示词 → LLM → 助手输出 → Memory 存储对话历史

示例代码：

from langchain.memory import ConversationBufferMemory
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain.chains import LLMChain

# 初始化 Memory
memory = ConversationBufferMemory(memory_key="chat_history", return_messages=True)

# 初始化 Chat Prompt Template
prompt_template = ChatPromptTemplate.from_messages([
    SystemMessage(content="你是一只戴着帽子的、会说话的柴犬，说话要可爱、幽默，用‘汪汪’结尾。"),
    MessagesPlaceholder(variable_name="chat_history"),
    HumanMessage(content="{user_input}"),
])

# 初始化 LLM Chain（把 Prompt Template、LLM、Memory 串联起来）
llm_chain = LLMChain(llm=llm, prompt=prompt_template, memory=memory, verbose=True)

# 第一次调用
response1 = llm_chain.invoke({"user_input": "你叫什么名字呀？"})
print("第一次输出：", response1["text"])
# 输出示例：我叫柴小帽！汪汪～

# 第二次调用（不需要手动传入对话历史，Memory 会自动处理）
response2 = llm_chain.invoke({"user_input": "那你今天想去哪里玩呀？"})
print("第二次输出：", response2["text"])
# 输出示例：我今天想去公园玩！可以和别的小狗一起跑步、玩飞盘，还能吃主人给的小零食呢！汪汪～

# 查看 Memory 存储的对话历史
print("Memory 存储的对话历史：", memory.load_memory_variables({}))
2.1.4 组件 4：Tool（工具组件）

生活化比喻：Tool 就像是“外卖员的工具包”——里面有各种工具（比如手机导航、充电宝、外卖箱、保温袋等），外卖员可以根据需要自主选择使用。

核心定义：Tool 是一个“LLM 可以调用的外部函数或 API”——每个 Tool 都有一个 name（工具名称）、一个 description（工具描述，用来告诉 LLM 这个工具是做什么的、什么时候用、怎么用）、一个 args_schema（参数模式，用来定义工具的输入参数类型）、一个 func（工具的执行函数）。

常见的 Tool 类型包括：

  • 搜索工具：比如 TavilySearchResults（Tavily 搜索引擎）、SerpAPI（SerpAPI 搜索引擎）、GoogleSearchAPIWrapper（Google 搜索 API）
  • Python 代码执行工具：比如 PythonREPLTool（本地 Python 解释器）、PythonAstREPLTool（本地 Python AST 解释器）
  • 数据库查询工具：比如 SQLDatabaseToolkit（SQL 数据库查询工具包）、PandasDataFrameToolkit（Pandas DataFrame 查询工具包）
  • API 调用工具：比如 Requests（HTTP 请求工具）、OpenAPISpec（OpenAPI 规范调用工具）

文本示意图：

用户输入 → LLM 推理是否需要调用工具 → 需要的话生成工具调用请求 → 执行工具 → 生成包含工具输出的提示词 → LLM → 助手输出

示例代码：

from langchain_community.tools.tavily_search import TavilySearchResults
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain.agents import AgentExecutor, create_openai_tools_agent

# 加载环境变量（需要在 .env 文件中设置 TAVILY_API_KEY）
load_dotenv()

# 初始化 Tool（Tavily 搜索引擎，最多返回 2 条结果）
search_tool = TavilySearchResults(max_results=2)
tools = [search_tool]

# 初始化 Chat Prompt Template（适合 OpenAI Tools Agent）
prompt_template = ChatPromptTemplate.from_messages([
    SystemMessage(content="你是一只戴着帽子的、会说话的柴犬，说话要可爱、幽默，用‘汪汪’结尾。你可以使用 Tavily 搜索引擎获取实时信息。"),
    MessagesPlaceholder(variable_name="chat_history", optional=True),
    HumanMessage(content="{input}"),
    MessagesPlaceholder(variable_name="agent_scratchpad"),  # 占位符：工具调用的中间过程
])

# 初始化 OpenAI Tools Agent（专门为 OpenAI 的 Tools API 设计的 Agent）
agent = create_openai_tools_agent(llm=llm, tools=tools, prompt=prompt_template)

# 初始化 Agent Executor（用来执行 Agent 的组件）
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

# 调用 Agent（查询今天的北京天气）
response = agent_executor.invoke({"input": "今天北京的天气怎么样呀？"})
print("输出：", response["output"])
# 输出示例：让我帮你查一下今天北京的天气！汪汪～
# （然后调用 Tavily 搜索引擎，获取结果，最后输出可爱的天气报告）
2.1.5 组件 5：Chain（链条组件）

生活化比喻：Chain 就像是“外卖员的硬编码路线图”——外卖员必须严格按照路线图走，不能调整。

核心定义：Chain 是一个“把多个 LangChain 组件串联起来的可执行工作流”——Chain 的输入是一个字典（包含所有动态变量值），输出是一个字典（包含所有组件的输出结果）。Chain 是线性的，没有分支、循环、并行、状态回溯等复杂逻辑。

常见的 Chain 类型包括：

  • LLMChain：最简单的 Chain，把 Prompt Template、LLM、Memory 串联起来
  • SequentialChain：顺序 Chain，把多个 Chain 串联起来，前一个 Chain 的输出作为后一个 Chain 的输入
  • RouterChain：路由 Chain，根据 LLM 的推理结果选择不同的子 Chain（这是 LangChain 中唯一支持分支逻辑的 Chain）
  • RetrievalQAChain：检索问答 Chain，把向量检索器、Prompt Template、LLM 串联起来，用来做 RAG（检索增强生成）应用

文本示意图：

输入字典 → Chain 1 → 输出字典 1 → Chain 2 → 输出字典 2 → ... → 最终输出字典
2.1.6 组件 6：Agent & Agent Executor（代理组件 & 代理执行器）

生活化比喻：Agent 就像是“外卖员的智能手机导航+语音助手”——导航会根据实时路况自主调整路线，语音助手会根据用户和商家的消息自主决定下一步操作；Agent Executor 就像是“外卖员的身体”——负责执行导航和语音助手的指令。

核心定义：

  • Agent：Agent 是一个“比 Chain 更高级的可执行工作流”——Agent 包含一个 LLM、一个 Prompt Template、一个 Tools List，它能根据 LLM 的推理结果自主选择工具、自主调整步骤、自主处理分支和循环，而不是硬编码的串联。Agent 的核心是“推理-行动循环”（ReAct Cycle）：
    1. 思考（Think）：LLM 根据当前的上下文（对话历史、工具输出）思考下一步该做什么
    2. 行动（Act）：如果需要调用工具，生成工具调用请求；如果不需要调用工具，生成最终的助手输出
    3. 观察（Observe）：执行工具，获取工具输出
    4. 重复（Repeat）：把工具输出加入上下文，回到第一步，直到生成最终的助手输出
  • Agent Executor：Agent Executor 是一个“用来执行 Agent 的组件”——它负责处理 Agent 的推理-行动循环，负责管理工具的调用，负责管理记忆的存储和检索。

常见的 Agent 类型包括：

  • OpenAI Tools Agent：专门为 OpenAI 的 Tools API 设计的 Agent（支持并行工具调用）
  • ReAct Agent：基于 ReAct 提示词工程方法设计的 Agent（通用型 Agent，适合所有 LLM）
  • Structured Chat Agent：支持结构化工具输入的 Agent（适合需要复杂参数的工具）
  • Zero-Shot React Description Agent：不需要示例，只根据工具描述就能自主选择工具的 Agent（通用型 Agent，适合所有 LLM）

文本示意图（ReAct Cycle）：

输入 → 思考（LLM） → 行动（生成工具调用请求/生成最终输出） → 如果是工具调用请求：执行工具 → 观察（获取工具输出） → 把观察加入上下文 → 回到思考 → 重复 → 最终输出
2.2 核心概念：LangGraph 生态的核心组件

接下来，我们来学习 LangGraph 生态的核心组件——注意，LangGraph 生态也分成了多个独立的包：

  • LangGraph Core：LangGraph 生态的核心库，包含了所有基础组件的定义（如 StateGraph、CompiledGraph、State、Node、Edge、Checkpointer 等）
  • LangGraph Checkpoint SQLite：基于 SQLite 的检查点实现（用来持久化状态，支持状态回溯、断点续传）
  • LangGraph Checkpoint Postgres：基于 PostgreSQL 的检查点实现（用来持久化状态，支持高并发）
  • LangGraph OpenAI：包含了 LangGraph 与 OpenAI 官方组件的集成（如 create_react_agent、create_openai_tools_agent 等）
  • LangGraph Anthropic：包含了 LangGraph 与 Anthropic 官方组件的集成
  • LangGraph Google：包含了 LangGraph 与 Google 官方组件的集成
  • LangGraph Studio：LangChain 官方维护的 LangGraph 可视化开发平台（提供了拖拽式构建图、调试图、监控图等功能）

在本文中，我们主要关注 LangGraph Core 的核心组件。

2.2.1 组件 1：State（状态）

生活化比喻：State 就像是“外卖配送调度系统的全局调度状态”——包括所有待分配的订单、所有外卖员的位置和状态、所有商家的备餐状态等，所有节点（部门）都可以读取和更新状态。

核心定义：State 是一个“统一的、结构化的对象”——用来存储整个 LangGraph 应用的全局状态。State 可以是任意类型的对象，但通常推荐使用 TypedDict（Python 3.8+ 内置）或 Pydantic BaseModel（更严格的类型检查）来定义，这样可以提高代码的可读性和可维护性，还能获得 IDE 的自动补全支持。

State 的更新机制是“合并更新”（Merge Update）——也就是说，每个节点只需要返回它想要更新的 State 的部分字段，LangGraph 会自动把这些部分字段合并到全局 State 中，而不需要返回整个 State。这大大简化了状态管理的复杂度。

文本示意图：

初始状态 → 节点 1 更新部分字段 → 合并后的状态 1 → 节点 2 更新部分字段 → 合并后的状态 2 → ... → 最终状态

示例代码（用 TypedDict 定义 State）：

from typing import TypedDict, Annotated, List
from langgraph.graph.message import add_messages  # 用来合并消息列表的函数

# 定义 State（用 TypedDict）
class State(TypedDict):
    # messages：消息列表，用 add_messages 函数合并（保留所有历史消息）
    messages: Annotated[List, add_messages]
    # user_name：用户名，直接覆盖更新
    user_name: str
    # order_status：订单状态，直接覆盖更新
    order_status: str

示例代码（用 Pydantic BaseModel 定义 State）：

from pydantic import BaseModel, Field
from typing import List
from langgraph.graph.message import add_messages

# 定义 State（用 Pydantic BaseModel）
class State(BaseModel):
    # messages：消息列表，用 add_messages 函数合并（保留所有历史消息）
    messages: List = Field(default_factory=list, description="消息列表")
    # user_name：用户名，直接覆盖更新
    user_name: str = Field(default="", description="用户名")
    # order_status：订单状态，直接覆盖更新
    order_status: str = Field(default="pending", description="订单状态")
    
    class Config:
        # 允许添加额外的字段
        extra = "allow"
    
    # 定义合并函数（必须实现 __add__ 方法）
    def __add__(self, other: "State") -> "State":
        # 合并消息列表
        merged_messages = add_messages(self.messages, other.messages)
        # 合并其他字段（直接覆盖）
        merged_data = {**self.dict(exclude={"messages"}), **other.dict(exclude={"messages"})}
        # 返回合并后的 State
        return State(messages=merged_messages, **merged_data)
2.2.2 组件 2：Node（节点）

生活化比喻：Node 就像是“外卖配送调度系统的各个部门”——比如订单接收节点、外卖员分配节点、路线规划节点、备餐监控节点、配送监控节点、投诉处理节点等，每个节点负责一个具体的“动作”。

核心定义：Node 是一个“可执行的函数”——输入是当前的全局 State，输出是想要更新的 State 的部分字段（或者是一个特殊的 END 节点，表示结束工作流）。Node 可以是任意类型的函数，但通常推荐使用异步函数（async def）来提高并发性能。

Node 的类型包括：

  • 普通节点（Normal Node）：最常见的节点，执行一个具体的动作，返回想要更新的 State 的部分字段
  • 条件节点（Conditional Node）：也叫“路由节点”（Router Node），根据当前的全局 State 返回下一个节点的名称（或者是一个节点名称列表，表示并行执行多个节点）
  • END 节点：特殊的节点，表示结束工作流
  • START 节点：特殊的节点，表示开始工作流（不需要手动定义，LangGraph 会自动添加）

文本示意图：

当前 State → 节点执行 → 返回更新的部分字段 → 合并到全局 State → 进入下一个节点

示例代码（普通节点）：

from langchain_core.messages import SystemMessage, HumanMessage, AIMessage
from langgraph.graph import StateGraph, END

# 定义普通节点：问候用户
def greet_user(state: State) -> State:
    # 获取当前的 State
    user_name = state.get("user_name", "")
    # 生成问候消息
    if user_name:
        greet_message = AIMessage(content=f"你好，{user_name}！我是柴小帽，很高兴为你服务！汪汪～")
    else:
        greet_message = AIMessage(content="你好！我是柴小帽，很高兴为你服务！请问你叫什么名字呀？汪汪～")
    # 返回想要更新的 State 的部分字段（合并问候消息到 messages 列表）
    return {"messages": [greet_message]}

# 定义普通节点：获取用户名
def get_user_name(state: State) -> State:
    # 获取当前的 State
    messages = state.get("messages", [])
    # 获取最后一条用户消息
    last_user_message = messages[-1].content if messages and isinstance(messages[-1], HumanMessage) else ""
    # 提取用户名（这里简化处理，直接用最后一条用户消息作为用户名）
    user_name = last_user_message.strip()
    # 返回想要更新的 State 的部分字段
    return {"user_name": user_name}

# 定义普通节点：结束对话
def end_conversation(state: State) -> State:
    # 生成结束消息
    end_message = AIMessage(content="好的，很高兴为你服务！下次见！汪汪～")
    # 返回想要更新的 State 的部分字段
    return {"messages": [end_message], "order_status": "completed"}
2.2.3 组件 3：Edge（边）

生活化比喻：Edge 就像是“外卖配送调度系统的工作流程规则”——比如订单接收节点 → 外卖员分配节点（无条件转移）、外卖员分配节点 → 等待节点（如果没有合适的外卖员）等，每条边定义了“动作之间的转移条件”。

核心定义：Edge 是一个“连接两个节点的有向线段”——用来定义节点之间的转移条件。Edge 的类型包括：

  • 无条件边（Unconditional Edge）：最常见的边，从一个节点无条件转移到另一个节点（比如 START → greet_user、greet_user → get_user_name）
  • 条件边（Conditional Edge）：也叫“路由边”（Router Edge），从一个节点根据条件转移到不同的节点（比如 get_user_name → greet_user 如果用户没有输入名字，get_user_name → end_conversation 如果用户输入了名字）
  • 并行边（Parallel Edge）：从一个节点并行转移到多个节点（需要先定义一个“并行节点组”，然后再用条件边或无条件边连接）

文本示意图：

节点 A → 条件判断 → 节点 B/节点 C/节点 D/...

示例代码（条件边的条件函数）：

# 定义条件函数：判断是否已经获取了用户名
def has_user_name(state: State) -> str:
    # 获取当前的 State
    user_name = state.get("user_name", "")
    # 如果已经获取了用户名，返回 "end_conversation"
    if user_name:
        return "end_conversation"
    # 如果没有获取用户名，返回 "greet_user"
    else:
        return "greet_user"
2.2.4 组件 4：StateGraph & CompiledGraph（状态图 & 编译后的状态图）

生活化比喻：StateGraph 就像是“外卖配送调度系统的设计图纸”——上面画了所有的节点（部门）和边（工作流程规则）；CompiledGraph 就像是“外卖配送调度系统的实际运行程序”——可以直接执行，用来处理实际的订单。

核心定义：

  • StateGraph：StateGraph 是一个“用来定义 LangGraph 应用的有向图”——开发者需要先定义 State，然后添加节点，最后添加边（连接节点），从而构建一个完整的 StateGraph。
  • CompiledGraph：CompiledGraph 是一个“编译后的 StateGraph”——开发者需要调用 StateGraph.compile() 方法来编译 StateGraph，得到 CompiledGraph。CompiledGraph 可以直接执行，用来处理实际的输入，还支持检查点（Checkpoint）、状态回溯、断点续传、可视化等功能。

文本示意图：

定义 State → 添加节点 → 添加边 → 构建 StateGraph → 编译 → 得到 CompiledGraph → 执行 → 得到最终 State

示例代码（构建和执行一个简单的 StateGraph）：

# 第一步：定义 State（用 TypedDict）
from typing import TypedDict, Annotated, List
from langgraph.graph.message import add_messages

class State(TypedDict):
    messages: Annotated[List, add_messages]
    user_name: str
    order_status: str

# 第二步：定义节点
from langchain_core.messages import SystemMessage, HumanMessage, AIMessage

def greet_user(state: State) -> State:
    user_name = state.get("user_name", "")
    if user_name:
        greet_message = AIMessage(content=f"你好，{user_name}！我是柴小帽，很高兴为你服务！汪汪～")
    else:
        greet_message = AIMessage(content="你好！我是柴小帽，很高兴为你服务！请问你叫什么名字呀？汪汪～")
    return {"messages": [greet_message]}

def get_user_name(state: State) -> State:
    messages = state.get("messages", [])
    last_user_message = messages[-1].content if messages and isinstance(messages[-1], HumanMessage) else ""
    user_name = last_user_message.strip()
    return {"user_name": user_name}

def end_conversation(state: State) -> State:
    end_message = AIMessage(content="好的，很高兴为你服务！下次见！汪汪～")
    return {"messages": [end_message], "order_status": "completed"}

# 第三步：定义条件函数
def has_user_name(state: State) -> str:
    user_name = state.get("user_name", "")
    if user_name:
        return "end_conversation"
    else:
        return "greet_user"

# 第四步：构建 StateGraph
from langgraph.graph import StateGraph, END

# 初始化 StateGraph
graph = StateGraph(State)

# 添加节点
graph.add_node("greet_user", greet_user)
graph.add_node("get_user_name", get_user_name)
graph.add_node("end_conversation", end_conversation)

# 添加边
# START → greet_user（无条件边）
graph.set_entry_point("greet_user")
# greet_user → get_user_name（无条件边）
graph.add_edge("greet_user", "get_user_name")
# get_user_name → 条件判断 → end_conversation/greet_user（条件边）
graph.add_conditional_edges(
    "get_user_name",
    has_user_name,
    {
        "end_conversation": "end_conversation",
        "greet_user": "greet_user",
    }
)
# end_conversation → END（无条件边）
graph.add_edge("end_conversation", END)

# 第五步：编译 StateGraph，得到 CompiledGraph
compiled_graph = graph.compile()

# 第六步：可视化 CompiledGraph（需要安装 graphviz 和 pygraphviz）
try:
    compiled_graph.get_graph().draw_mermaid_png(output_file_path="simple_graph.png")
    print("CompiledGraph 已保存为 simple_graph.png")
except Exception as e:
    print(f"可视化失败：{e}")
    # 输出 Mermaid 代码，可以在 https://mermaid.live/ 中查看
    print(compiled_graph.get_graph().draw_mermaid())

# 第七步：执行 CompiledGraph
# 第一次执行（用户没有输入名字）
initial_state1 = {"messages": [HumanMessage(content="")]}
final_state1 = compiled_graph.invoke(initial_state1)
print("第一次执行的最终 State：")
for message in final_state1["messages"]:
    print(f"{message.type
# rxjava # android
Logo
码道开发者社区

码道开发者社区，聚焦华为云码道 CodeArts 代码智能体，沉淀 Agent、Skill、鸿蒙开发实战内容，供开发者查阅资料、交流技术、分享工程实践

更多推荐

cover

状态丢失排障：Claude 持久记忆的跨会话局限与知芽 Notebook Skill 单元记忆如何沉淀可复用知识资产

avatar 码道开发者社区
cover

AI外贸建站06｜给网站穿上“公司制服”：配色、字体、logo和favicon，Codex一次应用全站

avatar 码道开发者社区
cover

Workbuddy中运行NetSuite MCP

avatar 码道开发者社区