Skip to content

Patterns and anti-patterns

Hooks are a small seam, and it is easy to use them well or badly. This page collects the shapes that hold up in production and the ones that bite.

Patterns

Audit log

Record every decision with enough to reconstruct it: the state, the questions, the answers, the model, the routing decision, usage and latency.

import json, sys

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)

A hook fires once per call, and a predict_batch call carries every state in it, so the record is written per decision: ctx.states and ctx.results are aligned by index. ctx.usage and ctx.elapsed_ms are totals for the whole call; each result carries its own usage.

Make it lenient if losing a log line must not fail a request: hooks_raise=False. Make it strict if the audit trail is a compliance requirement.

PII redaction

Redaction must happen in on_predict_start, before tokenization, or the model has already seen the data.

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)

A redaction hook is a policy hook: keep hooks_raise=True, because a silently broken redactor is a data leak.

Caching

A start hook checks the cache and calls ctx.skip(...); an end hook fills it. The forward pass is skipped on a 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)

Hooks fire once per call, so keying on one state is not enough on predict_batch: ctx.skip() replaces every result the call would have returned. Guard the cache with a lock when serving concurrently. On the Router the cached payload still gets a routing key, so the return shape is unchanged.

The same pair works on a single LangChain node via its hooks= argument, which is the way to cache one hot step in a graph without changing what every other caller of that agent sees.

Metrics

Counters and histograms from ctx.model, ctx.usage and ctx.elapsed_ms. Keep it lenient.

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

A policy hook raises to block a request. hooks_raise=True (the default) lets the block reach the caller; on_error and on_predict_end still run, so the audit trail records it.

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)

Test it against the state shape. A guard that only reads ctx.states[0] blocks a single call and lets a predict_batch call put every remaining state through the forward pass.

Confidence gating

An end hook rewrites a low-confidence answer to a safe fallback, or annotates it for downstream logic. This is a result mutation, not a rejection.

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)

Mutate through ctx.results, which holds one dict per state of the call: gating only the first one ships every other low-confidence answer unannotated.

Routing override

on_route may replace ctx.decision to pin a checkpoint for a class of traffic. It has to be a hook object: the convenience callables cover on_predict_start / on_predict_end only, so a bare function in hooks= is refused (see plain callables in hooks=).

from laya.router import RouteDecision

class Pin:
    def on_route(self, 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()])

Model lifecycle

on_load and on_evict observe checkpoints. Use them for warmup logs, memory accounting, or eviction alerts. They run outside the Router lock, so a hook may call back into the Router.

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 context

Thread a tenant id through by capturing it in the hook closure, or by reading it from a context-local. Do not store per-request state on the hook object without a 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"))

Composition

Several hooks of different kinds compose naturally. Within one scope, hooks=[...] entries run in list order and the on_predict_start= / on_predict_end= convenience callables follow. Across scopes, process-wide default hooks come before an instance's hooks, and an instance's hooks before per-call hooks.

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
)

Keep the ordering deliberate and documented, because a later hook sees the mutations of an earlier one.

Scoped instrumentation

Attach a tracer or debug hook only for the code that needs it, instead of reconstructing the agent. On exit -- even if the block raises -- hooks_installed removes one occurrence of each hook it installed, and nothing else: a hook added inside the block with add_hook outlives it.

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

add_hook/remove_hook do the same without a block, for a tracer that lives as long as the process.

Process-wide instrumentation

A tracer or metrics hook that every decision should see can be registered once, instead of being passed to each Agent and Router. Defaults run before the instance and per-call hooks.

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

This is global state, so scope it deliberately: set it once at startup, and clear_default_hooks() in tests so one test cannot leak a hook into the next.

Token-budget shaping

A start hook can raise the token budget for one call, for example when a question has many options and the default head budget would collapse the labels. Four details decide whether the hook helps or quietly makes the call worse:

  • A start hook's ctx.head_max_len replaces the budget for the call. What is in force before it is the caller's own per-call value, or the checkpoint default in ctx.agent.cfg -- so compare against that: writing a plain number can lower the budget a caller already set.
  • One call answers every question it carries, so size on the widest of them rather than on whichever happens to come first.
  • Once the options no longer fit the head, laya/common.py gives each of them max(4, (head_max_len - 16) // k) tokens. 16 + 4 * k therefore lands exactly on that floor: every label is still cut down to the tokens it shares with the others, which is the collapse the hook was written to avoid. 16 + 8 * k leaves them distinguishable.
  • The window defaults to max(64, max_len - head_max_len - 8), floored at 64, and is then capped at the room the questions actually leave for the state. So widening the head stops shaving the default once it reaches the floor -- at max_len=512 a head_max_len=448 gives a 64-token window, not the 56 the unfloored subtraction gives -- while the room keeps shrinking underneath it, and a question whose own head leaves less than 64 scans at that room. The floor does not rescue the state from its options: widen max_len with the head, which is what the need + 8 + 64 below does. And a head that fills the whole sequence is refused outright -- window_budget raises rather than scanning at the zero the subtraction would leave.
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)

This does not touch the shared agent config, so concurrent calls are unaffected. The same knobs are available per call: agent.system_one(state, questions, head_max_len=512, max_len=1024).

Widening is not free: a longer window means a larger tensor, and the checkpoints were trained at 512 (laya) and 1,024 tokens. Past that, narrowing the candidates with predict_shortlist beats stretching the budget.

Anti-patterns

Blocking work

Hooks run on the calling thread, and laya.serve uses a single inference worker. A hook that sleeps, waits on a network round-trip, or calls input() stalls every other request behind it.

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

If you must do slow work, set hooks_concurrent=False to at least keep the hook itself from overlapping, and run laya.serve behind a queue.

Raising from end hooks for control flow

on_predict_end runs after inference. Raising there throws away a computed result and, on the success path, surfaces to the caller. Use a start hook to block before paying for inference, or rewrite ctx.results to change the answer.

Shared mutable state without a lock

The same hook instance runs on many threads. self.counter += 1 races.

# 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

Silent failure

hooks_raise=False warns when a hook fails -- one line per distinct message and call site under Python's default filter, so a hook that fails the same way repeatedly prints once, not once per failure (Errors). A hook that catches everything itself hides real problems.

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

If a hook is optional, let hooks_raise=False handle it and watch the warnings. If it is not, let it raise.

Retaining contexts

A hook that appends ctx to a list keeps the whole state, questions, results and agent alive.

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

Redacting too late

By on_predict_end the model has already tokenized the state. Redact in on_predict_start.

Per-question logic in a per-call hook

There is one PredictContext per call, and one forward pass answers every question. There are no per-question events. Iterate the answers inside on_predict_end, and iterate the states too: on a batch, one context carries every state of the call.

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)

Recursive predict

A hook that calls agent.predict/system_one runs the hooks again. Without a depth guard this recurses.

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

Plain callables in hooks=

hooks= takes hook objects; a bare callable does not say which event it is for, so it is rejected. 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)

Assuming results exist in end hooks

On the failure path ctx.results is None unless a start hook set it. Always check.

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

Order-dependent hooks

A hook that reads a mutation from another hook is fragile unless the order is pinned. Installed hooks run in list order, then convenience callables; document any coupling, or merge the coupled hooks into one object.

See also

  • Errors: the failure matrix behind several of these anti-patterns.
  • Examples: fuller versions of the patterns above.