jevlaya
Deutsch
Dokumentation

Beispiele

Rezepte zum Kopieren und Einfügen. Jeder Ausschnitt ist eigenständig, abgesehen von den Helfern, die er nennt (ship, CACHE und so weiter), die du lieferst.

Schnellstart

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

Der browser-use-Anwendungsfall: jede Entscheidung erfassen und an einen externen Dienst senden.

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)

Ein Hook-Aufruf deckt den gesamten Aufruf ab, sodass die Schleife einen Datensatz pro Entscheidung schreibt; siehe Batch für dieselbe Form bei predict_batch.

Eine vollständig ausführbare Version findest du in examples/hooks/audit.py.

PII-Schwärzung

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)

Siehe 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

Siehe examples/hooks/cache.py.

Metriken

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)

Siehe examples/hooks/otel.py.

Leitplanke

Blockiere eine Anfrage, indem du aus einem Start-Hook eine Ausnahme auslöst.

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

Ein Start-Hook sieht jeden Zustand des Aufrufs, also prüfe sie alle: Nur ctx.states[0] zu lesen, lässt den Rest eines predict_batch-Aufrufs durch.

Konfidenz-Gate

Schreibe eine Antwort mit niedriger Konfidenz um oder annotiere sie.

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 enthält ein dict pro Zustand des Aufrufs, sodass die Schleife jede Antwort annotiert, die den Schwellenwert verfehlt, nicht nur die des ersten Zustands.

Routing-Pinning

Erzwinge einen Checkpoint für eine Klasse von Datenverkehr.

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

Pro Aufruf, ohne ihn zu installieren:

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

Lebenszyklus

Beobachte den Aufbau und die Verdrängung von 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

Komposition

Zuerst die installierten Hooks, dann die Convenience-Callables; alle teilen einen Kontext.

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 pro Aufruf

Überschreibe oder erweitere Hooks für einen einzelnen Aufruf.

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

Batch

Hooks werden einmal pro Agent.predict_batch-Aufruf ausgelöst, wobei ctx.states jeden Zustand enthält. Router.predict_batch führt stattdessen seine Router-Hooks einmal pro Anfrage aus, jeweils mit einem Zustand und einer eigenen run_id, sodass derselbe Hook dort einen Datensatz pro Aufruf des Hooks schreibt.

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)

HTTP-Server

Router-Hooks werden für laya.serve automatisch ausgelöst, weil der Server Router.predict aufruft.

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 stellt nur die Ereignisse der predict-Ebene bereit.

from laya.onnx_agent import ONNXAgent

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

Registrierung zur Laufzeit

Hänge Hooks nach der Konstruktion an, löse sie oder beschränke ihren Geltungsbereich.

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

Basisklasse und prozessweite Standardwerte

Leite von BaseHook ab, um nur das zu überschreiben, was du brauchst, und registriere etwas einmal für den gesamten Prozess, statt es an jeden Agent und Router zu übergeben.

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

Token-Budget

Forme das Token-Budget für einen Aufruf, aus einem Hook oder einem Argument pro Aufruf. Der Wert eines Hooks ersetzt das geltende Budget, also muss er dieses Budget zuerst lesen: dimensioniere an der breitesten Frage des Aufrufs, bleibe über der Token-Untergrenze, die der Kern auf die Optionen anwendet, und erweitere max_len zusammen mit head_max_len, damit der Zustand ein Fenster behält.

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)

Anpassung des Token-Budgets enthält die Arithmetik hinter jeder Zeile, und predict_shortlist ist die Option, wenn eine Menge von Labels nicht einmal in ein erweitertes Fenster passt.

Asynchrone Hooks

Verpacke einen asynchronen Hook in AsyncHook; jede Coroutine läuft im synchronen Kern vollständig durch, egal ob der Aufrufer synchron ist oder sich bereits in einer Ereignisschleife befindet.

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

Ein einfacher asynchroner Callable funktioniert ebenfalls:

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

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

Hook-Timeout

Begrenze jeden Hook-Aufruf, damit ein hängender Hook eine bediente Anfrage nicht aufhängen kann:

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)

Ein Hook, der das Zeitlimit überschreitet, löst TimeoutError aus (oder warnt, wenn hooks_raise=False). Der Hook läuft im Hintergrund weiter, gib also auch Netzwerkaufrufen ihr eigenes Zeitlimit. Siehe Fehler.

Hooks testen

Prüfe, was ein Hook gesehen hat, ohne ein Modell: steuere predict_batch mit den encode/forward/decode-Helfern als Stubs, wie es tests/test_hooks.py tut.

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

Die API-Oberfläche wird durch tests/test_hooks_api.py festgelegt.

Siehe auch