Criar um plugin de provedor de memória

Plugins de provedor de memória dão ao Hermes Agent conhecimento persistente entre sessões, além do MEMORY.md e do USER.md embutidos. Este guia mostra como construir um.

dica

Provedores de memória são um dos dois tipos de plugin de provedor. O outro é o plugin de motor de contexto, que substitui o compressor de contexto embutido. Os dois seguem o mesmo padrão: seleção única, guiados por configuração, gerenciados por hermes plugins.

Formatos de instalação

O Hermes descobre provedores de memória em quatro fontes, nesta ordem de precedência:

Fonte	Local	Observações
Embutido	plugins/memory/<name>/	Vem junto com o Hermes. Fechado para novos provedores — veja o CONTRIBUTING.
Usuário	$HERMES_HOME/plugins/<name>/	Colocado pelo usuário, por perfil.
Projeto	./.hermes/plugins/<name>/	Opcional, via HERMES_ENABLE_PROJECT_PLUGINS=1.
Pacote	Entry point hermes_agent.memory_providers	pip install, nada para copiar.

As fontes anteriores vencem em caso de colisão de nome, então um diretório largado numa árvore de trabalho nunca consegue encobrir um provedor que vem de fábrica.

observação

Isso é o inverso da ordem do sistema geral de plugins, em que o último vence. Um provedor de memória é ativado pelo nome (memory.provider), então encobrir redirecionaria silenciosamente a memória do agente, em vez de apenas sobrescrever uma ferramenta.

A descoberta só enumera — ela nunca importa um provedor. Nada roda até que memory.provider cite o provedor pelo nome.

Provedor em diretório

Um provedor em diretório fica em plugins/memory/<name>/ quando vem junto com o Hermes, em $HERMES_HOME/plugins/<name>/ quando o usuário instala, ou em ./.hermes/plugins/<name>/ quando é local do projeto:

plugins/memory/meu-provedor/
├── __init__.py      # Implementação de MemoryProvider + entry point register()
├── plugin.yaml      # Metadados (nome, descrição, hooks)
└── README.md        # Instruções de instalação, referência de configuração, ferramentas

Provedor empacotado

Um provedor instalado por pip publica um entry point no grupo hermes_agent.memory_providers. O nome do entry point é o nome do provedor que os usuários escolhem em memory.provider; o valor aponta para a função register(ctx) do provedor:

pyproject.toml

[project.entry-points."hermes_agent.memory_providers"]
my-provider = "my_provider:register"

Aponte o entry point para o pacote, ou para um register(ctx) dentro dele, e mantenha a implementação, as skills e os outros recursos no layout normal de pacote Python. Não é preciso copiar nada para $HERMES_HOME/plugins/.

Um entry point de pacote ganha tudo que uma instalação em diretório ganha, inclusive os dois arquivos que o Hermes lê do disco em vez de importar — config_schema.py (o painel de configuração do dashboard) e cli.py (os seus subcomandos hermes <provider>). Os dois são procurados ao lado do __init__.py do seu pacote, então aponte o entry point para um pacote, e não para um módulo único, se você entrega qualquer um dos dois.

A ABC MemoryProvider

Seu plugin implementa a classe base abstrata MemoryProvider, de agent/memory_provider.py:

from agent.memory_provider import MemoryProvider

class MyMemoryProvider(MemoryProvider):
    @property
    def name(self) -> str:
        return "my-provider"

    def is_available(self) -> bool:
        """Verifica se este provedor pode ativar. SEM chamadas de rede."""
        return bool(os.environ.get("MY_API_KEY"))

    def initialize(self, session_id: str, **kwargs) -> None:
        """Chamado uma vez, na inicialização do agente.

        kwargs sempre inclui:
          hermes_home (str): caminho ativo de HERMES_HOME. Use para armazenamento.
        """
        self._api_key = os.environ.get("MY_API_KEY", "")
        self._session_id = session_id

    # ... implemente os demais métodos

Métodos obrigatórios

Ciclo de vida principal

Método	Quando é chamado	Precisa implementar?
name (propriedade)	Sempre	Sim
is_available()	Init do agente, antes da ativação	Sim — sem chamadas de rede
initialize(session_id, **kwargs)	Inicialização do agente	Sim
get_tool_schemas()	Depois do init, para injeção de ferramentas	Sim
handle_tool_call(tool_name, args, **kwargs)	Quando o agente usa suas ferramentas	Sim (se você tem ferramentas)

Configuração

Método	Para que serve	Precisa implementar?
get_config_schema()	Declara os campos de configuração para hermes memory setup	Sim
save_config(values, hermes_home)	Escreve a configuração não secreta no local nativo	Sim (a menos que só use variáveis de ambiente)

Hooks opcionais

Método	Quando é chamado	Caso de uso
system_prompt_block()	Montagem do prompt de sistema	Informação estática do provedor
prefetch(query, *, session_id="")	Antes de cada chamada de API	Devolve o contexto recuperado
queue_prefetch(query, *, session_id="")	Depois de cada turno	Preaquece para o turno seguinte
sync_turn(user, assistant, *, session_id="", messages=None)	Depois de cada turno completo	Persiste a conversa
on_session_end(messages)	A conversa termina	Extração/descarga final
on_pre_compress(messages)	Antes da compressão de contexto	Salva insights antes do descarte
on_memory_write(action, target, content)	Escritas na memória embutida	Espelha para o seu backend
shutdown()	Saída do processo	Fecha conexões

Checkpoints de pré-compressão (fail-closed)

Por padrão, on_pre_compress() é de melhor esforço: se o seu provedor lança exceção, o host registra a falha e a compressão segue. Esse é o padrão certo para extração de insights — e o errado para um provedor cujo trabalho é arquivar evidência da transcrição num armazenamento durável antes da reescrita com perda. Para esse caso, o host oferece um contrato de checkpoint opcional (API v2):

from agent.memory_provider import MemoryProvider

class MyArchivingProvider(MemoryProvider):
    # Adesão: todo retorno bem-sucedido de on_pre_compress() significa que o
    # checkpoint durável foi confirmado. Lance exceção em qualquer falha — não
    # devolva sucesso parcial. A versão 1 (o padrão herdado) é o contrato
    # histórico implícito: semântica de melhor esforço, lista de mensagens crua.
    pre_compress_checkpoint_api_version = 2

    def on_pre_compress(self, messages):
        ids = self._archive(messages)   # precisa ser durável antes de retornar
        return f"checkpoint: {ids}"     # encaminhado para o prompt de resumo

Os operadores habilitam a imposição por implantação:

compression:
  checkpoint_required: true   # padrão: false

Com a trava ligada, a compressão falha fechada antes de qualquer reescrita com perda, a menos que um provedor ativo que anuncie a API tenha concluído o checkpoint: a transcrição não comprimida é preservada, a tentativa de compactação dá erro com BLOCKED_MISSING_PREREQUISITE, e dá para tentar de novo quando o seu armazenamento se recuperar. Com a trava desligada (padrão), nada muda para os provedores existentes.

A trava vale para toda autoridade de compactação, não só para o sumarizador do Hermes: a compactação nativa do lado do servidor (compression.codex_responses_native) fica suprimida enquanto a trava está armada, a microcompactação pós-turno (compression.micro_compact) é forçada a desligar na inicialização do agente (ela absorve trocas antigas num resumo rolante, sem hook de checkpoint no caminho), e o modo de API codex_app_server é recusado na inicialização do agente — o agente codex compacta a própria thread sem uma fronteira honesta de pré-compactação, então não dá para garantir um checkpoint obrigatório ali. O compressor do Hermes, que entende checkpoints, continua sendo a única autoridade com perda.

O que o seu provedor recebe depende da versão de API que ele declara. Provedores de versão 1 (o padrão implícito — todo provedor que já existia) mantêm o contrato histórico: a lista de mensagens crua, exatamente como antes. Provedores de checkpoint na versão 2 recebem evidência direta normalizada: só linhas de texto de usuário/assistente — resultados de ferramenta, mensagens de sistema, o payload tool_calls das mensagens do assistente (a prosa delas é mantida) e resumos de compactações anteriores são filtrados no host. Resumos anteriores são reconhecidos por um marcador persistente de mensagem _compressed_summary, que sobrevive a reinícios de processo, então uma sessão retomada nunca devolve resumos derivados para dentro do seu arquivo.

Checkpoints precisam ser idempotentes. Depois de um bloqueio fail-closed, a próxima tentativa de compactação chama on_pre_compress() de novo com a mesma transcrição — e uma transcrição que cresceu pouco produz evidência em grande parte repetida. Chaveie as escritas do seu arquivo por conteúdo (um digest da transcrição, por exemplo) e faça upsert, para que repetições e sobreposições sejam deduplicadas em vez de acumular arquivos duplicados.

Testes de contrato: tests/agent/test_pre_compress_checkpoint_contract.py.

Esquema de configuração

get_config_schema() devolve uma lista de descritores de campo usada pelo hermes memory setup:

def get_config_schema(self):
    return [
        {
            "key": "api_key",
            "description": "Chave de API do Meu Provedor",
            "secret": True,           # → escrito no .env
            "required": True,
            "env_var": "MY_API_KEY",   # nome explícito da variável de ambiente
            "url": "https://my-provider.com/keys",  # onde conseguir
        },
        {
            "key": "region",
            "description": "Região do servidor",
            "default": "us-east",
            "choices": ["us-east", "eu-west", "ap-south"],
        },
        {
            "key": "project",
            "description": "Identificador do projeto",
            "default": "hermes",
        },
    ]

Campos com secret: True e env_var vão para o .env. Campos não secretos são passados para save_config().

Esquema mínimo x completo

Todo campo de get_config_schema() vira uma pergunta durante o hermes memory setup. Provedores com muitas opções devem manter o esquema mínimo — inclua só os campos que o usuário precisa configurar (chave de API, credenciais obrigatórias). Documente os ajustes opcionais numa referência de arquivo de configuração (por exemplo, $HERMES_HOME/myprovider.json) em vez de perguntar todos durante a instalação. Isso mantém o assistente rápido sem abrir mão da configuração avançada. Veja o provedor Supermemory como exemplo — ele só pergunta a chave de API; todas as outras opções ficam em supermemory.json.

Salvar a configuração

def save_config(self, values: dict, hermes_home: str) -> None:
    """Escreve a configuração não secreta no seu local nativo."""
    import json
    from pathlib import Path
    config_path = Path(hermes_home) / "my-provider.json"
    config_path.write_text(json.dumps(values, indent=2))

Em provedores que só usam variáveis de ambiente, deixe o no-op padrão.

Entry point do plugin

def register(ctx) -> None:
    """Chamado pelo sistema de descoberta de plugins de memória."""
    ctx.register_memory_provider(MyMemoryProvider())

Um provedor também pode expor skills somente de leitura a partir do mesmo callback. As skills são qualificadas pelo nome do entry point e só são carregadas quando aquele provedor de memória está ativo:

from pathlib import Path

SKILLS_DIR = Path(__file__).parent / "skills"

def register(ctx) -> None:
    ctx.register_memory_provider(MyMemoryProvider())
    ctx.register_skill(
        "maintenance",
        SKILLS_DIR / "maintenance" / "SKILL.md",
        "Cuida do armazenamento de memória do provedor",
    )

Com o entry point my-provider ativo, a skill fica disponível como my-provider:maintenance pelo skill_view().

plugin.yaml

name: my-provider
version: 1.0.0
description: "Descrição curta do que este provedor faz."
hooks:
  - on_session_end    # liste os hooks que você implementa

Contrato de threads

sync_turn() NÃO PODE bloquear. Se o seu backend tem latência (chamadas de API, processamento por LLM), rode o trabalho numa thread daemon:

def sync_turn(self, user_content, assistant_content, *, session_id="", messages=None):
    def _sync():
        try:
            self._api.ingest(user_content, assistant_content, session_id=session_id, messages=messages)
        except Exception as e:
            logger.warning("Sync failed: %s", e)

    if self._sync_thread and self._sync_thread.is_alive():
        self._sync_thread.join(timeout=5.0)
    self._sync_thread = threading.Thread(target=_sync, daemon=True)
    self._sync_thread.start()

messages é o contexto opcional da conversa, no estilo OpenAI, no momento em que o turno terminou. Quando está presente, inclui mensagens de usuário/assistente, chamadas de ferramenta do assistente e mensagens de resultado de ferramenta. Provedores que não precisam do contexto cru do turno podem omitir o parâmetro messages; o Hermes vai continuar chamando eles com a assinatura antiga.

Provedores em nuvem devem documentar quais partes de messages saem do dispositivo. Chamadas de ferramenta e resultados de ferramenta podem conter caminhos de arquivo, saída de comandos ou outros dados do espaço de trabalho.

Isolamento entre perfis

Todos os caminhos de armazenamento precisam usar o kwarg hermes_home de initialize(), e não um ~/.hermes fixo no código:

# CERTO — no escopo do perfil
from hermes_constants import get_hermes_home
data_dir = get_hermes_home() / "my-provider"

# ERRADO — compartilhado entre todos os perfis
data_dir = Path("~/.hermes/my-provider").expanduser()

Testes

Veja tests/agent/test_memory_provider.py e os testes de memória vizinhos (tests/agent/test_memory_session_switch.py, tests/agent/test_memory_user_id.py, tests/run_agent/test_memory_provider_init.py) para padrões de ponta a ponta.

from agent.memory_manager import MemoryManager

mgr = MemoryManager()
mgr.add_provider(my_provider)
mgr.initialize_all(session_id="test-1", platform="cli")

# Testa o roteamento de ferramentas
result = mgr.handle_tool_call("my_tool", {"action": "add", "content": "test"})

# Testa o ciclo de vida
mgr.sync_all("user msg", "assistant msg")
mgr.on_session_end([])
mgr.shutdown_all()

Adicionar comandos de CLI

Plugins de provedor de memória podem registrar a própria árvore de subcomandos de CLI (por exemplo, hermes my-provider status, hermes my-provider config). Isso usa um sistema de descoberta por convenção — nenhuma mudança nos arquivos do núcleo é necessária.

Como funciona

1. Adicione um arquivo cli.py no diretório do seu plugin
2. Defina uma função register_cli(subparser) que monta a árvore do argparse
3. O sistema de plugins de memória descobre isso na inicialização, via discover_plugin_cli_commands()
4. Seus comandos aparecem sob hermes <provider-name> <subcommand>

Trava de provedor ativo: seus comandos de CLI só aparecem quando o seu provedor é o memory.provider ativo na configuração. Se o usuário não configurou o seu provedor, os comandos não aparecem em hermes --help.

Exemplo

# plugins/memory/my-provider/cli.py

def my_command(args):
    """Handler despachado pelo argparse."""
    sub = getattr(args, "my_command", None)
    if sub == "status":
        print("Provider is active and connected.")
    elif sub == "config":
        print("Showing config...")
    else:
        print("Usage: hermes my-provider <status|config>")

def register_cli(subparser) -> None:
    """Monta a árvore de argparse do hermes my-provider.

    Chamado por discover_plugin_cli_commands() na hora de montar o argparse.
    """
    subs = subparser.add_subparsers(dest="my_command")
    subs.add_parser("status", help="Show provider status")
    subs.add_parser("config", help="Show provider config")
    subparser.set_defaults(func=my_command)

Implementação de referência

Veja plugins/memory/honcho/cli.py para um exemplo completo, com 13 subcomandos, gerenciamento entre perfis (--target-profile) e leitura/escrita de configuração.

Estrutura de diretórios com CLI

plugins/memory/meu-provedor/
├── __init__.py      # Implementação de MemoryProvider + register()
├── plugin.yaml      # Metadados
├── cli.py           # register_cli(subparser) — comandos de CLI
└── README.md        # Instruções de instalação

Regra do provedor único

Só um provedor externo de memória pode ficar ativo por vez. Se o usuário tentar registrar um segundo, o MemoryManager recusa com um aviso. Isso evita inchaço no esquema de ferramentas e backends em conflito.

---