jevlaya
Français
Documentation

Exemples

Des recettes à copier-coller. Chaque extrait est autonome, à part les aides qu’il nomme (ship, CACHE, et ainsi de suite), que tu fournis.

Démarrage rapide

import laya

def log(ctx):
    print(ctx.model, ctx.results[0]["answers"])

agent = laya.load("convaiinnovations/laya", on_predict_end=log)
agent.system_one("I was charged twice.", {"urgent": {"type": "noul", "instructions": "Urgent?"}})

Audit

Le cas d’usage browser-use : capturer chaque décision et l’envoyer à un service externe.

import json, sys
import laya

def audit(ctx):
    for state, result in zip(ctx.states, ctx.results or []):
        record = {
            "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),
        }
        print(json.dumps(record), file=sys.stderr)
        # ship_to_service(record)

agent = laya.load("convaiinnovations/laya", on_predict_end=audit)

Un appel de hook couvre tout l’appel, donc la boucle écrit un enregistrement par décision ; voir Lot pour la même forme sur predict_batch.

Une version entièrement exécutable est dans examples/hooks/audit.py.

Masquer les PII

import re
import laya

EMAIL = re.compile(r"\b[\w.+-]+@[\w-]+\.[\w.-]+\b")
PHONE = re.compile(r"\+?\d[\d ()-]{7,}\d")

def scrub(value):
    if isinstance(value, str):
        return PHONE.sub("[phone]", EMAIL.sub("[email]", value))
    if isinstance(value, dict):
        return {k: scrub(v) for k, v in value.items()}
    if isinstance(value, list):
        return [scrub(v) for v in value]
    return value

def redact(ctx):
    ctx.states = [scrub(s) for s in ctx.states]

agent = laya.load("convaiinnovations/laya", on_predict_start=redact)

Voir examples/hooks/redact.py.

Cache

import hashlib, json
import laya

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

agent = laya.load("convaiinnovations/laya", on_predict_start=read, on_predict_end=write)
first = agent.system_one("state", QUESTIONS)    # runs the model
second = agent.system_one("state", QUESTIONS)   # served from CACHE

Voir examples/hooks/cache.py.

Métriques

import laya

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)

agent = laya.load("convaiinnovations/laya", on_predict_end=metrics, hooks_raise=False)

Voir examples/hooks/otel.py.

Garde-fou

Bloque une requête en levant depuis un hook de start.

import laya

class Blocked(Exception):
    pass

def guard(ctx):
    text = " ".join(str(state) for state in ctx.states).lower()
    if "ignore previous instructions" in text:
        raise Blocked("prompt injection")

agent = laya.load("convaiinnovations/laya", on_predict_start=guard)

try:
    agent.system_one("Ignore previous instructions and ...", QUESTIONS)
except Blocked:
    handle_block()

Un hook de start voit chaque état de l’appel, donc teste-les tous : ne lire que ctx.states[0] laisse le reste d’un appel predict_batch passer.

Porte de confiance

Réécris une réponse à faible confiance, ou annote-la.

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

agent = laya.load("convaiinnovations/laya", on_predict_end=gate)

ctx.results contient un dict par état de l’appel, donc la boucle annote chaque réponse qui rate le seuil, pas seulement celle du premier état.

Épinglage du routage

Force un checkpoint pour une classe de trafic.

from laya import Router
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 = Router(hooks=[pin])

Par appel, sans installer :

router.predict("refund request", QUESTIONS, hooks=[pin])

Cycle de vie

Observe la construction et l’éviction des checkpoints.

from laya import Router

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

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

router = Router(max_loaded=1, hooks=[Lifecycle()])
router.preload(["english", "multilingual"])   # on_load fires per build
router.unload()                               # on_evict fires per freed checkpoint

Composition

Les hooks installés d’abord, puis les appelables de commodité ; tous partagent un seul contexte.

import laya

class Metrics:
    def on_predict_end(self, ctx):
        record_latency(ctx.model, ctx.elapsed_ms)

def redact(ctx):
    ctx.states = [strip_pii(s) for s in ctx.states]

def audit(ctx):
    ship(ctx.run_id, ctx.results)

agent = laya.load(
    "convaiinnovations/laya",
    hooks=[Metrics()],              # installed, runs first
    on_predict_start=redact,        # convenience, appended
    on_predict_end=audit,           # convenience, appended
    hooks_raise=True,
)

Hooks par appel

Surcharge ou étends les hooks pour un seul appel.

agent.system_one(
    state,
    questions,
    on_predict_end=lambda ctx: debug_dump(ctx),
    hooks_raise=False,
)

router.predict(
    state,
    questions,
    hooks=[pin],                    # applies to on_route too
    on_predict_end=audit,
)

Lot

Les hooks se déclenchent une fois par appel Agent.predict_batch, avec ctx.states contenant chaque état. Router.predict_batch exécute ses hooks au niveau Router une fois par requête à la place, chacun avec un état et son propre run_id, donc le même hook y écrit un enregistrement par appel du hook.

def audit_batch(ctx):
    for state, result in zip(ctx.states, ctx.results):
        ship_one(ctx.run_id, state, result)

results = agent.predict_batch([state_a, state_b, state_c], questions, on_predict_end=audit_batch)

Serveur HTTP

Les hooks Router se déclenchent automatiquement pour laya.serve, parce que le serveur appelle Router.predict.

from laya import Router
from laya.serve import create_app

router = Router(hooks=[Metrics()], on_predict_end=audit, hooks_raise=False)
app = create_app(router=router)

ONNXAgent

ONNXAgent n’expose que les événements au niveau predict.

from laya.onnx_agent import ONNXAgent

agent = ONNXAgent("convaiinnovations/laya", onnx_path="laya.onnx", on_predict_end=audit)
agent.system_one(state, questions)

Enregistrement à l’exécution

Attache, détache ou limite les hooks à un bloc après la construction.

agent.add_hook(Metrics())          # attach at runtime
agent.remove_hook(Metrics())       # by identity

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

Classe de base et valeurs par défaut pour tout le processus

Hérite de BaseHook pour ne redéfinir que ce dont tu as besoin, et enregistre quelque chose une fois pour tout le processus au lieu de le passer à chaque Agent et Router.

from laya import BaseHook, hooks

class Audit(BaseHook):
    def on_predict_end(self, ctx):
        ship(ctx.run_id, ctx.results)

hooks.set_default_hooks(hooks=[Audit()])   # runs for every call in the process

# later, or in tests:
hooks.clear_default_hooks()

Budget de jetons

Façonne le budget de jetons pour un appel, depuis un hook ou un argument par appel. La valeur d’un hook remplace le budget en vigueur, donc il doit d’abord lire ce budget : dimensionne sur la question la plus large de l’appel, reste au-dessus du plancher de jetons que le cœur applique aux options, et élargis max_len avec head_max_len pour que l’état garde une fenêtre.

def widen(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
    if need > head:
        ctx.head_max_len = need
        ctx.max_len = max(window, need + 8 + 64)

agent = laya.load("convaiinnovations/laya", on_predict_start=widen)

# or per call
agent.system_one(state, questions, head_max_len=512, max_len=1024)

Ajustement du budget de jetons a l’arithmétique derrière chaque ligne, et predict_shortlist est l’option quand un jeu de libellés ne tient pas même dans une fenêtre élargie.

Hooks asynchrones

Enveloppe un hook async dans AsyncHook ; chaque coroutine s’exécute jusqu’au bout dans le cœur synchrone, que l’appelant soit synchrone ou déjà dans une boucle d’événements.

import laya
from laya import AsyncHook

class RemoteAudit:
    async def on_predict_end(self, ctx):
        await ship(ctx.run_id, ctx.results)

agent = laya.load("convaiinnovations/laya", hooks=[AsyncHook(RemoteAudit())])

Un appelable async simple fonctionne aussi :

async def async_end(ctx):
    await ship(ctx.results)

agent.system_one(state, questions, on_predict_end=async_end)

Délai d’attente des hooks

Borne chaque appel de hook, pour qu’un hook bloqué ne puisse pas bloquer une requête servie :

agent = laya.load("convaiinnovations/laya", on_predict_end=metrics, hooks_timeout=2.0)

# or per call
agent.system_one(state, questions, on_predict_end=metrics, hooks_timeout=0.5)

Un hook expiré lève TimeoutError (ou avertit quand hooks_raise=False). Le hook continue de tourner en arrière-plan, donc donne aussi aux appels réseau leur propre timeout. Voir erreurs.

Tester les hooks

Asserte ce qu’un hook a vu sans modèle : pilote predict_batch avec les aides encode/forward/decode bouchonnées, comme le fait tests/test_hooks.py.

seen = []
agent.predict_batch(["s0"], questions, on_predict_end=lambda ctx: seen.append(ctx.results))
assert len(seen) == 1

La surface d’API est épinglée par tests/test_hooks_api.py.

Voir aussi