Создаём AI-агента для конференций с памятью на Engram

Зачем нужен AI-агент с памятью для конференций?

Представьте: вы на конференции, стенд вашей компании привлекает много посетителей, но вы физически не можете ответить каждому. Посетители подходят, видят пустое место и уходят. Решение — AI-агент, который общается с гостями, отвечает на вопросы и запоминает всё, что сказано. Но стандартная память таких агентов слишком мала для сотен диалогов. В этой статье я покажу, как расширить память AI-агента с помощью сервиса Engram, чтобы он помнил каждого посетителя и помогал вам анализировать интерес к продукту.

Архитектура конференц-агента

Наш агент будет работать на ноутбуке у стенда. Посетители подходят, говорят с ним, а агент запоминает их вопросы и предпочтения. Владелец стенда позже может через Telegram спросить у агента, что интересовало конкретного гостя. Для этого нужны четыре компонента:

• AI-агент (Hermes)
• Интерфейс для посетителей (Hermes CLI с голосовым режимом)
• Долговременная память (Engram)
• Интерфейс для владельца (Hermes Messaging Gateway через Telegram)

Hermes покрывает всё, кроме памяти. Встроенная память Hermes хранит общие факты в файле Memory.md (около 800 токенов) и информацию о пользователях в User.md (около 500 токенов). Этого недостаточно для сотен посетителей. Здесь на помощь приходит Engram — управляемый сервис памяти от Weaviate.

Что такое Engram и как он работает?

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

Настройка Engram

Engram доступен бесплатно. Зайдите в Weaviate Console, зарегистрируйтесь и создайте API-ключ. Затем установите Python SDK:

pip install weaviate-engram

Создайте клиент:

from engram import EngramClient

client = EngramClient(api_key="your-api-key")

Добавлять память можно двумя способами: строкой или списком сообщений (conversation).

# Добавление строки
run = client.memories.add(
    "Alice предпочитает асинхронный Python и избегает Java.",
    user_id="hermes"
)

# Добавление диалога
run = client.memories.add([
    {"role": "user", "content": "Как лучше обрабатывать повторные попытки?"},
    {"role": "assistant", "content": "Экспоненциальная задержка с джиттером — стандартный подход."},
    {"role": "user", "content": "Понял, буду использовать в HTTP-клиенте."}
], user_id="hermes")

Поиск памяти:

results = client.memories.search(
    query="Что Alice думает о Python?",
    user_id="hermes"
)
for memory in results:
    print(memory.content)

Параметр user_id позволяет разделять память по пользователям — каждый посетитель будет иметь свою область памяти.

Создание плагина памяти для Hermes

В Hermes плагин памяти должен наследовать класс MemoryProvider. Структура плагина состоит из двух файлов: __init__.py (реализация) и plugin.yaml (метаданные).

Начнём с каркаса без логики:

import json
import os
from typing import Any, Dict, List
from agent.memory_provider import MemoryProvider
from engram import EngramClient

class Engram(MemoryProvider):
    @property
    def name(self) -> str:
        return "engram"

    def is_available(self) -> bool:
        return bool(os.environ.get("ENGRAM_API_KEY"))

    def initialize(self, session_id: str, **kwargs) -> None:
        pass

    def get_config_schema(self):
        pass

    def on_memory_write(self, action: str, target: str, content: str) -> None:
        pass

    def on_session_end(self, messages: List[Dict[str, Any]]) -> None:
        pass

    def get_tool_schemas(self) -> List[Dict[str, Any]]:
        pass

    def handle_tool_call(self, tool_name: str, args: Dict[str, Any], **kwargs) -> str:
        pass

Теперь реализуем методы.

Методы initialize и get_config_schema

def initialize(self, session_id: str, **kwargs) -> None:
    self._client = EngramClient(api_key=os.environ["ENGRAM_API_KEY"])
    self._user_id = session_id

def get_config_schema(self):
    return [
        {
            "key": "api_key",
            "description": "Engram API key",
            "secret": True,
            "required": True,
            "env_var": "ENGRAM_API_KEY",
            "url": "https://console.weaviate.cloud/engram"
        }
    ]

В initialize мы создаём клиент Engram и сохраняем session_id как user_id. Конфигурация требует только API-ключ.

Хуки on_memory_write и on_session_end

Хук on_memory_write вызывается, когда Hermes записывает память в Markdown-файлы. Мы отправляем этот контент в Engram.

def on_memory_write(self, action: str, target: str, content: str) -> None:
    self._client.memories.add(content, user_id=self._user_id)

Хук on_session_end вызывается при завершении сессии. Мы отправляем весь диалог в Engram.

def on_session_end(self, messages: List[Dict[str, Any]]) -> None:
    parsed_message = []
    for message in messages:
        if message['role'] == 'user':
            parsed_message.append({'role': 'user', 'content': message['content']})
        if message['role'] == 'assistant':
            parsed_message.append({'role': 'assistant', 'content': message['content']})
    self._client.memories.add(parsed_message, user_id=self._user_id)

Оба хука используют session_id как user_id, чтобы память каждого посетителя хранилась отдельно.

Инструменты для поиска памяти

Определяем схему инструмента для поиска:

SEARCH_SCHEMA = {
    "name": "engram_search",
    "description": "Поиск памяти в Engram",
    "parameters": {
        "type": "object",
        "properties": {
            "query": {
                "type": "string",
                "description": "Что искать в памяти Engram"
            },
            "user_id": {
                "type": "string",
                "description": "ID пользователя, чью память искать"
            }
        },
        "required": ["query", "user_id"]
    }
}

Регистрируем схему через get_tool_schemas:

def get_tool_schemas(self) -> List[Dict[str, Any]]:
    return [SEARCH_SCHEMA]

Реализуем вызов инструмента:

def handle_tool_call(self, tool_name: str, args: Dict[str, Any], **kwargs) -> str:
    if tool_name == "engram_search":
        query = args["query"]
        user_id = args["user_id"]
        results = self._client.memories.search(query=query, user_id=user_id)
        text = [result.content for result in results]
        return json.dumps({"result": "\n".join(text)})
    return json.dumps({"error": f"Неизвестный инструмент {tool_name}"})

Регистрация плагина

В конце файла добавляем функцию регистрации:

def register(ctx) -> None:
    """Вызывается системой обнаружения плагинов памяти."""
    ctx.register_memory_provider(Engram())

Как это работает вместе

Когда посетитель подходит к стенду, Hermes запускает сессию. Плагин Engram создаёт клиент и сохраняет session_id. Во время разговора Hermes записывает факты в Markdown, а хук on_memory_write отправляет их в Engram. При завершении сессии весь диалог отправляется через on_session_end. Владелец стенда может через Telegram попросить агента найти, что интересовало конкретного посетителя — агент вызывает инструмент engram_search, который ищет в памяти Engram.

Практический вывод

Попробуйте сами: зарегистрируйтесь в Weaviate Console, получите API-ключ, установите SDK и создайте плагин памяти по нашему шаблону. Затем настройте Hermes с этим плагином и запустите агента на ближайшей конференции. Вы увидите, как легко управлять памятью сотен посетителей с помощью Engram.