Muster und Antimuster

Hooks sind eine kleine Schnittstelle, und es ist leicht, sie gut oder schlecht zu nutzen. Diese Seite sammelt die Muster, die in Produktion halten, und die, die beißen.

• Muster
  • Audit-Log
  • PII-Schwärzung
  • Caching
  • Metriken
  • Leitplanken
  • Konfidenz-Gating
  • Routing-Override
  • Modell-Lebenszyklus
  • Multi-Tenant-Kontext
  • Komposition
  • Lokale Instrumentierung
  • Prozessweite Instrumentierung
  • Anpassung des Token-Budgets
• Antimuster
  • Blockierende Arbeit
  • Ausnahmen in End-Hooks für den Kontrollfluss
  • Gemeinsamer veränderlicher Zustand ohne Sperre
  • Stiller Fehler
  • Festhalten von Kontexten
  • Zu spätes Schwärzen
  • Logik pro Frage in einem Hook pro Aufruf
  • Rekursives predict
  • Einfache Callables in hooks=
  • Ergebnisse in End-Hooks voraussetzen
  • Reihenfolgeabhängige Hooks

Muster

Audit-Log

Zeichne jede Entscheidung mit genug Informationen auf, um sie zu rekonstruieren: den Zustand, die Fragen, die Antworten, das Modell, die Routing-Entscheidung, Nutzung und Latenz.

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)

Ein Hook wird einmal pro Aufruf ausgelöst, und ein predict_batch-Aufruf trägt jeden Zustand in sich, sodass der Datensatz pro Entscheidung geschrieben wird: ctx.states und ctx.results sind per Index ausgerichtet. ctx.usage und ctx.elapsed_ms sind Summen für den gesamten Aufruf; jedes Ergebnis trägt seine eigene usage.

Mache ihn nachsichtig, wenn der Verlust einer Log-Zeile keine Anfrage scheitern lassen darf: hooks_raise=False. Mache ihn strikt, wenn der Audit-Trail eine Compliance-Anforderung ist.

PII-Schwärzung

Die Schwärzung muss in on_predict_start erfolgen, vor der Tokenisierung, sonst hat das Modell die Daten bereits gesehen.

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)

Ein Schwärzungs-Hook ist ein Policy-Hook: Behalte hooks_raise=True, denn ein still kaputter Schwärzer ist ein Datenleck.

Caching

Ein Start-Hook prüft den Cache und ruft ctx.skip(...) auf; ein End-Hook füllt ihn. Bei einem Treffer wird der Forward-Pass übersprungen.

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)

Hooks werden einmal pro Aufruf ausgelöst, also reicht es nicht, auf einen Zustand zu schlüsseln, bei predict_batch: ctx.skip() ersetzt jedes Ergebnis, das der Aufruf zurückgegeben hätte. Schütze den Cache mit einer Sperre, wenn du nebenläufig bedienst. Beim Router bekommt die gecachte Payload weiterhin einen routing-Schlüssel, sodass die Rückgabeform unverändert bleibt.

Dasselbe Paar funktioniert auf einem einzelnen LangChain-Knoten über sein hooks=-Argument, was der Weg ist, einen heißen Schritt in einem Graphen zu cachen, ohne zu ändern, was jeder andere Aufrufer dieses Agenten sieht.

Metriken

Zähler und Histogramme aus ctx.model, ctx.usage und ctx.elapsed_ms. Halte ihn nachsichtig.

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)

Leitplanken

Ein Policy-Hook löst eine Ausnahme aus, um eine Anfrage zu blockieren. hooks_raise=True (der Standard) lässt die Blockierung den Aufrufer erreichen; on_error und on_predict_end laufen weiterhin, sodass der Audit-Trail sie aufzeichnet.

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 ihn gegen die Form des Zustands. Eine Leitplanke, die nur ctx.states[0] liest, blockiert einen einzelnen Aufruf und lässt einen predict_batch-Aufruf jeden verbleibenden Zustand durch den Forward-Pass schicken.

Konfidenz-Gating

Ein End-Hook schreibt eine Antwort mit niedriger Konfidenz in einen sicheren Fallback um oder annotiert sie für nachgelagerte Logik. Das ist eine Ergebnisänderung, keine Ablehnung.

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)

Mutiere über ctx.results, das ein dict pro Zustand des Aufrufs enthält: Nur den ersten zu gaten, liefert jede andere Antwort mit niedriger Konfidenz unannotiert aus.

Routing-Override

on_route kann ctx.decision ersetzen, um einen Checkpoint für eine Klasse von Datenverkehr festzulegen.

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

Modell-Lebenszyklus

on_load und on_evict beobachten Checkpoints. Nutze sie für Warmup-Logs, Speicherbilanzierung oder Verdrängungswarnungen. Sie laufen außerhalb der Router-Sperre, sodass ein Hook in den Router zurückrufen darf.

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

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

Router(hooks=[Lifecycle()])

Multi-Tenant-Kontext

Reiche eine Mandanten-ID durch, indem du sie im Closure des Hooks einfängst oder aus einem kontextlokalen Speicher liest. Speichere keinen Zustand pro Anfrage im Hook-Objekt ohne Sperre.

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

Komposition

Mehrere Hooks verschiedener Art komponieren sich auf natürliche Weise; installierte Hooks laufen zuerst, in Reihenfolge.

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
)

Halte die Reihenfolge bewusst und dokumentiert, denn ein späterer Hook sieht die Mutationen eines früheren.

Lokale Instrumentierung

Hänge einen Tracer- oder Debug-Hook nur für den Code an, der ihn braucht, statt den Agenten neu aufzubauen. hooks_installed stellt die vorherige Liste beim Verlassen wieder her, auch wenn der Block eine Ausnahme auslöst.

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

add_hook/remove_hook machen dasselbe ohne einen Block, für einen Tracer, der so lange lebt wie der Prozess.

Prozessweite Instrumentierung

Ein Tracer- oder Metrik-Hook, den jede Entscheidung sehen soll, kann einmal registriert werden, statt ihn an jeden Agent und Router zu übergeben. Standardwerte laufen vor den Hooks der Instanz und des Aufrufs.

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

Das ist globaler Zustand, also grenze ihn bewusst ein: Setze ihn einmal beim Start, und rufe clear_default_hooks() in Tests auf, damit ein Test keinen Hook in den nächsten durchsickern lässt.

Anpassung des Token-Budgets

Ein Start-Hook kann das Token-Budget für einen Aufruf erhöhen, zum Beispiel wenn eine Frage viele Optionen hat und das Standard-Head-Budget die Labels kollabieren ließe. Vier Details entscheiden, ob der Hook hilft oder den Aufruf still verschlechtert:

• Der ctx.head_max_len eines Start-Hooks ersetzt das Budget für den Aufruf. Was davor gilt, ist der eigene Pro-Aufruf-Wert des Aufrufers oder der Checkpoint-Standard in ctx.agent.cfg – also vergleiche damit, denn eine einfache Zahl zu schreiben, kann das Budget senken, das ein Aufrufer bereits gesetzt hat.
• Ein Aufruf beantwortet jede Frage, die er trägt, also dimensioniere an der breitesten von ihnen statt an der, die zufällig zuerst kommt.
• Sobald die Optionen nicht mehr in den Head passen, gibt laya/common.py jedem von ihnen max(4, (head_max_len - 16) // k) Tokens. 16 + 4 * k landet daher genau auf dieser Untergrenze: jedes Label wird trotzdem auf die Tokens reduziert, die es mit den anderen teilt, was der Kollaps ist, den der Hook vermeiden sollte. 16 + 8 * k lässt sie unterscheidbar.
• Der Zustand erhält max_len - head_max_len - 8 Tokens, also muss ein erweiterter Head max_len mit erweitern, sonst verliert der Zustand sein Fenster.

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)

Das berührt die geteilte Agent-Konfiguration nicht, sodass nebenläufige Aufrufe unberührt bleiben. Dieselben Stellschrauben sind pro Aufruf verfügbar: agent.system_one(state, questions, head_max_len=512, max_len=1024).

Erweitern ist nicht kostenlos: ein längeres Fenster bedeutet einen größeren Tensor, und die Checkpoints wurden mit 512 (laya) und 1.024 Tokens trainiert. Darüber hinaus schlägt das Verengen der Kandidaten mit predict_shortlist das Strecken des Budgets.

Antimuster

Blockierende Arbeit

Hooks laufen im aufrufenden Thread, und laya.serve verwendet einen einzelnen Inferenz-Worker. Ein Hook, der schläft, auf einen Netzwerk-Roundtrip wartet oder input() aufruft, blockiert jede andere Anfrage dahinter.

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

Wenn du langsame Arbeit verrichten musst, setze hooks_concurrent=False, um wenigstens zu verhindern, dass sich der Hook selbst überlappt, und betreibe laya.serve hinter einer Queue.

Ausnahmen in End-Hooks für den Kontrollfluss

on_predict_end läuft nach der Inferenz. Dort eine Ausnahme auszulösen, verwirft ein berechnetes Ergebnis und tritt auf dem Erfolgspfad an den Aufrufer. Nutze einen Start-Hook, um zu blockieren, bevor du für die Inferenz zahlst, oder schreibe ctx.results um, um die Antwort zu ändern.

Gemeinsamer veränderlicher Zustand ohne Sperre

Dieselbe Hook-Instanz läuft in vielen Threads. self.counter += 1 führt zu Race Conditions.

# 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

Stiller Fehler

hooks_raise=False warnt einmal pro Fehler, aber ein Hook, der selbst alles abfängt, verbirgt echte Probleme.

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

Wenn ein Hook optional ist, lass hooks_raise=False ihn behandeln und beobachte die Warnungen. Wenn nicht, lass ihn die Ausnahme auslösen.

Festhalten von Kontexten

Ein Hook, der ctx an eine Liste anhängt, hält den gesamten Zustand, die Fragen, die Ergebnisse und den Agenten am Leben.

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

Zu spätes Schwärzen

Bis on_predict_end hat das Modell den Zustand bereits tokenisiert. Schwärze in on_predict_start.

Logik pro Frage in einem Hook pro Aufruf

Es gibt einen PredictContext pro Aufruf, und ein Forward-Pass beantwortet jede Frage. Es gibt keine Ereignisse pro Frage. Iteriere die Antworten innerhalb von on_predict_end und iteriere auch die Zustände: In einem Batch trägt ein Kontext jeden Zustand des Aufrufs.

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)

Rekursives predict

Ein Hook, der agent.predict/system_one aufruft, führt die Hooks erneut aus. Ohne eine Tiefenbegrenzung rekursiert das.

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

Einfache Callables in hooks=

hooks= nimmt Hook-Objekte; ein nackter Callable sagt nicht, für welches Ereignis er gilt, also wird er abgelehnt. Verwende 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)

Ergebnisse in End-Hooks voraussetzen

Auf dem Fehlerpfad ist ctx.results None, es sei denn, ein Start-Hook hat es gesetzt. Prüfe immer.

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

Reihenfolgeabhängige Hooks

Ein Hook, der eine Mutation von einem anderen Hook liest, ist fragil, es sei denn, die Reihenfolge ist festgelegt. Installierte Hooks laufen in Listenreihenfolge, dann die Convenience-Callables; dokumentiere jede Kopplung oder führe die gekoppelten Hooks zu einem Objekt zusammen.

Siehe auch

• Fehler: die Fehlermatrix hinter mehreren dieser Antimuster.
• Beispiele: vollständigere Versionen der obigen Muster.