预测钩子

预测钩子

钩子让你观察或塑造 Laya 做出的每一个决策，而无需 fork 它。

它们是每个真实部署都需要的东西的扩展缝：审计日志、推理前脱敏 PII、缓存、指标、置信度门控、 路由覆盖，以及把决策转发给外部服务。它们是选用的：不配置任何钩子时，Agent、Router 和 ONNXAgent 的行为不变。

它们不只用于直接调用。每个 LangChain 与 LangGraph runnable 都接受同样五个逐调用 参数，所以钩子可以挂到图里的某一个节点上，而不是整个智能体上。

这个文件夹是完整的参考。从这里开始，然后深入你需要的那一页：

快速开始

import laya

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

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

一个对象可以实现生命周期事件中的任意子集：

class Audit:
    def on_predict_start(self, ctx):
        print("start", ctx.run_id)

    def on_predict_end(self, ctx):
        print("end", ctx.run_id, ctx.usage, ctx.elapsed_ms)

    def on_error(self, ctx):
        print("failed", ctx.run_id, ctx.error)

laya.load("convaiinnovations/laya", hooks=[Audit()])

钩子也可以在之后添加，或者限定在一段代码块内：

agent.add_hook(tracer)               # attach at runtime
with agent.hooks_installed(debug):   # installed for the block, removed on exit
    agent.system_one(state, questions)

见运行时注册。对于一个应该处处生效、又不想穿过每次调用的 钩子，用进程级默认值注册一次：

from laya import hooks

hooks.set_default_hooks(hooks=[Tracer()])

心智模型

有三个概念。

1. 钩子是一个可调用对象或一个对象。 一个普通函数适合一个事件；一个对象适合多个。两者都传给 hooks= / on_predict_start= / on_predict_end=。

2. 一次调用的每个钩子共享同一个可变的 PredictContext。 它携带状态、问题、结果、路由决策、 模型名、usage、计时和任何错误。一次调用可以同时携带很多状态（predict_batch），所以一个想 覆盖每一个决策的钩子必须遍历 ctx.states 和 ctx.results；ctx.usage 和 ctx.elapsed_ms 是整次调用的合计。因为上下文是可变的，钩子可以塑造这次调用，而不只是旁观：脱敏状态、改写 问题、替换结果，或用缓存答案跳过推理。

3. 有两个作用域。 Agent 钩子包住一次前向传播；Router 钩子包住路由加推理，还能看到模型 生命周期（on_route、on_load、on_evict）。这对应其他智能体框架里「run hooks」与 「agent hooks」的区分。

                             Router.predict(state, questions)
   ┌──────────────────────────────────────────────────────────────────────────┐
   │  route()                                                                 │
   │    ├─ detect language / workflow                                         │
   │    └─ on_route          ctx.decision  (a hook may replace it)            │
   │                                                                          │
   │  load(decision.model)                                                    │
   │    ├─ build checkpoint on first use ──► on_load    ctx.model, ctx.agent  │
   │    └─ evict LRU checkpoint ───────────► on_evict   ctx.model             │
   │                                                                          │
   │  on_predict_start       ctx.states, ctx.questions, ctx.decision          │
   │    │                                                                     │
   │    ├── ctx.skip(results)? ──► skip the forward pass                      │
   │    │                                                                     │
   │    └── Agent.system_one(...)  ──►  Agent-level hooks run here            │
   │           on_predict_start  ─►  forward  ─►  on_predict_end              │
   │                                                                          │
   │  result["routing"] = decision                                            │
   │  on_predict_end         ctx.results, ctx.usage, ctx.elapsed_ms           │
   └──────────────────────────────────────────────────────────────────────────┘
                 any failure on the way ──► on_error, then on_predict_end

作用域一览

laya.serve 和 MCP 服务器调用 Router.predict，所以 Router 钩子对它们自动生效。Router 运行一个 挂载的或内置的 agent 时，Agent 钩子就生效。

兼容性

• 不配置钩子就没有任何行为变化。未设置的路径有回归测试。
• 所有钩子参数都是带默认值的关键字参数，所以现有的调用继续可用。
• laya/hooks.py 是纯 Python：import laya 不会因为它而拉进 torch。
• 钩子默认是同步的。一个 async def 事件可以包进 AsyncHook，或者 作为一个普通的异步可调用对象传入，它会替你运行到完成。
• hooks_timeout 给慢钩子设上界，使它无法挂起一个正在服务的请求。
• 保持钩子快且非阻塞；关于对 laya.serve 的后果见错误和 模式。

另见

• examples/hooks/：可运行的审计、脱敏、缓存和指标钩子。
• tests/test_hooks.py：行为规范。
• tests/test_hooks_api.py：API 稳定性的守卫。
• laya/hooks.py：实现。