Agent 分层记忆架构实战：解决长对话上下文管理的性能瓶颈

1次阅读

没有评论

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

在大模型应用中，长对话场景下的上下文管理一直是个棘手的问题。随着对话轮次的增加，传统的全量上下文记忆方式会带来两个主要问题：

内存爆炸：假设每轮对话平均占用 2KB 内存，500 轮对话就会占用 1MB。对于高并发的服务来说，这会迅速耗尽服务器内存。
检索效率低下 ：全量上下文会导致 attention 计算复杂度呈 O(n²) 增长。实验数据显示，当上下文长度超过 2048 tokens 时，推理延迟会增加 300% 以上。

我们采用三层记忆结构：

短期记忆 (Short-term Memory)：保存最近 5 -10 轮对话，使用环形缓冲区(Circular Buffer) 实现，读写复杂度 O(1)
长期记忆(Long-term Memory)：存储压缩后的关键信息，采用时间序列数据库结构，支持按时间范围检索
元记忆(Meta Memory)：记录对话主题、用户偏好等元信息，使用键值对存储

短期记忆：collections.deque实现环形缓冲区，自动淘汰最早记录
长期记忆：pandas.DataFrame结构，包含 timestamp/keyword/embedding 三列
元记忆：Python 标准字典，配合 json 序列化持久化

采用基于 TF-IDF 的关键信息提取策略：

对每轮对话进行分词和词频统计
计算对话轮次间的词频变化率
保留变化率超过阈值的词条及其上下文

from collections import deque
from typing import Dict, List, Optional
import pandas as pd

class HierarchicalMemory:
    """
    分层记忆系统实现
    Attributes:
        short_term: deque 短期记忆队列
        long_term: pd.DataFrame 长期记忆存储
        meta: dict 元信息存储
        max_short_term: int 短期记忆容量
    """

    def __init__(self, max_short_term: int = 10):
        self.short_term = deque(maxlen=max_short_term)
        self.long_term = pd.DataFrame(columns=['timestamp', 'content', 'keywords'])
        self.meta = {'topics': set(), 'preferences': dict()}

    def add_short_term(self, utterance: str):
        """添加短期记忆"""
        self.short_term.append({'timestamp': pd.Timestamp.now(),
            'content': utterance
        })

    def compress_to_long_term(self):
        """压缩短期记忆到长期存储"""
        if not self.short_term:
            return

        # 关键信息提取逻辑
        recent = ''.join([item['content'] for item in self.short_term])
        keywords = self._extract_keywords(recent)

        self.long_term = pd.concat([
            self.long_term,
            pd.DataFrame([{'timestamp': pd.Timestamp.now(),
                'content': recent[-500:],  # 截断保留尾部
                'keywords': ','.join(keywords)
            }])
        ], ignore_index=True)