LangGraph 详解

---

目录

1. 什么是 LangGraph

2. 核心概念：State、Node、Edge、Graph

3. 完整构建流程

4. State 状态定义

5. Node 节点

6. Edge 边

7. State 更新策略：Reducer 机制详解

8. 节点重试策略

9. 节点缓存机制

10. 带参数节点的正确写法

11. 图的可视化

12. 与大模型集成

---

1. 什么是 LangGraph

LangGraph 是 LangChain 生态中的一个库，用于构建有状态的多代理应用。它的核心思想是将业务逻辑建模为一张有向图（Directed Graph），图由节点（Node）和边（Edge）组成，数据以状态（State）在节点间流转。

START → Node A → Node B → Node C → END
         (State 在每个节点之间传递和更新)

LangGraph 的灵魂只有四个要素：

要素	作用
State	图在执行过程中共享的数据结构（本质是一个字典）
Node	节点函数，接收当前 State，处理后返回更新
Edge	边，定义节点之间的执行顺序和数据流向
Graph	图，将所有节点和边组装起来编译成可执行的应用

为什么要用 LangGraph？

• 比 LangChain LCEL 链式调用更灵活，支持条件分支、循环、并行执行

• 天然适合多步骤工作流、智能体（Agent）、RAG等复杂场景

• 状态在节点间持久化，便于调试和断点恢复

---

2. 核心概念：State、Node、Edge、Graph

2.1 State（状态）

State 是贯穿整个图执行过程的共享数据容器，本质是一个 Python 字典（通常用 TypedDict 定义）。

from typing import TypedDict
​
class HelloState(TypedDict):
    name: str
    greeting: str

• 每个节点接收同一个 State 对象

• 节点返回的字典会合并到 State 中（如何合并由 Reducer 决定）

• 初始状态在 invoke() 时传入：app.invoke({"name": "z3"})

2.2 Node（节点）

节点是一个** Python 函数**，接收当前 State，返回更新后的部分 State。

def greet(state: HelloState) -> dict:
    name = state["name"]
    return {"greeting": f"你好, {name}"}

• 函数签名：def 节点名(state: State类型) -> dict

• 返回值只需要返回需要更新的字段，不必返回整个 State

• 节点可以调用大模型、API、工具，或者做纯粹的数值计算

2.3 Edge（边）

边定义了执行顺序和数据流向。有三种边：

边类型	方法	说明
普通边	add_edge(A, B)	A 执行完无条件执行 B
条件边	add_conditional_edges(A, fn, mapping)	A 执行完后根据函数返回值决定下一步
入口/出口	set_entry_point() / set_finish_point()	自动绑定 START/END

2.4 Graph（图）

图是 LangGraph 的顶层对象，通过 StateGraph 构建器将所有节点和边组装起来，最后 compile() 编译成可执行的应用。

graph = StateGraph(HelloState)   # 创建图构建器
graph.add_node("greet", greet)    # 添加节点
graph.add_edge(START, "greet")    # 添加边
graph.add_edge("greet", END)      # 添加边
app = graph.compile()              # 编译成可执行应用
result = app.invoke({"name": "z3"})  # 调用执行

---

3. 完整构建流程

LangGraph 的图构建遵循固定的 6 步流程（BuildWholeGraphSummary.py）：

# 1. 定义状态
class GraphState(TypedDict):
    process_data: dict
​
# 2. 定义节点函数
def input_node(state: GraphState):
    return {"process_data": {"input": "input_value"}}
​
def process_node(state: dict) -> dict:
    return {"process_data": {"process": "process_value9527"}}
​
def output_node(state: GraphState) -> GraphState:
    return {"process_data": state.get("process_data")}
​
# 3. 创建图构建器（指定 State 类型）
graph = StateGraph(GraphState)
​
# 4. 添加节点
graph.add_node("input", input_node)
graph.add_node("process", process_node)
graph.add_node("output", output_node)
​
# 5. 添加边（定义执行顺序）
graph.add_edge(START, "input")
graph.add_edge("input", "process")
graph.add_edge("process", "output")
graph.add_edge("output", END)
​
# 6. 编译图
app = graph.compile()
​
# 执行
result = app.invoke({"process_data": {"name": "测试数据", "value": 123456}})

执行流程：

START
  │
  ▼
input_node     (写入 process_data: {"input": "input_value"})
  │
  ▼
process_node   (写入 process_data: {"process": "process_value9527"})
  │
  ▼
output_node    (透传 process_data)
  │
  ▼
 END

---

4. State 状态定义

4.1 基本定义

用 TypedDict 定义状态结构：

from typing import TypedDict
​
class BasicState(TypedDict):
    user_input: str
    response: str
    count: int
    process_data: dict

invoke() 接收一个字典作为初始状态：

app.invoke({
    "user_input": "a",
    "response": "resp",
    "count": 25,
    "process_data": {"k1": "v1"}
})

4.2 输入输出模式（Input/Output Schema）

可以分别定义输入模式和输出模式，实现图的只读入参和只读出参：

from typing_extensions import TypedDict
​
class InputState(TypedDict):
    question: str
​
class OutputState(TypedDict):
    answer: str
​
class OverallState(InputState, OutputState):
    pass
​
# 在构建图时指定输入输出模式
builder = StateGraph(
    OverallState,
    input_schema=InputState,   # 调用时只需传入 InputState
    output_schema=OutputState  # 结果只暴露 OutputState 字段
)

这样图在调用时：

# 只需要传 question
result = graph.invoke({"question": "你好"})
# 结果只包含 answer
# {"question": "你好", "answer": "你好"}

4.3 无节点的最简图

StateGraph 允许只有 START → END 的空图，用于验证 State 定义：

basicState = StateGraph(BasicState)
basicState.add_edge(START, END)
app = basicState.compile()
result = app.invoke(initial_state)  # 直接从 START 到 END

---

5. Node 节点

5.1 基础节点

节点函数接收 State，返回部分更新字典：

class AtguiguState(TypedDict):
    value: int
    step: str
​
def node_a(state: AtguiguState) -> dict:
    return {"value": state["value"] + 1, "step": "A执行完毕"}

5.2 带参数的节点

节点函数可以携带额外参数，使用 functools.partial 绑定：

from functools import partial
​
def process_node(state: dict, param1: int, param2: str) -> dict:
    print(state, param1, param2)
    return {"process_data": {"process": "process_value"}}
​
process_with_params = partial(process_node, param1=100, param2="test")
graph.add_node("process", process_with_params)

5.3 节点与 Messages 列表

在接大模型的场景中，消息列表是最常见的状态字段，用 add_messages 作为 reducer：

from typing import Annotated, List, TypedDict
from langgraph.graph.message import add_messages
from langchain_core.messages import HumanMessage
​
class AtguiguState(TypedDict):
    messages: Annotated[List, add_messages]  # 追加而非覆盖
​
def model_node(state: AtguiguState):
    reply = llm.invoke(state["messages"])
    return {"messages": [reply]}  # 自动追加到列表末尾

add_messages reducer 的效果：每次节点返回消息时，自动将新消息追加到已有消息列表，而不是覆盖整个列表。

---

6. Edge 边

6.1 普通边（固定流向）

无条件从 A 跳转到 B：

graph.add_edge("node_a", "node_b")    # A 执行完 → 立即执行 B
graph.add_edge(START, "node_a")        # 图入口 → A
graph.add_edge("node_c", END)          # C 执行完 → 图结束

典型顺序链：START → A → B → C → END

6.2 入口点与出口点

set_entry_point() 和 set_finish_point() 是普通边的语法，底层等效于 add_edge(START, node) 和 add_edge(node, END)：

# 以下两段代码完全等价：
​
# 方式1：普通边
graph.add_edge(START, "node_a")
graph.add_edge("node_a", "node_b")
graph.add_edge("node_b", END)
​
# 方式2：语法糖
graph.set_entry_point("node_a")
graph.add_edge("node_a", "node_b")
graph.set_finish_point("node_b")

6.3 条件边（Conditional Edges）

根据节点执行结果动态决定下一步走向：

from langgraph.constants import START, END
from langgraph.graph import StateGraph
​# 定义状态
class MyState(BaseModel):
    x: int
    result: Optional[str] = None
​
# 判断函数：返回 True/False
def is_even(state: MyState) -> bool:
    return state.x % 2 == 0
​
# 定义节点
def handle_even(state: MyState) -> MyState:
    return MyState(x=state.x, result="even")
​
def handle_odd(state: MyState) -> MyState:
    return MyState(x=state.x, result="odd")
​# 定义图
builder = StateGraph(MyState)
builder.add_node("check_x", check_x)
builder.add_node("handle_even", handle_even)
builder.add_node("handle_odd", handle_odd)
​
# 条件边：从 check_x 出发，根据 is_even() 返回值决定去向
builder.add_conditional_edges(
    "check_x",                # 从哪个节点出发
    is_even,                  # 条件判断函数（返回 bool）
    {True: "handle_even", False: "handle_odd"}  # 返回值 → 目标节点映射
)
​
builder.add_edge(START, "check_x")
builder.add_edge("handle_even", END)
builder.add_edge("handle_odd", END)
​
graph = builder.compile()

执行逻辑：

输入 x=4（偶数）: START → check_x → is_even(True) → handle_even → END
输入 x=3（奇数）: START → check_x → is_even(False) → handle_odd → END

6.4 条件入口点（Conditional Entry Point）

从 START 开始就根据初始状态动态选择第一个节点（无需经过 check_x 节点做路由）：

def route_input(state: SimpleState) -> str:
    """根据输入内容返回路由键"""
    text = state["user_input"].lower()
    if "hello" in text or "hi" in text:
        return "greeting"
    elif "bye" in text or "exit" in text:
        return "farewell"
    else:
        return "question"
​
graph.add_conditional_edges(
    START,              # 从图的入口开始
    route_input,        # 路由函数（返回字符串作为键）
    {                  # 路由键到节点名的映射
        "greeting": "greeting_node",
        "farewell": "farewell_node",
        "question": "question_node"
    }
)

执行逻辑：

START → route_input() → "greeting" → greeting_node → END
START → route_input() → "farewell" → farewell_node → END
START → route_input() → "question" → question_node → END

6.5 条件边的多返回值映射

条件函数也可以返回多值（如 1/2/3），映射到不同的处理节点：

def route_by_value(state: AtguiguState) -> str:
    flag = state["x"]
    if flag == 1:
        return "condition_1"
    elif flag == 2:
        return "condition_2"
    else:
        return "condition_3"
​
graph.add_conditional_edges(
    START,
    route_by_value,
    {
        "condition_1": "node1",  # x == 1 → node1
        "condition_2": "node2",  # x == 2 → node2
        "condition_3": "node3"   # x == 3 → node3
    }
)

---

7. State 更新策略：Reducer 机制详解

7.1 为什么需要 Reducer？

在 LangGraph 中，多个节点可能对同一个状态字段产生更新。Reducer 决定了这些更新"如何合并"到 State 中。

没有 Reducer 之前：每个节点返回什么，State 就被覆盖成什么。

有了 Reducer 之后：可以控制是"追加"、"累加"还是"覆盖"。

这是通过 Annotated[type, reducer] 语法在 State 定义时声明的。

from typing import Annotated, List, TypedDict
​
class MyState(TypedDict):
    # 语法: Annotated[类型, Reducer函数]
    messages: Annotated[List[str], add_messages]   # 追加
    count:    Annotated[int, operator.add]          # 累加
    name:     str                                   # 无Reducer → 覆盖（默认行为）

7.2 默认 Reducer（覆盖）

不写 Reducer 时，等效于"后到者覆盖先到者"：

class DefaultState(TypedDict):
    foo: int
    bar: List[str]
    baz: str
​
# 初始: {"foo": 1, "bar": ["hi"], "baz": "hello"}
# node1 返回: {"foo": 22, "baz": "world"}     → foo/baz 被覆盖，bar 不变
# node2 返回: {"bar": ["bye1", "bye2"]}       → bar 被覆盖，foo/baz 不变
# 最终: {"foo": 22, "bar": ["bye1", "bye2"], "baz": "world"}

覆盖模式下，每个节点的返回值是全量替换，而不是部分更新。

7.3 add_messages Reducer（聊天消息追加）

专门为聊天消息列表设计，使用 Annotated[List, add_messages]：

from typing import Annotated, List
from langgraph.graph.message import add_messages
​
class ChatState(TypedDict):
    messages: Annotated[List, add_messages]

核心行为：追加而非覆盖

• 初始：{"messages": [HumanMessage("Hello")]}

• 节点 A 返回：{"messages": [AIMessage("Hi there!")]}

• 节点 B 返回：{"messages": [AIMessage("How can I help?")]}

• 最终：

{"messages": [
    HumanMessage("Hello"),
    AIMessage("Hi there!"),
    AIMessage("How can I help?")
]}

即使节点并行执行（两个节点都从 START 开始），各自的返回值也都会被追加，不会互相覆盖。这在多 Agent 协作时非常有用。

7.4 operator.add Reducer（列表拼接 / 数值累加 / 字符串连接）

operator.add 就是 Python 的 + 运算符，对不同类型有不同含义：

列表拼接：

class ListState(TypedDict):
    data: Annotated[List[int], operator.add]
​
# 初始: {"data": [0]}
# node1 返回: {"data": [1, 2]}   → [0] + [1, 2] = [0, 1, 2]
# node2 返回: {"data": [3, 4]}   → [0, 1, 2] + [3, 4] = [0, 1, 2, 3, 4]
# 最终: {"data": [0, 1, 2, 3, 4]}

数值累加：

class CountState(TypedDict):
    count: Annotated[int, operator.add]
​
# 初始: {"count": 10}
# node1 返回: {"count": 5}   → 10 + 5 = 15
# node2 返回: {"count": 3}   → 15 + 3 = 18
# 最终: {"count": 18}

字符串连接：

class TextState(TypedDict):
    text: Annotated[str, operator.add]
​
# 初始: {"text": "Say: "}
# node1 返回: {"text": "Hello "}   → "Say: " + "Hello " = "Say: Hello "
# node2 返回: {"text": "World!"}   → "Say: Hello " + "World!" = "Say: Hello World!"
# 最终: {"text": "Say: Hello World!"}

7.5 operator.mul Reducer 与初始值的陷阱

LangGraph 的 Reducer 在图执行正式开始前，会用类型的默认值做一次预计算。这个设计会导致 operator.mul 出现一个常见陷阱：

class MultiplyState(TypedDict):
    factor: Annotated[float, operator.mul]
​
def multiplier(state):
    return {"factor": 2.0}
​
result = graph.invoke({"factor": 5.0})
print(result)  # 期望 10.0，实际却是 0.0 ！！

原因分析：

步骤	说明
invoke 传入	{"factor": 5.0}
LangGraph 预计算	float 默认值 0.0 × 5.0 = 0.0
multiplier 节点执行	返回 {"factor": 2.0}
再次计算	0.0 × 2.0 = 0.0
最终结果	{"factor": 0.0} ← 错误！

解决方案：使用自定义 Reducer

def safe_mul(current: float, update: float) -> float:
    """处理 0.0 默认值问题的乘法 reducer"""
    if current == 0.0:
        return update  # 跳过初始 0.0，用 update 作为起点
    return current * update
​
class MultiplyState(TypedDict):
    factor: Annotated[float, safe_mul]
​
# 初始: {"factor": 5.0}
# multiplier 返回: {"factor": 2.0}
# 计算: 5.0 * 2.0 = 10.0
# 最终: {"factor": 10.0} ✓

经验法则：使用加减法以外的操作（乘、除、拼接自定义结构）时，务必使用自定义 Reducer，并在 Reducer 内部处理默认值 0.0 / 空列表 / 空字符串等边界情况。

7.6 自定义 Reducer 完整范式

Reducer 的签名固定为：(current_value, update_value) -> merged_value

# 示例：取最大值的 reducer
def take_max(current: int, update: int) -> int:
    return max(current, update)
​
class MyState(TypedDict):
    score: Annotated[int, take_max]
​
# 初始: {"score": 10}
# node1 返回: {"score": 25}   → max(10, 25) = 25
# node2 返回: {"score": 20}   → max(25, 20) = 25
# 最终: {"score": 25}

7.7 同一 State 中多字段多策略混用

LangGraph 支持 State 中不同字段使用不同的 Reducer：

class ChatState(TypedDict):
    messages: Annotated[list, add_messages]   # 消息追加
    tags:     Annotated[List[str], operator.add]  # 标签拼接
    score:    Annotated[float, operator.add]      # 分数累加
    name:     str                                 # 无Reducer → 覆盖
​
def process(state: ChatState) -> dict:
    return {
        "messages": [AIMessage("回答内容")],
        "tags": ["processed"],
        "score": 1.0
    }
    # name 字段不在返回值中 → 保持原值不变

并行执行时（START → node1, START → node2）：

# node1 返回: {"messages": [...], "tags": ["processed"], "score": 1.0}
# node2 返回: {"messages": [...], "tags": ["positive"], "score": 0.5}
​
# 最终效果：
# messages: 两个节点的结果都被追加（add_messages）
# tags:     ["processed"] + ["positive"] = 拼接（operator.add）
# score:    1.0 + 0.5 = 1.5（operator.add）

---

8. 节点重试策略

8.1 为什么需要重试？

在实际生产环境中，API 调用、网络请求、外部工具调用都可能因为临时故障而失败。重试机制让节点在失败后自动重试，提升工作流的健壮性。

8.2 RetryPolicy 完整参数

from langgraph.types import RetryPolicy
​
retry_policy = RetryPolicy(
    max_attempts=5,       # 最大尝试次数（含首次），超过则抛异常
    initial_interval=1.0,  # 首次重试前等待的秒数
    backoff_factor=2.0,    # 每次失败后，等待时间 = interval × backoff_factor^retry_count
    jitter=True,           # True=在上述基础上加随机抖动，避免多节点同时重试造成雪崩
    retry_on=[Exception]   # 仅对列表中的异常类型重试；其他异常立即抛出
)

退避时间计算示例（initial_interval=1, backoff_factor=2, jitter=False）：

尝试	等待时间
第1次重试	1.0 秒
第2次重试	2.0 秒
第3次重试	4.0 秒
第4次重试	8.0 秒

加上 jitter 后，每次的实际等待时间会在计算值附近浮动。

8.3 默认重试策略 vs 自定义重试策略

LangGraph 内置的默认重试策略（不指定 retry 参数时）：

max_attempts=5
retry_on=default_retry_on(Exception)

default_retry_on 的行为：

异常类型	是否重试
Exception（所有普通异常）	✅ 重试
ValueError	❌ 不重试
TypeError	❌ 不重试
RuntimeError	❌ 不重试
ImportError	❌ 不重试
OSError	❌ 不重试

8.4 三种 retry_on 配置方式

方式一：指定异常类型列表（最简单）

from requests import RequestException, Timeout
​
builder.add_node(
    "api_node",
    call_external_api,
    retry=RetryPolicy(
        max_attempts=3,
        retry_on=[RequestException, Timeout, ConnectionError]
    )
)

方式二：传入过滤函数（最灵活）

def selective_retry(exception: Exception) -> bool:
    """
    自定义过滤逻辑：只对包含特定关键词的异常重试
    """
    msg = str(exception)
    # API 服务端错误（500系列）→ 重试
    if "500" in msg or "502" in msg or "503" in msg:
        return True
    # 认证失败（401/403）→ 不重试（无法通过重试解决）
    if "401" in msg or "403" in msg:
        return False
    # 其他网络错误 → 重试
    if isinstance(exception, (ConnectionError, Timeout)):
        return True
    return False
​
builder.add_node(
    "api_node",
    call_external_api,
    retry=RetryPolicy(max_attempts=5, retry_on=selective_retry)
)

方式三：完全不禁用重试（关闭）

builder.add_node(
    "no_retry_node",
    risky_operation,
    retry=RetryPolicy(max_attempts=1)  # 只尝试1次，即禁用重试
)

8.5 完整重试演示

from langgraph.types import RetryPolicy
​
attempt_counter = 0
​
def unstable_api(state):
    global attempt_counter
    attempt_counter += 1
    print(f"第 {attempt_counter} 次调用")
    if attempt_counter < 3:
        raise Exception(f"模拟失败 (第{attempt_counter}次)")
    return {"result": f"成功，经过 {attempt_counter} 次尝试"}
​
builder = StateGraph(State)
builder.add_node("api", unstable_api, retry=RetryPolicy(max_attempts=5))
builder.add_edge(START, "api")
builder.add_edge("api", END)
graph = builder.compile()
​
# 输出:
# 第 1 次调用  → 失败，等待1秒
# 第 2 次调用  → 失败，等待2秒
# 第 3 次调用  → 成功！
# result: {"result": "成功，经过 3 次尝试"}

---

9. 节点缓存机制

9.1 为什么需要缓存？

某些节点执行代价高昂（如调用付费 API、大模型推理、复杂计算），而实际运行时经常用相同的输入反复调用。缓存可以在 TTL（时间-to-live）内对相同输入直接返回缓存结果，大幅降低成本和延迟。

9.2 两层缓存配置

LangGraph 的缓存分两层：

第一层：节点级缓存（cache_policy）

在 add_node() 时为单个节点设置：

from langgraph.types import CachePolicy
​
builder.add_node(
    "expensive_node",
    expensive_func,
    cache_policy=CachePolicy(ttl=60)  # 该节点的缓存有效期 60 秒
)

第二层：图级缓存存储（cache=）

在 compile() 时指定缓存后端：

from langgraph.cache.memory import InMemoryCache
​
app = builder.compile(cache=InMemoryCache())  # 内存缓存

两层必须同时配置才生效。cache_policy 定义缓存策略，cache= 定义存储后端。

9.3 缓存生效的完整示例

import time
from langgraph.cache.memory import InMemoryCache
from langgraph.types import CachePolicy
​
class State(TypedDict):
    x: int
    result: int
​
def expensive_node(state: State) -> dict:
    time.sleep(3)  # 模拟耗时3秒
    return {"result": state["x"] * 2}
​
builder = StateGraph(State)
builder.add_node(
    "expensive_node",
    expensive_node,
    cache_policy=CachePolicy(ttl=8)  # 缓存有效期8秒
)
builder.set_entry_point("expensive_node")
builder.set_finish_point("expensive_node")
​
app = builder.compile(cache=InMemoryCache())
​
# 第1次调用：耗时3秒（真实计算）
start = time.time()
result1 = app.invoke({"x": 5})
print(f"第1次: {result1}, 耗时 {time.time()-start:.1f}s")
​
# 第2次调用（相同输入）：瞬间返回
start = time.time()
result2 = app.invoke({"x": 5})
print(f"第2次: {result2}, 耗时 {time.time()-start:.1f}s")
​
# 等待8秒，缓存过期
time.sleep(8)
​
# 第3次调用：重新计算
start = time.time()
result3 = app.invoke({"x": 5})
print(f"第3次: {result3}, 耗时 {time.time()-start:.1f}s")

典型输出：

第1次: {'x': 5, 'result': 10}, 耗时 3.0s
第2次: {'x': 5, 'result': 10}, 耗时 0.0s   ← 命中缓存
第3次: {'x': 5, 'result': 10}, 耗时 3.0s   ← 缓存过期，重新计算

9.4 缓存的 key 是如何计算的？

缓存 key 由节点的输入 State 决定。LangGraph 会自动序列化输入状态作为缓存键。因此：

• {"x": 5} → 有缓存

• {"x": 6} → 无缓存（key 不同）

• 节点内部修改了其他字段 → 只要输入相同，仍命中缓存

9.5 内置缓存后端

后端	说明
InMemoryCache()	进程内存存储，重启后清空，适合开发调试
（可扩展）	支持自定义后端（如 Redis、SQLite）

---

10. 带参数节点的正确写法

节点函数可以通过 functools.partial 绑定额外参数：

from functools import partial
​
def process_node(state: dict, param1: int, param2: str) -> dict:
    print(f"参数: param1={param1}, param2={param2}")
    return {"process_data": {"result": param1 + len(param2)}}
​
# 绑定参数
bounded_node = partial(process_node, param1=100, param2="hello")
​
builder = StateGraph(State)
builder.add_node("process", bounded_node, retry=RetryPolicy(max_attempts=3))

partial 的作用是预先填充 param1 和 param2，使得 bounded_node(state) 调用时等价于 process_node(state, 100, "hello")。

参数绑定 + 重试 + 缓存的组合

三个机制可以同时作用于一个节点：

from functools import partial
from langgraph.types import RetryPolicy, CachePolicy
​
def heavy_api_call(state, api_key: str, model: str):
    ...
​
bounded = partial(heavy_api_call, api_key="xxx", model="gpt-4")
​
builder.add_node(
    "api",
    bounded,
    retry=RetryPolicy(max_attempts=3, retry_on=[RequestException]),
    cache_policy=CachePolicy(ttl=300)
)

---

12. 图的可视化

LangGraph 内置三种可视化方式：

9.1 ASCII 可视化

print(app.get_graph().print_ascii())

输出示例：

    +---------+     +----------------+     +---------+
    |  START  | --> |     node_a     | --> |  node_b |
    +---------+     +----------------+     +---------+

9.2 Mermaid 语法

print(app.get_graph().draw_mermaid())

输出 Mermaid 格式的图描述代码，可粘贴到 https://www.processon.com/mermaid 或 mermaid.live 渲染成图片。

9.3 直接导出 PNG

png_bytes = app.get_graph().draw_mermaid_png(
    max_retries=5,
    retry_delay=2.0,
    # draw_method=MermaidDrawMethod.PYPPETEER  # 本地浏览器渲染（需安装 pyppeteer）
)
with open("langgraph.png", "wb") as f:
    f.write(png_bytes)

注意：draw_mermaid_png() 依赖 mermaid.ink 在线服务，网络不稳定时会失败，可设置 max_retries 重试。

---

13. 与大模型集成

13.1 接入阿里云 DashScope（qwen-plus）

LangGraph 中的节点可以调用大模型，实现有状态的多轮对话：

from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage
​
class AtguiguState(TypedDict):
    messages: Annotated[List, add_messages]
​
# 初始化大模型（阿里云百炼平台）
llm = init_chat_model(
    model="qwen-plus",
    model_provider="openai",
    api_key=os.getenv("aliQwen-api"),  # 从环境变量读取
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
)
​
# 节点：调用大模型
def model_node(state: AtguiguState):
    reply = llm.invoke(state["messages"])
    return {"messages": [reply]}  # 追加模型回复
​
# 构建图：START → model → END
graph = StateGraph(AtguiguState)
graph.add_node("model", model_node)
graph.add_edge(START, "model")
graph.add_edge("model", END)
app = graph.compile()
​
# 调用
result = app.invoke({
    "messages": "请用一句话解释什么是 LangGraph。"
})
print(result["messages"][-1].content)

执行流程：

初始 messages: "请用一句话解释什么是 LangGraph。"
        ↓
   model 节点调用 qwen-plus 大模型
        ↓
返回: messages = [HumanMessage(...), AIMessage("LangGraph 是...")]

13.2 聊天机器人的状态流

用户输入 → messages 列表追加 HumanMessage
         ↓
      model 节点调用 LLM
         ↓
      messages 列表追加 AIMessage
         ↓
      返回最终 state（包含完整对话历史）

关键点：使用 Annotated[List, add_messages] 作为消息列表的类型，节点的返回值会被追加而非覆盖，这样多轮对话时历史消息不会丢失。

---

总结：LangGraph 核心 API 一览

操作	API
创建图	StateGraph(StateClass)
添加节点	add_node("名称", 函数)
带参数节点	add_node("名称", partial(fn, param=value))
普通边	add_edge("A", "B")
条件边	add_conditional_edges("A", 函数, 映射)
入口/出口	set_entry_point() / set_finish_point()
编译	compile()
执行	invoke(initial_state)
Reducer — 默认（覆盖）	field: T（不写 Annotated）
Reducer — 消息追加	Annotated[List, add_messages]
Reducer — 列表拼接	Annotated[List, operator.add]
Reducer — 数值累加	Annotated[int, operator.add]
Reducer — 字符串连接	Annotated[str, operator.add]
Reducer — 自定义	Annotated[T, def reducer(current, update)]
重试策略	add_node(..., retry=RetryPolicy(...))
缓存策略	add_node(..., cache_policy=CachePolicy(ttl=N)) + compile(cache=InMemoryCache())
可视化	get_graph().print_ascii() / draw_mermaid() / draw_mermaid_png()

核心公式：Annotated[类型, Reducer] — 状态字段通过 Annotated 声明合并策略，不写则默认覆盖。

---

适用场景

• 多步骤工作流：数据清洗 → 转换 → 存储

• 带分支的业务流程：根据输入内容走不同的处理路径

• 多轮对话 Agent：在节点中调用工具，状态保留对话历史

• RAG 检索流程：查询 → 检索 → 生成答案

• 循环监控：节点执行 → 判断是否重复 → 重复则回环

---