Как устроена Memory в Harness (in-memory, fs, vector stores, relay)?

Краткий тезис

Memory в архитектуре Harness — это многоуровневая система хранения и доступа к контексту для AI-агентов. Она включает четыре уровня: in-memory (быстрая, но непостоянная память текущего диалога), filesystem (fs) (долговременное хранение сессий), vector stores (поиск|семантический поиск по истории) и relay (передача памяти между агентами). Ключевой механизм — фильтрация через sessionId, которая изолирует данные разных сессий, предотвращая их смешивание. Такая архитектура обеспечивает баланс между скоростью, масштабируемостью и возможностью поиска релевантного контекста.

1. Термин: Memory в контексте Agentic RAG

Memory (память) — это подсистема агента, отвечающая за сохранение, извлечение и обновление информации о прошлых взаимодействиях, состоянии выполнения задач и внешних данных. В отличие от обычного RAG, где памятью часто служит только векторное хранилище, в Agentic RAG агент должен удерживать контекст многошагового диалога, вспоминать предыдущие решения и передавать знания другим агентам. Harness — это специализированный фреймворк (или слой в архитектуре агента), который предоставляет единый интерфейс для работы с разными типами памяти. Он состоит из четырёх логических уровней, каждый со своей скоростью, надёжностью и областью применения.

Зачем нужна многоуровневая память?

• Скорость: in-memory даёт микросекундный доступ к текущему контексту.
• Постоянство: fs не теряет данные при перезапуске.
• Поиск: vector stores позволяют находить релевантные прошлые диалоги по смыслу.
• Координация: relay обеспечивает бесшовную передачу контекста между агентами (например, при эскалации задачи).

2. Уровень 1: In-memory (оперативная память)

Назначение

Хранение текущего диалога и временных данных внутри процесса. Это самый быстрый слой, но volatile (теряется при падении или перезапуске агента).

Как устроено

• Структура: обычно простой dict или OrderedDict с ключом sessionId.
• Содержимое: список сообщений (role, content, timestamp), состояние выполнения шагов, метаданные сессии.
• Ограничение: размер буфера — например, последние 10–20 сообщений, чтобы не переполнить память.

Пример реализации на Python

import time

class InMemoryStore:
    def __init__(self, max_messages=20):
        self._data = {}
        self.max_messages = max_messages

    def append(self, session_id: str, role: str, content: str):
        if session_id not in self._data:
            self._data[session_id] = []
        self._data[session_id].append({
            "role": role,
            "content": content,
            "timestamp": time.time()
        })
        # trim to max_messages
        if len(self._data[session_id]) > self.max_messages:
            self._data[session_id] = self._data[session_id][-self.max_messages:]

    def get_context(self, session_id: str):
        return self._data.get(session_id, [])

Плюсы и минусы

Типовой сценарий: чтение последних сообщений для формирования промпта LLM.

3. Уровень 2: Filesystem (fs) — файловая система

Назначение

Долговременное хранение сессий в виде файлов (JSON, SQLite, Parquet). Данные сохраняются при перезапуске агента, но поиск по содержанию неэффективен.

Как устроено

• Путь: каталог, например /memory_sessions/{session_id}.json.
• Формат: каждый файл — полный лог диалога с метаданными.
• Фильтрация: загрузка нужной сессии по sessionId напрямую.

Пример записи/чтения

import json
import os

MEMORY_DIR = "./sessions"

def save_session(session_id: str, data: dict):
    os.makedirs(MEMORY_DIR, exist_ok=True)
    path = os.path.join(MEMORY_DIR, f"{session_id}.json")
    with open(path, "w") as f:
        json.dump(data, f)

def load_session(session_id: str) -> dict | None:
    path = os.path.join(MEMORY_DIR, f"{session_id}.json")
    if os.path.exists(path):
        with open(path) as f:
            return json.load(f)
    return None

Плюсы и минусы

Типовой сценарий: восстановление контекста после перезапуска агента или передача всей сессии для долгосрочного анализа.

4. Уровень 3: Vector stores — векторные хранилища

Назначение

Семантический поиск по истории диалогов. Позволяет находить релевантные сообщения или фрагменты даже если они из разных сессий.

Как устроено

• Эмбеддинги: каждое сообщение (или суммарный контекст шага) преобразуется в вектор с помощью embedding model.
• Индекс: векторы помещаются в векторное хранилище (FAISS, Chroma, Pinecone, Qdrant).
• Поиск: по запросу строится query embedding и выполняется ANN-поиск (приблизительные ближайшие соседи) с фильтром по sessionId (опционально).

Пример с FAISS

import faiss
import numpy as np
from sentence_transformers import SentenceTransformer

model = SentenceTransformer("all-MiniLM-L6-v2")
dim = model.get_sentence_embedding_dimension()
index = faiss.IndexFlatL2(dim)

session_ids = []  # список sessionIds для каждого вектора

def add_message(session_id: str, text: str):
    vec = model.encode([text])
    index.add(np.array(vec))
    session_ids.append(session_id)

def search(query: str, k=5, session_id_filter=None):
    q_vec = model.encode([query])
    distances, indices = index.search(np.array(q_vec), k)
    results = []
    for i in indices[0]:
        if session_id_filter is None or session_ids[i] == session_id_filter:
            results.append((session_ids[i], distances[0][0]))
    return results

Плюсы и минусы

Типовой сценарий: агент получает вопрос "А что мы делали с API два дня назад?" и находит подходящий контекст из другой сессии.

5. Уровень 4: Relay — передача памяти между агентами

Назначение

Обеспечивает cross-context relay — передачу контекста (памяти) от одного агента другому без потери информации. Используется в мультиагентных системах при эскалации, кооперации или передачи заданий.

Как устроено

• Механизм: сериализуется дайджест памяти (или ссылка на неё) и отправляется через relay bus (очередь сообщений, HTTP-коллбек, gRPC-стрим).
• Содержимое: sessionId, ключевые факты, последние шаги, векторные представления (опционально).
• Фильтрация: принимающий агент может принять только часть памяти, например, только релевантные факты, отфильтрованные по топику.

Пример схемы relay

Агент A (текущее выполнение) → Relay Service → Агент B (принимает контекст)

• Реализация через REST: POST /relay/memory { sessionId, digest: [...], vector_id: "..." }
• Реализация через очередь RabbitMQ/Kafka: публикация события memory.relay.

Плюсы и минусы

Типовой сценарий: пользовательский запрос "найди ошибку и исправь её" — сначала агент-анализатор находит баг, затем relay передаёт контекст агенту-исправителю, который получает готовые описания и код.

6. Ключевой механизм: фильтр sessionId

sessionId — уникальный идентификатор диалога или сессии взаимодействия. Он пронизывает все уровни памяти:

• In-memory: sessionId — ключ словаря.
• Filesystem: sessionId — имя файла.
• Vector stores: sessionId хранится как метаданные вместе с вектором; при поиске можно применить фильтр, чтобы искать только внутри одной сессии или по всем.
• Relay: sessionId передаётся в пакете, чтобы принимающий агент мог привязать контекст к своей локальной памяти.

Зачем изоляция?

• Конфиденциальность: данные одного пользователя не смешиваются с другим.
• Производительность: уменьшает объём поиска в vector store.
• Отказоустойчивость: при сбое можно восстановить конкретную сессию.

7. Взаимодействие уровней в Harness

Слой памяти Harness обычно реализует fallback-стратегию:

1. Попытка in-memory: если нужный контекст уже в оперативной памяти — возвращаем его мгновенно.
2. Если нет → fs: загружаем последнюю сессию с диска.
3. Если требуется семантический поиск → vector store: выполняем ANN-поиск, возможно, с фильтром sessionId.
4. Если контекст запрошен другим агентом → relay: подписываемся на relay-канал и получаем память удалённо.

Такой pipeline позволяет достичь низкой задержки для частых запросов и не жертвовать полнотой для редких.

8. Сравнительная таблица уровней памяти

9. Типовой сценарий использования в Agentic RAG

Предположим, пользователь начал диалог с агентом по технической поддержке.

• Шаг 1: Агент пишет первые сообщения в in-memory (быстрый доступ).
• Шаг 2: После каждого сообщения делается snapshot в fs (на случай сбоя).
• Шаг 3: Когда пользователь спрашивает "напомни, что мы уже обсуждали про базу данных", агент запускает vector store поиск по всем сессиям с этим userId (или текущим sessionId), находит релевантные фрагменты.
• Шаг 4: Если задача требует привлечения другого агента (например, специалиста по БД), то через relay передаётся дайджест текущего контекста, чтобы второй агент мог продолжить без потери нити.

10. Best Practices для настройки памяти Harness

• Размер in-memory: держите не более нескольких тысяч сессий на один процесс, при превышении используйте LRU-эвакцию.
• Индексация vector store: используйте IVF или HNSW для больших коллекций; обновляйте индекс асинхронно.
• Сессионная изоляция: всегда храните sessionId в метаданных векторного индекса, чтобы при необходимости ограничить поиск.
• Relay: шифруйте пакеты памяти (TLS/SSL) и используйте acknowledgment, чтобы избежать потери контекста.
• Комбинирование: для долгих сессий периодически формируйте summary (через LLM) и сохраняйте его в fs или vector store — это уменьшит объём и ускорит поиск.

Пет-проект для закрепления

Задача: Реализовать упрощённую версию памяти Harness для агента-помощника, который может продолжать диалог после перезапуска и находить релевантные прошлые сообщения.

Инструменты: Python, SQLite (как fs), FAISS, Sentence Transformers, Flask (для relay).

Шаги:

1. Создать класс HarnessMemory, который инициализирует in-memory dict, SQLite-базу (таблица sessions с колонками session_id, payload) и FAISS-индекс с метаданными.
2. Реализовать методы append(session_id, role, content) — записывает во все уровни.
3. get_latest_context(session_id) — возвращает последние 10 сообщений из in-memory (если нет — из SQLite).
4. search_semantic(query, k=5) — выполняет ANN-поиск по всем сообщениям, возвращает session_id и текст.
5. relay_to_agent(target_url, session_id) — сериализует контекст в JSON и отправляет POST-запрос на target_url.
6. Написать простой тест: имитация диалога с 20 сообщениями, потом запрос "Что я говорил про погоду?" и проверка, что семантический поиск находит правильный фрагмент.

Ожидаемый результат: Консольный скрипт, который запускает тестовый агент, сохраняет память, перезапускает "агента" (новый объект) и успешно восстанавливает контекст, а также выполняет семантический поиск.

Связь с другими вопросами

Навигация

• Предыдущий: 749
• Следующий: 751
• Индекс: 00. Индекс разборов