Agent 状态持久化方案：Redis、数据库和文件存储的选型对比

baronbool

895人浏览 · 2026-07-17 14:14:34

baronbool · 2026-07-17 14:14:34 发布

Agent 状态持久化方案：Redis、数据库和文件存储的选型对比

一、为什么你的 Agent 会话总是"失忆"

Agent 执行多步任务时，最大的坑不是模型能力不足，而是状态丢失。

比如一个数据分析 Agent 需要先查库、再做聚合、最后生成报告。如果每一步都要重新推理上下文，响应延迟会从 3 秒飙到 15 秒，Token 消耗直接翻倍。

更糟的是，用户中途关掉页面再打开，Agent 可能完全不记得之前聊了什么。

这些问题都指向同一个根因：缺乏可靠的状态持久化机制。

Agent 的状态通常包含：对话历史、任务进度、中间推理结果、工具调用栈。这些数据的读写模式各不相同，所以不存在"一把梭"的存储方案。

二、三种方案的底层原理与数据流

先看一张典型的多步 Agent 执行流程中，状态读写发生的时机：

三种存储方案在读写模式上的核心差异：

特性	Redis	PostgreSQL	文件存储
读延迟	<1ms	1-5ms	5-50ms
写延迟	<1ms	1-10ms	10-100ms
数据结构	KV/Hash/List	关系表	文件/对象
持久化	可配置	强持久化	强持久化
查询能力	Key 精确匹配	SQL 查询	路径匹配

三、生产级分层存储实现

以下是 Go 语言实现的分层状态管理器：

package agent

import (
    "context"
    "encoding/json"
    "fmt"
    "time"

    "github.com/go-redis/redis/v8"
    "gorm.io/gorm"
)

// AgentState Agent 执行状态
type AgentState struct {
    SessionID  string            `json:"session_id" gorm:"primaryKey"`
    TaskID     string            `json:"task_id" gorm:"index"`
    StepIndex  int               `json:"step_index"`
    Context    json.RawMessage   `json:"context" gorm:"type:jsonb"`     // 当前推理上下文
    ToolStack  []ToolCallRecord  `json:"tool_stack" gorm:"type:jsonb"` // 工具调用栈
    Status     string            `json:"status"`                         // running/completed/failed
    CreatedAt  time.Time         `json:"created_at"`
    UpdatedAt  time.Time         `json:"updated_at"`
}

// ToolCallRecord 工具调用记录
type ToolCallRecord struct {
    ToolName   string          `json:"tool_name"`
    Arguments  json.RawMessage `json:"arguments"`
    Result     json.RawMessage `json:"result,omitempty"`
    CalledAt   time.Time       `json:"called_at"`
}

// StateManager 分层状态管理器
type StateManager struct {
    redis *redis.Client   // 热状态缓存层
    db    *gorm.DB        // 持久化层
}

// GetState 先查 Redis 缓存，miss 则回源数据库
func (sm *StateManager) GetState(ctx context.Context, sessionID string) (*AgentState, error) {
    // 第一层：尝试从 Redis 读取热状态
    key := fmt.Sprintf("agent:state:%s", sessionID)
    data, err := sm.redis.Get(ctx, key).Bytes()
    if err == nil {
        var state AgentState
        if err := json.Unmarshal(data, &state); err != nil {
            return nil, fmt.Errorf("反序列化 Redis 缓存失败: %w", err)
        }
        return &state, nil
    }

    // 第二层：Redis miss，回源数据库
    var state AgentState
    if err := sm.db.WithContext(ctx).Where("session_id = ?", sessionID).
        First(&state).Error; err != nil {
        if err == gorm.ErrRecordNotFound {
            return nil, fmt.Errorf("会话 %s 不存在", sessionID)
        }
        return nil, fmt.Errorf("查询数据库失败: %w", err)
    }

    // 回填 Redis 缓存，过期时间根据活跃度设定
    cached, _ := json.Marshal(state)
    sm.redis.Set(ctx, key, cached, 30*time.Minute)

    return &state, nil
}

// SaveState 写入数据库并更新 Redis 缓存
func (sm *StateManager) SaveState(ctx context.Context, state *AgentState) error {
    state.UpdatedAt = time.Now()

    // 先写数据库（强持久化）
    if err := sm.db.WithContext(ctx).Save(state).Error; err != nil {
        return fmt.Errorf("保存状态到数据库失败: %w", err)
    }

    // 再更新 Redis 缓存（允许失败，缓存是优化而非必需）
    key := fmt.Sprintf("agent:state:%s", state.SessionID)
    cached, _ := json.Marshal(state)
    sm.redis.Set(ctx, key, cached, 30*time.Minute)

    return nil
}

// ArchiveState 归档已完成会话——迁移到冷存储
func (sm *StateManager) ArchiveState(ctx context.Context, sessionID string) error {
    // 归档时删除 Redis 缓存，释放内存
    key := fmt.Sprintf("agent:state:%s", sessionID)
    sm.redis.Del(ctx, key)

    // 更新数据库状态标记为已归档
    return sm.db.WithContext(ctx).
        Model(&AgentState{}).
        Where("session_id = ?", sessionID).
        Update("status", "archived").Error
}

使用时只需组合这三种存储能力：