示例

可複製貼上的配方。除了它點名的那些輔助函式（ship、CACHE 等等，由你提供），每個片段都是 自包含的。

• 快速開始
• 審計
• 脫敏 PII
• 快取
• 指標
• 防護欄
• 置信度門控
• 釘住路由
• 生命週期
• 組合
• 逐呼叫鉤子
• 批處理
• HTTP 伺服器
• ONNXAgent
• 執行時註冊
• 基類與程序級預設值
• 非同步鉤子
• 鉤子超時
• token 預算
• 測試鉤子

快速開始

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

審計

browser-use 那個用例：捕獲每一個決策併發給一個外部服務。

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)

一次鉤子呼叫覆蓋整次呼叫，所以這個迴圈每個決策寫一條記錄；predict_batch 上的同樣形狀見 批處理。

完整可執行的版本在 examples/hooks/audit.py。

脫敏 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)

見 examples/hooks/redact.py。

快取

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

見 examples/hooks/cache.py。

指標

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)

見 examples/hooks/otel.py。

防護欄

從一個 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()

一個 start 鉤子能看到這次呼叫的每個狀態，所以要把它們全測一遍：只讀 ctx.states[0] 會讓 predict_batch 呼叫裡剩下的部分通過。

置信度門控

改寫一個低置信度的答案，或者給它加標註。

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 為這次呼叫的每個狀態儲存一個 dict，所以這個迴圈給每一個沒到閾值的答案都加標註， 而不只是第一個狀態的。

釘住路由

為某一類流量強制一個 checkpoint。

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

逐呼叫，不安裝：

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

生命週期

觀察 checkpoint 的構建和驅逐。

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

組合

已安裝的鉤子在前，然後是便捷可呼叫物件；全部共享一個上下文。

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

逐呼叫鉤子

為單次呼叫覆蓋或擴充套件鉤子。

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

批處理

鉤子每次 Agent.predict_batch 呼叫觸發一次，ctx.states 持有每個狀態。Router.predict_batch 改為每請求執行一次它的 Router 級鉤子，每個帶一個狀態和自己的 run_id，所以那裡同一個鉤子每次 鉤子呼叫寫一條記錄。

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 伺服器

Router 鉤子對 laya.serve 自動生效，因為伺服器呼叫 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 只暴露 predict 級事件。

from laya.onnx_agent import ONNXAgent

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

執行時註冊

在構造之後掛上、摘下或限定鉤子的作用域。

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

基類與程序級預設值

繼承 BaseHook 只覆蓋你需要的東西，併為整個程序註冊一次某個鉤子，而不是把它傳給每個 Agent 和 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()

token 預算

為一次呼叫塑造 token 預算，可以從鉤子裡做，也可以用逐呼叫參數。鉤子的值替換生效中的預算，所以它 必須先讀那個預算：按這次呼叫裡最寬的問題定尺寸，不要低於核心給選項施加的 token 下限，並在拓寬 head_max_len 時一併拓寬 max_len，這樣狀態才留得住視窗。

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)

token 預算的塑造有每一行背後的算術，而 predict_shortlist是當一組標籤連拓寬後的視窗都裝不下時的選項。

非同步鉤子

把非同步鉤子包進 AsyncHook；每個協程都在同步核心裡執行到完成，無論呼叫方是同步的，還是已經在一 個事件迴圈裡。

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

普通的非同步可呼叫物件也行：

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

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

鉤子超時

給每次鉤子呼叫設上界，這樣一個卡住的鉤子無法掛起一個正在服務的請求：

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)

超時的鉤子拋 TimeoutError（或 hooks_raise=False 時警告）。鉤子會在後臺繼續執行，所以也要給 網路呼叫它們自己的超時。見錯誤。

測試鉤子

不用模型就能斷言鉤子看到了什麼：把 encode/forward/decode 這些輔助函式打樁，然後驅動 predict_batch，就像 tests/test_hooks.py 做的那樣。

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

API 表面由 tests/test_hooks_api.py 釘住。

另見

• 鏈路追蹤：run_id、span、OpenTelemetry。
• 模式與反模式：這些配方背後的推理。