Agent对话窗口上下文不够？从架构设计到代码实现的全面解决方案

1次阅读

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

在构建 Agent 对话系统时，上下文窗口的大小直接影响对话的连贯性和智能程度。传统的全量存储方式虽然简单，但随着对话轮次的增加，内存占用会线性增长，最终导致系统响应变慢甚至崩溃。尤其是在处理长对话场景时，这个问题尤为突出。

优点：实现简单，对话上下文完整
缺点：内存占用大，无法处理长对话

优点：内存占用可控
缺点：可能丢失重要上下文，导致对话不连贯

优点：内存占用可控，保持对话连贯性
缺点：实现复杂度较高

class ChunkedMemory:
    def __init__(self, chunk_size=5):
        """
        初始化分块内存
        :param chunk_size: 每个块存储的对话轮次数
        """
        self.chunk_size = chunk_size
        self.memory = []  # 存储所有块
        self.current_chunk = []  # 当前活跃块

    def add_message(self, role, content):
        """添加新消息到内存"""
        message = {'role': role, 'content': content}
        self.current_chunk.append(message)

        # 当当前块满时，存入内存并创建新块
        if len(self.current_chunk) >= self.chunk_size:
            self._compress_and_store()

    def _compress_and_store(self):
        """压缩当前块并存储"""
        if not self.current_chunk:
            return

        # 简单压缩：保留最后一条用户消息和 AI 回复
        compressed = [self.current_chunk[-2], self.current_chunk[-1]] if len(self.current_chunk) > 1 else [self.current_chunk[-1]]
        self.memory.append(compressed)
        self.current_chunk = []

    def get_context(self):
        """获取完整上下文"""
        context = []
        for chunk in self.memory:
            context.extend(chunk)
        context.extend(self.current_chunk)
        return context

from sentence_transformers import SentenceTransformer
import numpy as np

class SemanticCompressor:
    def __init__(self):
        self.model = SentenceTransformer('paraphrase-MiniLM-L6-v2')

    def compress(self, messages):
        """基于语义相似度压缩消息"""
        if len(messages) <= 2:
            return messages

        # 计算所有消息的嵌入向量
        contents = [msg['content'] for msg in messages]
        embeddings = self.model.encode(contents)

        # 计算相似度矩阵
        sim_matrix = np.zeros((len(messages), len(messages)))
        for i in range(len(messages)):
            for j in range(i+1, len(messages)):
                sim = np.dot(embeddings[i], embeddings[j]) / (np.linalg.norm(embeddings[i]) * np.linalg.norm(embeddings[j]))
                sim_matrix[i][j] = sim

        # 保留最不相似的几条消息
        avg_sims = np.mean(sim_matrix, axis=1)
        top_indices = np.argsort(avg_sims)[:2]  # 保留 2 条最不相似的消息

        return [messages[i] for i in sorted(top_indices)]

我们在不同对话长度下测试了三种方案的内存占用和响应时间：