Проблема, которая бесит всех: почему ваш AI-агент - золотая рыбка
Вы только что обсудили с агентом детальный план проекта, расписали архитектуру, обговорили техдолг. Через 20 сообщений спрашиваете: "А про что мы говорили в начале?" И получаете классическое: "Извините, я не могу вспомнить предыдущий контекст".
Ограничение контекста - не техническая мелочь. Это фундаментальный барьер. Когда я запускаю локальные модели на трёх 3090, меня бесит не потребление памяти GPU. Меня бесит, что я плачу за железо, а агент ведёт себя как пациент с амнезией.
Типичная ошибка: пытаться запихнуть всю историю в промпт. Результат? Либо OOM, либо качество ответов падает до нуля, потому что модель теряется в тысячах токенов.
Проблема глубже, чем кажется. Это не просто "добавить больше RAM". Архитектура трансформеров экспоненциально усложняется с ростом контекста. KV-cache ломается, внимание рассеивается, и вместо умного агента получаем нейросеть, которая пережевывает собственный хвост.
Скользящее окно: не решение, а костыль, который работает
Скользящее окно контекста - это как пытаться читать "Войну и мир" через замочную скважину. Вы видите только последнюю страницу, но делаете вид, что поняли весь роман.
Технически это просто: храним N последних токенов, старые выбрасываем. Всё. Но вот загвоздка - что именно выбрасывать? Случайные токены? Ключевые детали? Инструкции системы?
# Наивная реализация - так НЕ делать
class BadSlidingWindow:
def __init__(self, max_tokens=4000):
self.max_tokens = max_tokens
self.history = []
def add(self, text):
tokens = tokenize(text)
self.history.extend(tokens)
# Просто отрезаем начало - ужасная идея
while len(self.history) > self.max_tokens:
self.history.pop(0) # Прощай, системный промпт!Видите проблему? Первым делом мы теряем системные инструкции. Агент забывает, кто он и что должен делать. Типичная история: через час работы ваш кодирующий ассистент начинает писать стихи на Python.
1 Разделяй и властвуй: сегментируй контекст
Первое правило: не храни всё в одной куче. Разбей контекст на слои:
| Слой | Что содержит | Приоритет удаления |
|---|---|---|
| Системный | Роль, инструкции, запреты | Никогда |
| Ядро диалога | Последние 5-10 обменов | Низкий |
| Исторический | Старые сообщения (суммаризованные) | Средний |
| Мета-размышления | "Думаю о...", "Планирую..." | Высокий |
# Правильная сегментация
class SegmentedContext:
def __init__(self):
self.system_prompt = [] # Никогда не трогаем
self.recent_dialog = deque(maxlen=10) # Последние обмены
self.summarized_history = [] # Сжатая история
self.agent_thoughts = [] # Первые на удаление
def compress_if_needed(self):
total = self.count_tokens()
if total > MAX_TOKENS:
# Сначала убираем мета-размышления
while self.agent_thoughts and total > MAX_TOKENS:
removed = self.agent_thoughts.pop(0)
total -= count_tokens(removed)
# Потом суммаризируем старую историю
if total > MAX_TOKENS:
self.compress_history()Агентские подпрограммы: когда одна нейросеть управляет другой
Вот где начинается магия. Скользящее окно - механизм. Агентские подпрограммы - интеллект. Идея проста: создаём специализированного агента, чья единственная работа - управлять памятью основного агента.
Это не просто суб-агент. Это архитектурный паттерн. Memory Agent работает постоянно в фоне, как сборщик мусора в JVM, только умный.
2 Memory Agent: что он делает и почему он нужен
Memory Agent решает три задачи:
- Суммаризация: Сжимает старые диалоги в плотные выжимки
- Извлечение: Находит релевантные куски истории по запросу
- Приоритизация: Решает, что сохранить, а что выбросить
Ключевой момент: Memory Agent работает на той же модели, что и основной агент. Зачем? Консистентность. Если основной агент - CodeLlama 13B, а Memory Agent - TinyLlama 1B, они будут говорить на разных языках. Суммаризации будут бесполезны.
# Архитектура Memory Agent
class MemoryAgent:
def __init__(self, llm):
self.llm = llm # Та же модель, что у основного агента
self.memory_store = {} # Временное хранилище
async def summarize_chunk(self, dialog_chunk):
"""Сжимает фрагмент диалога в 10% от оригинала"""
prompt = f"""Суммаризируй этот диалог, сохранив:
1. Ключевые решения
2. Изменения в требованиях
3. Технические детали
Диалог:
{dialog_chunk}
"""
summary = await self.llm.generate(prompt)
return compress_to_tokens(summary, target_ratio=0.1)
def retrieve_relevant(self, query, history):
"""Находит релевантные части истории"""
# Простейший semantic search
query_embed = embed(query)
scores = []
for idx, memory in enumerate(history):
mem_embed = embed(memory)
score = cosine_similarity(query_embed, mem_embed)
if score > 0.7: # Порог релевантности
scores.append((idx, memory, score))
return sorted(scores, key=lambda x: x[2], reverse=True)[:3]Memory Agent должен работать асинхронно. Пока основной агент думает над ответом, Memory Agent уже суммаризирует предыдущий обмен. Задержка должна быть меньше времени генерации основного ответа.
Гибридный подход: скользящее окно + семантический поиск
Вот где становится интересно. Мы комбинируем:
- Скользящее окно для немедленного контекста (последние 2K токенов)
- Семантический поиск по суммаризованной истории (ещё 2K токенов)
- Динамическую подгрузку по требованию
Когда пользователь спрашивает: "А что мы решили насчёт базы данных?", система:
- Ищет в суммаризованной истории по "база данных"
- Если находит - загружает этот фрагмент в контекст
- Автоматически вытесняет менее релевантные части
class HybridMemorySystem:
def __init__(self, main_llm, embedder):
self.main_context = [] # Текущее скользящее окно
self.memory_agent = MemoryAgent(main_llm)
self.embedder = embedder
self.memory_index = {} # {embedding: (summary, metadata)}
async def process_query(self, query):
"""Обработка запроса с гибридной памятью"""
# Шаг 1: Поиск в долговременной памяти
relevant = self.search_long_term(query)
# Шаг 2: Добавление релевантного в контекст
context = self.build_context(query, relevant)
# Шаг 3: Генерация ответа
response = await self.main_llm.generate(context)
# Шаг 4: Асинхронное обновление памяти
asyncio.create_task(
self.update_memory(query, response)
)
return response
def search_long_term(self, query):
"""Семантический поиск по embeddings"""
query_embed = self.embedder.encode(query)
results = []
for mem_embed, (summary, meta) in self.memory_index.items():
similarity = cosine_similarity(query_embed, mem_embed)
if similarity > 0.65:
results.append({
'summary': summary,
'similarity': similarity,
'metadata': meta
})
return sorted(results, key=lambda x: x['similarity'], reverse=True)Ошибки, которые сломают вашу систему
Я видел десятки реализаций. 90% падают на этих граблях:
1. Циклическая суммаризация
Memory Agent суммаризирует диалог. Потом суммаризирует суммаризацию. Ещё раз. В итоге получаем: "Мы говорили о чём-то важном". Вся конкретика утеряна.
Решение: храните оригинальные ключевые фрагменты. Суммаризация - для поиска, оригиналы - для подгрузки. И никогда не суммаризируйте суммаризации.
2. Забывание системного промпта
Самая частая ошибка. Через час работы агент забывает свою роль. Начинает отвечать как чат-бот, а не как эксперт по коду.
# КАТЕГОРИЧЕСКИ НЕЛЬЗЯ
class TerribleMemory:
def trim_context(self):
# Так теряется личность агента
if len(self.context) > MAX_LEN:
self.context = self.context[-MAX_LEN:] # Прощай, системный промпт!3. Слепой semantic search
Ищете по embeddings без учёта времени? Получаете релевантные, но устаревшие ответы. Пользователь спрашивает про текущую архитектуру, а система подгружает решения недельной давности.
Решение: весовой коэффициент времени. Чем свежее память, тем выше приоритет.
Практическая реализация: с нуля за 200 строк
Вот минимальная рабочая система. Используем локальную модель через llama.cpp:
import asyncio
from typing import List, Dict
from dataclasses import dataclass
from datetime import datetime
import numpy as np
@dataclass
class MemoryChunk:
text: str
embedding: np.ndarray
timestamp: datetime
chunk_type: str # 'system', 'dialog', 'summary', 'thought'
class PracticalMemorySystem:
"""Полная система за 200 строк"""
def __init__(self, llm, embed_model, max_tokens=8000):
self.llm = llm
self.embed_model = embed_model
self.max_tokens = max_tokens
# Иерархическая память
self.system_chunks: List[MemoryChunk] = [] # Не удаляем никогда
self.recent_chunks: List[MemoryChunk] = [] # Скользящее окно
self.long_term_chunks: List[MemoryChunk] = [] # Суммаризованное
# Текущий счётчик токенов
self.current_tokens = 0
async def add_interaction(self, user_msg: str, agent_msg: str):
"""Добавляем новый обмен в память"""
# 1. Сохраняем как есть в recent
dialog = f"User: {user_msg}\nAgent: {agent_msg}"
embed = self.embed_model.encode(dialog)
chunk = MemoryChunk(
text=dialog,
embedding=embed,
timestamp=datetime.now(),
chunk_type='dialog'
)
self.recent_chunks.append(chunk)
self.current_tokens += self.count_tokens(dialog)
# 2. Проверяем лимит
await self.enforce_limit()
# 3. Асинхронно суммаризируем старые диалоги
if len(self.recent_chunks) > 10:
asyncio.create_task(self.summarize_old_dialogs())
async def enforce_limit(self):
"""Умное управление лимитом токенов"""
while self.current_tokens > self.max_tokens:
# Удаляем в порядке приоритета
if self.recent_chunks:
# Ищем мета-размышления
thoughts = [c for c in self.recent_chunks
if c.chunk_type == 'thought']
if thoughts:
chunk = thoughts[0]
self.recent_chunks.remove(chunk)
self.current_tokens -= self.count_tokens(chunk.text)
continue
# Удаляем самый старый диалог
chunk = self.recent_chunks.pop(0)
self.current_tokens -= self.count_tokens(chunk.text)
# Перемещаем в долговременную (уже суммаризованную)
if chunk.chunk_type == 'dialog':
await self.move_to_long_term(chunk)
async def move_to_long_term(self, chunk: MemoryChunk):
"""Суммаризация и перемещение в долговременную память"""
summary = await self.summarize_chunk(chunk.text)
summary_embed = self.embed_model.encode(summary)
summary_chunk = MemoryChunk(
text=summary,
embedding=summary_embed,
timestamp=chunk.timestamp,
chunk_type='summary'
)
self.long_term_chunks.append(summary_chunk)
# Ограничиваем долговременную память
if len(self.long_term_chunks) > 100:
# Удаляем наименее релевантные (по времени и использованию)
self.long_term_chunks.sort(key=lambda x: x.timestamp)
self.long_term_chunks = self.long_term_chunks[-50:]
async def query(self, question: str) -> str:
"""Поиск в памяти с гибридным подходом"""
question_embed = self.embed_model.encode(question)
# Ищем в долговременной памяти
relevant = []
for chunk in self.long_term_chunks:
similarity = self.cosine_sim(question_embed, chunk.embedding)
if similarity > 0.7:
relevant.append((chunk, similarity))
# Сортируем по релевантности
relevant.sort(key=lambda x: x[1], reverse=True)
# Собираем контекст
context_parts = []
# 1. Системный промпт (всегда первый)
for chunk in self.system_chunks:
context_parts.append(chunk.text)
# 2. Релевантная долговременная память
for chunk, _ in relevant[:3]: # Топ-3
context_parts.append(f"[Из памяти] {chunk.text}")
# 3. Недавний диалог
for chunk in self.recent_chunks[-5:]:
context_parts.append(chunk.text)
# 4. Текущий вопрос
context = "\n\n".join(context_parts)
context += f"\n\nUser: {question}\nAgent:"
# Генерация ответа
response = await self.llm.generate(context)
# Сохраняем взаимодействие
await self.add_interaction(question, response)
return response
@staticmethod
def cosine_sim(a, b):
return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))
@staticmethod
def count_tokens(text: str) -> int:
# Упрощённо - 1 токен ≈ 4 символа
return len(text) // 4Зачем всё это, если есть RAG?
Справедливый вопрос. RAG (Retrieval-Augmented Generation) - модное слово. Но у него другая задача. RAG ищет в документах. Наша система работает с диалогом, с течением времени, с изменяющимся контекстом.
RAG не понимает, что решение от вчера уже устарело. RAG не отличает системные инструкции от пользовательских запросов. RAG не управляет скользящим окном в реальном времени.
Это как сравнивать библиотекаря (RAG) и личного секретаря (наша система). Библиотекарь находит книги. Секретарь помнит, что вы говорили на прошлой встрече, что изменилось, и что важно прямо сейчас.
Что будет, когда контекст станет бесконечным?
Производители обещают модели с 1M токенов. Заманчиво? Опасно. Потому что проблема не в хранении, а в поиске. Хранить миллион токенов - легко. Найти в них нужную информацию за разумное время - нет.
Мой прогноз: будущее за гибридными системами. Сверхдлинный контекст для фонового хранения. Умные агенты для извлечения. И иерархическая структура, где каждый слой оптимизирован под свою задачу.
Пока индустрия гонится за длиной контекста, умные разработчики строят архитектуру. Потому что железо упирается в физические limits (апгрейд памяти GPU становится экстремальным спортом), а архитектура - нет.
Ваш агент всё ещё забывает, о чём вы говорили вчера? Пора перестать ждать чуда от hardware и начать строить систему, которая работает с тем, что есть. Потому что проблема не в том, что у модели мало памяти. Проблема в том, что мы неправильно ей управляем.
