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æåºå®çä¸ä¸æé¿åº¦éå¶ã
更多推荐

所有评论(0)