1.2 状态图(State Graph)详解

文档摘要

1.2 状态图(State Graph)详解本节导读：通过本节学习，你将深入理解LangGraph的核心概念——状态图，掌握StateGraph的创建方法和使用技巧，能够设计出高效、可维护的智能体执行流程。学习目标理解状态图的概念和作用掌握StateGraph的创建和配置方法学会设计合理的图结构和工作流理解不同编译选项对执行的影响核心概念什么是状态图？状态图（State Graph）是LangGraph的核心组件，它定义了智能体的执行流程和状态管理机制。与传统编程中的线性执行不同，状态图通过节点（Nodes）和边（Edges）来构建复杂的执行路径。

1.2 状态图(State Graph)详解

本节导读：通过本节学习，你将深入理解LangGraph的核心概念——状态图，掌握StateGraph的创建方法和使用技巧，能够设计出高效、可维护的智能体执行流程。

学习目标

理解状态图的概念和作用
掌握StateGraph的创建和配置方法
学会设计合理的图结构和工作流
理解不同编译选项对执行的影响

核心概念

什么是状态图？

状态图（State Graph）是LangGraph的核心组件，它定义了智能体的执行流程和状态管理机制。与传统编程中的线性执行不同，状态图通过**节点（Nodes）和边（Edges）**来构建复杂的执行路径。

状态图的核心要素

状态（State）：贯穿整个执行过程的数据结构
节点（Nodes）：执行单元，包含具体的业务逻辑
边（Edges）：连接节点的路径，定义执行流向
入口点（Entry Point）：图执行的起始位置
出口点（Exit Point）：图执行的结束位置

状态设计原则

良好的状态设计是构建高效状态图的关键：


from typing import TypedDict, Annotated, Sequence
from langgraph.graph import MessagesState

# 推荐的状态设计
class AgentState(TypedDict):
    # 核心消息流
    messages: Annotated[Sequence[dict], add_messages]
    
    # 执行状态
    current_step: str
    is_complete: bool
    
    # 业务数据
    user_input: str
    processing_result: dict
    error_info: str | None
    
    # 元数据
    created_at: str
    last_updated: str

环境准备 / 前置知识

基础安装


pip install -U langgraph langchain langchain-openai

前置要求

理解Python类型注解
掌握基本的面向对象编程
了解异步编程概念
熟悉图论基础

分步实战

步骤 1：创建基础状态图


from langgraph.graph import StateGraph, START, END
from typing import TypedDict, Annotated, Sequence
from langgraph.graph import add_messages

# 定义状态结构
class ChatState(TypedDict):
    messages: Annotated[Sequence[dict], add_messages]
    user_name: str
    session_id: str

# 创建简单的聊天节点
def greeting_node(state: ChatState):
    """问候节点"""
    greeting = f"你好，{state['user_name']}！我是智能助手。"
    return {
        "messages": [{"role": "assistant", "content": greeting}],
        "session_id": f"session_{hash(state['user_name']) % 10000}"
    }

def response_node(state: ChatState):
    """响应节点"""
    last_message = state["messages"][-1]["content"]
    response = f"您说：'{last_message}'，我理解了您的需求。"
    return {"messages": [{"role": "assistant", "content": response}]}

# 构建图结构
graph = StateGraph(ChatState)
graph.add_node("greeting", greeting_node)
graph.add_node("response", response_node)

# 设置边关系
graph.add_edge(START, "greeting")
graph.add_edge("greeting", "response")
graph.add_edge("response", END)

# 编译图
compiled_graph = graph.compile()

# 测试执行
result = compiled_graph.invoke({
    "user_name": "张三",
    "messages": [],
    "session_id": ""
})
print(result["messages"][-1]["content"])

步骤 2：构建复杂业务流程图


from langchain_openai import ChatOpenAI
import time
from datetime import datetime

class OrderState(TypedDict):
    messages: Annotated[Sequence[dict], add_messages]
    order_id: str
    current_step: str
    order_data: dict
    is_complete: bool
    error_message: str | None

# 订单处理节点
def validate_order(state: OrderState):
    """订单验证节点"""
    order_data = state.get("order_data", {})
    
    # 验证订单信息
    if not order_data.get("user_id"):
        return {
            "messages": [{"role": "assistant", "content": "缺少用户ID，订单验证失败"}],
            "current_step": "validation",
            "is_complete": True,
            "error_message": "缺少用户ID"
        }
    
    if not order_data.get("products"):
        return {
            "messages": [{"role": "assistant", "content": "订单中缺少商品信息"}],
            "current_step": "validation",
            "is_complete": True,
            "error_message": "缺少商品信息"
        }
    
    # 生成订单ID
    order_id = f"ORD{int(time.time()) % 1000000:06d}"
    
    return {
        "messages": [{"role": "assistant", "content": f"订单验证通过，订单ID: {order_id}"}],
        "order_id": order_id,
        "current_step": "validated",
        "order_data": order_data,
        "is_complete": False
    }

def process_payment(state: OrderState):
    """支付处理节点"""
    order_id = state["order_id"]
    total_amount = sum(item["price"] * item["quantity"] for item in state["order_data"]["products"])
    
    # 模拟支付处理
    time.sleep(1)
    
    return {
        "messages": [{"role: "assistant", "content": f"订单{order_id}支付成功，金额: ¥{total_amount}"}],
        "current_step": "payment_complete",
        "is_complete": False
    }

def ship_order(state: OrderState):
    """发货节点"""
    order_id = state["order_id"]
    
    # 模拟发货处理
    time.sleep(1)
    
    return {
        "messages": [{"role": "assistant", "content": f"订单{order_id}已发货，预计3-5天送达"}],
        "current_step": "shipped",
        "is_complete": True
    }

def handle_error(state: OrderState):
    """错误处理节点"""
    error_msg = state.get("error_message", "未知错误")
    
    return {
        "messages": [{"role": "assistant", "content": f"处理失败：{error_msg}"}],
        "current_step": "error",
        "is_complete": True
    }

# 构建订单处理图
order_graph = StateGraph(OrderState)
order_graph.add_node("validate_order", validate_order)
order_graph.add_node("process_payment", process_payment)
order_graph.add_node("ship_order", ship_order)
order_graph.add_node("handle_error", handle_error)

# 设置边关系
order_graph.add_edge(START, "validate_order")

# 条件边：根据验证结果决定下一步
order_graph.add_conditional_edges(
    "validate_order",
    lambda state: "handle_error" if state.get("is_complete") else "process_payment",
    {"handle_error": "handle_error", "process_payment": "process_payment"}
)

order_graph.add_edge("process_payment", "ship_order")
order_graph.add_edge("ship_order", END)
order_graph.add_edge("handle_error", END)

# 编译订单处理图
compiled_order_graph = order_graph.compile()

# 测试订单处理流程
test_order = {
    "messages": [],
    "order_id": "",
    "current_step": "start",
    "order_data": {
        "user_id": "user_123",
        "products": [
            {"name": "商品A", "price": 99.99, "quantity": 2},
            {"name": "商品B", "price": 149.99, "quantity": 1}
        ]
    },
    "is_complete": False,
    "error_message": None
}

result = compiled_order_graph.invoke(test_order)
for msg in result["messages"]:
    print(f"{msg['role']}: {msg['content']}")

步骤 3：使用条件边构建智能路由


from langchain_openai import ChatOpenAI

class SupportState(TypedDict):
    messages: Annotated[Sequence[dict], add_messages]
    user_query: str
    query_category: str
    resolved: bool

def classify_query(state: SupportState):
    """查询分类节点"""
    query = state["user_query"]
    
    # 简单的关键词分类
    if "退款" in query or "退货" in query:
        category = "refund"
    elif "技术" in query or "bug" in query:
        category = "technical"
    elif "账户" in query or "登录" in query:
        category = "account"
    else:
        category = "general"
    
    return {
        "messages": [{"role": "assistant", "content": f"您的查询已分类为：{category}"}],
        "query_category": category,
        "resolved": False
    }

def handle_refund(state: SupportState):
    """退款处理节点"""
    return {
        "messages": [{"role": "assistant", "content": "退款处理：请联系客服提供订单号，我们将在3-5个工作日内处理"}],
        "resolved": True
    }

def handle_technical(state: SupportState):
    """技术支持节点"""
    return {
        "messages": [{"role": "assistant", "content": "技术支持：请提供详细的问题描述，我们的技术团队将尽快为您解决"}],
        "resolved": True
    }

def handle_account(state: SupportState):
    """账户问题节点"""
    return {
        "messages": [{"role": "assistant", "content": "账户问题：请检查您的登录信息，或重置密码"}],
        "resolved": True
    }

def handle_general(state: SupportState):
    """一般咨询节点"""
    return {
        "messages": [{"role": "assistant", "content": "一般咨询：感谢您的咨询，我们将为您提供相关信息"}],
        "resolved": True
    }

# 构建客服支持图
support_graph = StateGraph(SupportState)
support_graph.add_node("classify_query", classify_query)
support_graph.add_node("handle_refund", handle_refund)
support_graph.add_node("handle_technical", handle_technical)
support_graph.add_node("handle_account", handle_account)
support_graph.add_node("handle_general", handle_general)

# 设置边关系
support_graph.add_edge(START, "classify_query")

# 条件边：根据查询类别路由到不同处理节点
support_graph.add_conditional_edges(
    "classify_query",
    lambda state: state["query_category"],
    {
        "refund": "handle_refund",
        "technical": "handle_technical", 
        "account": "handle_account",
        "general": "handle_general"
    }
)

# 所有处理节点都结束
for handler in ["handle_refund", "handle_technical", "handle_account", "handle_general"]:
    support_graph.add_edge(handler, END)

compiled_support_graph = support_graph.compile()

# 测试不同类型的查询
test_queries = [
    "我想要退款，订单号是12345",
    "登录不了账户，密码忘记了",
    "程序出现bug，无法正常使用",
    "我想了解一下产品信息"
]

for query in test_queries:
    result = compiled_support_graph.invoke({
        "messages": [],
        "user_query": query,
        "query_category": "",
        "resolved": False
    })
    print(f"用户: {query}")
    print(f"客服: {result['messages'][-1]['content']}")
    print("-" * 50)

步骤 4：循环结构处理


class AnalysisState(TypedDict):
    messages: Annotated[Sequence[dict], add_messages]
    data_points: list
    current_index: int
    is_complete: bool
    analysis_results: dict

def data_collection(state: AnalysisState):
    """数据收集节点"""
    data_points = state["data_points"]
    current_index = state["current_index"]
    
    if current_index >= len(data_points):
        return {
            "messages": [{"role": "assistant", "content": "数据收集完成"}],
            "is_complete": True
        }
    
    # 收集当前数据点
    current_data = data_points[current_index]
    result = f"已收集数据点 {current_index + 1}/{len(data_points)}: {current_data}"
    
    return {
        "messages": [{"role: "assistant", "content": result}],
        "current_index": current_index + 1,
        "is_complete": False
    }

def analysis_node(state: AnalysisState):
    """分析节点"""
    data_points = state["data_points"]
    results = []
    
    for i, data in enumerate(data_points):
        # 模拟数据分析
        analysis = f"数据点{i+1}分析结果：{data * 2}"
        results.append(analysis)
    
    return {
        "messages": [{"role": "assistant", "content": "\n".join(results)}],
        "analysis_results": {"analyses": results},
        "is_complete": True
    }

# 构建带循环的分析图
analysis_graph = StateGraph(AnalysisState)
analysis_graph.add_node("data_collection", data_collection)
analysis_graph.add_node("analysis", analysis_node)

# 设置边关系
analysis_graph.add_edge(START, "data_collection")

# 条件边：根据数据收集状态决定下一步
analysis_graph.add_conditional_edges(
    "data_collection",
    lambda state: "analysis" if state["is_complete"] else "data_collection"
)

analysis_graph.add_edge("analysis", END)

compiled_analysis_graph = analysis_graph.compile()

# 测试循环处理
test_data = [10, 20, 30, 40, 50]
result = compiled_analysis_graph.invoke({
    "messages": [],
    "data_points": test_data,
    "current_index": 0,
    "is_complete": False,
    "analysis_results": {}
})

高级特性

1. 自定义条件函数


def custom_router(state: ChatState):
    """自定义路由函数"""
    last_message = state["messages"][-1]["content"]
    
    if "紧急" in last_message:
        return "urgent_handler"
    elif "查询" in last_message:
        return "query_handler"
    else:
        return "general_handler"

# 使用自定义路由
graph.add_conditional_edges(
    "classifier",
    custom_router,
    {
        "urgent_handler": "urgent_handler",
        "query_handler": "query_handler", 
        "general_handler": "general_handler"
    }
)

2. 动态图修改


def dynamic_graph_builder(user_preferences: dict):
    """根据用户偏好动态构建图"""
    graph = StateGraph(ChatState)
    
    # 根据偏好添加不同节点
    if user_preferences.get("support_mode"):
        graph.add_node("support_handler", support_handler)
        graph.add_edge(START, "support_handler")
    else:
        graph.add_node("general_handler", general_handler)
        graph.add_edge(START, "general_handler")
    
    graph.add_edge("support_handler", END)
    graph.add_edge("general_handler", END)
    
    return graph.compile()

最佳实践与避坑

最佳实践

状态结构简洁：避免过度嵌套的状态结构
节点职责单一：每个节点只做一件事
错误处理完备：为每个关键节点添加错误处理
日志记录：记录关键状态变化便于调试

常见坑点

循环引用：确保没有循环依赖导致死循环
状态冲突：多个节点同时修改同一状态字段
内存泄漏：长时间运行任务未正确清理状态
异步问题：异步节点处理不当导致状态同步问题

性能优化

1. 状态优化


# 避免大对象存储在状态中
class OptimizedState(TypedDict):
    # 只存储必要的核心数据
    messages: Annotated[Sequence[dict], add_messages]
    metadata: dict  # 元数据单独存储
    
    # 大数据存储在外部
    data_ref: str  # 外部存储引用

2. 节点优化


# 使用缓存避免重复计算
from functools import lru_cache

@lru_cache(maxsize=128)
def expensive_calculation(data: str):
    """耗时的计算使用缓存"""
    return complex_calculation(data)

本节小结

通过本节学习，我们深入理解了LangGraph的核心组件——状态图，掌握了StateGraph的创建方法和使用技巧。状态图通过节点和边的组合，可以构建出灵活、高效的智能体执行流程，支持复杂的业务逻辑和条件路由。

下一节我们将探讨LangGraph的另一个核心组件——节点(Node)与边(Edge)的编程方法。

延伸阅读

关键词：状态图, StateGraph, 节点, 边, 条件路由, 智能体执行流程
难度：进阶
预计阅读：30分钟