Padrões e antipadrões

Os hooks são uma costura pequena, e é fácil usá-los bem ou mal. Esta página reúne os formatos que se sustentam em produção e os que mordem.

• Padrões
  • Log de auditoria
  • Censura de PII
  • Cache
  • Métricas
  • Guardrails
  • Gating por confiança
  • Override de roteamento
  • Ciclo de vida do modelo
  • Contexto multi-tenant
  • Composição
  • Instrumentação com escopo
  • Instrumentação para todo o processo
  • Ajuste do orçamento de tokens
• Antipadrões
  • Trabalho bloqueante
  • Lançar de hooks de fim para controle de fluxo
  • Estado mutável compartilhado sem lock
  • Falha silenciosa
  • Retenção de contextos
  • Censurar tarde demais
  • Lógica por pergunta em um hook por chamada
  • Predict recursivo
  • Callables simples em hooks=
  • Presumir que results existem em hooks de fim
  • Hooks dependentes de ordem

Padrões

Log de auditoria

Registre cada decisão com o suficiente para reconstruí-la: o estado, as perguntas, as respostas, o modelo, a decisão de roteamento, o uso e a latência.

import json

def audit(ctx):
    for state, result in zip(ctx.states, ctx.results or []):
        json.dump({
            "run_id": ctx.run_id,
            "model": ctx.model,
            "state": state,
            "routing": result.get("routing"),
            "answers": result["answers"],
            "usage": result.get("usage"),
            "call_usage": ctx.usage,
            "call_elapsed_ms": round(ctx.elapsed_ms or 0.0, 3),
        }, sys.stdout)
        sys.stdout.write("\n")

laya.load("convaiinnovations/laya", on_predict_end=audit)

Um hook dispara uma vez por chamada, e uma chamada de predict_batch carrega todos os estados nela, então o registro é gravado por decisão: ctx.states e ctx.results ficam alinhados por índice. ctx.usage e ctx.elapsed_ms são totais da chamada inteira; cada resultado carrega seu próprio usage.

Torne-o tolerante se perder uma linha de log não pode falhar uma solicitação: hooks_raise=False. Torne-o estrito se a trilha de auditoria é um requisito de conformidade.

Censura de PII

A censura precisa acontecer em on_predict_start, antes da tokenização, ou o modelo já viu os dados.

import re
EMAIL = re.compile(r"\b[\w.+-]+@[\w-]+\.[\w.-]+\b")

def redact(ctx):
    ctx.states = [
        EMAIL.sub("[email]", s) if isinstance(s, str) else s
        for s in ctx.states
    ]

laya.load("convaiinnovations/laya", on_predict_start=redact)

Um hook de censura é um hook de política: mantenha hooks_raise=True, porque um censurador quebrado em silêncio é um vazamento de dados.

Cache

Um hook de início verifica o cache e chama ctx.skip(...); um hook de fim o preenche. A passada direta é pulada em um acerto.

import hashlib, json

CACHE = {}

def key(ctx, index):
    # Not sort_keys=True: criteria order is positional, so two orders are two questions,
    # and the checkpoint and token budget change the answer too.
    payload = json.dumps([ctx.states[index], ctx.questions, ctx.model,
                          ctx.max_len, ctx.head_max_len], default=str)
    return hashlib.sha256(payload.encode()).hexdigest()

def read(ctx):
    hits = [CACHE.get(key(ctx, i)) for i in range(len(ctx.states))]
    if all(hit is not None for hit in hits):
        ctx.skip(hits)   # one per state: skip replaces the whole call

def write(ctx):
    for i, result in enumerate(ctx.results or []):
        CACHE[key(ctx, i)] = result

laya.load("convaiinnovations/laya", on_predict_start=read, on_predict_end=write)

Os hooks disparam uma vez por chamada, então usar como chave um único estado não basta em predict_batch: ctx.skip() substitui todo resultado que a chamada teria retornado. Proteja o cache com um lock ao servir de forma concorrente. No Router a carga útil em cache ainda ganha uma chave routing, então o formato do retorno não muda.

O mesmo par funciona em um único nó LangChain através do seu argumento hooks=, que é a forma de cachear um passo quente em um grafo sem mudar o que todo outro chamador daquele agente vê.

Métricas

Contadores e histogramas a partir de ctx.model, ctx.usage e ctx.elapsed_ms. Mantenha-o tolerante.

COUNTS, LATENCIES = {}, []

def metrics(ctx):
    COUNTS[ctx.model] = COUNTS.get(ctx.model, 0) + 1
    if ctx.elapsed_ms is not None:
        LATENCIES.append(ctx.elapsed_ms)

laya.load("convaiinnovations/laya", on_predict_end=metrics, hooks_raise=False)

Guardrails

Um hook de política lança para bloquear uma solicitação. hooks_raise=True (o padrão) deixa o bloqueio chegar ao chamador; on_error e on_predict_end ainda rodam, então a trilha de auditoria o registra.

class Blocked(Exception):
    pass

def guard(ctx):
    if any("ssn" in str(state).lower() for state in ctx.states):
        raise Blocked("possible PII in state")

laya.load("convaiinnovations/laya", on_predict_start=guard)

Teste-o contra o formato do estado. Um guard que só lê ctx.states[0] bloqueia uma chamada única e deixa uma chamada predict_batch colocar todos os estados restantes pela passada direta.

Gating por confiança

Um hook de fim reescreve uma resposta de baixa confiança para um fallback seguro, ou a anota para lógica a jusante. Isto é uma mutação de resultado, não uma rejeição.

def gate(ctx):
    for result in ctx.results or []:
        answer = result["answers"].get("dept")
        if answer and answer["confidence"] < 0.6:
            answer["choice"] = "human-review"
            answer["gated"] = True

laya.load("convaiinnovations/laya", on_predict_end=gate)

Mute através de ctx.results, que guarda um dict por estado da chamada: aplicar gating só ao primeiro entrega todas as outras respostas de baixa confiança sem anotação.

Override de roteamento

on_route pode substituir ctx.decision para fixar um checkpoint para uma classe de tráfego.

from laya.router import RouteDecision

def pin(ctx):
    if "refund" in str(ctx.states[0]).lower():
        ctx.decision = RouteDecision(
            model="typed-decisions",
            repo="convaiinnovations/laya/typed-decisions",
            reason="refund workflow",
            detection=None,
            workflow=None,
        )

Router(hooks=[pin])

Ciclo de vida do modelo

on_load e on_evict observam checkpoints. Use-os para logs de aquecimento, contabilidade de memória ou alertas de remoção. Eles rodam fora do lock do Router, então um hook pode chamar de volta o Router.

class Lifecycle:
    def on_load(self, ctx):
        print("loaded", ctx.model)

    def on_evict(self, ctx):
        print("evicted", ctx.model)

Router(hooks=[Lifecycle()])

Contexto multi-tenant

Passe um id de tenant por fio capturando-o no closure do hook, ou lendo-o de um context-local. Não guarde estado por solicitação no objeto de hook sem um lock.

def make_audit(tenant):
    def audit(ctx):
        ship(tenant, ctx.run_id, ctx.results)
    return audit

agent = laya.load("convaiinnovations/laya", on_predict_end=make_audit("acme"))

Composição

Vários hooks de tipos diferentes se compõem naturalmente; os hooks instalados rodam primeiro, em ordem.

agent = laya.load(
    "convaiinnovations/laya",
    hooks=[Metrics(), Guardrail()],     # metrics first, then policy
    on_predict_start=redact,            # convenience callables appended after hooks
    hooks_raise=True,                   # policy failures are fatal
)

Mantenha a ordenação deliberada e documentada, porque um hook posterior vê as mutações de um anterior.

Instrumentação com escopo

Anexe um tracer ou hook de debug apenas para o código que precisa dele, em vez de reconstruir o agente. hooks_installed restaura a lista anterior na saída, mesmo que o bloco lance.

with agent.hooks_installed(DebugDump()):
    agent.system_one(state, questions)   # DebugDump only here

add_hook/remove_hook fazem o mesmo sem um bloco, para um tracer que vive tanto quanto o processo.

Instrumentação para todo o processo

Um tracer ou hook de métricas que toda decisão deveria ver pode ser registrado uma vez, em vez de ser passado a cada Agent e Router. Os padrões rodam antes dos hooks da instância e por chamada.

from laya import BaseHook, hooks

class Metrics(BaseHook):
    def on_predict_end(self, ctx):
        record(ctx.model, ctx.elapsed_ms)

hooks.set_default_hooks(hooks=[Metrics()])

Isto é estado global, então dê escopo deliberadamente: defina-o uma vez na inicialização, e clear_default_hooks() nos testes para que um teste não vaze um hook para o próximo.

Ajuste do orçamento de tokens

Um hook de início pode aumentar o orçamento de tokens de uma chamada, por exemplo quando uma pergunta tem muitas opções e o orçamento padrão da cabeça colapsaria os rótulos. Quatro detalhes decidem se o hook ajuda ou deixa a chamada pior em silêncio:

• O ctx.head_max_len de um hook de início substitui o orçamento da chamada. O que está em vigor antes dele é o valor por chamada do próprio chamador, ou o padrão do checkpoint em ctx.agent.cfg – então compare com isso: escrever um número simples pode baixar um orçamento que um chamador já tinha definido.
• Uma chamada responde toda pergunta que carrega, então dimensione pela mais larga delas, não pela que por acaso vier primeiro.
• Quando as opções já não cabem na cabeça, laya/common.py dá a cada uma delas max(4, (head_max_len - 16) // k) tokens. 16 + 4 * k portanto cai exatamente nesse piso: todo rótulo ainda é cortado para os tokens que compartilha com os outros, que é o colapso que o hook foi escrito para evitar. 16 + 8 * k os deixa distinguíveis.
• O estado recebe max_len - head_max_len - 8 tokens, então uma cabeça alargada tem que alargar max_len junto ou o estado perde sua janela.

def widen_for_high_cardinality(ctx):
    k = max((len(q.get("criteria", {}) or {}) for q in ctx.questions.values()), default=0)
    if k < 50:
        return
    cfg = getattr(ctx.agent, "cfg", None) or {}
    head = ctx.head_max_len if ctx.head_max_len is not None else cfg.get("head_max_len", 192)
    window = ctx.max_len if ctx.max_len is not None else cfg.get("max_len", 512)
    need = 16 + 8 * k                          # 8 tokens per label, not the core's floor of 4
    if need > head:                            # only ever widen, never lower
        ctx.head_max_len = need
        ctx.max_len = max(window, need + 8 + 64)   # 8 reserved, then room for the state

agent = laya.load("convaiinnovations/laya", on_predict_start=widen_for_high_cardinality)

Isto não toca na config compartilhada do agente, então chamadas concorrentes não são afetadas. Os mesmos ajustes estão disponíveis por chamada: agent.system_one(state, questions, head_max_len=512, max_len=1024).

Alargar não é de graça: uma janela mais longa significa um tensor maior, e os checkpoints foram treinados em 512 (laya) e 1.024 tokens. Além disso, estreitar os candidatos com predict_shortlist supera esticar o orçamento.

Antipadrões

Trabalho bloqueante

Os hooks rodam na thread que chama, e o laya.serve usa um único worker de inferência. Um hook que dorme, espera por uma ida e volta de rede, ou chama input() trava toda outra solicitação atrás dele.

# bad: blocks the whole server
def audit(ctx):
    requests.post("https://slow.example/decisions", json=..., timeout=30)

# better: enqueue, let a background worker ship it
def audit(ctx):
    QUEUE.put_nowait(record(ctx))

Se você precisa fazer trabalho lento, defina hooks_concurrent=False para ao menos impedir que o próprio hook se sobreponha, e rode o laya.serve atrás de uma fila.

Lançar de hooks de fim para controle de fluxo

on_predict_end roda depois da inferência. Lançar ali joga fora um resultado calculado e, no caminho de sucesso, aparece para o chamador. Use um hook de início para bloquear antes de pagar pela inferência, ou reescreva ctx.results para mudar a resposta.

Estado mutável compartilhado sem lock

A mesma instância de hook roda em muitas threads. self.counter += 1 tem corrida.

# bad
class Count:
    def __init__(self): self.n = 0
    def on_predict_end(self, ctx): self.n += 1

# good
import threading
class Count:
    def __init__(self):
        self.n = 0
        self._lock = threading.Lock()
    def on_predict_end(self, ctx):
        with self._lock:
            self.n += 1

Falha silenciosa

hooks_raise=False avisa uma vez por falha, mas um hook que captura tudo por conta própria esconde problemas reais.

# bad: no one will ever know the audit trail stopped
def audit(ctx):
    try:
        ship(record(ctx))
    except Exception:
        pass

Se um hook é opcional, deixe hooks_raise=False cuidar dele e fique de olho nos avisos. Se não é, deixe-o lançar.

Retenção de contextos

Um hook que anexa ctx a uma lista mantém vivo todo o estado, as perguntas, os resultados e o agente.

# bad: unbounded memory growth
SEEN = []
def audit(ctx):
    SEEN.append(ctx)

# good: keep only what you need
SEEN = []
def audit(ctx):
    SEEN.append((ctx.run_id, ctx.model, ctx.elapsed_ms))

Censurar tarde demais

Em on_predict_end o modelo já tokenizou o estado. Censure em on_predict_start.

Lógica por pergunta em um hook por chamada

Há um PredictContext por chamada, e uma passada direta responde toda pergunta. Não há eventos por pergunta. Itere as respostas dentro de on_predict_end, e itere os estados também: em um lote, um contexto carrega todos os estados da chamada.

def flag(ctx):
    for result in ctx.results or []:
        for qid, answer in result["answers"].items():
            if answer.get("confidence", 1.0) < 0.5:
                alert(qid, ctx.run_id)

Predict recursivo

Um hook que chama agent.predict/system_one roda os hooks de novo. Sem uma proteção de profundidade isto recursa.

# bad
def enrich(ctx):
    ctx.results = [agent.predict(ctx.states[0], EXTRA_QUESTIONS)]

# good: guard, or use a separate agent with no hooks
def enrich(ctx):
    if getattr(ctx, "_enriched", False):
        return
    ctx._enriched = True
    ctx.results = [enricher.predict(state, EXTRA_QUESTIONS) for state in ctx.states]

Callables simples em hooks=

hooks= aceita objetos de hook; um callable puro não diz para qual evento ele serve, então é rejeitado. Use on_predict_start= / on_predict_end=.

# bad: TypeError
laya.load("convaiinnovations/laya", hooks=[lambda ctx: None])

# good
laya.load("convaiinnovations/laya", on_predict_end=lambda ctx: None)

Presumir que results existem em hooks de fim

No caminho de falha ctx.results é None, a menos que um hook de início o tenha definido. Verifique sempre.

def audit(ctx):
    if ctx.results is None:
        log_failure(ctx.run_id, ctx.error)
        return
    log_success(ctx.run_id, ctx.results)

Hooks dependentes de ordem

Um hook que lê uma mutação de outro hook é frágil a menos que a ordem esteja fixada. Os hooks instalados rodam na ordem da lista, depois os callables de conveniência; documente qualquer acoplamento, ou junte os hooks acoplados em um único objeto.

Veja também

• Erros: a matriz de falhas por trás de vários destes antipadrões.
• Exemplos: versões mais completas dos padrões acima.