Motifs et anti-motifs

Les hooks sont une petite couture, et il est facile de bien ou mal les utiliser. Cette page rassemble les formes qui tiennent en production et celles qui mordent.

• Motifs
  • Journal d’audit
  • Masquage des PII
  • Mise en cache
  • Métriques
  • Garde-fous
  • Gating par confiance
  • Surcharge du routage
  • Cycle de vie du modèle
  • Contexte multi-locataire
  • Composition
  • Instrumentation limitée à un bloc
  • Instrumentation à l’échelle du processus
  • Ajustement du budget de jetons
• Anti-motifs
  • Travail bloquant
  • Lever depuis les hooks de fin pour le flux de contrôle
  • État partagé mutable sans verrou
  • Échec silencieux
  • Retenir les contextes
  • Masquer trop tard
  • Logique par question dans un hook par appel
  • Predict récursif
  • Appelables simples dans hooks=
  • Supposer que les résultats existent dans les hooks de fin
  • Hooks dépendants de l’ordre

Motifs

Journal d’audit

Enregistre chaque décision avec de quoi la reconstruire : l’état, les questions, les réponses, le modèle, la décision de routage, l’usage et la latence.

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 déclenche une fois par appel, et un appel predict_batch porte tous ses états, donc l’enregistrement est écrit par décision : ctx.states et ctx.results sont alignés par index. ctx.usage et ctx.elapsed_ms sont des totaux pour tout l’appel ; chaque résultat porte son propre usage.

Rends-le indulgent si perdre une ligne de log ne doit pas faire échouer une requête : hooks_raise=False. Rends-le strict si la piste d’audit est une exigence de conformité.

Masquage des PII

Le masquage doit avoir lieu dans on_predict_start, avant la tokenisation, sinon le modèle a déjà vu les données.

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 masquage est un hook de politique : garde hooks_raise=True, parce qu’un masqueur cassé silencieusement est une fuite de données.

Mise en cache

Un hook de start consulte le cache et appelle ctx.skip(...) ; un hook de fin le remplit. La passe avant est sautée en cas de 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)

Les hooks se déclenchent une fois par appel, donc indexer sur un seul état ne suffit pas sur predict_batch : ctx.skip() remplace chaque résultat que l’appel aurait renvoyé. Protège le cache avec un verrou quand tu sers en parallèle. Sur le Router, la charge utile en cache reçoit quand même une clé routing, donc la forme de retour est inchangée.

La même paire fonctionne sur un seul nœud LangChain via son argument hooks=, ce qui est la façon de mettre en cache une étape chaude d’un graphe sans changer ce que voit chaque autre appelant de cet agent.

Métriques

Des compteurs et des histogrammes à partir de ctx.model, ctx.usage et ctx.elapsed_ms. Garde-le indulgent.

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)

Garde-fous

Un hook de politique lève une erreur pour bloquer une requête. hooks_raise=True (le défaut) laisse le blocage atteindre l’appelant ; on_error et on_predict_end tournent quand même, donc la piste d’audit l’enregistre.

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-le contre la forme de l’état. Une garde qui ne lit que ctx.states[0] bloque un appel unique et laisse un appel predict_batch faire passer tous les états restants dans la passe avant.

Gating par confiance

Un hook de fin réécrit une réponse à faible confiance vers un fallback sûr, ou l’annote pour la logique en aval. C’est une mutation de résultat, pas un rejet.

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 via ctx.results, qui contient un dict par état de l’appel : ne gater que le premier livre toutes les autres réponses à faible confiance non annotées.

Surcharge du routage

on_route peut remplacer ctx.decision pour épingler un checkpoint pour une classe de trafic.

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

Cycle de vie du modèle

on_load et on_evict observent les checkpoints. Utilise-les pour des logs de préchauffage, de la comptabilité mémoire ou des alertes d’éviction. Ils tournent hors du verrou du Router, donc un hook peut rappeler le Router.

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

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

Router(hooks=[Lifecycle()])

Contexte multi-locataire

Fais passer un id de locataire en le capturant dans la fermeture du hook, ou en le lisant depuis une variable de contexte locale. Ne stocke pas d’état par requête sur l’objet hook sans verrou.

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

Composition

Plusieurs hooks de types différents se composent naturellement ; les hooks installés tournent en premier, dans l’ordre.

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
)

Garde l’ordre délibéré et documenté, parce qu’un hook tardif voit les mutations d’un hook antérieur.

Instrumentation limitée à un bloc

Attache un traceur ou un hook de debug seulement pour le code qui en a besoin, au lieu de reconstruire l’agent. hooks_installed restaure la liste précédente à la sortie, même si le bloc lève une erreur.

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

add_hook/remove_hook font la même chose sans bloc, pour un traceur qui vit aussi longtemps que le processus.

Instrumentation à l’échelle du processus

Un traceur ou un hook de métriques que chaque décision doit voir peut être enregistré une fois, au lieu d’être passé à chaque Agent et Router. Les valeurs par défaut tournent avant les hooks de l’instance et par appel.

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

C’est de l’état global, alors limite-le délibérément : définis-le une fois au démarrage, et clear_default_hooks() dans les tests pour qu’un test ne puisse pas laisser fuir un hook dans le suivant.

Ajustement du budget de jetons

Un hook de start peut augmenter le budget de jetons d’un appel, par exemple quand une question a beaucoup d’options et que le budget de tête par défaut écraserait les libellés. Quatre détails décident si le hook aide ou aggrave silencieusement l’appel :

• Le ctx.head_max_len d’un hook de start remplace le budget de l’appel. Ce qui est en vigueur avant lui, c’est la valeur par appel de l’appelant, ou le défaut du checkpoint dans ctx.agent.cfg – donc compare par rapport à ça, car écrire un nombre nu peut abaisser un budget que l’appelant avait déjà fixé.
• Un appel répond à chaque question qu’il porte, donc dimensionne sur la plus large d’entre elles plutôt que sur celle qui se trouve être la première.
• Une fois que les options ne tiennent plus dans la tête, laya/common.py donne à chacune max(4, (head_max_len - 16) // k) jetons. 16 + 4 * k tombe donc exactement sur ce plancher : chaque libellé est quand même réduit aux jetons qu’il partage avec les autres, ce qui est l’écrasement que le hook a été écrit pour éviter. 16 + 8 * k les laisse distinguables.
• L’état reçoit max_len - head_max_len - 8 jetons, donc une tête élargie doit élargir max_len avec elle, sinon l’état perd sa fenêtre.

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)

Cela ne touche pas la config d’agent partagée, donc les appels concurrents ne sont pas affectés. Les mêmes réglages sont disponibles par appel : agent.system_one(state, questions, head_max_len=512, max_len=1024).

Élargir n’est pas gratuit : une fenêtre plus longue signifie un tenseur plus grand, et les checkpoints ont été entraînés à 512 (laya) et 1 024 jetons. Au-delà, rétrécir les candidats avec predict_shortlist vaut mieux qu’étirer le budget.

Anti-motifs

Travail bloquant

Les hooks tournent sur le thread appelant, et laya.serve utilise un seul worker d’inférence. Un hook qui dort, attend un aller-retour réseau, ou appelle input() bloque chaque autre requête derrière lui.

# 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 tu dois faire du travail lent, mets hooks_concurrent=False pour au moins empêcher le hook lui-même de se chevaucher, et fais tourner laya.serve derrière une file.

Lever depuis les hooks de fin pour le flux de contrôle

on_predict_end tourne après l’inférence. Lever là-dedans jette un résultat calculé et, sur le chemin du succès, remonte à l’appelant. Utilise un hook de start pour bloquer avant de payer l’inférence, ou réécris ctx.results pour changer la réponse.

État partagé mutable sans verrou

La même instance de hook tourne sur de nombreux threads. self.counter += 1 entre en course.

# 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

Échec silencieux

hooks_raise=False avertit une fois par échec, mais un hook qui attrape tout lui-même cache de vrais problèmes.

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

Si un hook est optionnel, laisse hooks_raise=False s’en occuper et surveille les avertissements. Sinon, laisse-le lever.

Retenir les contextes

Un hook qui ajoute ctx à une liste garde en vie tout l’état, les questions, les résultats et l’agent.

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

Masquer trop tard

À on_predict_end, le modèle a déjà tokenisé l’état. Masque dans on_predict_start.

Logique par question dans un hook par appel

Il y a un PredictContext par appel, et une passe avant répond à chaque question. Il n’y a pas d’événement par question. Itère les réponses à l’intérieur de on_predict_end, et itère aussi les états : sur un lot, un contexte porte chaque état de l’appel.

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 récursif

Un hook qui appelle agent.predict/system_one relance les hooks. Sans garde de profondeur, cela récursionne.

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

Appelables simples dans hooks=

hooks= prend des objets hook ; un appelable nu ne dit pas pour quel événement il est, donc il est rejeté. Utilise 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)

Supposer que les résultats existent dans les hooks de fin

Sur le chemin d’échec, ctx.results est None sauf si un hook de start l’a défini. Vérifie toujours.

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

Hooks dépendants de l’ordre

Un hook qui lit une mutation d’un autre hook est fragile sauf si l’ordre est épinglé. Les hooks installés tournent dans l’ordre de la liste, puis les appelables de commodité ; documente tout couplage, ou fusionne les hooks couplés en un seul objet.

Voir aussi

• Erreurs : la matrice de défaillance derrière plusieurs de ces anti-motifs.
• Exemples : des versions plus complètes des motifs ci-dessus.