Jean's Blog

一个专注软件测试开发技术的个人博客

0%

LangGraph介绍

发表于 更新于 分类于
学习LangGraph学习笔记第一讲

什么是智能体

智能体（AI Agents或Agents）是指能够自主感知环境、做出决策并执行行动的系统或程序。根据IBM的定义，智能体是“能够通过设计其工作流和利用可用工具，代表用户或其他系统自主执行任务的系统或程序”[IBM]。英伟达则将智能体描述为“AI 智能体是先进的 AI 系统，旨在根据高级目标自主进行推理，制定计划并执行复杂任务。”，代表了“从简单自动化向能够管理复杂工作流的自主系统过渡”的演进方向[NVIDIA]。

在更专业的学术定义中，百度百科指出：“智能体是指能够感知环境并自主采取行动以实现特定目标的实体。这一概念最早由“人工智能之父”马文·明斯基提出，他认为某些问题可经由社会中的一些个体经过协商后解决，这些个体就是智能体。”[百度百科]

智能体具有以下基本特征：

  • 自主性(Autonomy)：智能体能够在没有人类或其他实体的直接干预下运行，并对其行动和内部状态具有某种程度的控制。
  • 反应性(Reactivity)：智能体能够感知其环境，并对环境变化做出实时响应。
  • 交互性/社交性(Socialability)：智能体能够与其他智能体或人类进行交互和协作。
  • 适应性/主动性(Proactivity)：智能体能够根据环境变化主动调整其行为策略，适应新的情况。
  • 学习能力：许多智能体具有通过经验或数据学习和改进的能力。

在 LangChain 的语境下，智能体是一个利用 LLM 作为推理引擎的系统。它不仅能回答问题，还能根据目标自主决定调用哪些工具（如搜索、运行代码、查询数据库）来完成任务。

智能体的核心公式：

Agent = LLM（大脑） + Planning（规划） + Memory（记忆） + Tool Use（工具使用）

  • 规划： 将复杂任务拆解为小步骤。
  • 记忆： 记住之前的对话上下文或中间操作结果。
  • 工具： 能够访问外部 API、计算器、搜索引擎等。

单智能体 (Single-Agent)

概念： 只有一个核心控制循环。所有的推理、规划和执行都由一个 LLM 驱动。

在 LangChain 中的实现

LangChain 最早的火爆就是因为提供了各种 Agent 类型。最经典的模式是 ReAct (Reason + Act)：

  1. Thought（思考）： LLM 决定下一步做什么。
  2. Action（行动）： LLM 选择一个工具并提供参数。
  3. Observation（观察）： 运行工具并返回结果给 LLM。
  4. 循环： 重复上述步骤直到任务完成。

优缺点：

  • 优点： 结构简单，易于调试，适合处理逻辑线性、目标明确的任务。
  • 缺点： 随着任务复杂度增加，LLM 容易陷入循环、丢失目标（上下文过长）或在复杂决策中“断片”。

多智能体系统 (Multi-Agent System, MAS)

概念： 多个智能体协同工作。每个智能体通常扮演不同的角色（Role），就像一个公司里的不同部门（如：程序员、测试员、产品经理）。

优缺点：

  • 优点： 专业化： 每个智能体只需要精通一件事（提示词更短、准确度更高）。
    • 鲁棒性： 可以引入“审核员”角色，大幅降低幻觉。
  • 缺点： 架构复杂，Token 消耗量大，存在通信开销。

image-20250917100224849

常见的多智能体架构

多智能体架构中的网状结构、监管者模式、分级架构和自定义模式，并通过类比人类工作方式说明多智能体如何协作完成复杂任务。

image-20250917100621793

  • 网状结构：任何一个智能体都可以进行决策
  • 监督者结构：由主管来决策下一步操作
  • 监管者架构（工具）：智能体作为工具，接受一个LLM主管的调用
  • 分级架构：多级架构每级都有一个监管者
  • 自定义：只有部分智能体具备决策权

LangGraph介绍

官方地址：https://docs.langchain.com/oss/python/langgraph/overview

LangGraph 是一个专门用来构建「有状态、多角色 AI 应用」的框架。

它的核心设计思路是用「图结构」来管理 AI 的工作流，特别擅长处理需要循环、条件分支的复杂场景。

和传统线性的 AI 工作流不同，LangGraph 可以让你的 AI 系统拥有动态决策能力，而不是一条路走到黑。

核心优势

优势 解读
循环执行 支持「思考 - 行动 - 观察」的循环模式，这是 AI 智能体的核心能力。比如 AI 可以先思考要做什么 → 调用工具执行 → 观察结果 → 再重新思考，直到完成目标。
状态持久化 会自动维护对话历史和上下文信息。你不用自己写复杂的代码来保存对话状态，它会帮你管理，AI 能记住之前的交互内容。
可视化调试 可以和 LangSmith 无缝集成，提供强大的可视化和监控功能。你能直观看到 AI 的工作流每一步是怎么跑的，方便排查问题、优化流程。
类型安全 支持类似 TypeScript 的类型注解，能减少运行时错误。就像给你的 AI 应用加上了一层 “语法校验”，降低因数据类型错误导致的 bug 概率。

LangGraph 的框架价值

把复杂的 AI 工作流，转化成清晰的图结构，让 AI 应用真正具备动态决策能力。

简单来说，LangGraph 解决了传统线性 AI 流程的 “死板” 问题，让 AI 可以像人一样：

  • 遇到问题可以反复尝试、调整方案（循环）
  • 能根据不同情况走不同的处理路径（分支）
  • 全程能被追踪、调试，方便开发维护

环境安装

1
pip install -U langgraph

LangGraph核心概念详解

图 (Graph) 的基本组成

这是 LangGraph 最基础的结构定义，也是整个框架的 “骨架”。

核心概念

LangGraph 的图结构由三个核心元素构成：

节点（Nodes） + 边（Edges） + 状态（State）

代码示例解读

1
2
3
4
5
# 图的骨架：节点 + 边 + 状态
from langgraph.graph import StateGraph

# 创建图容器
workflow_builder = StateGraph(MessagesState)
  • 这里用 StateGraph 创建了一个图的 “容器”，并且绑定了 MessagesState（也就是后面会讲的状态）。
  • 后续你所有的节点、边，都是往这个 workflow_builder 里添加。

通俗比喻

就像一条披萨制作流水线：

  • 整个图 = 从准备面团 → 加配料 → 烘烤 → 出货的完整流程
  • 它定义了所有步骤（节点）之间如何连接、按什么顺序执行（边），以及全程跟着流转的订单信息（状态）。

状态 (State)：节点间的流通载体

State 是 LangGraph 的灵魂，也是整个工作流的 “共享内存”。

核心概念

State 是在整个图执行过程中持久化的数据，它会在节点之间流动，并且每个节点都可以读取、修改它。

你可以把它理解为：所有节点都能访问和修改的 “全局变量”，但它的结构是强定义、可校验的。

代码示例解读

1
2
3
4
5
from typing import TypedDict, List

class MessagesState(TypedDict):
    """消息状态：记录对话历史"""
    messages: List[dict]
  • 这里用 TypedDict 定义了一个状态结构 MessagesState，里面有一个 messages 列表，用来存对话历史。
  • 后续所有节点，都可以读取 state["messages"]，也可以往里面追加新消息，这些修改会自动被 LangGraph 保存下来。

通俗比喻

还是披萨流水线：

  • State 就像一张跟着披萨走的订单表，上面写着顾客的需求（比如加不加辣、要不要双倍芝士）。
  • 每个工序（节点）都会看这张订单表，按要求处理披萨，甚至修改订单信息（比如中途顾客加了配料），下一个工序拿到的就是更新后的订单。

节点 (Nodes) 与边 (Edges)

这是 LangGraph 控制流程的核心，决定了工作流 “做什么” 和 “怎么走”。

1. 节点（Nodes）

  • 节点是基本处理单元，就是一个个具体要执行的任务。
  • 比如：调用大模型生成回复、调用工具查询数据、判断是否需要继续执行，这些都可以做成一个节点。

2. 边（Edges）

边定义了节点之间的连接关系，相当于控制流的方向盘，决定了工作流的执行路径。

它分为两种：

① 普通边：固定执行路径
1
builder.add_edge("node_a", "node_b")
  • 含义：执行完 node_a 之后，固定跳转到 node_b。
  • 就像流水线里 “面团准备好后，必须直接送到加配料的工序”，路径是固定的。
② 条件边：动态决策路径
1
2
3
4
def should_continue(state):
    if state["needs_tool"]:
        return "tool_node"
    return END
  • 含义：根据当前 state 的值，动态决定下一步走哪条路。
    • 如果 needs_tool 为 True，就去执行 tool_node（调用工具）
    • 否则就直接结束流程（END）
  • 这就是 LangGraph 能实现 “动态决策” 的关键！比如 AI 判断自己回答不上来，就自动去调用工具查资料，查完再回来继续回答。

总结一下三者的关系

元素 作用 披萨流水线比喻
State 共享数据，全程流转、可修改 订单表，记录所有需求和状态
Nodes 具体的处理步骤 准备面团、加配料、烘烤等工序
Edges 步骤之间的连接规则 工序之间的传送带，分固定和分支两种

LangGraph 就是靠这三者的配合，实现了传统线性工作流做不到的：循环、分支、动态决策。

LangGraph 状态更新机制

核心：Annotated[list, operator.add]

这是 LangGraph 实现增量更新的关键语法，也是它状态管理的核心优势。

它解决了什么问题？

默认情况下，如果你直接定义一个 list 类型的状态字段，节点返回新列表时，会直接覆盖掉原来的列表。

而 Annotated[list, operator.add] 会告诉 LangGraph：

新数据不是用来覆盖旧列表的，而是用 + 操作追加到旧列表的末尾。

举个例子

  • 原始状态：messages = [{"role": "user", "content": "你好"}]

  • 节点返回：{"messages": [{"role": "assistant", "content": "你好呀"}]}

  • 用 Annotated[list, operator.add] 后，最终状态：

    1
    messages = [{"role": "user", "content": "你好"}, {"role": "assistant", "content": "你好呀"}]

  • 如果不用这个语法，结果会变成：

    1
    messages = [{"role": "assistant", "content": "你好呀"}]

    （直接覆盖，历史对话丢了）

这就是为什么对话历史、聊天记录这类场景，必须用追加更新模式。

三种状态更新模式对比

图里的表格，把 LangGraph 的状态更新分成了三类，我给你整理成更清晰的对比表：

更新类型 语法 / 实现方式 适用场景 核心特点
追加更新 Annotated[list, operator.add] 消息历史、对话记录、日志列表 不覆盖旧数据，把新数据追加到列表末尾
覆盖更新 默认行为 计数器、状态标志、当前步骤 节点返回新值后，直接替换掉旧值
自定义合并 自定义 reducer 函数 复杂数据结构（如嵌套字典、自定义对象） 自己写逻辑决定新旧数据怎么合并

状态流转示例

pizza_workflow.py

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
from typing import Literal, TypedDict
from langgraph.graph import StateGraph, START, END

# 定义状态结构：披萨订单
class PizzaState(TypedDict):
    order_id: str       # 订单号
    size: str           # 披萨尺寸
    toppings: list      # 配料列表
    current_step: str   # 当前制作步骤

# 节点函数：准备面团
def prepare_dough(state: PizzaState) -> PizzaState:
    """准备面团节点"""
    print(f"准备 {state['size']} 尺寸的面团")
    # 更新当前步骤状态（覆盖更新模式）
    state["current_step"] = "dough_ready"
    return state

# 1. 构建流程图
builder = StateGraph(PizzaState)

# 2. 添加节点
builder.add_node("prepare_dough", prepare_dough)

# 3. 定义流转路径
builder.add_edge(START, "prepare_dough")
builder.add_edge("prepare_dough", END)

# 4. 编译图（可执行版本）
graph = builder.compile()

在这个例子中，PizzaState对象像一个生产订单，在各个节点间传递，每个节点都会查看订单信息并更新生产状态。以下是代码详解：

  1. 定义状态 PizzaState

    1
    2
    3
    4
    5
    class PizzaState(TypedDict):
        order_id: str
        size: str
        toppings: list
        current_step: str

    • 这是一个用 TypedDict 定义的状态结构，相当于披萨的生产订单。

    • 里面记录了订单号、尺寸、配料、当前步骤，这些信息会在节点间传递和更新。

  1. 定义节点 prepare_dough

    1
    2
    3
    4
    5
    def prepare_dough(state: PizzaState) -> PizzaState:
        """准备面团节点"""
        print(f"准备 {state['size']} 尺寸的面团")
        state["current_step"] = "dough_ready"
        return state

    • 这是一个 “准备面团” 的处理节点。

    • 它接收当前的 state（订单信息），读取 size 字段，然后把 current_step 更新为 dough_ready，最后返回更新后的 state。

    • 这里的 current_step 字段，用的就是覆盖更新模式：每次节点返回新值，就会替换掉旧值。

  1. 构建流程图

    1
    2
    3
    4
    5
    6
    7
    # 构建流程图
    builder = StateGraph(PizzaState)
    builder.add_node("prepare_dough", prepare_dough)

    # 定义流转路径
    builder.add_edge(START, "prepare_dough")
    builder.add_edge("prepare_dough", END)

    • 用 StateGraph(PizzaState) 创建了一个绑定了 PizzaState 的图容器。

    • 把 prepare_dough 函数注册成了一个节点。

    • 定义了路径：从 START 开始，执行 prepare_dough，然后直接到 END 结束。

  1. 示例的核心逻辑

    • PizzaState 就像一张跟着披萨走的订单，每个节点都会查看订单信息，并更新订单状态。

    • 这个例子里只有一个节点，执行完就结束，非常直观地展示了 “状态在节点间传递并被修改” 的过程。

LangGraph 的状态更新机制，核心就是这三点：

  1. 追加更新：解决了对话历史这类列表数据的 “增量保存” 问题，不会丢历史。
  2. 覆盖更新：适合简单的状态标志、计数器，用起来简单直接。
  3. 自定义合并：给复杂场景留了扩展空间，你可以自己控制数据合并逻辑。

而整个状态流转，就像披萨订单一样，在节点间传递、被修改，最终带着所有处理信息完成整个流程。

工具调用机制深度解析

工具定义与绑定：给 LLM 装 “外挂”

这部分讲的是：怎么定义一个工具，并让大模型知道它的存在。

核心概念

在 LangGraph/LangChain 中，工具是扩展 LLM 能力的关键。工具调用允许 LLM 把自己做不了的任务 “外包” 给外部函数或 API，比如数学计算、查天气、查数据库等。

代码示例解读（tool_definition.py）

1
2
3
4
5
6
7
8
9
10
11
12
from langchain.tools import tool

# 定义算术工具
@tool
def multiply(a: int, b: int) -> int:
    """Multiply two numbers."""
    return a * b

tools = [multiply]

# 绑定工具到模型
model_with_tools = model.bind_tools(tools)
  1. @tool 装饰器：把普通的 Python 函数标记成 LangChain 能识别的 “工具”，让它能被 LLM 调用。
  2. multiply 函数：一个简单的乘法工具，接收两个整数，返回乘积。
  3. tools = [multiply]：把定义好的工具放进列表，方便后续绑定。
  4. model.bind_tools(tools)：把工具列表绑定到 LLM 模型上，让模型知道 “我有这些工具可以用”，并学会生成调用工具的指令。

工具定义的 3 个关键要素

要素 作用 示例
清晰的 docstring LLM 是根据工具的描述，来判断什么时候、该不该调用这个工具。描述越清楚，模型调用越准。 """Multiply two numbers."""
类型提示 帮助 LLM 理解参数的类型和格式，避免生成错误的参数。 a: int, b: int
工具名称 必须简洁明确，能被模型识别，后续调用时会用到这个名称。 multiply

LLM 工具调用机制：从 “说” 到 “做”

这部分讲的是：LLM 怎么生成工具调用指令，以及系统怎么识别并执行它。

核心流程

  1. 用户提问（比如 Add 3 and 4，虽然示例里是加法，但绑定的是乘法工具，这里是演示格式）
  2. LLM 判断需要调用工具，生成 tool_calls 字段
  3. 下一个节点（比如 LangGraph 里的工具节点）检查 tool_calls 字段，调用对应的工具
  4. 工具执行完，把结果返回给 LLM，生成最终回答

代码示例解读（tool_invocation.py）

1
2
3
4
5
6
7
8
9
10
11
# 当用户输入"Add 3 and 4"时，LLM返回：
response = model_with_tools.invoke("Add 3 and 4")
print(response.tool_calls)

# 输出类似：
# [{
#   'name': 'add',
#   'args': {'a': 3, 'b': 4},
#   'id': 'call_123',
#   'type': 'tool_call'
# }]
  • 这里 model_with_tools.invoke() 调用了绑定好工具的模型，模型会根据用户问题，生成工具调用的指令。
  • 指令存在 response.tool_calls 里，是一个列表（支持一次调用多个工具）。

tool_calls 字段深度解析

字段 作用
name 要调用的工具名称，必须和你注册的工具名称完全匹配，不然找不到工具。
args 调用工具所需的参数字典，键是参数名，值是模型生成的参数值。
id 工具调用的唯一标识符，用来追踪这次调用的结果，后续可以把结果和调用对应起来。
type 固定为 tool_call，用来标识这是一个工具调用指令。

完整流程总结

  1. 定义工具：用 @tool 装饰器写函数，加上 docstring 和类型提示。
  2. 绑定工具：把工具列表绑定到 LLM 模型上。
  3. 模型生成调用指令：用户提问后，模型生成 tool_calls。
  4. 节点执行工具：LangGraph 里的节点读取 tool_calls，找到对应的工具并执行。
  5. 返回结果给模型：工具执行结果返回给 LLM，模型根据结果生成最终回答。

LangGraph1.1核心更新概览

版本核心定位：类型安全大升级

LangGraph 1.1 的核心目标，是解决之前版本在类型安全上的短板，让开发者构建 AI 智能体时，能获得更可靠的类型检查、更少的运行时错误，整体开发体验更稳健。

关键更新详解

更新点 通俗解释 实际开发收益
类型安全的流处理和调用 stream/invoke 这类核心方法，现在支持完整的类型检查，输入 / 输出类型会被约束和校验 减少因类型不匹配导致的运行时崩溃，IDE 能直接提示类型错误，不用等到运行才发现问题
修复父图和子图的重放行为 子图（subgraph）嵌套的场景下，历史执行记录 / 重放逻辑存在的异常被修复 智能体的状态回溯、调试和重试功能更稳定，嵌套复杂图的行为可预测性大幅提升
输出类型强制转换 流处理和调用的结果，会被自动 / 强制转换为定义好的类型，避免类型丢失或不一致 下游代码可以安全地使用返回结果，不用手动做类型断言或判断，减少冗余代码和潜在 bug

核心新特性：version="v2" 流格式

这是本次更新的 “王牌功能”：

  • 它是一种可选的新流处理格式，需要在调用 stream()/astream()/invoke()/ainvoke() 时通过 version="v2" 启用。
  • 启用后，这四个核心方法会获得端到端的类型安全保障，包括输入、流事件、最终输出的类型都能被静态检查和推断。
  • 旧版流格式会保持兼容，开发者可以按需迁移，不用一次性修改所有代码。

给开发者的实际影响

  1. 调试效率提升：IDE 可以直接在编码阶段提示类型错误，不用再靠日志排查类型问题。
  2. 嵌套图更稳定：复杂智能体（多子图嵌套）的执行、重放、重试行为更可靠，适合构建生产级应用。
  3. 迁移成本低：v2 是可选格式，你可以先在关键模块启用，逐步替换旧版代码。
  4. 类型推断更友好：输出结果会自动匹配你定义的状态 / 返回类型，减少 Any 类型和手动类型转换。

类型安全的流处理和调用

LangGraph 1.1 引入的 version="v2" 流格式，核心目的是给 stream()/astream()/invoke()/ainvoke() 这几个核心方法，加上完整的类型安全支持，同时不破坏旧版代码的兼容性。

旧版本（v1）

  • 默认启用，无需额外配置，和之前的行为保持一致。

  • stream() 的问题：产生的是 “裸元组”（比如 (stream_mode, data) 或者直接 data），IDE / 类型检查器没法推断出它的具体结构。

  • invoke() 的问题：直接返回普通 Python 字典，没有固定结构；中断（interrupt）会被混在字典里的 __interrupt__ 字段里，非常混乱。

代码示例：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
from langgraph.graph import StateGraph, START, END
from pydantic import BaseModel

class MyState(BaseModel):
    answer: str
    count: int

def hello_node(state):
    return {"answer": "Hello World", "count": state.count + 1}

graph = StateGraph(MyState)
graph.add_node("hello", hello_node)
graph.add_edge(START, "hello")
graph.add_edge("hello", END)
compiled = graph.compile()

# v1：返回普通字典
result = compiled.invoke({"answer": "", "count": 0})
print(type(result))  # <class 'dict'>

LangGraph 1.1（v2）

  • 可选启用，调用时加 version="v2" 参数即可。

  • stream() 的改进：产生的是强类型的 StreamPart 字典，包含 type/ns/data/interrupts 字段，结构清晰且可被类型检查。

  • invoke() 的改进：返回 GraphOutput 对象，通过 .value 访问状态数据，.interrupts 单独访问中断信息，结构非常清晰。

  • 自动类型转换：当你的状态是 Pydantic BaseModel 或 dataclass 时，输出会自动强制转换成正确的类型，不用自己手动解析。

代码示例：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
from langgraph.graph import StateGraph, START, END
from langgraph.types import GraphOutput  # v2 引入的类型
from pydantic import BaseModel

class MyState(BaseModel):
    answer: str
    count: int

def hello_node(state):
    return {"answer": "Hello World", "count": state.count + 1}

graph = StateGraph(MyState)
graph.add_node("hello", hello_node)
graph.add_edge(START, "hello")
graph.add_edge("hello", END)
compiled = graph.compile()

# v2：返回 GraphOutput 对象
result = compiled.invoke({"answer": "", "count": 0}, version="v2")
print(type(result))          # <class 'langgraph.types.GraphOutput'>
print(type(result.value))    # 会自动转为 MyState 类型，而不是 dict
print(result.interrupts)     # 中断信息单独在这里，不会混在结果里

类型安全流处理的核心优势

优势 通俗解释 开发体验提升
✅ 完整的类型安全 IDE / 类型检查器能准确推断类型，提前发现类型错误 编码阶段就能发现 bug，不用等运行时报错
✅ 自动类型转换 Pydantic/dataclass 状态的输出会自动转为对应类型 不用再写一堆手动类型断言 / 解析代码
✅ 结构化流数据 流数据是标准化的 StreamPart 字典，字段固定 处理流数据时不用猜结构，代码可读性大幅提升
✅ 清晰的中断处理 中断不再混在结果字典里，而是 GraphOutput 的独立属性 调试和处理中断逻辑时，代码更干净，不容易出错
✅ 向后兼容 默认还是 version="v1"，现有代码不用改就能跑 可以按需逐步迁移到 v2，不用一次性重构
  1. 新项目直接用 v2：在 invoke/stream 调用里加上 version="v2"，直接享受类型安全带来的便利。
  2. 旧项目按需迁移：核心模块先改成 v2，其他模块保持 v1，不影响整体运行。
  3. 结合 Pydantic 使用：用 Pydantic 定义状态模型，v2 会自动帮你做类型校验和转换，效果最好。

类型安全的流和调用与输出类型强制转换

LangGraph 1.1 里，version="v2" 带来的一个关键特性：当你用 Pydantic 模型定义状态时，框架会自动把返回结果转换成你定义的类型，而不是返回普通字典。

什么是「类型强制转换」？

你告诉 LangGraph：“我的状态是一个 Pydantic 模型”，然后 LangGraph 会确保返回的数据也是这个模型类型，而不是一个普通的字典。

版本 调用返回结果 你需要做什么 问题 / 优势
以前（v1） {"answer": "...", "count": 1}（普通 Python 字典） 手动写 MyState(**result)转换成模型 手动转换容易出错，IDE 无法自动补全，类型检查器也没法帮你
现在（v2） MyState(answer="...", count=1)（直接就是你的 Pydantic 模型实例） 直接用，不用转换 IDE 能自动补全字段，类型检查器能发现错误，代码更安全直观

代码示例

旧版（v1）写法：需要手动转换

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
from langgraph.graph import StateGraph, START, END
from pydantic import BaseModel

class MyState(BaseModel):
    answer: str
    count: int

def hello_node(state):
    return {"answer": "Hello", "count": state.count + 1}

graph = StateGraph(MyState)
graph.add_node("hello", hello_node)
graph.add_edge(START, "hello")
graph.add_edge("hello", END)
compiled = graph.compile()

# v1：返回普通字典
result_dict = compiled.invoke({"answer": "", "count": 0})
print(type(result_dict))  # <class 'dict'>

# 必须手动转换成 Pydantic 模型
result_state = MyState(**result_dict)
print(result_state.answer)  # 现在才能安全访问字段

新版（v2）写法：自动转换，一步到位

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
from langgraph.graph import StateGraph, START, END
from pydantic import BaseModel

class MyState(BaseModel):
    answer: str
    count: int

def hello_node(state):
    return {"answer": "Hello", "count": state.count + 1}

graph = StateGraph(MyState)
graph.add_node("hello", hello_node)
graph.add_edge(START, "hello")
graph.add_edge("hello", END)
compiled = graph.compile()

# v2：直接返回 GraphOutput，里面的 .value 已经是 MyState 实例
output = compiled.invoke({"answer": "", "count": 0}, version="v2")
print(type(output.value))  # <class '__main__.MyState'>

# 直接访问，不用手动转换
print(output.value.answer)  # IDE 自动补全，类型安全
print(output.value.count)

特性的核心价值

  1. 消除手动转换的冗余代码：不用再写 MyState(**result) 这种模板代码。
  2. IDE 自动补全和类型检查：你在写代码的时候，就能看到 .answer/.count 这些字段提示，也能发现拼写错误。
  3. 减少运行时错误：如果返回的数据结构不符合你定义的 Pydantic 模型，会直接在转换时抛出错误，而不是等到后面用的时候才崩溃。
  4. 代码更直观、可读性更强：看到 MyState 就知道这是什么类型的数据，不用再猜字典里有哪些 key。

使用条件

要启用这个特性，需要同时满足两个条件：

  1. 你的状态是用 Pydantic BaseModel（或 dataclass）定义的。
  2. 调用 invoke()/stream() 时，加上参数 version="v2"。

invoke()和stream()类型转换示例

invoke() 类型转换示例

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
from langgraph.graph import StateGraph, START, END
from langgraph.types import GraphOutput
from pydantic import BaseModel

# 定义状态模型
class MyState(BaseModel):
    answer: str
    count: int

# 定义处理节点
def process(state: MyState) -> MyState:
    return state.model_copy(update={"answer": "Hello World", "count": state.count + 1})

# 构建图
graph = StateGraph(MyState)
graph.add_node("process", process)
graph.add_edge(START, "process")
graph.add_edge("process", END)
compiled = graph.compile()

# 使用 v2 版本调用
result = compiled.invoke({"answer": "", "count": 0}, version="v2")

# 查看类型转换效果
print(type(result))          # GraphOutput
print(type(result.value))    # MyState

# 直接使用模型属性
print(result.value.answer)   # "Hello World"
print(result.value.count)    # 1

stream() 类型转换示例

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
from langgraph.graph import StateGraph, START, END
from langgraph.types import StreamPart
from pydantic import BaseModel

# 定义状态模型
class MyState(BaseModel):
    answer: str
    count: int

# 定义处理节点
def process(state: MyState) -> MyState:
    return state.model_copy(update={"answer": "Hello World", "count": state.count + 1})

# 构建图
graph = StateGraph(MyState)
graph.add_node("process", process)
graph.add_edge(START, "process")
graph.add_edge("process", END)
compiled = graph.compile()

# 使用 v2 版本流处理
for part in compiled.stream({"answer": "", "count": 0}, version="v2"):
    if part["type"] == "values":
        # part["data"] 已经是 MyState 类型
        print(part["data"])          # MyState
        print(part["data"].answer)   # "Hello World"
        print(part["data"].count)    # 1

总结

LangGraph 1.1 版本更新的最终总结页，把前面讲的所有新特性做了系统梳理，并说明了这些更新给开发者带来的实际价值。

更新点 通俗解释 开发收益
1. 类型安全的流处理和调用 引入 version="v2" 流格式，给 stream()/astream()/invoke()/ainvoke() 这四个核心方法加上了完整的类型安全保障 IDE 能自动补全、类型检查器能提前发现错误，减少运行时崩溃
2. 父图和子图的重放行为修复 修复了嵌套子图时，执行历史记录、重放 / 重试功能的异常问题 复杂多子图嵌套的智能体，行为更稳定、可预测，调试和状态回溯更可靠
3. 输出类型强制转换 当你的状态是 Pydantic 模型或 dataclass 时，v2 模式会自动把输出转换成你定义的类型，不再返回普通字典 不用手动写类型转换代码，代码更简洁，也避免了手动转换可能带来的 bug
4. 向后兼容性 默认仍然是 version="v1"，旧代码完全不用修改就能正常运行 可以按需逐步迁移到 v2，不用一次性重构所有项目，风险可控

这些改进本质上是让 LangGraph 从一个 “能用的工具”，变成了一个 “更工程化、更适合生产级开发” 的框架 ：

  • 更好的 IDE 支持和类型检查 → 编码阶段就能发现问题，调试效率大幅提升
  • 更少的运行时错误 → 智能体系统的稳定性和可靠性显著提高
  • 更高的代码质量和可维护性 → 类型安全的代码更容易阅读、重构和协作
  • 低风险的升级路径 → 现有项目可以平滑过渡，不会被强制绑定新版本

LangGraph 1.1 的核心目标，是通过类型安全升级，解决之前版本在复杂智能体开发中遇到的类型不匹配、调试困难、嵌套图不稳定等痛点，同时通过向后兼容的设计，让开发者可以零成本、按需升级，最终让构建复杂 AI 智能体工作流变得更轻松、更可靠。