Patrones y antipatrones

Los hooks son un punto de extensión pequeño, y es fácil usarlos bien o mal. Esta página reúne las formas que resisten en producción y las que muerden.

• Patrones
  • Registro de auditoría
  • Censura de PII
  • Caché
  • Métricas
  • Guardarraíles
  • Gating por confianza
  • Anulación del enrutamiento
  • Ciclo de vida del modelo
  • Contexto multiinquilino
  • Composición
  • Instrumentación local
  • Instrumentación global
  • Ajuste del presupuesto de tokens
• Antipatrones
  • Trabajo bloqueante
  • Lanzar excepciones desde hooks de fin para el flujo de control
  • Estado mutable compartido sin bloqueo
  • Fallo silencioso
  • Retener contextos
  • Censurar demasiado tarde
  • Lógica por pregunta en un hook por llamada
  • Predict recursivo
  • Callables simples en hooks=
  • Asumir que los resultados existen en los hooks de fin
  • Hooks dependientes del orden

Patrones

Registro de auditoría

Registra cada decisión con lo suficiente para reconstruirla: el estado, las preguntas, las respuestas, el modelo, la decisión de enrutamiento, el uso y la latencia.

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)

Un hook se dispara una vez por llamada, y una llamada a predict_batch lleva dentro todos los estados, así que el registro se escribe por decisión: ctx.states y ctx.results están alineados por índice. ctx.usage y ctx.elapsed_ms son totales de toda la llamada; cada resultado lleva su propio usage.

Hazlo permisivo si perder una línea de log no debe hacer fallar una solicitud: hooks_raise=False. Hazlo estricto si el rastro de auditoría es un requisito de cumplimiento.

Censura de PII

La censura debe ocurrir en on_predict_start, antes de la tokenización, o el modelo ya habrá visto los datos.

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)

Un hook de censura es un hook de política: mantén hooks_raise=True, porque un censurador que falla en silencio es una fuga de datos.

Caché

Un hook de inicio consulta la caché y llama a ctx.skip(...); un hook de fin la rellena. La pasada hacia adelante se omite cuando hay acierto.

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)

Los hooks se disparan una vez por llamada, así que usar como clave un solo estado no basta en predict_batch: ctx.skip() reemplaza todos los resultados que la llamada habría devuelto. Protege la caché con un bloqueo cuando sirvas en concurrencia. En el Router, el payload cacheado igual recibe una clave routing, así que la forma del valor devuelto no cambia.

El mismo par funciona en un solo nodo de LangChain mediante su argumento hooks=, que es la forma de cachear un paso caliente de un grafo sin cambiar lo que ve cualquier otro llamador de ese agente.

Métricas

Contadores e histogramas a partir de ctx.model, ctx.usage y ctx.elapsed_ms. Hazlo permisivo.

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)

Guardarraíles

Un hook de política lanza una excepción para bloquear una solicitud. hooks_raise=True (el valor por defecto) deja que el bloqueo llegue al llamador; on_error y on_predict_end se siguen ejecutando, así que el rastro de auditoría lo 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)

Pruébalo contra la forma del estado. Un guardarraíl que solo lee ctx.states[0] bloquea una sola llamada y deja que una llamada a predict_batch meta todos los estados restantes por la pasada hacia adelante.

Gating por confianza

Un hook de fin reescribe una respuesta de baja confianza por un fallback seguro, o la anota para la lógica posterior. Esto es una mutación del resultado, no un rechazo.

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 a través de ctx.results, que contiene un dict por cada estado de la llamada: aplicar gating solo al primero entrega todas las demás respuestas de baja confianza sin anotar.

Anulación del enrutamiento

on_route puede reemplazar ctx.decision para fijar un checkpoint a una clase de tráfico.

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 del modelo

on_load y on_evict observan checkpoints. Úsalos para logs de calentamiento, contabilidad de memoria o alertas de desalojo. Se ejecutan fuera del bloqueo del Router, así que un hook puede volver a llamar al 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 multiinquilino

Propaga un id de inquilino capturándolo en el closure del hook, o leyéndolo de un contexto local. No guardes estado por solicitud en el objeto hook sin un bloqueo.

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"))

Composición

Varios hooks de distinto tipo se componen de forma natural; los hooks instalados se ejecutan primero, en orden.

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én el orden deliberado y documentado, porque un hook posterior ve las mutaciones de uno anterior.

Instrumentación local

Adjunta un hook de trazado o depuración solo al código que lo necesita, en lugar de reconstruir el agente. hooks_installed restaura la lista anterior al salir, incluso si el bloque lanza una excepción.

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

add_hook/remove_hook hacen lo mismo sin un bloque, para un trazador que dura tanto como el proceso.

Instrumentación global

Un hook de trazado o de métricas que toda decisión deba ver se puede registrar una vez, en lugar de pasarlo a cada Agent y cada Router. Los valores por defecto se ejecutan antes de los hooks de la instancia y los de la llamada.

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()])

Esto es estado global, así que delimítalo a conciencia: establételo una vez al arrancar, y llama a clear_default_hooks() en las pruebas para que una prueba no filtre un hook a la siguiente.

Ajuste del presupuesto de tokens

Un hook de inicio puede subir el presupuesto de tokens de una llamada, por ejemplo cuando una pregunta tiene muchas opciones y el presupuesto de head por defecto colapsaría las etiquetas. Cuatro detalles deciden si el hook ayuda o si empeora la llamada en silencio:

• El ctx.head_max_len de un hook de inicio reemplaza el presupuesto de la llamada. Lo que rige antes es el valor por llamada del propio llamador, o el valor por defecto del checkpoint en ctx.agent.cfg – así que compáralo con eso: escribir un número fijo puede bajar el presupuesto que el llamador ya había fijado.
• Una llamada responde todas las preguntas que lleva, así que dimensiona según la más ancha, no según la que llegue primero.
• Cuando las opciones ya no caben en el head, laya/common.py da a cada una max(4, (head_max_len - 16) // k) tokens. Por lo tanto, 16 + 4 * k cae justo en ese mínimo: cada etiqueta igual se recorta a los tokens que comparte con las demás, que es el colapso que el hook se escribió para evitar. 16 + 8 * k las deja distinguibles.
• El estado recibe max_len - head_max_len - 8 tokens, así que un head ensanchado tiene que ensanchar max_len con él o el estado pierde su ventana.

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)

Esto no toca la configuración compartida del agente, así que las llamadas concurrentes no se ven afectadas. Los mismos ajustes están disponibles por llamada: agent.system_one(state, questions, head_max_len=512, max_len=1024).

Ensanchar no es gratis: una ventana más larga implica un tensor más grande, y los checkpoints se entrenaron a 512 (laya) y 1,024 tokens. Más allá de eso, estrechar los candidatos con predict_shortlist supera a estirar el presupuesto.

Antipatrones

Trabajo bloqueante

Los hooks se ejecutan en el hilo llamador, y laya.serve usa un único worker de inferencia. Un hook que duerme, espera un viaje de ida y vuelta por la red o llama a input() atasca todas las demás solicitudes detrás de él.

# 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))

Si tienes que hacer trabajo lento, pon hooks_concurrent=False para al menos evitar que el propio hook se solape, y ejecuta laya.serve detrás de una cola.

Lanzar excepciones desde hooks de fin para el flujo de control

on_predict_end se ejecuta después de la inferencia. Lanzar una excepción ahí descarta un resultado ya calculado y, en la ruta de éxito, aflora al llamador. Usa un hook de inicio para bloquear antes de pagar la inferencia, o reescribe ctx.results para cambiar la respuesta.

Estado mutable compartido sin bloqueo

La misma instancia de hook se ejecuta en muchos hilos. self.counter += 1 tiene condiciones de carrera.

# 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

Fallo silencioso

hooks_raise=False avisa una vez por cada fallo, pero un hook que atrapa todo por su cuenta oculta problemas reales.

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

Si un hook es opcional, deja que hooks_raise=False se encargue y vigila los avisos. Si no lo es, deja que lance la excepción.

Retener contextos

Un hook que agrega ctx a una lista mantiene vivos todo el estado, las preguntas, los resultados y el 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

Para cuando se ejecuta on_predict_end, el modelo ya tokenizó el estado. Censura en on_predict_start.

Lógica por pregunta en un hook por llamada

Hay un solo PredictContext por llamada, y una pasada hacia adelante responde todas las preguntas. No hay eventos por pregunta. Itera las respuestas dentro de on_predict_end, e itera también los estados: en un lote, un contexto lleva todos los estados de la llamada.

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

Un hook que llama a agent.predict/system_one ejecuta los hooks otra vez. Sin una guarda de profundidad, esto recurre.

# 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 en hooks=

hooks= acepta objetos hook; un callable simple no dice para qué evento es, así que se rechaza. 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)

Asumir que los resultados existen en los hooks de fin

En la ruta de fallo, ctx.results es None a menos que un hook de inicio lo establezca. Comprueba siempre.

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

Hooks dependientes del orden

Un hook que lee una mutación de otro hook es frágil a menos que el orden esté fijado. Los hooks instalados se ejecutan en el orden de la lista, y después los callables de conveniencia; documenta cualquier acoplamiento, o combina los hooks acoplados en un solo objeto.

Ver también

• Errores: la matriz de fallos detrás de varios de estos antipatrones.
• Ejemplos: versiones más completas de los patrones anteriores.