Criar um plugin de motor de contexto

Plugins de motor de contexto substituem o ContextCompressor embutido por uma estratégia alternativa de gerenciar o contexto da conversa. Por exemplo, um motor de Lossless Context Management (LCM) que monta um DAG de conhecimento em vez de resumir com perda.

Como funciona

O gerenciamento de contexto do agente é construído sobre a ABC ContextEngine (agent/context_engine.py). O ContextCompressor embutido é a implementação padrão. Motores em plugin precisam implementar a mesma interface.

Só um motor de contexto pode estar ativo por vez. A escolha vem da configuração:

# config.yaml
context:
  engine: "compressor"    # embutido padrão
  engine: "lcm"           # ativa um motor de plugin chamado "lcm"

Motores em plugin nunca são ativados automaticamente — o usuário precisa definir context.engine explicitamente com o nome do plugin.

Estrutura de diretórios

Cada motor de contexto vive em plugins/context_engine/<nome>/:

plugins/context_engine/lcm/
├── __init__.py      # exporta a subclasse de ContextEngine
├── plugin.yaml      # metadados (nome, descrição, versão)
└── ...              # quaisquer outros módulos que o seu motor precise

A ABC ContextEngine

Seu motor precisa implementar estes métodos obrigatórios:

from agent.context_engine import ContextEngine

class LCMEngine(ContextEngine):

    @property
    def name(self) -> str:
        """Identificador curto, por exemplo 'lcm'. Tem que bater com o valor no config.yaml."""
        return "lcm"

    def update_from_response(self, usage: dict) -> None:
        """Chamado depois de cada chamada de LLM, com o dict de uso.

        Atualize self.last_prompt_tokens, self.last_completion_tokens,
        self.last_total_tokens a partir da resposta.
        """

    def should_compress(self, prompt_tokens: int = None) -> bool:
        """Devolve True se a compactação deve disparar neste turno."""

    def compress(self, messages: list, current_tokens: int = None,
                 focus_topic: str = None) -> list:
        """Compacta a lista de mensagens e devolve uma nova lista (possivelmente menor).

        A lista devolvida precisa ser uma sequência de mensagens válida no formato OpenAI.

        ``focus_topic`` é uma string opcional de tópico vinda do
        ``/compress <foco>`` manual; motores que suportam compressão guiada devem
        priorizar a preservação da informação relacionada a ele, os outros podem ignorá-lo.
        """

Atributos de classe que seu motor precisa manter

O agente lê estes diretamente, para exibição e log:

last_prompt_tokens: int = 0
last_completion_tokens: int = 0
last_total_tokens: int = 0
threshold_tokens: int = 0        # quando a compressão dispara
context_length: int = 0          # janela de contexto completa do modelo
compression_count: int = 0       # quantas vezes compress() rodou

Métodos opcionais

Estes têm padrões razoáveis na ABC. Sobrescreva conforme a necessidade:

Método	Padrão	Sobrescreva quando
on_session_start(session_id, **kwargs)	Não faz nada	Você precisa carregar estado persistido (DAG, banco)
on_session_end(session_id, messages)	Não faz nada	Você precisa gravar estado, fechar conexões
on_session_reset()	Zera os contadores de tokens	Você tem estado por sessão para limpar
update_model(model, context_length, ...)	Atualiza context_length + limiar	Você precisa recalcular orçamentos ao trocar de modelo
get_tool_schemas()	Devolve []	Seu motor oferece ferramentas que o agente pode chamar (por exemplo, lcm_grep)
handle_tool_call(name, args, **kwargs)	Devolve um JSON de erro	Você implementa tratadores de ferramenta
should_compress_preflight(messages)	Devolve False	Você consegue fazer uma estimativa barata antes da chamada de API
get_status()	Dict padrão de tokens/limiar	Você tem métricas próprias para expor
select_context(request_messages, *, conversation_messages, incoming_message, budget_tokens)	Devolve None (não faz nada)	Você seleciona/roteia qual contexto entra nesta requisição (recuperação, roteamento por tópico) — veja abaixo
on_turn_complete(messages, usage=None, **kwargs)	Não faz nada	Você ingere/indexa/observa o turno concluído — veja abaixo

Seleção e observação de contexto por turno

O compress() responde a "o contexto está longo demais → encurte". Dois ganchos opcionais, que por padrão não fazem nada, cobrem o eixo ortogonal de seleção / observação, para que um motor não precise mais forçar should_compress() a True e abusar do compress() como callback por turno:

def select_context(self, request_messages, *, conversation_messages=None,
                   incoming_message=None, budget_tokens=0):
    """Escolhe/substitui o contexto DESTA requisição, antes do despacho.

    Devolva uma nova lista de mensagens para usar nesta única chamada ao provedor
    (recuperação, roteamento por tópico, troca de papel/ramo), ou None para deixá-la
    inalterada. Só vale para a requisição: o histórico persistido nunca é alterado.
    """

def on_turn_complete(self, messages, usage=None, **kwargs):
    """Observa um turno concluído, depois que o laço de assistente/ferramenta termina.

    Recebe uma cópia rasa da transcrição finalizada mais o dict canônico de uso do
    turno (ou None se nenhuma resposta de provedor foi alcançada), para que o motor
    possa ingerir/indexar/resumir para o próximo select_context(). O valor de retorno
    é ignorado.
    """

Contrato:

• Não faz nada por padrão, falha aberta. Os dois têm return None como padrão. Um gancho ausente, uma exceção ou um valor de retorno inválido deixam a requisição intacta — então um motor que falha nunca é pior do que não instalar nenhum. O host ainda compara a identidade com o padrão herdado da ABC e o pula inteiramente, então motores que não implementam (incluindo o compressor embutido) não pagam trabalho algum por requisição.
• select_context() só vale para a requisição. A lista devolvida substitui as mensagens de uma única chamada ao provedor; o histórico persistido nunca é gravado. Devolver None, [], algo que não seja lista, ou uma lista com itens que não sejam dicts, tudo isso cai de volta na requisição sem modificação.
• Ordem / estabilidade do cache. O gancho roda antes do controle de cache do prompt e de todos os sanitizadores de requisição, então (a) uma substituição ainda passa pela mesma validação de qualquer requisição, e (b) o padrão que não faz nada deixa a requisição byte a byte idêntica — o comportamento do cache de prompt não muda para motores que não implementam. Um motor que substitui a lista muda só o próprio prefixo de cache. É avaliado por requisição ao provedor (roda de novo nas tentativas repetidas).
• on_turn_complete() é só observação pós-turno; trate messages como somente leitura. A cobertura é de melhor esforço: ele dispara na costura padrão de finalização de turno. Alguns caminhos anormais de retorno antecipado no laço (por exemplo, um bloqueio por política de conteúdo ou uma falha terminal do provedor) persistem e retornam sem passar pela finalização, então hoje não emitem esse gancho — trate-o como observação de melhor esforço para turnos concluídos, não como callback garantido para toda saída antecipada. Unificar todos os caminhos terminais atrás de uma única costura de finalização é um trabalho à parte.

Quando usar esses ganchos — e quando NÃO usar

• Implemente select_context() só quando o seu motor precisar substituir o contexto por requisição — seleção com recuperação, roteamento por tópico/ramo, troca de papel. É o único verbo capaz de trocar quais mensagens entram numa requisição: o gancho de plugin pre_llm_call é só de injeção, por decisão documentada (ele acrescenta à mensagem do usuário e nunca reescreve a lista, para preservar o prefixo do cache de prompt). Se você não precisa substituir, não implemente.
• Se o seu plugin só precisa de observação/ingestão pós-turno (indexação, sincronia de memória, análise), implemente um provedor de memória (sync_turn() — veja Plugins de Provedor de Memória) em vez de um motor de contexto. Um motor de contexto assume a política de compactação da sessão; um provedor de memória observa os turnos sem assumir nada. O on_turn_complete() existe como espelho de observação para motores que já precisam do select_context() — para que o mesmo componente aprenda com o turno que acabou de rotear — e não como callback de turno de uso geral.
• Impacto de um select_context() de verdade no cache de prompt. Uma seleção que realmente faz algo naturalmente muda o prefixo do cache de prompt nos turnos em que muda a seleção — o prefixo daquela requisição deixa de bater com o prefixo em cache no provedor, então esses turnos regravam o cache em vez de lê-lo. Motores devem devolver seleções estáveis quando nada mudou (o mesmo objeto ou uma lista igual) e só remodelar o contexto quando a decisão de roteamento realmente for diferente; uma seleção que muda a cada turno abre mão silenciosamente do reuso de cache em todo turno.

Ferramentas do motor

Motores de contexto podem expor ferramentas que o agente chama diretamente. Devolva os esquemas em get_tool_schemas() e trate as chamadas em handle_tool_call():

def get_tool_schemas(self):
    return [{
        "name": "lcm_grep",
        "description": "Search the context knowledge graph",
        "parameters": {
            "type": "object",
            "properties": {
                "query": {"type": "string", "description": "Search query"}
            },
            "required": ["query"],
        },
    }]

def handle_tool_call(self, name, args, **kwargs):
    if name == "lcm_grep":
        results = self._search_dag(args["query"])
        return json.dumps({"results": results})
    return json.dumps({"error": f"Unknown tool: {name}"})

As ferramentas do motor são injetadas na lista de ferramentas do agente na inicialização e despachadas automaticamente — não precisa registrar em registry nenhum.

Registro

Por diretório (recomendado)

Coloque seu motor em plugins/context_engine/<nome>/. O __init__.py precisa exportar uma subclasse de ContextEngine. O sistema de descoberta acha e instancia sozinho.

Pelo sistema geral de plugins

Um plugin comum também pode registrar um motor de contexto:

def register(ctx):
    engine = LCMEngine(context_length=200000)
    ctx.register_context_engine(engine)

Só um motor pode ser registrado. Um segundo plugin que tente registrar é rejeitado com um aviso.

Ciclo de vida

1. Motor instanciado (carga do plugin ou descoberta por diretório)
2. on_session_start() — a conversa começa
3. update_from_response() — depois de cada chamada de API
4. should_compress() — checado a cada turno
5. compress() — chamado quando should_compress() devolve True
6. on_session_end() — fim da sessão (saída da CLI, /reset, expiração no gateway)

O on_session_reset() é chamado em /new ou /reset para limpar o estado por sessão sem desligar tudo.

Configuração

Os usuários escolhem o seu motor por hermes plugins → Provider Plugins → Context Engine, ou editando o config.yaml:

context:
  engine: "lcm"   # tem que bater com a propriedade name do seu motor

O bloco de configuração compression (compression.threshold, compression.protect_last_n etc.) é específico do ContextCompressor embutido, com uma exceção explícita: compression.model_thresholds (limiares sobrescritos por modelo) faz parte do contrato de motor de contexto. O host atribui o mapa resolvido a engine.model_thresholds antes da chamada inicial de update_model(), e o update_model() da classe base o aplica (casamento pela maior substring, caindo para o limiar configurado do motor). Motores que sobrescrevem update_model() são donos da própria política de compactação e podem honrar ou ignorar o mapa — use from agent.context_compressor import resolve_model_threshold para reaproveitar a mesma lógica de resolução. Para todo o resto, seu motor deve definir o próprio formato de configuração, se precisar, lendo do config.yaml durante a inicialização.

Testes

from agent.context_engine import ContextEngine

def test_engine_satisfies_abc():
    engine = YourEngine(context_length=200000)
    assert isinstance(engine, ContextEngine)
    assert engine.name == "your-name"

def test_compress_returns_valid_messages():
    engine = YourEngine(context_length=200000)
    msgs = [{"role": "user", "content": "hello"}]
    result = engine.compress(msgs)
    assert isinstance(result, list)
    assert all("role" in m for m in result)

Veja tests/agent/test_context_engine.py para a suíte completa de testes do contrato da ABC.

Segurança entre threads

Quando compression.context_timeout_seconds > 0 (o padrão), o Hermes roda a passagem inteira de compressão — incluindo o compress() do seu motor e os callbacks de fronteira, e o on_pre_compress / on_session_switch de qualquer provedor de memória — numa thread daemon de um pool, com tempo limite do lado do host. Seu motor precisa, portanto, assumir que:

• As chamadas podem chegar em qualquer thread do pool. Não conte com afinidade de thread nem com estado em threading.local compartilhado com a thread da conversa.
• A lista de mensagens que você recebe é um instantâneo profundo e privado; alterá-la no lugar é permitido (contrato legado), mas a alteração só fica visível se a passagem for confirmada. Depois de um tempo limite do host, o trabalho que ainda estava rodando é descartado — nunca publique em estado externo/durável fora da confirmação.
• Passagens de sessões diferentes podem rodar ao mesmo tempo em threads irmãs do pool; uma única instância de motor/provedor compartilhada entre sessões precisa ser segura entre threads.

Veja também

• Compressão e Cache de Contexto — como o compressor embutido funciona
• Plugins de Provedor de Memória — sistema análogo de plugin de seleção única, para memória
• Plugins — visão geral do sistema de plugins

---