Agent 数据存储实战：从短暂存储到长储存的平滑迁移方案

1次阅读

共计 1745 个字符，预计需要花费 5 分钟才能阅读完成。

最近在开发一个任务调度 Agent 时，遇到了数据丢失的坑：内存中的任务状态在程序重启后全部消失。这才意识到，不同的存储方案对 Agent 的可靠性影响巨大。本文会从实际案例出发，带你理解从短暂存储到持久化存储的完整升级路径。

先看三种常见存储方式的特性对比：

内存字典
```
cache = {'task1': 'running'}
```
优点：纳秒级读写，零依赖
致命伤：程序退出即消失

Redis

import redis
r = redis.Redis(host='localhost', port=6379)
r.set('task1', 'running')

优点：微秒级响应，支持 TTL 自动过期
注意点：需要维护连接池

数据库 (MySQL/PostgreSQL)

import sqlite3
conn = sqlite3.connect('agent.db')
conn.execute('INSERT INTO tasks VALUES (?,?)', ('task1', 'running'))

优点：数据永不丢失
代价：毫秒级延迟

推荐分阶段使用存储策略，这是我们的 Python 实现模板：

class HybridStorage:
    def __init__(self):
        self.memory_cache = {}
        self.redis_client = redis.Redis()
        self.db_conn = sqlite3.connect('agent.db')

    def save(self, key, value, persist_level=1):
        """
        :param persist_level: 
            1- 仅内存  2- 内存 +Redis  3- 全量存储
        """
        self.memory_cache[key] = value

        if persist_level >= 2:
            self.redis_client.set(key, value)

        if persist_level == 3:
            self.db_conn.execute("INSERT OR REPLACE INTO tasks VALUES (?,?)", 
                (key, value)
            )

使用时根据数据重要性选择级别：

临时计算中间结果用 level1
需要跨进程共享选 level2
关键业务数据必须用 level3

避免每次操作都新建连接（错误示范）：

# 错误写法：每次创建新连接
def get_from_redis(key):
    r = redis.Redis()  # 新建连接
    return r.get(key)

应该复用连接池：

# 正确写法
pool = redis.ConnectionPool(max_connections=10)

def get_from_redis(key):
    r = redis.Redis(connection_pool=pool)
    return r.get(key)

直接存储 Python 对象会报错：

# 错误示例
r.set('task', {'status': 'running'})  # TypeError!

需要先序列化：

import pickle

# 序列化存储
r.set('task', pickle.dumps({'status': 'running'}))

# 反序列化读取
data = pickle.loads(r.get('task'))

问题现象	可能原因	解决方案
随机读取变慢	Redis 内存不足触发淘汰	增加 maxmemory 或优化数据过期策略
数据库 CPU 飙升	缺少索引或大事务	EXPLAIN 分析慢查询，添加适当索引
内存持续增长	缓存未设置 TTL	对非核心数据设置过期时间