LangGraph工作流编排：从状态机到动态路由实战

1. LangGraph 核心概念解析

LangGraph 作为 LangChain 生态中的工作流编排引擎，其核心设计理念是将复杂业务逻辑拆解为可组合、可重用的节点单元，通过状态传递和路由控制实现灵活的业务流程。理解其基础架构是掌握高级用法的前提。

1.1 状态图模型

StateGraph（状态图）是 LangGraph 的核心抽象，它借鉴了有限状态机（FSM）的概念，但增加了更灵活的状态管理和路由能力。一个典型的状态图包含以下要素：

• 状态对象：作为数据载体在节点间传递，支持两种定义方式：

  python复制# 方式1：使用TypedDict（灵活但弱类型）
  from typing import TypedDict
  class MyState(TypedDict):
      counter: int
      log: list[str]
  
  # 方式2：使用Pydantic（强类型校验）
  from pydantic import BaseModel
  class MyState(BaseModel):
      counter: int
      log: list[str] = []
  

• 节点函数：每个节点都是纯函数，接收当前状态并返回新状态。关键设计原则：

  • 保持无副作用：避免在节点内修改外部状态
  • 单一职责：每个节点只完成一个明确的任务
  • 幂等性：相同输入应产生相同输出

1.2 执行生命周期

LangGraph 工作流的完整执行流程遵循严格的生命周期：

1. 初始化阶段：

   • 解析状态定义
   • 验证节点函数签名
   • 检查边连接的合法性

2. 编译阶段：

   • 将Python对象转换为可执行图结构
   • 优化节点执行顺序
   • 生成运行时校验逻辑

3. 执行阶段：

   mermaid复制graph TD
     A[输入初始状态] --> B[查找起始节点]
     B --> C{是否条件边?}
     C -->|是| D[执行路由函数]
     C -->|否| E[执行固定边]
     D --> F[选择目标节点]
     E --> G[跳转到指定节点]
     F --> H[执行节点逻辑]
     G --> H
     H --> I{是否到达END?}
     I -->|否| C
     I -->|是| J[返回最终状态]
   

重要提示：LangGraph 默认采用深度优先执行策略，当遇到并行分支（如Send操作）时会创建子执行上下文。开发者可通过runtime_config参数调整最大递归深度等参数。

2. 基础工作流构建实战

2.1 线性流程实现

最常见的业务场景是顺序执行链式任务。以下是一个完整的数据处理流水线示例：

python复制from langgraph.graph import StateGraph

# 状态定义
class DataPipelineState(TypedDict):
    raw_data: dict
    cleaned_data: dict
    analysis_result: float

# 节点实现
def data_loader(_) -> dict:
    print("加载数据源...")
    return {"raw_data": {"temp": [22.1, 23.4, None, 25.0]}}

def data_cleaner(state: DataPipelineState) -> dict:
    print("清洗数据...")
    cleaned = [x for x in state["raw_data"]["temp"] if x is not None]
    return {"cleaned_data": {"temp": cleaned}}

def data_analyzer(state: DataPipelineState) -> dict:
    print("分析数据...")
    avg = sum(state["cleaned_data"]["temp"])/len(state["cleaned_data"]["temp"])
    return {"analysis_result": avg}

# 构建流程
builder = StateGraph(DataPipelineState)
builder.add_node("load", data_loader)
builder.add_node("clean", data_cleaner)
builder.add_node("analyze", data_analyzer)

# 定义边关系
builder.add_edge(START, "load")
builder.add_edge("load", "clean")
builder.add_edge("clean", "analyze")
builder.add_edge("analyze", END)

# 编译执行
pipeline = builder.compile()
result = pipeline.invoke({})
print(f"平均温度: