LangGraph 基础：

LangGraph 是一个用于构建、管理和部署长运行、具备状态的 agent 的 low-level orchestration framework and runtime；它关注的是 agent orchestration，也就是如何把模型调用、工具调用、状态保存、控制流跳转以及人工介入组织成一个可持续运行的流程

LangGraph 经常与 LangChain 的 models 和 tools 一起使用；但 LangGraph 并不依赖 LangChain 才能使用；更准确地说，LangChain 负责提供模型集成、工具集成与可组合组件，也包含构建在 LangGraph 之上的 agent abstraction；而 LangGraph 负责更底层的编排能力；LangSmith 则补上 tracing、debugging、evaluation、observability 与 deployment 相关能力；如果希望快速使用更高层的 agent abstraction，可以从 LangChain 的 agents 入手；如果希望显式控制状态、执行路径和运行时行为，则应学习 LangGraph

简介：

LangGraph 的核心不是 "替你写好一个固定形态的智能体"，而是提供一套更底层的运行机制，让开发者自己组织 long-running、stateful 的 agent 或 workflow；这也是为什么会反复强调 durable execution、streaming、human-in-the-loop 和 memory 这些能力，它们都属于 "编排与运行" 层面的能力，而不是单次模型调用层面的能力

理解 LangGraph，可以先抓住三个最重要的对象：

State：运行过程中的共享数据；图中的各个步骤通过读写 State 协同工作；在一个 agent 执行期间，State 会持续存在，并随着节点执行不断被更新；

Node：真正执行工作的函数；一个节点通常负责一件明确的事情，例如调用模型、执行工具、整理结果，或者根据当前状态做出下一步决策；

Edge：节点之间的连接关系；在 Graph API 中，开发者通过 add_edge 或 add_conditional_edges 显式声明执行路径，让图知道下一步该走向哪里；

LangGraph 当前提供两种主要写法：

Graph API：把流程明确拆成节点和连接关系，适合需要精确控制执行路径的场景

Functional API：把同样的控制逻辑写进一个函数，用普通的循环和条件判断组织流程，适合希望用单函数方式表达 agent 行为的场景

如果把 LangChain 和 LangGraph 放在一起理解，可以把两者的职责区分为两层；LangChain 侧重模型、工具和组件集成，帮助我们把 LLM 应用的基础部件接起来；而 LangGraph 侧重状态化编排，让这些部件按照可追踪、可恢复、可暂停、可继续的方式运行

因此，LangGraph 不是把 LangChain 替换掉，而是在需要更细粒度控制时，把应用从 "能调用模型" 推进到 "能可靠运行一条复杂流程"

从学习路径上看，当前官方文档通常先从 Quickstart 入手，先搭起一个最小可运行的 graph；然后再继续学习 persistence、streaming、interrupts、memory 和 subgraphs；这条路径非常符合 LangGraph 的定位：先理解 State、Node、Edge 和 compile 之后的执行方式，再逐步给图增加可恢复执行、实时输出、人工介入和多图组合这些能力

快速体验：

要理解 LangGraph 的作用，最直接的办法不是先记很多概念，而是先看 "直接调用模型" 和 "把同一类任务组织成 graph" 在写法上有什么差别

先看最基础的模型调用；这一层更多是在使用 LangChain 提供的模型接口；如果沿用官方 quickstart 的示例模型，需要先在运行环境中配置 ANTHROPIC_API_KEY

from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage

# 初始化聊天模型
# model: 指定要使用的聊天模型名称；这里沿用官方 quickstart 示例中的模型
# temperature: 控制输出随机性；设为 0 表示尽量稳定，适合教学示例
model = init_chat_model(
    "claude-sonnet-4-6",
    temperature=0,
)

# 直接调用模型
# 这里传入的是一个消息列表；列表中的每个元素都表示一条对话消息
response = model.invoke(
    [
        # HumanMessage 表示“用户消息”
        # content: 用户真正输入给模型的文本内容
        HumanMessage(content="Add 3 and 4.")
    ]
)

# 打印模型最终返回的文本内容
print(response.content)

这段代码已经可以完成一次模型调用；但如果任务不再只是 "问一次，答一次"，而是要在模型和工具之间循环、把中间结果保存在状态里、根据结果决定下一步节点、或者在中途暂停等待外部输入，就需要额外组织运行过程，LangGraph 的价值，正是在这里开始体现

在进入官方 quickstart 的 Graph API agent 之前，先看一个更小的预热示例；它来自官方 overview，只有一个节点，目的只是先看懂 graph 最基本的执行方式：

from langgraph.graph import StateGraph, MessagesState, START, END

# 这是一个最小节点示例    
# state: 当前图运行时的共享状态；这里使用内置的 MessagesState
# MessagesState: {"messages": Annotated[list[AnyMessage], add_messages]}
# 返回值必须是“状态更新字典”
def mock_llm(state: MessagesState):
    # 这里返回一条 AI 消息，表示这个节点执行后的输出会写回 state["messages"]
    return {
        "messages": [
            {
                "role": "ai",
                "content": "hello world",
            }
        ]
    }

# 创建一张图
# MessagesState: 指定这张图使用什么状态结构
# 内置的 MessagesState 适合消息驱动的最小示例
graph = StateGraph(MessagesState)

# 注册节点
# node 参数是一个节点函数
# 如果只传函数本身，LangGraph 会使用函数名作为节点名
graph.add_node(mock_llm)

# 添加执行边
# add_edge(source, target):
#   source: 起点节点名
#   target: 终点节点名
# START: 图的统一入口
# END: 图的统一结束点
graph.add_edge(START, "mock_llm")
graph.add_edge("mock_llm", END)

# 编译图
# compile(): 把“图的定义”转换成“可执行对象”
graph = graph.compile()

# 执行图
# invoke(input):
# input 是这次运行的初始状态
# 这里把一条用户消息放入 messages，作为图的输入
result = graph.invoke(
    {
        "messages": [
            {
                "role": "user",
                "content": "hi!",
            }
        ]
    }
)

print(result)

这个预热示例只需要看清四个点：

  • StateGraph(...) 用来声明这张图围绕哪种 State 运行；
  • add_node(...) 用来告诉图有哪些节点；
  • add_edge(...) 用来告诉图节点之间如何连接；
  • compile() 之后，图才能通过 invoke() 真正执行；

有了这层基础，再看官方 quickstart 的最小 agent 就容易很多；下面这段代码仍然是 Graph API 的标准思路，但会按职责拆开解释

先看整体结构；这个 agent 的核心不是语法，而是三个对象之间的关系：

  • State 负责保存消息和辅助状态；
  • llm_call 与 tool_node 是两个主要节点；
  • Edge 和条件路由决定下一步是继续调用工具，还是直接结束；

定义模型和工具：

准备 agent 后面要用到的基础部件；模型是推理核心，工具是模型可以选择调用的外部能力：

from langchain.chat_models import init_chat_model
from langchain.tools import tool

# 初始化模型
# 后面的节点都会复用这个模型实例
model = init_chat_model(
    "claude-sonnet-4-6",
    temperature=0,
)

# 这是一个可被模型调用的工具
# @tool 会把函数转换成工具对象，并保留函数签名和说明文档
@tool
def multiply(a: int, b: int) -> int:
    """
    Multiply `a` and `b`.
    """
    return a * b

@tool
def add(a: int, b: int) -> int:
    """
    Adds `a` and `b`.
    """
    return a + b

@tool
def divide(a: int, b: int) -> float:
    """
    Divide `a` and `b`.
    """
    return a / b

# 把所有工具放进列表，便于统一管理
tools = [add, multiply, divide]

# 建立“工具名 -> 工具对象”的映射
# 后面工具节点执行时，会根据 tool_call["name"] 找到对应工具
tools_by_name = {tool.name: tool for tool in tools}

# 绑定工具到模型，绑定之后，模型的输出中可能出现 tool_calls
model_with_tools = model.bind_tools(tools)

到这里时，可以先把它理解成一句话：先准备好 "模型能调用哪些工具"，后面的 graph 只是负责把这件事组织起来；

定义 State：

定义整个 graph 运行期间共享的数据结构；这里最需要关注的是 messages 字段；而 llm_calls 只是一个辅助字段，用来观察模型被调用了几次：

import operator

from langchain.messages import AnyMessage
from typing_extensions import Annotated, TypedDict


class MessagesState(TypedDict):
    # messages: 保存整个 agent 运行期间的消息历史
    # Annotated[list[AnyMessage], operator.add]:
    #   这里不是普通的类型声明，还附带了 reducer 信息
    #   operator.add 表示当节点返回新的 messages 时，不会覆盖旧列表，而是把新消息追加到原列表后面，这样多轮执行时，消息上下文才能一直保留
    messages: Annotated[list[AnyMessage], operator.add]

    # 记录模型被调用了多少次，这是辅助观察字段，不是主流程必需字段
    llm_calls: int

定义模型节点：

读取当前状态里的消息，把消息交给模型，然后把模型最新返回的结果写回状态：

from langchain.messages import SystemMessage

# “模型节点”
def llm_call(state: MessagesState):
    # 读取当前状态中的消息，然后调用模型，最后把模型输出写回状态
    return {
        "messages": [
            model_with_tools.invoke(
                [
                    # SystemMessage 用来提供系统级指令，它告诉模型当前任务的角色和目标
                    SystemMessage(
                        content=(
                            "You are a helpful assistant tasked with performing "
                            "arithmetic on a set of inputs."
                        )
                    )
                ]
                # 把系统消息和当前消息历史拼接在一起，作为模型输入
                + state["messages"]
            )
        ],
        # 更新模型调用次数
        # 如果 state 中还没有这个字段，就从 0 开始计数
        "llm_calls": state.get("llm_calls", 0) + 1,
    }

这一步可以理解成 "让模型基于当前上下文做一次判断"；如果模型决定直接回答，流程就会走向结束；如果模型返回了 tool_calls，流程就会进入工具节点

定义工具节点：

读取模型上一条消息里的 tool_calls，逐个执行对应工具，再把工具结果包装成 ToolMessage 写回状态：

from langchain.messages import ToolMessage

# “工具节点”
# 它负责读取上一条模型消息中的 tool_calls，执行这些工具，再把工具执行结果写回消息历史
def tool_node(state: MessagesState):
    result = []

    # state["messages"][-1]: 取当前消息历史中的最后一条消息，它是刚刚由模型节点返回的 AI 消息
    # .tool_calls: 模型决定要调用工具时，会把工具调用信息放在这里
    for tool_call in state["messages"][-1].tool_calls:
        # 根据工具名找到真实工具对象
        tool = tools_by_name[tool_call["name"]]

        # tool_call["args"]: 模型为这次工具调用生成的参数字典
        observation = tool.invoke(tool_call["args"])

        # 工具执行后的结果必须包装成 ToolMessage，这样下一轮模型调用时，模型才能读取到工具输出
        result.append(
            ToolMessage(
                # 工具返回的结果内容
                content=observation,
                # 对应哪一次工具调用；用于把结果和原始 tool_call 对上
                tool_call_id=tool_call["id"],
            )
        )

    # 返回新的 messages 更新，这里返回的 ToolMessage 会被追加到原来的消息历史中
    return {"messages": result}

这里工具执行完并不是流程结束，而是要把结果重新写回 messages；只有这样，下一轮模型调用才能基于工具输出继续推理

定义条件路由：

这一段不做业务处理，只负责决定流程往哪里走；这正是 Edge 的意义所在：它不负责生成数据，而是负责控制执行路径：

from typing import Literal
from langgraph.graph import END

# 判断下一步路由
def should_continue(state: MessagesState) -> Literal["tool_node", END]:

    # 取最后一条消息；它通常是 llm_call 节点刚写入的模型回复
    last_message = state["messages"][-1]

    # 如果模型返回了 tool_calls，说明它希望先调用工具
    if last_message.tool_calls:
        return "tool_node"

    # 如果没有 tool_calls，说明模型已经可以直接给出最终回答
    return END

组装并执行 graph：

前面的代码都还是 "零件"，这一段才是真正把零件装成一张图：

from langchain.messages import HumanMessage
from langgraph.graph import StateGraph, START, END


# 创建图
# MessagesState: 指定这张图的共享状态结构
agent_builder = StateGraph(MessagesState)

# 注册节点
# add_node(name, action):
#   name: 节点名称
#   action: 节点函数
agent_builder.add_node("llm_call", llm_call)
agent_builder.add_node("tool_node", tool_node)

# 添加入口边
# 图执行开始后，先进入 llm_call 节点
agent_builder.add_edge(START, "llm_call")

# 添加条件路由
# add_conditional_edges(source, path, path_map):
#   source: 从哪个节点之后开始判断路由
#   path: 路由函数，这里是 should_continue
#   path_map: 路由函数可能返回的目标节点集合
agent_builder.add_conditional_edges(
    "llm_call",
    should_continue,
    ["tool_node", END],
)

# 添加边，如果执行了工具节点，就回到模型节点，这样模型才能继续读取工具结果并决定下一步
agent_builder.add_edge("tool_node", "llm_call")

# 编译图
agent = agent_builder.compile()

# 执行图
# 初始输入只需要提供一条用户消息
result = agent.invoke(
    {
        "messages": [
            HumanMessage(content="Add 3 and 4.")
        ]
    }
)

# 打印整个执行后积累下来的消息历史
for message in result["messages"]:
    message.pretty_print()

整个流程：用户消息先进入 messages，llm_call 读取消息并调用模型；如果模型返回 tool_calls，就进入 tool_node；tool_node 执行工具，并把工具结果写回 messages；流程再回到 llm_call；如果模型不再请求工具，graph 就走向 END

直接模型调用只关心 "给模型什么输入，拿回什么输出"；而 LangGraph 会把执行过程拆成 State、Node 和路由逻辑；模型是否调用工具、工具执行后的结果如何回到状态、接下来是继续循环还是结束，都被写成显式的运行结构

这也是 LangGraph 最值得学习的地方；它让 "调用模型" 这件事从一次请求，变成一条可组织、可追踪、可扩展的执行流程

如果不想显式编写节点和边，官方 quickstart 还给出了 Functional API 的写法；它用 @entrypoint() 和 @task 把同样的 agent 写成一个函数，在函数内部使用普通的 while 循环和条件判断来控制流程；两种写法表达的是同一类运行机制，只是组织代码的方式不同

单 Agent 构建：

Agent 简介：

在 LangGraph 的官方定义里，LangGraph 是一个面向 long-running、stateful agents 的 low-level orchestration framework and runtime；因此，这里的 Agent 不是单次调用大模型的包装器，而是一段可以持续运行、能够保存状态、能够根据条件切换执行路径的流程

如果从实现角度理解，一个 LangGraph Agent 至少包含四类基本能力：

  • 使用 model 完成语言理解与生成；
  • 在需要时调用 tools 处理外部动作或外部数据；
  • 通过 shared state 在多个步骤之间保存上下文；
  • 在关键节点通过 persistence 和 interrupts 暂停、恢复并接受人工输入；

这也是 LangGraph 与普通 "模型调用脚本" 的根本区别：前者强调的是 orchestration，也就是如何把多个步骤、多个判断点和多轮状态变化组织成一个可以可靠运行的 agent；后者通常只负责完成一次输入到输出的请求

在具体实现上，LangGraph 并不会预先规定你的 agent 必须长成什么样；官方文档给出的通用做法，是先把要自动化的过程拆成离散步骤，再把每个步骤实现成 node，最后通过 edge 和 shared state 把它们连接起来；因此，理解 Agent 的最好方式，不是先记住某个固定 API，而是理解 "状态如何流动、节点如何执行、分支如何选择、执行如何暂停与恢复"

构建 Agent：

要构建一个聊天 Agent，最自然的起点仍然是基础的模型调用；在 LangChain 中通常先初始化一个 chat model，然后把用户消息传给模型：

from langchain_community.chat_models.tongyi import ChatTongyi
from langchain.messages import HumanMessage

# 初始化聊天模型；这里沿用 LangGraph quickstart 中的模型写法
model = ChatTongyi(
    model="qwen3-max",
    temperature=0,
)

# 直接调用模型；这一层只负责单次请求，没有状态编排和流程控制
response = model.invoke(
    [HumanMessage(content="你好，你是谁；你可以帮我做什么")]
)

# 模型返回结果
print(response.content)

而在 LangGraph 中，一个最小的聊天 Agent，通常会先把 "调用模型" 这件事放进 graph；下面这段代码沿用官方文档的 MessagesState 写法，把聊天过程组织成一张最小可执行图：

from langchain_community.chat_models.tongyi import ChatTongyi
from langgraph.graph import StateGraph, MessagesState, START, END


# 初始化模型；后面的节点会复用这个模型实例
model = ChatTongyi(
    model="qwen3-max",
    temperature=0,
)


def call_model(state: MessagesState):
    # 读取当前消息历史并调用模型
    # state["messages"] 中会包含本轮输入以及已有的上下文消息
    response = model.invoke(state["messages"])

    # 节点返回的是对状态的更新；
    # 这里把新的 AIMessage 追加回 messages，供后续步骤继续使用
    return {"messages": [response]}


# 创建 graph；MessagesState 是 LangGraph 内置的消息状态结构
builder = StateGraph(MessagesState)

# 把调用模型的函数注册成节点
builder.add_node("call_model", call_model)

# 定义执行路径；从 START 进入 call_model，执行完后到 END
builder.add_edge(START, "call_model")
builder.add_edge("call_model", END)

# compile 之后，图才会变成真正可执行的对象
graph = builder.compile()


# 执行 graph；输入仍然从 messages 进入
# result 是图执行结束后的最终 state，通常长这样：
# {
#     "messages": [
#         HumanMessage(content="你好，你是谁；你可以帮我做什么"),
#         AIMessage(content="我是...")
#     ]
# }

result = graph.invoke(
    {
        "messages": [
            {"role": "user", "content": "你好，你是谁；你可以帮我做什么"}
        ]
    }
)

# 输出模型的回复
print(result["messages"][-1].content)

运行结束后，result["messages"] 会在原始用户消息的基础上多出一条新的 AI 消息；这说明模型调用虽然仍然只发生了一次，但它现在已经被放进了 graph 的状态流转里，而不再只是一个孤立的函数调用

这个示例虽然很小，但已经具备了 Agent 的基本轮廓：

  • 输入不再只是传给模型，而是先进入 MessagesState；
  • call_model 成为图中的一个 node；
  • graph 负责编排这个 node 的执行；
  • 模型输出会被写回 messages，成为后续步骤的上下文；

在这个基础上，Agent 还可以继续增加流式输出，按照官方 streaming 文档，最常用的几种 stream_mode 包括：

  • values：按步骤输出完整 state；
  • updates：按步骤输出各节点返回的状态更新；
  • messages：输出 LLM 调用期间的 token 与 metadata；
  • custom：输出节点通过 get_stream_writer() 主动写出的自定义事件；

例如，下面的代码同时监听 updates 和 messages；这样一边可以看到节点更新了什么状态，一边可以看到模型生成过程中的 token 输出：

# 初始化模型；后面的节点会复用这个模型实例
model = ChatTongyi(
    model="qwen3-max",
    temperature=0,
    streaming=True,
)

# ...

for chunk in graph.stream(
    {
        "messages": [
            {"role": "user", "content": "请介绍一下你自己"}
        ]
    },
    stream_mode=["updates", "messages"], # 指定需要流式消费的内容类型；updates 是节点状态更新，messages 是模型返回的 token
    version="v2", # 指定使用 v2 版本的流式 API
    # 在 version="v2" 下，chunk 类似：
    # {
    #     "type": "updates" or "messages", 
    #     "ns": (),
    #     "data": ...
    # }
    # type: 事件类型，ns: 命名空间，data: 事件数据
):
    if chunk["type"] == "updates":
        # 可能有多个节点的状态更新，这里逐个处理
        for node_name, state_update in chunk["data"].items():
            # 处理节点返回的状态更新
            print(f"Node {node_name} state update: {state_update}")
    # updates 类型的 chunk 类似：
    # {
    #    "type": "updates",
    #    "ns": (),
    #    "data": {
    #       "call_model": {
    #           "messages": [
    #               AIMessage(content="我是助手", ...)
    #           ]
    #       }
    #    }
    # }

    elif chunk["type"] == "messages":
        # 模型返回的 token 以及相关的元信息
        token, metadat = chunk["data"]
        if token.content:
            # 消费流式 token
            print(token.content, end="", flush=True)
    # messages 类型的 chunk 类似：            
    # {
    #    "type": "messages",
    #    "ns": (),
    #    "data": (
    #        AIMessageChunk(content="我", ...),
    #        {
    #            "langgraph_node": "call_model",
    #            "langgraph_step": 1,
    #            "ls_provider": "fakelistchatmodel",
    #            "ls_model_type": "chat",
    #            ...
    #        }
    #    )
    # }

从这里开始，聊天 Agent 就不再只是 "调用一次模型"；而是一个可以继续接入 tools、memory 与 interrupts 的执行框架

工具调用：

工具调用是 Agent 最关键的增强能力之一；当模型需要查外部信息、做确定性计算、调用外部系统或执行真实动作时，单靠模型生成文本并不够，这时就需要让模型发出 tool call，再由程序执行真实工具

按照官方 quickstart 的最小示例，一个具备工具调用能力的 Agent，通常包含下面几个部分：

定义 model 和 tools；定义保存消息历史的 State；定义模型节点，让模型决定是否调用工具；定义工具节点，负责真正执行工具；定义条件路由，决定下一步是继续调用工具，还是结束；把这些节点和边组装成 graph

下面的示例沿用官方 quickstart 的 Graph API 写法，构建一个最小可运行的计算 Agent：

import operator
from typing import Literal

from langchain_community.chat_models.tongyi import ChatTongyi
from langchain.messages import AnyMessage, HumanMessage, SystemMessage, ToolMessage
from langchain.tools import tool
from langgraph.graph import StateGraph, START, END
from typing_extensions import Annotated, TypedDict

# 初始化模型；后续模型节点会反复使用它
model = ChatTongyi(
    model="qwen3-max",
    temperature=0,
)


# 定义工具；这些函数会暴露给模型，供模型在推理时决定是否调用
@tool
def multiply(a: int, b: int) -> int:
    """
    Multiply `a` and `b`.
    """
    return a * b

@tool
def add(a: int, b: int) -> int:
    """
    Adds `a` and `b`.
    """
    return a + b

@tool
def divide(a: int, b: int) -> float:
    """
    Divide `a` and `b`.
    """
    return a / b

# 工具列表
tools = [add, multiply, divide]
# 方便后续根据名字查找工具
tools_by_name = {tool.name: tool for tool in tools}

# 返回一个绑定了工具的模型实例
model_with_tools = model.bind_tools(tools)

class MessagesState(TypedDict):
    # messages 是共享消息历史；
    # operator.add 表示每次返回的新消息都会追加到旧消息后面，而不是覆盖
    messages: Annotated[list[AnyMessage], operator.add]

    # 记录模型被调用了多少次，便于观察 agent 的执行过程
    llm_calls: int

# “模型节点”
# 读取当前 state 中的消息历史，让模型基于上下文继续推理
# 因为 model_with_tools 绑定了工具，模型这一步既可能直接回答，也可能先返回一个包含 tool_calls 的 AIMessage，交给后续工具节点处理
def llm_call(state: MessagesState):
    # LangGraph 节点返回的是“状态更新字典”
    return {
        # 这里更新的是 messages 字段：把本轮模型新产生的一条 AIMessage 追加到历史里
        "messages": [
            # invoke(...) 的返回值是 AIMessage，而不是普通字符串
            # 传给模型的是一整个消息列表：
            model_with_tools.invoke(
                [
                    # 预设一个 SystemMessage，告诉模型它可以使用工具进行计算；后续的对话历史会追加在这个 SystemMessage 后面
                    SystemMessage(
                        content=(
                            "You are a helpful assistant tasked with performing "
                            "arithmetic on a set of inputs."
                        )
                    )
                ]
                # 拼接当前已有的对话历史
                + state["messages"]
            )
        ],
        # 记录模型被调用的次数
        "llm_calls": state.get("llm_calls", 0) + 1,
    }

# “工具节点”
def tool_node(state: MessagesState):
    result = []
    # 读取上一条 AIMessage 里的 tool_calls，逐个执行模型请求的工具
    for tool_call in state["messages"][-1].tool_calls:
        # tool_call 通常长这样：
        # {"name": "add", "args": {"a": 3, "b": 4}, "id": "call_xxx", ...}
        # 根据工具名找到真正的工具对象
        tool = tools_by_name[tool_call["name"]]
        # 用模型给出的参数执行工具，得到工具返回结果
        observation = tool.invoke(tool_call["args"])
        result.append(
            ToolMessage(
                # 把工具结果包装成 ToolMessage，供下一轮模型读取
                content=observation,
                # tool_call_id 必须对应回模型刚才发起的那次工具调用，这样模型在下一轮才能知道这个结果属于哪个 tool_call。
                tool_call_id=tool_call["id"],
            )
        )
    # 返回的是状态更新字典，不是最终答案，result 会被 LangGraph 自动追加到历史消息列表后面
    return {"messages": result}

# “条件路由函数”
# 每次 llm_call 执行完后，检查最新一条消息，决定图下一步该走哪里
def should_continue(state: MessagesState) -> Literal["tool_node", END]:
    last_message = state["messages"][-1]
    if last_message.tool_calls:
        # 返回 tool_node，对应后续要跳转到的节点名
        return "tool_node"

    # 如果没有 tool_calls，说明模型已经可以直接给出答案，当前这轮 agent 执行可以结束
    return END

# 创建 agent 的图构建器，整个执行流程都基于这个状态结构展开
agent_builder = StateGraph(MessagesState)

# 注册两个核心节点：一个负责调用模型，一个负责执行工具
agent_builder.add_node("llm_call", llm_call)
agent_builder.add_node("tool_node", tool_node)

# 从 START 先进入模型节点
agent_builder.add_edge(START, "llm_call")
# 模型节点执行后，根据 should_continue 的判断决定：有工具调用就去 tool_node，没有就直接结束
agent_builder.add_conditional_edges(
    "llm_call",
    should_continue,
    ["tool_node", END],
)
# 工具执行完后，再回到模型节点读取工具结果并继续推理
agent_builder.add_edge("tool_node", "llm_call")

# compile 之后，builder 才会变成真正可执行的 agent
agent = agent_builder.compile()

result = agent.invoke(
    {"messages": [HumanMessage(content="Add 3 and 4, then divide the result by 7.")]}
)
print(result["messages"][-1].content) # The result of adding 3 and 4 is 7. Dividing 7 by 7 gives 1.0.

最终状态里的 messages 不会只有最后一句答案；通常会依次保留用户消息、模型发起工具调用的消息、工具执行后的 ToolMessage，以及模型根据工具结果生成的最终回答

从学习角度看，这一段最有价值的地方正是 "完整消息链路" 而不是某一条孤立输出

这段代码最重要的地方，不在于代码量，而在于执行机制：

  • llm_call 让模型决定是否发起工具调用；
  • tool_node 执行模型请求的工具；
  • add_conditional_edges(...) 根据最新一条模型消息里的 tool_calls 决定下一步；
  • tool_node -> llm_call 这条边让工具结果重新回到模型，形成 "模型判断 -> 工具执行 -> 模型继续判断" 的闭环；

如果工具执行过程中出错，官方 Thinking in LangGraph 给出的思路也很明确；不同类型的错误要分别处理：

  • transient errors 交给系统自动重试；
  • LLM-recoverable errors 写回 state，让模型看到错误后自行调整；
  • user-fixable errors 用 interrupt() 暂停，等待用户补充信息；
  • unexpected errors 直接抛出，便于调试；

例如，当工具错误属于模型有机会自行修正的可恢复错误时，可以把错误信息写回状态，并跳转回模型决策节点，让模型在下一轮基于该错误继续推理、调整参数或更换策略：

from typing import Literal, TypedDict
from langgraph.types import Command


class ToolError(Exception):
    """工具执行失败时抛出的自定义异常"""
    pass


class AgentState(TypedDict, total=False):
    # 模型决定要调用的工具描述
    tool_call: dict
    # 工具执行后的结果，成功或失败都会写回这里
    tool_result: str


def run_tool(tool_call: dict) -> str:
    """根据 tool_call 执行具体工具"""
    name = tool_call["name"]
    args = tool_call["args"]

    if name == "add":
        return str(args["a"] + args["b"])

    if name == "divide":
        if args["b"] == 0:
            raise ToolError("division by zero")
        return str(args["a"] / args["b"])

    raise ToolError(f"unknown tool: {name}")


def execute_tool(state: AgentState) -> Command[Literal["agent"]]:
    # 从 state 里取出本轮要执行的工具调用
    tool_call = state["tool_call"]

    try:
        # 真正执行工具
        result = run_tool(tool_call)

        # 返回 Command：LangGraph 的控制对象，用来同时表达“更新状态”和“下一步跳转”
        # update：要写回 state 的内容
        # goto：当前节点执行完后要跳到的目标节点
        return Command(
            update={"tool_result": result},
            goto="agent",
        )

    except ToolError as e:
        # 如果工具出错，也不要直接中断流程；
        # 把错误写回 state，让下一轮模型看到后自己调整策略
        return Command(
            update={"tool_result": f"Tool error: {e}"},
            goto="agent",
        )

这类处理方式的重点，是让 agent 的错误处理继续留在图的控制流里，而不是把所有异常都当成简单的函数报错

记忆机制：

消息记忆是多轮对话和长流程执行的基础；在 LangGraph 中，memory（记忆）不是某个单独的聊天缓存类，而是建立在 persistence（持久化）和 store（存储）之上的两层能力

  • 短期记忆依赖 checkpointer（检查点保存器），把 graph state（图状态）以 checkpoint（检查点/状态快照）的形式保存到 thread（会话线程）中；
  • 长期记忆依赖 store，把用户级或应用级的数据跨对话保存下来；

官方 persistence 文档明确指出，只要 graph 编译时挂上 checkpointer，LangGraph 就会在每个 super-step（超步，一轮状态推进）边界保存一个 checkpoint；这些 checkpoint 会按 thread 组织；因此，只要后续继续使用同一个 thread_id（线程 ID），图就能在同一条执行线程中继续累积状态

短期记忆（Checkpointer）：

下面的示例使用 InMemorySaver（内存保存器）为图增加短期记忆；第一次调用时告诉模型 "我叫 Zhu"，第二次继续使用同一个 thread_id 提问 "我叫什么名字"，图就会复用同一线程里的历史消息：

from langchain_community.chat_models.tongyi import ChatTongyi
from langgraph.graph import StateGraph, MessagesState, START
from langgraph.checkpoint.memory import InMemorySaver


model = ChatTongyi(model="qwen3-max", temperature=0)

def call_model(state: MessagesState):
    response = model.invoke(state["messages"])
    return {"messages": [response]}

# 创建一个内存中的 checkpointer，用于保存和加载图的状态
checkpointer = InMemorySaver() 

builder = StateGraph(MessagesState)
builder.add_node("call_model", call_model)
builder.add_edge(START, "call_model")

# 挂上 checkpointer 后，图的状态会按 thread 保存为 checkpoints 
graph = builder.compile(checkpointer=checkpointer)

config = {
    "configurable": {
        "thread_id": "1",
    }
}

# 第一次对话；把用户信息写进同一个 thread 的状态里
graph.invoke(
    {
        "messages": [
            {"role": "user", "content": "Hi! Remember: my name is Zhu"}
        ]
    },
    config,
)

# 第二次对话；继续使用相同 thread_id，图会读取之前保存的状态
result = graph.invoke(
    {
        "messages": [
            {"role": "user", "content": "What is my name"}
        ]
    },
    config,
)

print(result["messages"][-1].content) # My name is Zhu.

这个例子真正要观察的不是第二次回答的具体措辞，而是它为什么能够回答出来；因为前后两次调用使用的是同一个 thread_id，所以第二次执行时，graph 会先从 checkpointer 中恢复该线程已保存的状态，再在这个基础上继续追加新的用户消息和模型回复

这里有两个关键点：

  • thread_id 是短期记忆的索引；不用同一个 thread_id，就不会命中同一条历史线程；
  • 保存下来的不是 "最后一句文本"，而是整个 graph state 在每个 super-step 边界的快照；（LangGraph 持久化的不是某条孤立回复，而是每一轮状态推进结束时的整个 graph state 快照；因此恢复时恢复的是完整上下文，而不只是聊天内容）

当对话很长时，短期记忆还需要管理；官方 add-memory 文档给出了两条最常用的路径：trim 和 summarize

第一种是 trim；也就是在每次真正调用 LLM 之前，先把过长的消息历史裁剪到允许的 token 范围内：

from langchain_core.messages.utils import (
    trim_messages, # “裁剪消息列表”的工具函数
    count_tokens_approximately, # “近似估算 token 数”的工具函数
)


def call_model(state: MessagesState):
    # 先按 token 预算裁剪历史消息，优先保留最近的上下文
    # 这里只裁剪本轮发给模型的输入，不会改掉 state 里的完整历史
    messages = trim_messages(
        state["messages"],
        strategy="last",  # 超长时优先保留最后面的消息
        token_counter=count_tokens_approximately,  # 近似估算 token 数
        max_tokens=128,  # 限制传给模型的上下文长度
        start_on="human",  # 裁剪后尽量从用户消息开始
        end_on=("human", "tool"),  # 裁剪后尽量在 human 或 tool 消息处结束
    )

    # 用裁剪后的消息调用模型
    response = model.invoke(messages)

    # 把本轮模型回复追加回状态
    return {"messages": [response]}

第二种是 summarize；也就是把较早的消息总结成摘要，再把摘要和最近的消息一起交给模型；官方示例使用 SummarizationNode 完成这件事：

from typing import TypedDict

from langchain.messages import AnyMessage
from langchain_core.messages.utils import count_tokens_approximately
from langchain_community.chat_models.tongyi import ChatTongyi
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import StateGraph, START, MessagesState
from langmem.short_term import SummarizationNode, RunningSummary

# 主模型
model = ChatTongyi(model="qwen3-max", temperature=0)
# 专门用于生成摘要的模型配置，限制摘要输出长度
summarization_model = model.bind(max_tokens=128)


class State(MessagesState):
    # 摘要节点的内部上下文：用来保存滚动摘要信息，记住“已经总结到哪里了”
    context: dict[str, RunningSummary]


class LLMInputState(TypedDict):
    # summarize 节点产出的压缩版消息，主模型实际读取的是它
    summarized_messages: list[AnyMessage]
    # 滚动摘要状态，也会随着 thread 一起保存
    context: dict[str, RunningSummary]


# 摘要节点：
# 一般先读取完整历史 messages，如果历史还不长，summarized_messages 基本就是原始 messages；
# 如果历史过长，就把较早的内容压成摘要，再和最近几条原话一起输出
summarization_node = SummarizationNode(
    token_counter=count_tokens_approximately, # 用于估算消息长度
    model=summarization_model,      # 用于生成摘要的模型
    max_tokens=256,                 # 压缩后最终给主模型的上下文预算
    max_tokens_before_summary=256,  # 超过这个长度时开始触发摘要
    max_summary_tokens=128,         # 摘要文本本身的长度上限
)


def call_model(state: LLMInputState):
    # 主模型不直接用完整历史 messages，而是用 summarize 节点处理后的 summarized_messages
    response = model.invoke(state["summarized_messages"])
    return {"messages": [response]}


checkpointer = InMemorySaver()

builder = StateGraph(State)
builder.add_node("summarize", summarization_node)
builder.add_node("call_model", call_model)

# 先做摘要，再把压缩后的上下文交给主模型
builder.add_edge(START, "summarize")
builder.add_edge("summarize", "call_model")

# 配上 checkpointer 后，同一个 thread_id 下的消息历史和滚动摘要都能延续
graph = builder.compile(checkpointer=checkpointer)

运行这段代码后，真正发生的变化是：graph 在进入 call_model 之前，会先经过 summarize 节点；如果消息历史已经超过设定的 token 阈值，较早的对话会先被压缩成摘要，再把摘要和最近的消息一起交给模型；因此这里的 "结果" 重点不是某条固定输出，而是模型所看到的输入上下文已经变成了 "摘要 + 最近消息" 的形式

从使用角度看，trim 更直接；summarize 更适合需要尽量保留长期上下文线索的对话；两种方法都属于对短期记忆的管理，而不是长期记忆本身

长期记忆（Store）：

长期记忆用于保存跨对话、跨线程仍然需要保留的数据；例如用户偏好、用户资料、应用级配置等；这类数据不属于单条 thread 内的会话历史，而属于更长期的 store

官方 add-memory 文档推荐的做法，是在 graph 编译时传入 store，然后通过 Runtime 在 node 内访问 store；下面的示例展示了一个最小的长期记忆读写流程：

Runtime 是 LangGraph 在节点执行时自动注入的运行时对象，用来提供当前执行所需的上下文和能力；它不属于图的业务状态 state，而是节点访问外部上下文、长期记忆存储以及执行信息的统一入口；

当图通过 context_schema 定义了运行时上下文后，节点可以通过 runtime.context 读取这些外部信息

在长期记忆场景中，Runtime 最重要的作用是暴露 runtime.store；节点可以通过它访问编译图时挂上的 store，从而读取和写入跨对话、跨线程保存的长期数据；也就是说，store 是长期记忆的存储层，而 runtime 是节点访问这个存储层的入口

from dataclasses import dataclass
import uuid

from langchain_community.chat_models.tongyi import ChatTongyi
from langgraph.graph import StateGraph, MessagesState, START
from langgraph.runtime import Runtime
from langgraph.store.memory import InMemoryStore


# 主模型：真正负责回答用户问题
model = ChatTongyi(model="qwen3-max", temperature=0)

@dataclass # 定义简单的数据容器
class Context:
    # 运行时上下文：不属于 thread 内消息历史，而是每次执行时额外传入的信息
    # 这里用它标识“当前是哪个用户”
    user_id: str


def call_model(
    state: MessagesState,
    runtime: Runtime[Context],
    # Runtime 提供运行时能力：
    # runtime.context 读取当前执行上下文
    # runtime.store 访问跨对话保存的长期记忆
):
    user_id = runtime.context.user_id

    # namespace 用来区分长期记忆的作用域
    # 这里按 user_id 组织记忆，所以不同 thread 也能共享同一用户的长期数据
    namespace = ("memories", user_id) # 按“数据类型 -> 用户”组织

    # 用当前用户最后一句话去检索相关长期记忆
    memories = runtime.store.search(
        namespace, # 指定检索哪个命名空间的数据
        query=str(state["messages"][-1].content), # 检索关键词
    )

    # 把检索到的长期记忆拼进 system prompt，让模型在当前对话中“想起”这些信息
    info = "\n".join([d.value["data"] for d in memories])
    system_msg = f"You are a helpful assistant talking to the user. User info: {info}"

    # 如果用户要求系统“记住”某件事，就把它写入长期 store
    last_message = state["messages"][-1]
    if "remember" in last_message.content.lower():
        # 这里是最小示例，写入内容是演示用的固定值
        memory = "User name is Zhu"
        # 把一条新的长期记忆写入当前用户对应的存储空间，并为它生成一个唯一 id
        runtime.store.put(namespace, str(uuid.uuid4()), {"data": memory})

    # 当前这条 thread 的历史消息仍然来自 state["messages"]
    # 长期记忆通过 system message 注入，和 thread 内短期历史一起交给模型
    response = model.invoke(
        [{"role": "system", "content": system_msg}] + state["messages"]
    )
    return {"messages": [response]}


# 长期记忆的存储层；这里用内存版 store 作为最小示例
store = InMemoryStore()

# graph 的主状态仍然是 MessagesState；
# 额外声明 context_schema 后，这个图允许并期望在 invoke(...) 时传入一个 Context 类型的 context
builder = StateGraph(MessagesState, context_schema=Context)
builder.add_node(call_model)
builder.add_edge(START, "call_model")

# 编译时挂上 store，节点运行时才能通过 runtime.store 访问长期记忆
graph = builder.compile(store=store)


# 第一次对话：
# thread_id="1" 表示这是第一条短期会话线程；
# context 里传入 user_id，用来定位这个用户的长期记忆作用域
graph.invoke(
    {"messages": [{"role": "user", "content": "Hi! Remember: my name is Bob"}]},
    {"configurable": {"thread_id": "1"}},
    context=Context(user_id="user_129"),
)

# 第二次对话：
# 换了 thread_id，说明这是另一条短期会话；
# 但 user_id 没变，所以仍然可以命中同一个用户的长期记忆
result = graph.invoke(
    {"messages": [{"role": "user", "content": "What is my name"}]},
    {"configurable": {"thread_id": "2"}},
    context=Context(user_id="user_129"),
)


# 读取当前 thread 中最后一条模型回复
print(result["messages"][-1].content)

这个例子里真正值得关注的结果，是 "同一个用户在不同 thread 中仍然能被识别"；因为长期记忆不是跟着 thread_id 走，而是存放在 store 中，并通过 namespace 和 user_id 检索；所以即便换了新的执行线程，只要还是同一个用户上下文，graph 依然可以把相关长期记忆重新取出来

这个示例体现了短期记忆和长期记忆最关键的区别；

  • 短期记忆依赖 thread_id，强调 "同一执行线程" 的连续性；
  • 长期记忆依赖 namespace 和 store，强调 "跨线程仍然可访问" 的持久数据；

如果只需要本地演示，可以用 InMemoryStore；如果是生产环境，官方文档建议使用数据库后端的 store

Time Travel 时间回溯：

Time Travel 依赖 checkpoint 历史，LangGraph 会按 super-step 保存状态快照，因此可以通过 get_state_history() 找到图运行过程中的每个检查点；update_state() 会基于某个 checkpoint 创建一个新的分支，而不是修改原始 checkpoint；随后再用新的 config 重新 invoke，就能从选定位置重演后续步骤

下面的示例先推荐一位作家，再根据作家风格写一个短笑话；然后选择中间的 checkpoint，把作者改成另一位，再重放后续步骤：

from typing import TypedDict
from typing_extensions import NotRequired
from uuid import uuid4

from langchain_community.chat_models.tongyi import ChatTongyi
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, StateGraph

model = ChatTongyi(
    model="qwen3-max",
    temperature=0,
)

class State(TypedDict):
    # author 和 joke 都不是初始输入
    # 它们会在不同节点中逐步补齐
    author: NotRequired[str] # 这个 key 可以不存在，但如果存在，它的值必须是 str
    joke: NotRequired[str]

def author_node(state: State):
    # 第一步只负责推荐作家
    prompt = "请推荐一位受欢迎的作家；只返回作家名字"
    author = model.invoke(prompt).content
    return {"author": author}


def joke_node(state: State):
    # 第二步读取上一节点的 author，再根据这个作者的风格生成短笑话
    prompt = f"用作家 {state['author']} 的风格，写一个 100 字以内的笑话"
    joke = model.invoke(prompt).content
    return {"joke": joke}


builder = StateGraph(State)
builder.add_node("author_node", author_node)
builder.add_node("joke_node", joke_node)
builder.add_edge(START, "author_node")
builder.add_edge("author_node", "joke_node")
builder.add_edge("joke_node", END)

checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)

# 每次运行建议使用新的 thread_id，避免和旧运行历史混在一起
config = {"configurable": {"thread_id": str(uuid4())}}

# 先正常执行一次
graph.invoke({}, config)

# 获取当前线程的所有 checkpoint，按时间倒序排列，最新的在最前面
states = list(graph.get_state_history(config))

# 选中 author_node 执行完成后的快照
selected_state = states[1]

# 基于这个快照创建一个新分支，把作者改成别的名字，再重放后续步骤
new_config = graph.update_state(
    selected_state.config, # 包含 thread_id、checkpoint_id 等配置信息
    values={"author": "孙笑川"},
)

# 继续从新分支往后执行
result = graph.invoke(None, new_config)
print(result["joke"])

get_state_history() 返回的 snapshot 是按时间倒序排列的，最新的在最前面；因此 states[1] 代表 author_node 执行完后的那一刻；把 author 改成孙笑川以后，后面的 joke_node 会重新根据这个作者生成内容；这就是时间回溯最常见的用法

人类监督（Human-in-the-loop）：

Human-in-the-loop 是 LangGraph 的核心能力之一；当 Agent 即将执行关键动作，或者在某一步骤上必须等待外部输入时，图可以通过 interrupt() 暂停，把等待的信息暴露给调用方；当人类给出新的输入后，再通过 Command(resume=...) 从原来的执行位置继续

官方 interrupts 文档强调了三个基本前提：

  • 图必须配置 checkpointer，才能保存中断时的状态；
  • 调用时必须传入 thread_id，这样恢复时才能找到正确的状态；
  • interrupt() 的参数和 Command(resume=...) 的值都必须是 JSON-serializable；

下面的示例实现了一个 "预订宾馆前先审批" 的最小 Human-in-the-loop 流程：

from typing import Literal
from typing_extensions import TypedDict

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import StateGraph, START, END
from langgraph.types import interrupt, Command


class BookingState(TypedDict):
    # 当前要预订的酒店名称
    hotel_name: str
    # 人工审批结果；开始时为空，审批后会变成 True / False
    approved: bool | None
    # 最终执行结果
    result: str | None

# Human-in-the-loop 审核节点：
def human_review(state: BookingState) -> Command[Literal["book_hotel", END]]:
    
    # 在真正执行预订前先暂停，等待外部人工决定是否继续
    # interrupt(...) 里传入的字典会作为“中断信息”返回给调用方
    approved = interrupt(
        {
            "question": "是否批准执行预订动作",
            "hotel_name": state["hotel_name"],
        }
    )

    # 恢复执行时，Command(resume=...) 中的值会成为 interrupt(...) 的返回值
    if approved:
        return Command(
            # 审批通过：更新状态，并继续进入真正执行预订的节点
            update={"approved": True},
            goto="book_hotel",
        )

    # 审批拒绝：更新状态，并直接结束图
    return Command(
        update={"approved": False},
        goto=END,
    )

# 只有 human_review 批准后，才会进入这个节点
def book_hotel(state: BookingState):
    return {
        "result": f"成功预订 {state['hotel_name']}",
    }


builder = StateGraph(BookingState)
builder.add_node("human_review", human_review)
builder.add_node("book_hotel", book_hotel)
builder.add_edge(START, "human_review")
builder.add_edge("book_hotel", END)

# 需要 checkpointer 才能在 interrupt() 处暂停，并在之后用同一个 thread 恢复
graph = builder.compile(checkpointer=InMemorySaver())


# thread_id 用来标识这次被暂停 / 恢复的是哪一条执行线程
config = {"configurable": {"thread_id": "thread-1"}}

# 第一次执行会在 interrupt() 处暂停
# version="v2" 时，返回结果可以通过 paused.interrupts 查看中断信息
paused = graph.invoke(
    {
        "hotel_name": "两年半宾馆",
        "approved": None,
        "result": None,
    },
    config=config,
    version="v2",
)

# 这里可以自行检查 paused.interrupts 中的中断信息
print(paused.interrupts)
# (Interrupt(value={'question': '是否批准执行预订动作', 'hotel_name': '两年半宾馆'}, id='e1d6d30e518a59a399c419c9b08b3ef2'),)

# 人工批准后，用相同的 thread_id 恢复执行
# 这里的 resume=True 会回填给 human_review 里的 approved 变量
resumed = graph.invoke(
    Command(resume=True),
    config=config,
    version="v2",
)

# 读取恢复执行后的最终结果
print(resumed.value["result"]) # 成功预订 两年半宾馆

这段代码的执行顺序非常重要：human_review 节点先调用 interrupt(...)；图在这一点暂停，并把 interrupt payload 返回给调用方；checkpointer 保存当前状态；外部系统或人工界面拿到 payload，做出批准或拒绝；再次调用 graph.invoke(Command(resume=...)，config=同一个thread_id)；interrupt() 在节点内部拿到 resume 值，节点继续向下执行

官方文档还特别提醒了一点：恢复执行时，包含 interrupt() 的那个 node 会从头重新运行；因此，如果某个节点既要做真实副作用，又要中断等待批准，就必须把中断放在副作用之前，或者像上面的示例一样，把 "审批" 和 "真正执行动作" 拆成两个 node；这样结构更清晰，也更安全

如果这个流程需要和前端实时交互，可以结合 streaming 一起使用；官方 interrupts 文档建议在 Human-in-the-loop 场景中同时监听 "messages" 和 "updates"，这样既能实时看到模型输出，也能及时检测到图是否已经进入 interrupt 状态

Multi-Agent：

多 Agent 构建：

当单个 Agent 已经具备聊天、工具调用和记忆能力后，进一步的需求通常不是继续把所有逻辑都塞进同一个 Agent 或同一个 graph，而是把不同职责拆成多个专门组件，再由一个上层流程负责调度

在 LangChain / LangGraph 体系中，Multi-Agent（多智能体）并不是一套完全独立的新机制；它更像是一种组织复杂应用的架构方式：LangChain 主要负责模型、工具、检索和 Agent 能力，而LangGraph 主要负责复杂流程、状态流转和节点编排；两者结合后，就可以构建出具备 supervisor（调度者）、router（路由器）、subagent（子 agent）、handoff（移交）等能力的多智能体工作流

使用 Multi-agent 通常是为了解决几类问题：第一是上下文管理，让不同 Agent 只看到自己需要的信息；第二是职责拆分，让不同能力可以独立开发和维护；第三是并行处理，让多个专门 Agent 分别完成子任务后再汇总结果

在 LangGraph 里，常见做法是把每个专长 Agent 封装成独立节点或 subgraph（子图），再由父 graph 负责分发任务和汇总结果；官方 use-subgraphs 文档明确指出，subgraph 就是 "作为另一个 graph 中节点使用的 graph"；它的典型用途之一，就是构建 multi-agent systems

对于入门学习，先掌握两种最小组织方式就足够了：

固定流程编排：

固定流程编排适合步骤明确、顺序固定的任务；例如先让 research agent 整理资料，再让 writer agent 根据资料生成答案。这个流程每次都会按相同顺序执行，只是输入内容不同

下面的示例采用官方文档中的 "call a subgraph inside a node" 模式，也就是父图和子图使用不同的 state schema（状态结构），由父图节点负责把自己的状态转换成子图输入，再把子图结果映射回父图状态：

from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END


class ResearchState(TypedDict):
    # research 子图需要的问题输入
    question: str
    # research 子图产出的资料笔记
    notes: str

def collect_notes(state: ResearchState):
    # 第一个子图模拟 research agent；它只负责围绕问题整理资料
    return {
        "notes": f"围绕“{state['question']}”整理出的关键信息"
    }

# 第一张子图：负责“研究 / 整理资料”
research_builder = StateGraph(ResearchState)
research_builder.add_node("collect_notes", collect_notes)
research_builder.add_edge(START, "collect_notes")
research_builder.add_edge("collect_notes", END)
research_graph = research_builder.compile()


class WriterState(TypedDict):
    # writer 子图消费 research 阶段产出的 notes
    notes: str
    # writer 子图最终产出的答案
    answer: str

def write_answer(state: WriterState):
    # 第二个子图模拟 writer agent；输入是上一个子图产出的 notes
    return {
        "answer": f"根据资料整理出的最终回答：{state['notes']}"
    }

# 第二张子图：负责“根据资料写答案”
writer_builder = StateGraph(WriterState)
writer_builder.add_node("write_answer", write_answer)
writer_builder.add_edge(START, "write_answer")
writer_builder.add_edge("write_answer", END)
writer_graph = writer_builder.compile()


class ParentState(TypedDict):
    # 父图统一维护整体流程需要的字段
    question: str
    notes: str
    answer: str

def run_research_agent(state: ParentState):
    # 父图节点里调用 research 子图；
    # 先把父图 state 映射成子图需要的输入结构
    research_result = research_graph.invoke(
        {
            "question": state["question"],
            "notes": "",
        }
    )

    # 子图执行结束后，只把父图真正需要的结果字段写回父图 state
    return {"notes": research_result["notes"]}


def run_writer_agent(state: ParentState):
    # 第二个父图节点里调用 writer 子图；
    # 它继续消费父图里已经更新好的 notes
    writer_result = writer_graph.invoke(
        {
            "notes": state["notes"],
            "answer": "",
        }
    )
    return {"answer": writer_result["answer"]}


# 父图负责调度两个子图：先 research，再 writer
builder = StateGraph(ParentState)
builder.add_node("research_agent", run_research_agent)
builder.add_node("writer_agent", run_writer_agent)
builder.add_edge(START, "research_agent")
builder.add_edge("research_agent", "writer_agent")
builder.add_edge("writer_agent", END)

graph = builder.compile()


# 执行父图后，最终结果会同时包含 question / notes / answer
result = graph.invoke(
    {
        "question": "请概括 LangGraph 中 subgraph 的作用",
        "notes": "",
        "answer": "",
    }
)


print(result["answer"]) # 根据资料整理出的最终回答：围绕“请概括 LangGraph 中 subgraph 的作用”整理出的关键信息

这个示例里真正值得观察的，不是某一句固定输出，而是状态如何在父图和子图之间流转；父图先执行 research_agent 节点，把 question 交给 research 子图并取回 notes；然后再执行 writer_agent 节点，把 notes 交给 writer 子图并取回 answer；最终，父图只保留自己关心的三个字段，而两个子图可以各自维护独立的输入输出结构

这正是多 Agent 组织在 LangGraph 中最关键的地方；父图负责 orchestration（编排），子图负责各自职责；如果多个 Agent 共享同一批 state keys（状态字段），例如都围绕同一个 messages 通道工作，那么也可以像官方 use-subgraphs 文档展示的那样，直接把编译后的 subgraph 传给 add_node；如果父图和子图之间需要做字段转换，或者每个 Agent 需要自己的私有状态，那么像上面这样在 node 内部调用 subgraph 会更合适

Supervisor 与 Router 编排：

和固定流程不同，router 模式并不要求每次都按同样顺序执行所有 Agent；它会先根据用户输入判断任务类型，再只调用最合适的那个 Agent；例如先判断用户问题属于普通问答、写笑话还是路线规划，再把任务交给对应的专职 Agent，适合任务类型不同、需要动态分发的场景

下面示例加入一个简单的 supervisor，让它先判断应该调用哪个子 Agent；为了突出流程，示例中先用规则模拟分类，真实项目里可以把 supervisor 替换成 LLM 调用：

from typing import Literal
from typing_extensions import NotRequired, TypedDict
from langgraph.graph import StateGraph, START, END


class GeneralState(TypedDict):
    # general 子图接收用户问题
    question: str
    # general 子图产出普通回答
    answer: str

def general_reply(state: GeneralState):
    return {
        "answer": f"通用助手回答：{state['question']}"
    }

# 第一张子图：负责普通问答
general_builder = StateGraph(GeneralState)
general_builder.add_node("general_reply", general_reply)
general_builder.add_edge(START, "general_reply")
general_builder.add_edge("general_reply", END)
general_graph = general_builder.compile()


class JokeState(TypedDict):
    # joke 子图接收用户问题
    question: str
    # joke 子图产出笑话回答
    answer: str

def joke_reply(state: JokeState):
    return {
        "answer": f"笑话助手回答：围绕“{state['question']}”讲一个简短笑话"
    }

# 第二张子图：负责生成笑话
joke_builder = StateGraph(JokeState)
joke_builder.add_node("joke_reply", joke_reply)
joke_builder.add_edge(START, "joke_reply")
joke_builder.add_edge("joke_reply", END)
joke_graph = joke_builder.compile()


class ParentState(TypedDict):
    # 父图维护整体流程需要的字段
    question: str
    answer: str
    # route 保存 supervisor 的路由结果
    route: NotRequired[str] # 这个 key 可以不存在，但如果存在，它的值必须是 str

def supervisor(state: ParentState):
    # 入门示例中先用规则模拟 supervisor，真实项目里可以替换成 LLM 分类
    if "笑话" in state["question"] or "段子" in state["question"]:
        return {"route": "joke"}
    return {"route": "general"}

def route_next(state: ParentState) -> Literal["general", "joke"]:
    # 条件路由只关心 supervisor 写入的 route 字段
    route = state.get("route", "general")
    if route in {"general", "joke"}:
        return route
    # 如果路由结果不合法，就回退到 general
    return "general"

def run_general_agent(state: ParentState):
    # 父图节点里调用 general 子图，先把父图 state 映射成子图需要的输入结构
    result = general_graph.invoke(
        {
            "question": state["question"],
            "answer": "",
        }
    )
    # 子图执行结束后，只把父图关心的字段写回父图 state
    return {"answer": result["answer"]}

def run_joke_agent(state: ParentState):
    # 父图节点里调用 joke 子图，joke 子图可以拥有自己的状态结构和内部流程
    result = joke_graph.invoke(
        {
            "question": state["question"],
            "answer": "",
        }
    )
    return {"answer": result["answer"]}


# 父图负责编排：先 supervisor 判断路由，再调用对应子 Agent
builder = StateGraph(ParentState)

builder.add_node("supervisor", supervisor)
builder.add_node("general_agent", run_general_agent)
builder.add_node("joke_agent", run_joke_agent)

builder.add_edge(START, "supervisor")
builder.add_conditional_edges(
    "supervisor", # 从 supervisor 节点执行之后开始条件路由
    route_next, # 条件路由函数，根据返回的路由结果选择不同的子图
    {
        "general": "general_agent",
        "joke": "joke_agent",
    }, # 路由映射表
)
builder.add_edge("general_agent", END)
builder.add_edge("joke_agent", END)

graph = builder.compile()

result = graph.invoke(
    {
        "question": "给我讲一个关于 Python 的段子",
        "answer": "",
    }
)

print(result["answer"]) # 笑话助手回答：围绕“给我讲一个关于 Python 的段子”讲一个简短笑话

这个例子中，父图先执行 supervisor，把用户问题分类成 general 或 joke；然后 add_conditional_edges 根据 route_next 的返回值，决定进入哪个专职 Agent；真正生成回答的不是 supervisor，而是被路由选中的子 Agent

因此，supervisor / router 模式的重点不是 "多执行几个节点"，而是 "根据当前输入选择合适的处理路径"；如果后续要新增 couplet_agent、route_agent 或 rag_agent，主流程只需要增加新的节点和路由映射，对已有 Agent 的内部实现影响很小

在持久化方面，官方文档也明确建议，大多数 multi-agent 场景优先采用 per-invocation 的 subgraph persistence；也就是每次调用子图时从新的状态开始，但在单次调用内部仍然继承父图的 checkpointer 能力；这更适合把子 Agent 当作一次性专长处理器来使用

# 人工智能 # python
Logo
亚马逊云科技技术品牌专区

更多推荐

cover

基于STM32的全屋环境智能监测与预警终端设计

avatar 亚马逊云科技技术品牌专区

Keras：面向人类的深度学习库

Keras3支持多后端（TensorFlow、PyTorch、JAX），通过设置环境变量切换，兼容指定版本。pip安装后可用Sequential API快速构建CNN模型，如MNIST分类示例，实

avatar 亚马逊云科技技术品牌专区

AI行业日报｜2026-09-25：5个热点事件

本期行业动态聚焦于端侧算力重构、自主智能体工作流、开发工具经济模型、网络隔离架构设计与公共部门合规审查。以下结合一线技术场景进行结构化梳理，供 AI 研究员与软件工程团队参考。 PrismML 正将其轻量级语言模型适配至搭载高通处理器的智能眼镜终端 1 。该平台的核心战略指向开源权重模型的本地化运行，旨在最大化挖掘穿戴设备现有算力的利用效率 1 。对于嵌入式与边缘计算开发者而言，此类微型模型向低功

avatar 亚马逊云科技技术品牌专区