jevlaya
Español
Documentación

Ejemplos

Recetas para copiar y pegar. Cada fragmento es autónomo salvo por los helpers que nombra (ship, CACHE, y así sucesivamente), que aportas tú.

Inicio rápido

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

Auditoría

El caso de uso de browser-use: captura cada decisión y envíala a un servicio externo.

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)

Una llamada de hook cubre toda la llamada, así que el bucle escribe un registro por decisión; consulta Lote para la misma forma en predict_batch.

Hay una versión completa y ejecutable en examples/hooks/audit.py.

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

Consulta examples/hooks/redact.py.

Caché

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

Consulta examples/hooks/cache.py.

Métricas

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)

Consulta examples/hooks/otel.py.

Guardarraíl

Bloquea una solicitud lanzando una excepción desde un hook de inicio.

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 inicio ve cada estado de la llamada, así que pruébalos todos: leer solo ctx.states[0] deja pasar el resto de una llamada a predict_batch.

Puerta de confianza

Reescribe una respuesta de baja confianza, o anótala.

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 contiene un dict por estado de la llamada, así que el bucle anota cada respuesta que no alcanza el umbral, no solo la del primer estado.

Fijar el routing

Fuerza un checkpoint para una clase de tráfico.

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

Por llamada, sin instalarlo:

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

Ciclo de vida

Observa la construcción y la expulsión de 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

Composición

Primero los hooks instalados, luego los invocables de conveniencia; todos comparten un contexto.

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 por llamada

Sobrescribe o extiende los hooks para una sola llamada.

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

Lote

Los hooks se disparan una vez por llamada a Agent.predict_batch, con ctx.states conteniendo cada estado. En cambio, Router.predict_batch ejecuta sus hooks de nivel de Router una vez por solicitud, cada uno con un estado y su propio run_id, así que el mismo hook allí escribe un registro por llamada del 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)

Servidor HTTP

Los hooks del Router se disparan para laya.serve automáticamente, porque el servidor llama a 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 solo expone los eventos de nivel de predict.

from laya.onnx_agent import ONNXAgent

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

Registro en runtime

Adjunta, separa o limita el alcance de los hooks después de la construcción.

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

Clase base y valores predeterminados para todo el proceso

Hereda de BaseHook para sobrescribir solo lo que necesitas, y registra algo una vez para todo el proceso en lugar de pasarlo a cada Agent y 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()

Presupuesto de tokens

Da forma al presupuesto de tokens de una llamada, desde un hook o desde un argumento por llamada. El valor de un hook reemplaza el presupuesto vigente, así que tiene que leer primero ese presupuesto: dimensiónalo con la pregunta más ancha de la llamada, mantente por encima del piso de tokens que el núcleo aplica a las opciones, y amplía max_len junto con head_max_len para que el estado conserve una ventana.

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)

Ajuste del presupuesto de tokens tiene la aritmética detrás de cada línea, y predict_shortlist es la opción cuando un conjunto de etiquetas no cabe ni en una ventana ampliada.

Hooks asíncronos

Envuelve un hook asíncrono en AsyncHook; cada corrutina se ejecuta hasta completarse en el núcleo síncrono, ya sea que quien llama sea síncrono o ya esté dentro de un bucle de eventos.

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 invocable asíncrono simple también funciona:

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

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

Tiempo límite de hook

Limita cada llamada de hook, para que un hook atascado no pueda colgar una solicitud servida:

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 que agotó el tiempo lanza TimeoutError (o avisa cuando hooks_raise=False). El hook sigue ejecutándose en segundo plano, así que dale también su propio tiempo límite a las llamadas de red. Consulta errores.

Probar hooks

Comprueba lo que vio un hook sin un modelo: impulsa predict_batch con los helpers de encode/forward/decode sustituidos por stubs, como hace tests/test_hooks.py.

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

La superficie de la API está fijada por tests/test_hooks_api.py.

Véase también