概述

LangGraph 是 LangChain 生态中用于构建有状态、多步 Agent 工作流的核心库。与传统的 Chain 不同，LangGraph 使用图（Graph）的概念来建模 Agent 行为，其中节点（Node）表示计算步骤，边（Edge）表示控制流，状态（State）在节点间传递。

前置条件

• Python 3.10+
• LangGraph >= 0.0.20
• 安装命令：pip install langgraph

核心概念

1. State（状态）

State 是图的共享数据结构，包含整个工作流的当前信息：

from typing import TypedDict
from typing_extensions import Annotated
from operator import add

class AgentState(TypedDict):
    # 基本字段：直接覆盖更新
    messages: list[str]
    
    # 带 Reducer 的字段：累加更新
    history: Annotated[list[str], add]

2. Node（节点）

节点是执行特定逻辑的 Python 函数：

def process_node(state: AgentState) -> AgentState:
    """处理节点：接收当前状态，返回更新后的状态"""
    current_msg = state["messages"][-1]
    return {
        "messages": [f"Processed: {current_msg}"]
    }

3. Edge（边）

边决定下一个执行哪个节点：

# 普通边：从 node_a 直接到 node_b
builder.add_edge("node_a", "node_b")

# 条件边：根据状态决定路由
def routing_func(state: AgentState) -> str:
    if len(state["messages"]) > 5:
        return "end_node"
    return "continue_node"

builder.add_conditional_edges("router", routing_func)

完整示例：简单的对话 Agent

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

# 1. 定义状态
class ChatState(TypedDict):
    messages: list[str]
    turn_count: int

# 2. 创建图构建器
builder = StateGraph(ChatState)

# 3. 定义节点
def should_continue(state: ChatState) -> str:
    """判断是否继续对话"""
    if state["turn_count"] >= 3:
        return "end"
    return "continue"

def agent_node(state: ChatState) -> ChatState:
    """Agent 核心逻辑"""
    last_msg = state["messages"][-1]
    response = f"我收到了: {last_msg} (第 {state['turn_count'] + 1} 轮)"
    return {
        "messages": state["messages"] + [response],
        "turn_count": state["turn_count"] + 1
    }

# 4. 添加节点和边
builder.add_node("agent", agent_node)
builder.add_edge(START, "agent")
builder.add_conditional_edges(
    "agent",
    should_continue,
    {"continue": "agent", "end": END}
)

# 5. 编译图
graph = builder.compile()

# 6. 执行
result = graph.invoke({
    "messages": ["你好"],
    "turn_count": 0
})

print(result["messages"])
# ['你好', '我收到了: 你好 (第 1 轮)', '我收到了: 我收到了: 你好 (第 1 轮) (第 2 轮)', '我收到了: ... (第 3 轮)']

与 LangChain Agent 的区别

特性	LangChain Chain	LangGraph
工作流	线性/固定	图结构/可条件分支
状态管理	无内置状态	内置 State 共享
循环支持	不支持	支持任意循环
调试难度	较低	需要理解图执行模型

常见问题

Q1: State 的 Reducer 有什么用？

• 默认 Reducer 直接覆盖字段值
• 使用 Annotated[list, add] 可以实现列表累加
• 自定义 Reducer 可以实现复杂的合并逻辑

Q2: 如何处理多轮对话的记忆？

• 在 State 中定义 messages 列表存储对话历史
• 使用 add Reducer 保留所有历史消息
• 定期使用 LLM 总结历史信息以节省 Token

Q3: 条件边的返回值类型有什么要求？

• 返回值必须是字符串，对应目标节点的名称
• 可以返回 END 表示结束执行
• 可以返回 START 表示重新开始

参考资料

• LangGraph 官方文档
• Graph API 概述
• LangGraph GitHub 仓库