LangGraph与MySQL持久化存储实践-AI智能范式网

LangGraph与MySQL持久化存储实践

LangGraphMySQL持久化存储

|难度：普通|阅读约 16 分钟

1. LangGraph与MySQL持久化存储实践

作为一名长期从事AI应用开发的工程师，我最近在项目中遇到了一个典型问题：如何让LangGraph会话保持持久化状态，同时区分短期记忆和长期记忆。经过多次尝试，最终选择了MySQL作为存储方案。这个方案不仅稳定可靠，还能很好地支持高并发场景。

在实际开发中，会话状态的持久化是个关键需求。想象一下，当用户与AI系统进行多轮对话时，如果每次都要从头开始，体验会非常糟糕。而MySQL作为成熟的关系型数据库，提供了事务支持、索引优化等特性，非常适合这种需要频繁读写但又要求数据一致性的场景。

需要模型API调用？免费领10W Token，多模型网关一键接入 Claude、DeepSeek 等主流模型。

2. 环境准备与依赖安装

2.1 依赖包选择与安装

在开始之前，我们需要准备以下关键依赖包：

bash复制uv add langgraph-checkpoint-mysql pymysql aiomysql asyncmy cryptography

这个组合的选择经过了仔细考量：

langgraph-checkpoint-mysql：官方提供的MySQL检查点插件
pymysql：纯Python实现的MySQL客户端
aiomysql：异步IO版的MySQL客户端，适合高并发场景
asyncmy：另一个高性能异步MySQL驱动
cryptography：用于安全连接的必要加密库

提示：在实际部署环境中，建议固定这些依赖的版本号，避免因版本更新导致兼容性问题。我们团队就曾因为自动更新依赖而浪费了半天时间排查问题。

2.2 数据库配置

在MySQL中需要预先创建好数据库和表结构。以下是推荐的表设计：

sql复制CREATE DATABASE langgraph_sessions CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;

CREATE TABLE session_states (
    session_id VARCHAR(255) PRIMARY KEY,
    short_term_memory JSON NOT NULL,
    long_term_memory JSON,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
    metadata JSON
) ENGINE=InnoDB;

这个设计考虑了：

使用utf8mb4字符集支持完整的Unicode
为session_id建立主键索引
使用JSON类型存储灵活的记忆数据
自动维护创建和更新时间戳
额外的metadata字段用于扩展

3. 核心实现解析

3.1 状态管理架构设计

LangGraph的状态管理分为三个层次：

会话状态(AgentState)：包含当前对话的完整上下文
短期记忆：保存最近几轮的对话内容（通常3-5轮）
长期记忆：存储关键对话节点和摘要信息

python复制from typing import TypedDict, List
from langchain_core.messages import BaseMessage

class AgentState(TypedDict):
    messages: List[BaseMessage]
    short_term_memory: List[BaseMessage]
    long_term_memory: dict
    turn_count: int

3.2 MySQL检查点实现

检查点(Checkpointer)是持久化的核心组件，负责将会话状态保存到MySQL：

python复制from langgraph.checkpoint.mysql import MySQLCheckpointer
from langchain_core.runnables import RunnableConfig

checkpointer = MySQLCheckpointer(
    host="localhost",
    port=3306,
    user="langgraph",
    password="securepassword",
    database="langgraph_sessions",
    table_name="session_states",
    ssl_disabled=False
)

async def save_state(session_id: str, state: AgentState):
    config = RunnableConfig(configurable={"thread_id": session_id})
    await checkpointer.aput(config, state)

这个实现有几个关键点：

使用SSL加密连接确保数据安全
每个会话通过唯一的session_id标识
异步操作(aput)避免阻塞主线程

3.3 记忆管理策略

短期记忆处理最近的消息，采用滑动窗口策略：

python复制def update_short_term_memory(state: AgentState, new_message: BaseMessage) -> AgentState:
    max_short_term = 5  # 保留最近5条消息
    updated = state["short_term_memory"][-max_short_term + 1 :] + [new_message]
    return {**state, "short_term_memory": updated}

长期记忆则采用摘要和关键信息提取：

python复制from langchain_core.language_models import BaseLanguageModel

async def update_long_term_memory(
    llm: BaseLanguageModel, 
    state: AgentState
) -> AgentState:
    conversation = "\n".join([m.content for m in state["messages"]])
    prompt = f"""请从以下对话中提取关键信息作为长期记忆：
    {conversation}
    
    输出格式为JSON，包含：人物、地点、时间、重要事件等关键字段"""
    
    extracted = await llm.ainvoke(prompt)
    return {**state, "long_term_memory": extracted}

4. 完整工作流实现

4.1 初始化代理

python复制from langchain.agents import create_agent
from langchain.tools import Tool

def create_calculator_tool():
    return Tool(
        name="calculator",
        func=lambda x: str(eval(x)),
        description="用于数学表达式计算"
    )

agent = create_agent(
    llm=llm,
    tools=[create_calculator_tool()],
    checkpointer=checkpointer,
    interrupt_before=["calculator"]
)

4.2 会话处理循环

python复制async def handle_conversation(session_id: str, user_input: str):
    # 加载现有状态
    config = RunnableConfig(configurable={"thread_id": session_id})
    state = await checkpointer.aget(config) or {
        "messages": [],
        "short_term_memory": [],
        "long_term_memory": {},
        "turn_count": 0
    }
    
    # 更新状态
    state["turn_count"] += 1
    state["messages"].append(HumanMessage(content=user_input))
    
    # 执行代理
    result = await agent.ainvoke(state, config)
    
    # 更新记忆
    state = update_short_term_memory(state, result["messages"][-1])
    if state["turn_count"] % 5 == 0:  # 每5轮更新长期记忆
        state = await update_long_term_memory(llm, state)
    
    # 保存状态
    await checkpointer.aput(config, state)
    return result["messages"][-1].content

5. 性能优化与问题排查

5.1 连接池配置

高并发场景下，必须配置适当的连接池：

python复制from sqlalchemy.pool import QueuePool

checkpointer = MySQLCheckpointer(
    # ...其他参数...
    poolclass=QueuePool,
    pool_size=10,
    max_overflow=20,
    pool_timeout=30
)

5.2 常见错误与解决方案

错误现象	可能原因	解决方案
连接超时	网络问题或连接泄漏	检查连接池配置，添加连接健康检查
JSON解析失败	数据格式损坏	添加数据验证层，使用try-catch处理
性能下降	未使用索引或大事务	为session_id添加索引，分批提交事务
内存泄漏	未及时释放资源	确保所有连接使用with语句或显式关闭

5.3 监控指标建议

在生产环境中应该监控以下关键指标：

平均查询延迟（应<100ms）
连接池使用率（应<80%）
检查点操作成功率（应>99.9%）
会话状态大小（警惕异常增长）

6. 实际应用中的经验分享

在三个月的生产环境运行中，我们总结了以下宝贵经验：

批量操作优化：当需要处理大量会话状态时，使用executemany比单条插入快10倍以上。我们实现了一个批量检查点写入器，将写入吞吐量从500 QPS提升到了5000+ QPS。
记忆压缩技巧：发现某些会话的长期记忆会无限制增长，后来添加了以下压缩逻辑：

python复制def compress_memory(memory: dict) -> dict:
    if len(json.dumps(memory)) > 1024:  # 超过1KB时压缩
        return {k: v for k, v in memory.items() if k in KEY_FIELDS}
    return memory

故障恢复策略：实现了一个状态修复工具，当检测到损坏的会话状态时，会自动从最近的检查点恢复，并保留原始数据供分析。这个工具帮助我们快速解决了多次生产问题。
测试建议：一定要模拟网络分区和数据库故障场景。我们使用Chaos Engineering工具故意中断MySQL连接，发现了多个边缘情况下的bug。

这个MySQL持久化方案已经稳定运行了半年多，支持了日均百万级的会话处理。最关键的收获是：在AI系统中，状态管理不能只考虑功能实现，还必须重视性能、可靠性和可观测性。