跳到主要内容

概述与核心概念

LangGraph是什么​

LangGraph是一个低级代理编排框架，专为构建复杂、长时间运行且有状态的工作流而设计。与LangChain的高级API相比，LangGraph提供了更直接的控制和更大的灵活性。

核心概念​

1. Graph（图）​

图由节点和边组成，表示工作流的执行路径：

2. State（状态）​

状态是流经图的数据结构，包含工作流所需的所有信息：

from typing import TypedDict

class WorkflowState(TypedDict):
    messages: list           # 对话消息
    current_step: str        # 当前步骤
    results: dict           # 中间结果
    metadata: dict          # 元数据

3. Node（节点）​

节点是执行单元，接收状态并返回修改后的状态：

def process_node(state: WorkflowState) -> WorkflowState:
    # 处理状态
    state["current_step"] = "processed"
    return state

4. Edge（边）​

边定义了节点之间的连接：

  • 直连边：无条件连接
  • 条件边：基于状态的条件连接

执行流程​

Graph API vs Functional API​

Graph API（节点-边范式）​

优点：

  • 清晰的工作流结构
  • 可视化支持
  • 适合复杂的分支和循环

缺点：

  • 需要理解图论
  • 代码相对冗长

Functional API（函数式范式）​

优点：

  • 更像传统编程
  • 代码更简洁
  • 学习曲线平缓

缺点：

  • 不支持可视化
  • 适合线性工作流

与其他框架的对比​

特性LangGraphLangChain AgentCelery
代理支持完全完全否
状态管理内置简单外部
持久化支持否支持
人工中断支持否否
并行处理支持部分完全
学习曲线中等平缓陡峭

常见使用场景​

1. 多步骤工作流​

2. 条件路由​

3. 循环和重试​

4. 并行处理​

快速开始示例​

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

# 1. 定义状态
class SimpleState(TypedDict):
    value: int

# 2. 定义节点
def increment_node(state: SimpleState):
    return {"value": state["value"] + 1}

def double_node(state: SimpleState):
    return {"value": state["value"] * 2}

# 3. 构建图
graph = StateGraph(SimpleState)
graph.add_node("increment", increment_node)
graph.add_node("double", double_node)

graph.add_edge(START, "increment")
graph.add_edge("increment", "double")
graph.add_edge("double", END)

# 4. 编译
app = graph.compile()

# 5. 运行
result = app.invoke({"value": 5})
print(result)  # {"value": 12}

关键优势​

1. 完全控制​

# 自定义每一个执行步骤
def custom_node(state):
    # 前处理
    prepared = prepare_data(state)
    
    # 主逻辑
    result = execute_logic(prepared)
    
    # 后处理
    final = finalize_result(result)
    
    return final

2. 状态隔离​

# 每个节点只修改自己关心的状态部分
def node_a(state):
    state["a_result"] = compute_a()
    return state

def node_b(state):
    state["b_result"] = compute_b()
    return state

3. 易于测试​

# 可以独立测试每个节点
def test_node():
    test_state = {"value": 10}
    result = my_node(test_state)
    assert result["value"] == 20

性能特性​

  • 流式处理：逐步获取结果
  • 检查点：定期保存状态以便恢复
  • 并发：支持并行节点执行
  • 缓存：避免重复计算

最佳实践​

  1. 单一职责：每个节点只做一件事
  2. 清晰的状态：使用TypedDict明确定义状态结构
  3. 错误处理：在节点中实现错误处理
  4. 日志记录：添加日志便于调试
  5. 测试：编写单元测试验证节点逻辑

后续学习​

完成本节后，建议学习：