智能体短期记忆系统设计与Redis实现

1. 短期记忆系统概述

在智能体（Agent）系统中，短期记忆扮演着类似人类工作记忆的角色，负责临时存储和处理最近发生的事件信息。这种记忆机制使得智能体能够保持对话连贯性、跟踪任务进度，并对环境变化做出及时响应。

短期记忆与长期记忆的关键区别在于：

• 时间范围：通常保留几分钟到几小时内的信息
• 容量限制：受存储介质和检索效率约束
• 访问速度：需要毫秒级响应
• 信息粒度：记录原始事件细节而非抽象知识

2. 核心数据结构设计

2.1 事件基础模型

python复制class MemoryEvent:
    def __init__(self, 
                 timestamp: float,
                 session_id: str,
                 event_type: str,
                 content: str,
                 importance: float = 0.5,
                 tags: list = None):
        self.timestamp = timestamp  # 事件发生时间戳
        self.session_id = session_id  # 所属会话ID
        self.event_type = event_type  # 事件类型标识
        self.content = content  # 事件内容主体
        self.importance = importance  # 重要性评分(0-1)
        self.tags = tags or []  # 分类标签
        self.access_count = 0  # 访问计数

2.2 时间序列存储方案

Redis实现方案

python复制import redis
import json
from datetime import datetime

class RedisTimeSeriesMemory:
    def __init__(self, host='localhost', port=6379):
        self.conn = redis.Redis(host=host, port=port)
        self.event_key_prefix = "mem:event:"
        self.session_key_prefix = "mem:session:"
    
    def add_event(self, event):
        """存储事件到时间序列"""
        event_dict = {
            'timestamp': event.timestamp,
            'session_id': event.session_id,
            'event_type': event.event_type,
            'content': event.content,
            'importance': event.importance,
            'tags': json.dumps(event.tags)
        }
        
        # 使用Redis Stream存储
        self.conn.xadd(
            f"{self.session_key_prefix}{event.session_id}",
            event_dict,
            id=f"{int(event.timestamp*1000)}-0"
        )
        
        # 更新会话索引
        self.conn.zadd(
            "active_sessions",
            {event.session_id: event.timestamp}
        )

3. 关键功能实现

3.1 最近事件检索

python复制def get_recent_events(self, session_id, count=5):
    """获取指定会话的最近事件"""
    events = self.conn.xrevrange(
        f"{self.session_key_prefix}{session_id}",
        count=count
    )
    
    return [self._parse_event(e[1]) for e in events]

def _parse_event(self, event_data):
    """转换原始数据为MemoryEvent对象"""
    return MemoryEvent(
        timestamp=float(event_data[b'timestamp']),
        session_id=event_data[b'session_id'].decode(),
        event_type=event_data[b'event_type'].decode(),
        content=event_data[b'content'].decode(),
        importance=float(event_data[b'importance']),
        tags=json.loads(event_data[b'tags'].decode())
    )

3.2 时间窗口查询

python复制def get_events_by_time_range(self, session_id, start_time, end_time):
    """获取时间范围内的事件"""
    start_id = f"{int(start_time*1000)}-0"
    end_id = f"{int(end_time*1000)}-0"
    
    events = self.conn.xrange(
        f"{self.session_key_prefix}{session_id}",
        min=start_id,
        max=end_id
    )
    
    return [self._parse_event(e[1]) for e in events]

4. 性能优化策略

4.1 内存缓存热点数据

python复制from functools import lru_cache

class CachedMemory(RedisTimeSeriesMemory):
    @lru_cache(maxsize=1024)
    def get_recent_events(self, session_id, count=5):
        return super().get_recent_events(session_id, count)
    
    @lru_cache(maxsize=512)
    def get_events_by_time_range(self, session_id, start_time, end_time):
        return super().get_events_by_time_range(session_id, start_time, end_time)

4.2 数据分片策略

python复制def _get_shard_key(self, session_id):
    """根据会话ID计算数据分片"""
    hash_val = hash(session_id) % 16
    return f"shard_{hash_val}:{self.session_key_prefix}{session_id}"

5. 实际应用案例

5.1 对话系统上下文保持

python复制class ConversationManager:
    def __init__(self):
        self.memory = CachedMemory()
    
    def process_message(self, session_id, message):
        # 存储用户输入
        user_event = MemoryEvent(
            timestamp=time.time(),
            session_id=session_id,
            event_type="user_message",
            content=message,
            importance=0.7
        )
        self.memory.add_event(user_event)
        
        # 获取最近5条对话上下文
        context = self.memory.get_recent_events(session_id, count=5)
        
        # 生成响应逻辑...

6. 常见问题解决方案

6.1 内存溢出处理

python复制def clean_old_sessions(self, max_age_hours=24):
    """清理过期会话数据"""
    cutoff = time.time() - max_age_hours * 3600
    old_sessions = self.conn.zrangebyscore(
        "active_sessions", 0, cutoff
    )
    
    for session_id in old_sessions:
        self.conn.delete(f"{self.session_key_prefix}{session_id}")
        self.conn.zrem("active_sessions", session_id)

6.2 数据一致性保障

python复制def add_event_with_backup(self, event):
    """带事务处理的事件存储"""
    pipe = self.conn.pipeline()
    try:
        # 主存储
        pipe.xadd(
            f"{self.session_key_prefix}{event.session_id}",
            self._serialize_event(event)
        )
        
        # 备份存储
        pipe.zadd(
            "event_backup",
            {json.dumps(self._serialize_event(event)): event.timestamp}
        )
        
        pipe.execute()
    except redis.RedisError:
        pipe.discard()
        raise

7. 系统监控指标

python复制def get_memory_stats(self):
    """获取存储系统状态"""
    return {
        'total_sessions': self.conn.zcard("active_sessions"),
        'oldest_session': self.conn.zrange("active_sessions", 0, 0, withscores=True),
        'memory_usage': self.conn.info('memory')['used_memory_human']
    }

8. 扩展接口设计

8.1 Web服务接口

python复制from flask import Flask, request

app = Flask(__name__)
memory = CachedMemory()

@app.route('/events', methods=['POST'])
def add_event():
    data = request.json
    event = MemoryEvent(**data)
    memory.add_event(event)
    return {'status': 'success'}

@app.route('/events/recent')
def get_recent():
    session_id = request.args.get('session_id')
    count = int(request.args.get('count', 5))
    events = memory.get_recent_events(session_id, count)
    return {'events': [e.__dict__ for e in events]}

9. 测试验证方案

9.1 性能基准测试

python复制def test_performance():
    """写入和读取性能测试"""
    mem = RedisTimeSeriesMemory()
    start = time.time()
    
    # 写入测试
    for i in range(1000):
        event = MemoryEvent(
            timestamp=time.time(),
            session_id="perf_test",
            event_type="test",
            content=f"message {i}"
        )
        mem.add_event(event)
    
    # 读取测试
    for _ in range(100):
        mem.get_recent_events("perf_test", 10)
    
    duration = time.time() - start
    print(f"处理1000次写入+100次读取耗时: {duration:.2f}秒")

10. 部署架构建议

10.1 生产环境配置

yaml复制# docker-compose.yml示例
version: '3'
services:
  redis:
    image: redis:6
    ports:
      - "6379:6379"
    volumes:
      - redis_data:/data
    command: redis-server --save 60 1 --loglevel warning

  memory_service:
    build: .
    ports:
      - "5000:5000"
    environment:
      REDIS_HOST: redis
    depends_on:
      - redis

volumes:
  redis_data: