你的 AI 智能体到底需要多少记忆空间

+-------------------------------------------------------------------------+
|                           用户输入 / 任务查询                           |
+-------------------------------------------------------------------------+
                                     |
                                     v
+-------------------------------------------------------------------------+
| 1. 短期记忆 (工作内存 / 滑动窗口)                                       |
|    - 存放当前对话、系统提示词及最近几轮交互。                           |
+-------------------------------------------------------------------------+
                                     |
                                     v
+-------------------------------------------------------------------------+
| 2. 情境记忆 (压缩/摘要历史)                                             |
|    - 提炼过去的交互步骤，保留长期任务的状态与执行路径。                 |
+-------------------------------------------------------------------------+
                                     |
                                     v
+-------------------------------------------------------------------------+
| 3. 长期记忆 (向量数据库 / RAG / 语义内存)                               |
|    - 检索历史事实、用户偏好及外部知识库。                               |
+-------------------------------------------------------------------------+

import tiktoken

class DynamicMemoryController:
    def __init__(self, model_name="gpt-4o", max_short_term_tokens=4000, compression_threshold=0.8):
        self.encoder = tiktoken.encoding_for_model(model_name)
        self.max_short_term_tokens = max_short_term_tokens
        self.compression_threshold = int(max_short_term_tokens * compression_threshold)
        self.short_term_buffer = []
        self.episodic_archive = []

    def _calculate_tokens(self, text: str) -> int:
        return len(self.encoder.encode(text))

    def add_message(self, role: str, content: str):
        token_count = self._calculate_tokens(content)
        self.short_term_buffer.append({
            "role": role,
            "content": content,
            "tokens": token_count
        })
        self._manage_memory()

    def _manage_memory(self):
        total_tokens = sum(msg["tokens"] for msg in self.short_term_buffer)

        # 当 Token 总数超过压缩阈值时，将较旧的消息移入情境归档区
        if total_tokens > self.compression_threshold:
            print(f"[内存控制器] 触发阈值限制（当前 {total_tokens} 词元）。启动内存压缩...")
            while total_tokens > self.max_short_term_tokens * 0.5 and len(self.short_term_buffer) > 2:
                # 确保不删除索引为 0 的系统提示词 (System Prompt)
                target_index = 1 if self.short_term_buffer[0]["role"] == "system" else 0
                removed_message = self.short_term_buffer.pop(target_index)
                self.episodic_archive.append(removed_message)
                total_tokens = sum(msg["tokens"] for msg in self.short_term_buffer)

            self._trigger_episodic_summarization()

    def _trigger_episodic_summarization(self):
        # 在实际生产环境中，我们会将此处的归档内容发送给一个高性价比的模型
        # 例如通过 n1n.ai 路由到 DeepSeek-V3 进行后台摘要生成，并更新系统提示词
        archive_content = "\n".join([f"{m['role']}: {m['content']}" for m in self.episodic_archive])
        print(f"[情境归档器] 已将 {len(self.episodic_archive)} 条消息移交后台进行摘要压缩。")
        self.episodic_archive.clear()

    def get_active_context(self):
        return [{"role": m["role"], "content": m["content"]} for m in self.short_term_buffer]

# 示例运行
controller = DynamicMemoryController(max_short_term_tokens=1000)
controller.add_message("system", "你是一个高效的编程助手。")
for i in range(5):
    controller.add_message("user", f"这是第 {i} 条消息，包含一些测试数据，用于填充内存缓冲区。")
    controller.add_message("assistant", f"已收到第 {i} 条消息。")

print(f"当前缓冲区中的活动消息数: {len(controller.get_active_context())}")