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 as formas que se aguentam em produção e as que mordem.

• Padrões
  • Registo de auditoria
  • Censura de PII
  • Colocação em cache
  • Métricas
  • Guardrails
  • Gating por confiança
  • Substituição do encaminhamento
  • Ciclo de vida do modelo
  • Contexto multi-tenant
  • Composição
  • Instrumentação com âmbito
  • Instrumentação para todo o processo
  • Ajuste do orçamento de tokens
• Antipadrões
  • Trabalho bloqueante
  • Levantar de hooks de fim para fluxo de controlo
  • Estado mutável partilhado sem lock
  • Falha silenciosa
  • Retenção de contextos
  • Censurar demasiado tarde
  • Lógica por pergunta num hook por chamada
  • Predição recursiva
  • Invocáveis simples em hooks=
  • Assumir que os resultados existem em hooks de fim
  • Hooks dependentes da ordem

Padrões

Registo de auditoria

Registra cada decisão com o suficiente para a reconstruir: o estado, as perguntas, as respostas, o modelo, a decisão de encaminhamento, o usage 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 predict_batch transporta todos os estados nela, por isso o registo é escrito por decisão: ctx.states e ctx.results estão alinhados por índice. ctx.usage e ctx.elapsed_ms são totais para toda a chamada; cada resultado transporta o seu próprio usage.

Torna-o tolerante se perder uma linha de log não deve falhar um pedido: hooks_raise=False. Torna-o estrito se o rasto de auditoria for um requisito de conformidade.

Censura de PII

A censura tem de 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: mantém hooks_raise=True, porque um censurador silenciosamente avariado é uma fuga de dados.

Colocação em cache

Um hook de início verifica a cache e chama ctx.skip(...); um hook de fim enche-a. A passagem direta é saltada numa cache hit.

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, por isso basear a chave num só estado não chega no predict_batch: ctx.skip() substitui cada resultado que a chamada teria devolvido. Protege a cache com um lock quando serves em simultâneo. No Router, o payload em cache continua a receber uma chave routing, por isso a forma do retorno fica inalterada.

O mesmo par funciona num único nó LangChain através do seu argumento hooks=, que é a forma de colocar em cache um passo quente de um grafo sem alterar o que todos os outros autores de chamadas desse agente veem.

Métricas

Contadores e histogramas a partir de ctx.model, ctx.usage e ctx.elapsed_ms. Mantém-no 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 levanta exceção para bloquear um pedido. hooks_raise=True (a predefinição) deixa o bloqueio chegar ao autor da chamada; on_error e on_predict_end continuam a correr, por isso o rasto de auditoria registá-lo.

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)

Testa-o contra a forma do estado. Um guard que só lê ctx.states[0] bloqueia uma única chamada e deixa uma chamada predict_batch passar todos os restantes estados pela passagem direta.

Gating por confiança

Um hook de fim reescreve uma resposta de baixa confiança para um fallback seguro, ou anota-a 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)

Muta através de ctx.results, que contém um dict por estado da chamada: fazer gating só sobre o primeiro entrega todas as outras respostas de baixa confiança sem anotação.

Substituição do encaminhamento

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. Usa-os para logs de aquecimento, contabilidade de memória, ou alertas de libertação. Correm fora do lock do Router, por isso um hook pode chamar de volta para 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

Faz passar um id de tenant capturando-o na closure do hook, ou lendo-o de um contexto local. Não guardes estado por pedido no objeto 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 compõem-se naturalmente; os hooks instalados correm primeiro, por 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
)

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

Instrumentação com âmbito

Anexa um tracer ou hook de depuração apenas para o código que precisa dele, em vez de reconstruir o agente. hooks_installed restaura a lista anterior à saída, mesmo se o bloco levantar exceção.

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 a decisão deve ver pode ser registado uma vez, em vez de ser passado a cada Agent e Router. As predefinições correm 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, por isso delimita-o deliberadamente: define-o uma vez no arranque, e chama clear_default_hooks() nos testes para que um teste não deixe escapar um hook para o seguinte.

Ajuste do orçamento de tokens

Um hook de início pode aumentar o orçamento de tokens para uma chamada, por exemplo quando uma pergunta tem muitas opções e o orçamento de cabeça predefinido colapsaria as etiquetas. Quatro detalhes decidem se o hook ajuda ou piora silenciosamente a chamada:

• O ctx.head_max_len de um hook de início substitui o orçamento para a chamada. O que está em vigor antes dele é o valor por chamada do próprio autor da chamada, ou a predefinição do checkpoint em ctx.agent.cfg – por isso compara com isso: escrever um número simples pode baixar um orçamento que um autor de chamada já definiu.
• Uma chamada responde a todas as perguntas que transporta, por isso dimensiona pela mais larga delas, e não por aquela que por acaso vem primeiro.
• Assim que as opções deixam de caber na cabeça, laya/common.py dá a cada uma max(4, (head_max_len - 16) // k) tokens. 16 + 4 * k cai, portanto, exatamente nesse piso: cada etiqueta é ainda cortada para os tokens que partilha com as outras, que é o colapso que o hook foi escrito para evitar. 16 + 8 * k deixa-as distinguíveis.
• O estado recebe max_len - head_max_len - 8 tokens, por isso uma cabeça alargada tem de alargar max_len com ela, ou o estado perde a 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 partilhada do agente, por isso as chamadas concorrentes não são afetadas. Os mesmos botões estão disponíveis por chamada: agent.system_one(state, questions, head_max_len=512, max_len=1024).

Alargar não é grátis: uma janela mais longa significa um tensor maior, e os checkpoints foram treinados a 512 (laya) e 1,024 tokens. Para lá disso, estreitar os candidatos com predict_shortlist bate esticar o orçamento.

Antipadrões

Trabalho bloqueante

Os hooks correm 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 à rede, ou chama input() trava todos os outros pedidos 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 tens mesmo de fazer trabalho lento, define hooks_concurrent=False para pelo menos evitar que o próprio hook se sobreponha, e corre o laya.serve atrás de uma fila.

Levantar de hooks de fim para fluxo de controlo

on_predict_end corre depois da inferência. Levantar aí deita fora um resultado calculado e, no caminho de sucesso, aflora ao autor da chamada. Usa um hook de início para bloquear antes de pagar a inferência, ou reescreve ctx.results para alterar a resposta.

Estado mutável partilhado sem lock

A mesma instância de hook corre em muitas threads. self.counter += 1 corre a 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 apanha tudo ele próprio 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, deixa o hooks_raise=False tratá-lo e vigia os avisos. Se não é, deixa-o levantar.

Retenção de contextos

Um hook que acrescenta 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 demasiado tarde

Quando chega a on_predict_end, o modelo já tokenizou o estado. Censura em on_predict_start.

Lógica por pergunta num hook por chamada

Há um PredictContext por chamada, e uma passagem direta responde a todas as perguntas. Não há eventos por pergunta. Itera as respostas dentro de on_predict_end, e itera também os estados: num lote, um contexto transporta 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)

Predição recursiva

Um hook que chama agent.predict/system_one corre os hooks de novo. Sem uma guarda 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]

Invocáveis simples em hooks=

hooks= aceita objetos hook; um invocável nu não diz para que evento é, por isso é rejeitado. Usa 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)

Assumir que os resultados existem em hooks de fim

No caminho de falha, ctx.results é None a menos que um hook de início o tenha definido. Verifica 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 da ordem

Um hook que lê uma mutação de outro hook é frágil a menos que a ordem esteja fixada. Os hooks instalados correm pela ordem da lista, depois os invocáveis de conveniência; documenta qualquer acoplamento, ou funde os hooks acoplados num só objeto.

Ver também

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