AI Agent记忆系统设计：短期记忆与长期记忆的实现

AI Agent记忆系统设计：短期记忆与长期记忆的实现

在AI Agent的架构中，记忆系统（Memory System）是区分"智能助手"与"普通聊天机器人"çš„å ³é”®ç»„ä»¶ã€‚ä¸€ä¸ªæ²¡æœ‰è®°å¿†çš„Agentåªèƒ½å¤„ç†å•è½®å¯¹è¯ï¼Œè€Œå ·å¤‡å®Œå–„è®°å¿†ç³»ç»Ÿçš„Agentèƒ½å¤Ÿç†è§£ä¸Šä¸‹æ–‡ã€ç§¯ç´¯ç»éªŒã€æŒç»­è¿›åŒ–ã€‚æœ¬æ–‡å°†æ·±å ¥è®²è§£AI Agentè®°å¿†ç³»ç»Ÿçš„ä¸‰å¤§æ ¸å¿ƒç±»åž‹â€”â€”Buffer Memory、Summary Memory和Vector Memory，并给出完整的Pythonå®žçŽ°ä»£ç ã€‚

一、为什么记忆系统如此重要？

æƒ³è±¡ä½ åœ¨ä¸Žä¸€ä½åŠ©æ‰‹å¯¹è¯ï¼š

• ä½ è¯´ï¼š"我住在上海"
• äº”è½®ä¹‹åŽä½ é—®ï¼š"我住的城市今天天气如何？"

如果没有记忆系统，Agentå®Œå ¨æ— æ³•å›žç­”è¿™ä¸ªé—®é¢˜ã€‚è®°å¿†ç³»ç»Ÿèµ‹äºˆäº†Agent：

| 记忆类型 | 作用范围 | æ ¸å¿ƒåŠŸèƒ½ | |---------|---------|---------| | 短期记忆 | 当前会话 | 保持对话上下文连贯性 | | 长期记忆 | è·¨ä¼šè¯æŒä¹ åŒ– | 存储用户偏好、历史事实 | | 外部记忆 | 知识库/文档 | 扩展Agent的知识边界 |

二、短期记忆：Buffer Memory（缓冲记忆）

Buffer Memory是最基础的短期记忆形式，它直接保存最近的对话历史，通常以滑动窗口的形式维护。

2.1 原理与实现

Buffer Memoryçš„æ ¸å¿ƒæ€æƒ³å¾ˆç®€å•ï¼šä¿ç•™æœ€è¿‘Nè½®å¯¹è¯ï¼Œè¶ å‡ºçª—å£çš„è‡ªåŠ¨ä¸¢å¼ƒã€‚è¿™ç§æ–¹å¼è®¡ç®—å¼€é”€ä½Žã€å»¶è¿Ÿå°ï¼Œé€‚åˆå¯¹å®žæ—¶æ€§è¦æ±‚é«˜çš„åœºæ™¯ã€‚

from collections import deque from dataclasses import dataclass from typing import List, Optional from datetime import datetime @dataclass class Message: role: str # "user" 或 "assistant" content: str timestamp: datetime = None def __post_init__(self): if self.timestamp is None: self.timestamp = datetime.now() class BufferMemory: """滑动窗口缓冲记忆 保留最近 k 轮对话，适合维护短时上下文。 """ def __init__(self, max_turns: int = 10): self.max_turns = max_turns self.buffer = deque(maxlen=max_turns * 2) # æ¯è½®åŒ å«user+assistant def add_message(self, role: str, content: str) -> None: """æ·»åŠ ä¸€æ¡æ¶ˆæ¯åˆ°ç¼“å†²åŒº""" self.buffer.append(Message(role=role, content=content)) def get_history(self) -> List[Message]: """获取当前缓冲区中的所有消息""" return list(self.buffer) def get_formatted_history(self) -> str: """æ ¼å¼åŒ–ä¸ºå­—ç¬¦ä¸²ä¾›LLM使用""" lines = [] for msg in self.buffer: role_label = "用户" if msg.role == "user" else "助手" lines.append(f"[{role_label}] {msg.content}") return "\n".join(lines) def clear(self) -> None: """æ¸ ç©ºè®°å¿†""" self.buffer.clear() def is_empty(self) -> bool: return len(self.buffer) == 0 # 使用示例 buffer = BufferMemory(max_turns=3) # 模拟对话 buffer.add_message("user", "ä½ å¥½ï¼Œæˆ‘æƒ³è®¢ä¸€å¼ åŽ»åŒ—äº¬çš„æœºç¥¨") buffer.add_message("assistant", "好的，请问您想什么时候出发？") buffer.add_message("user", "明天上午") buffer.add_message("assistant", "明天上午的航班有CA1501和MU5101...") buffer.add_message("user", "我要CA1501") print("=== Buffer Memory å† å®¹ ===") print(buffer.get_formatted_history()) print(f"\n总消息数：{len(buffer.get_history())}")

2.2 Token感知的Buffer Memory

åœ¨å®žé™ ç”Ÿäº§çŽ¯å¢ƒä¸­ï¼Œæ›´é«˜çº§çš„åšæ³•æ˜¯æŒ‰Tokenæ•°é‡è€Œéžè½®æ•°æŽ§åˆ¶çª—å£å¤§å°ï¼Œå› ä¸ºä¸åŒLLM有固定的上下文长度限制。