LangGraph 状态机：复杂 Agent 任务流程管理实战

3. 状态持久化

为了保证系统的可靠性，特别是在分布式环境或长时间运行的任务中，我们需要持久化状态信息。LangGraph 内置了 Checkpointer 机制，也可以结合 Redis 等外部存储实现自定义持久化。

import redis
import json

class StateManager:
    def __init__(self):
        self.redis_client = redis.Redis(host='localhost', port=6379, db=0)
        
    def save_state(self, session_id: str, state: dict):
        """保存状态到 Redis"""
        self.redis_client.set(
            f"shopping_state:{session_id}",
            json.dumps(state),
            ex=3600  # 1 小时过期
        )
    
    def load_state(self, session_id: str) -> dict:
        """从 Redis 加载状态"""
        state_data = self.redis_client.get(f"shopping_state:{session_id}")
        return json.loads(state_data) if state_data else None

4. 错误恢复机制

任何步骤都可能出错，我们需要优雅地处理这些情况，包括重试、回滚和最终错误处理。

import asyncio

class ErrorHandler:
    def __init__(self):
        self.max_retries = 3
        
    async def with_retry(self, func, state: dict):
        """带重试机制的函数执行"""
        retries = 0
        while retries < self.max_retries:
            try:
                return await func(state)
            except Exception as e:
                retries += 1
                if retries == self.max_retries:
                    return self.handle_final_error(e, state)
                await self.handle_retry(e, state, retries)
    
    def handle_final_error(self, error, state: dict):
        """处理最终错误"""
        state["error"] = str(error)
        return self.rollback_to_last_stable_state(state)

实战案例：智能客服系统

让我们看一个实际的例子 - 智能客服系统。该系统需要维护对话历史、识别用户意图并确认问题解决状态。

from langgraph.graph import StateGraph
from typing import TypedDict, List

class CustomerServiceState(TypedDict):
    conversation_history: List[str]
    current_intent: str
    user_info: dict
    resolved: bool

class CustomerServiceGraph(StateGraph):
    def __init__(self):
        super().__init__(CustomerServiceState)
        
        # 初始化状态节点
        self.add_node("greeting", self.greet_customer)
        self.add_node("understand_intent", self.analyze_intent)
        self.add_node("handle_query", self.process_query)
        self.add_node("confirm_resolution", self.check_resolution)
        
        # 定义流程
        self.add_edge("greeting", "understand_intent")
        self.add_edge("understand_intent", "handle_query")
        self.add_edge("handle_query", "confirm_resolution")
        self.add_conditional_edges(
            "confirm_resolution",
            lambda s: "end" if s["resolved"] else "handle_query"
        )
        
    async def greet_customer(self, state: State):
        """欢迎客户"""
        response = await self.llm.generate(
            prompt=f"""
            历史对话：{state['conversation_history']}
            任务：生成合适的欢迎语
            要求：
            1. 保持专业友好
            2. 如果是老客户，表示认出了他们
            3. 询问如何帮助
            """
        )
        state['conversation_history'].append(f"Assistant: {response}")
        return state
        
    async def analyze_intent(self, state: State):
        """理解用户意图"""
        response = await self.llm.generate(
            prompt=f"""
            历史对话：{state['conversation_history']}
            任务：分析用户意图
            输出格式：
            {{
                "intent": "退款/咨询/投诉/其他",
                "confidence": 0.95,
                "details": "具体描述"
            }}
            """
        )
        state['current_intent'] = json.loads(response)
        return state

使用方法

# 初始化系统
graph = CustomerServiceGraph().compile()
state_manager = StateManager()
error_handler = ErrorHandler()

async def handle_customer_query(user_id: str, message: str):
    # 加载或创建状态
    state = state_manager.load_state(user_id) or {
        "conversation_history": [],
        "current_intent": None,
        "user_info": {},
        "resolved": False
    }
    
    # 添加用户消息
    state["conversation_history"].append(f"User: {message}")
    
    # 执行状态机流程
    try:
        result = await graph.ainvoke(state)
        # 保存状态
        state_manager.save_state(user_id, result)
        return result["conversation_history"][-1]
    except Exception as e:
        return await error_handler.with_retry(
            graph.ainvoke,
            state
        )

最佳实践

状态设计原则

• 保持简单：状态结构应清晰明了，避免过度嵌套。
• 最小化存储：只存储必要的信息，减少序列化开销。
• 可序列化：确保状态对象可以被 JSON 或其他序列化格式安全存储。

转换逻辑优化

• 条件转换：利用 add_conditional_edges 实现动态路由。
• 避免死循环：设置最大迭代次数或超时机制。
• 明确入口与出口：始终定义清晰的 set_entry_point 和结束节点。

错误处理策略

• 优雅降级：当 LLM 调用失败时，提供默认回复或人工介入通道。
• 详细日志：记录每一步的状态变更和异常堆栈。
• 回滚机制：在严重错误发生时，恢复到上一个稳定状态。

性能优化

• 异步操作：充分利用 await 和 ainvoke 提高并发处理能力。
• 状态缓存：对于频繁读取的状态，考虑本地缓存策略。
• 控制大小：限制对话历史长度，防止上下文窗口溢出。

常见陷阱和解决方案

状态爆炸

• 问题：状态数量过多导致维护困难，图结构过于复杂。
• 解决：合并相似状态，使用状态组合而不是创建新状态；引入分层状态机。

死锁情况

• 问题：状态转换循环导致任务卡住，无法到达结束节点。
• 解决：添加超时机制和强制退出条件；在代码层面增加计数器限制循环次数。

状态一致性

• 问题：分布式环境下，多个实例可能读取到不一致的状态快照。
• 解决：使用分布式锁和事务机制；利用 LangGraph 内置的 Checkpointer 保证原子性。

总结

LangGraph 状态机为复杂 AI Agent 任务流程管理提供了一个强大的解决方案。通过清晰的任务流程管理、可靠的状态持久化、完善的错误处理以及灵活的扩展能力，开发者可以构建更加健壮和智能的应用程序。在实际项目中，建议根据业务需求合理设计状态图，并重视异常处理和性能优化，以确保系统的稳定性和用户体验。