Три уровня памяти: Working (контекстное окно), Short-term (Redis/Postgres), Long-term (ChromaDB + эмбеддинги). Практическая архитектура с кодом.
╔════════════════════════════════════════════════════════╗ ║ 🧠 ПИРАМИДА ПАМЯТИ АГЕНТА ║ ╠════════════════════════════════════════════════════════╣ ║ ║ ║ ⚡ WORKING MEMORY (секунды — минуты) ║ ║ ▸ Контекстное окно LLM (последние N сообщений) ║ ║ ▸ Хранилище: оперативная память ║ ║ ▸ Размер: 4K–128K токенов ║ ║ ║ ║ 🕐 SHORT-TERM MEMORY (часы — дни) ║ ║ ▸ Полная история диалога ║ ║ ▸ Хранилище: Redis / PostgreSQL ║ ║ ▸ TTL: 24 часа (автоочистка) ║ ║ ║ ║ 📚 LONG-TERM MEMORY (недели — месяцы) ║ ║ ▸ Векторная БД с семантическим поиском ║ ║ ▸ Хранилище: ChromaDB / Qdrant / Pinecone ║ ║ ▸ Поиск по смыслу, а не по ключевым словам ║ ║ ║ ╚════════════════════════════════════════════════════════╝
import tiktoken class WorkingMemory: """Sliding window + summarization для контекстного окна.""" def __init__(self, max_tokens=4096, model="gpt-4"): self.max_tokens = max_tokens self.encoder = tiktoken.encoding_for_model(model) self.messages = [] self.summary = "" def count_tokens(self, text): return len(self.encoder.encode(text)) def add_message(self, role, content): self.messages.append({"role": role, "content": content}) self._trim_if_needed() def _trim_if_needed(self): """Sliding window: удаляем старые сообщения при переполнении.""" while self._total_tokens() > self.max_tokens: removed = self.messages.pop(1) # оставляем system prompt self.summary += f"[{removed['role']}]: {removed['content'][:200]}...\n" def _total_tokens(self): return sum(self.count_tokens(m["content"]) for m in self.messages) def get_context(self): """Возвращает сообщения с суммаризацией старых.""" if self.summary: summary_msg = {"role": "system", "content": f"Резюме предыдущего: {self.summary}"} return [self.messages[0], summary_msg] + self.messages[1:] return self.messages
import json import redis from datetime import timedelta class ShortTermMemory: """Redis-хранилище истории диалога с TTL.""" def __init__(self, host="localhost", port=6379, ttl_hours=24): self.r = redis.Redis(host=host, port=port, decode_responses=True) self.ttl = timedelta(hours=ttl_hours) def save_message(self, session_id, role, content): """Добавление сообщения в историю сессии.""" key = f"session:{session_id}:messages" msg = json.dumps({"role": role, "content": content, "ts": datetime.now().isoformat()}) self.r.rpush(key, msg) self.r.expire(key, self.ttl) def get_history(self, session_id, limit=50): """Получение последних N сообщений сессии.""" key = f"session:{session_id}:messages" raw = self.r.lrange(key, -limit, -1) return [json.loads(m) for m in raw] def clear_session(self, session_id): """Очистка сессии (logout, сброс контекста).""" self.r.delete(f"session:{session_id}:messages")
import chromadb from sentence_transformers import SentenceTransformer class LongTermMemory: """ChromaDB для долговременных воспоминаний.""" def __init__(self, db_path="./agent_memory"): self.client = chromadb.PersistentClient(path=db_path) self.encoder = SentenceTransformer( "intfloat/multilingual-e5-large") self.collection = self.client.get_or_create_collection( name="memories", metadata={"hnsw:space": "cosine"} ) def store(self, user_id, memory_text, metadata=None): """Сохранение воспоминания с эмбеддингом.""" import uuid mem_id = str(uuid.uuid4()) embedding = self.encoder.encode(memory_text).tolist() self.collection.add( documents=[memory_text], embeddings=[embedding], ids=[mem_id], metadatas=[{**({} if not metadata else metadata), "user_id": user_id, "timestamp": datetime.now().isoformat()}] ) return mem_id def recall(self, query, user_id=None, n=5, threshold=0.65): """Семантический поиск воспоминаний.""" query_emb = self.encoder.encode(query).tolist() where = {"user_id": user_id} if user_id else None results = self.collection.query( query_embeddings=[query_emb], n_results=n, where=where ) # Фильтрация по порогу релевантности filtered = [(doc, dist) for doc, dist in zip(results['documents'][0], results['distances'][0]) if dist > threshold] return filtered
from rank_bm25 import BM25Okapi class HybridRetriever: """Гибридный поиск: BM25 (ключевые слова) + векторный (смысл).""" def __init__(self, long_term, alpha=0.5): self.ltm = long_term self.alpha = alpha # вес векторного поиска (0..1) self._build_bm25_index() def _build_bm25_index(self): """Построение BM25 индекса по всем документам.""" all_docs = self.ltm.collection.get()['documents'] tokenized = [doc.lower().split() for doc in all_docs] self.bm25 = BM25Okapi(tokenized) self.all_docs = all_docs def search(self, query, top_k=5): """Гибридный поиск с reranking.""" # BM25 скоринг tokenized_query = query.lower().split() bm25_scores = self.bm25.get_scores(tokenized_query) # Векторный скоринг query_emb = self.ltm.encoder.encode(query) vec_results = self.ltm.collection.query( query_embeddings=[query_emb.tolist()], n_results=len(self.all_docs)) # Нормализация и комбинирование norm_bm25 = bm25_scores / (bm25_scores.max() + 1e-9) final_scores = (1 - self.alpha) * norm_bm25 + \ self.alpha * (1 - vec_results['distances'][0]) # Top-K по комбинированному скору top_indices = np.argsort(final_scores)[::-1][:top_k] return [self.all_docs[i] for i in top_indices]
class MemoryManager: """Единый интерфейс для всех трёх уровней памяти.""" def __init__(self, working, short_term, long_term): self.wm = working self.stm = short_term self.ltm = long_term def remember(self, session_id, role, content): """Запомнить сообщение на всех уровнях.""" self.wm.add_message(role, content) self.stm.save_message(session_id, role, content) # В долговременную — только факты (эвристика) if self._is_factual(content): self.ltm.store(session_id, content) def recall(self, session_id, query, strategy="hybrid"): """Извлечение релевантных воспоминаний.""" # 1. Проверяем Working Memory (самое быстрое) wm_matches = [m for m in self.wm.messages if query.lower() in m["content"].lower()] # 2. Short-term: быстрый поиск по ключевым словам stm_history = self.stm.get_history(session_id) # 3. Long-term: семантический поиск ltm_results = self.ltm.recall(query, user_id=session_id) return { "working": wm_matches, "short_term": stm_history, "long_term": ltm_results } def forget(self, session_id=None, memory_id=None): """Удаление воспоминаний.""" if memory_id: self.ltm.collection.delete(ids=[memory_id]) if session_id: self.stm.clear_session(session_id) self.wm.messages.clear()