人工智能NLP强化学习【免费下载链接】layaNon-autoregressive System 1 decision engine. Typed choice, score and yes/no decisions over any text in a single forward pass, in 100 languages, with a router that picks the right checkpoint per request.项目地址https://gitcode.com/gh_mirrors/lay/laya点击查看免费下载本文以 Laya 的官方钩子示例文档docs/hooks/examples.md为骨架逐一拆解 18 类可直接复制的钩子配方并结合laya/hooks.py源码、examples/hooks/下可运行的示例与tests/中的行为测试讲清每个配方的触发时机、上下文字段与最佳实践。读完你可以在不 fork Laya 的前提下为Agent、Router、ONNXAgent补齐审计日志、PII 脱敏、结果缓存、指标采集、安全网关、置信度门控、路由固定、令牌预算调节等生产级能力。钩子是 Laya 的扩展接缝它让你在不修改引擎代码的前提下观察或改写每一次决策。钩子默认是完全可选的——不配置任何钩子时Agent、Router、ONNXAgent的行为与原生完全一致该无钩子路径由回归测试覆盖。每个钩子都会共享同一个可变的PredictContexton_predict_start可以改写states/questions甚至用ctx.skip()跳过推理on_predict_end可以改写resultson_route可以改写路由决策decision。下面是 18 个即贴即用的配方。1. 快速上手一行挂载日志钩子最简形态是把一个普通函数通过on_predict_end传入laya.load每次推理结束都会回调它ctx里携带模型名与答案import laya def log(ctx): print(ctx.model, ctx.results[0][answers]) agent laya.load(convaiinnovations/laya, on_predict_endlog) agent.system_one(I was charged twice., {urgent: {type: noul, instructions: Urgent?}})这里的system_one(state, questions)是 Laya 的核心调用给定一段文本state和一组问题questions单次前向推理输出每个问题的类型化答案。on_predict_end是五个预测级事件之一完整事件集为on_predict_start、on_predict_end、on_route、on_load、on_evict、on_error定义在 laya/hooks.py 的HOOK_EVENTS元组中。2. 审计捕获每一次决策并外发典型的浏览器自动化browser-use场景把每次决策完整记录下来含路由信息、答案、token 用量、耗时再发送到外部服务。ctx中results为每个 state 的结果字典结构与system_one返回值一致。import json, sys import laya def audit(ctx): record { run_id: ctx.run_id, model: ctx.model, routing: ctx.results[0].get(routing) if ctx.results else None, answers: ctx.results[0][answers] if ctx.results else None, usage: ctx.usage, elapsed_ms: round(ctx.elapsed_ms or 0.0, 3), } print(json.dumps(record), filesys.stderr) # ship_to_service(record) agent laya.load(convaiinnovations/laya, on_predict_endaudit)几个值得注意的字段细节见 docs/hooks/api.md 的 PredictContext 表run_id每次公开调用生成一次uuid4().hex同一次调用的所有钩子共享是关联事件与 span 的纽带elapsed_ms整次调用的墙钟耗时毫秒在on_predict_end时才被填充usage为{input_tokens, output_tokens}由 aggregate_usage 对各 state 的用量求和得到失败路径上results为None所以审计钩子必须先判空。可运行的完整版见 examples/hooks/audit.py它还演示了同一钩子挂到Router上后ctx.decision路由决策也会出现在审计记录里。审计属于合规类钩子建议测试环境用hooks_raiseTrue默认生产环境按需放宽。3. 脱敏 PII在模型看到数据之前改写状态PII 脱敏必须发生在on_predict_start即在 tokenization 之前——一旦到了on_predict_end模型已经看过原文了。配方通过递归scrub处理字符串、字典与列表把邮箱和电话替换为占位符后整体替换ctx.statesimport 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_startredact)参考实现见 examples/hooks/redact.py。脱敏是策略钩子建议保持hooks_raiseTrue因为静默失效的脱敏器等于数据泄露见 docs/hooks/errors.md 的策略建议表。4. 缓存命中时用ctx.skip()跳过整个前向推理缓存配方的精妙之处在于ctx.skip(results)从on_predict_start调用它会直接设置ctx.results并跳过前向推理但on_predict_end仍然运行因此审计、指标等后续钩子照常工作。缓存键由 state 与 questions 的规范化 JSON 的 SHA-256 组成import hashlib, json import laya CACHE {} def key(state, questions): payload json.dumps([state, questions], sort_keysTrue, defaultstr) return hashlib.sha256(payload.encode()).hexdigest() def read(ctx): hit CACHE.get(key(ctx.states[0], ctx.questions)) if hit is not None: ctx.skip([hit]) def write(ctx): if ctx.results: CACHE[key(ctx.states[0], ctx.questions)] ctx.results[0] agent laya.load(convaiinnovations/laya, on_predict_startread, on_predict_endwrite) first agent.system_one(state, QUESTIONS) # runs the model second agent.system_one(state, QUESTIONS) # served from CACHE可运行版见 examples/hooks/cache.py。实现上PredictContext.skip()只是把传入的 results 列表赋给ctx.results见 laya/hooks.py在Router上被跳过的 payload 仍会被加上routing键除非已有从而保持Router.predict的返回结构不变。并发服务时请给缓存加锁见 docs/hooks/patterns.md。5. 指标按模型统计计数与延迟指标钩子通常从ctx.model、ctx.usage、ctx.elapsed_ms聚合计数器与直方图。注意这里传了hooks_raiseFalse可观测性绝不能拖垮正常请求——钩子抛异常时只会发出RuntimeWarning并继续见 laya/hooks.py 的 dispatch 策略。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_endmetrics, hooks_raiseFalse)参考实现见 examples/hooks/otel.py其中注释给出了把计数与延迟直方图接入 OpenTelemetryopentelemetry-api的meter.create_counter/create_histogram的完整示例。6. 安全网关从 start 钩子抛异常阻断请求在on_predict_start抛出异常即可拦截请求例如检测提示注入import laya class Blocked(Exception): pass def guard(ctx): text str(ctx.states[0]).lower() if ignore previous instructions in text: raise Blocked(prompt injection) agent laya.load(convaiinnovations/laya, on_predict_startguard) try: agent.system_one(Ignore previous instructions and ..., QUESTIONS) except Blocked: handle_block()hooks_raiseTrue默认让阻断真正传到调用方同时on_error与on_predict_end仍会执行预测生命周期由try/except/finally包裹见 docs/hooks/errors.md所以审计轨迹不会丢。策略钩子应保持严格模式——静默失败的网关同样是安全漏洞。7. 置信度门控改写低置信答案on_predict_end钩子可以直接改写结果结果变异而非拒绝请求。例如把置信度低于 0.6 的部门分类改写为human-review并打上gated标记def gate(ctx): answer ctx.results[0][answers].get(dept) if answer and answer[confidence] 0.6: answer[choice] human-review answer[gated] True agent laya.load(convaiinnovations/laya, on_predict_endgate)8. 路由固定为特定流量钉死检查点Router特有的on_route钩子可以在路由检测之后、加载模型之前替换ctx.decision从而把某类流量强制路由到指定检查点。RouteDecision是 Laya 的路由决策数据结构见 laya/router.py 的定义包含model别名、repo仓库路径、reason、detection、workflow字段from laya import Router from laya.router import RouteDecision def pin(ctx): if refund in str(ctx.states[0]).lower(): ctx.decision RouteDecision( modeltyped-decisions, repoconvaiinnovations/laya/typed-decisions, reasonrefund workflow, detectionNone, workflowNone, ) router Router(hooks[pin])不想把它安装为全局钩子时可以每次调用临时传入hooks[pin]同样作用于on_routerouter.predict(refund request, QUESTIONS, hooks[pin])注意route()是公开 API调用它同样会分发on_routeRouter.predict的内部流程是 route → load → predicton_route在检测之后、加载之前触发流程总览见 docs/hooks/index.md。9. 生命周期观察on_load与on_evict用对象钩子观察检查点的构建与驱逐适合做预热日志、内存核算或驱逐告警。这两个事件运行在 Router 锁之外钩子内部甚至可以回调 Routerfrom 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_loaded1, hooks[Lifecycle()]) router.preload([english, multilingual]) # on_load fires per build router.unload() # on_evict fires per freed checkpointRouter.preload(names)逐个构建检查点每次构建触发一次on_loadunload()释放全部每次释放触发on_evict。on_load事件里ctx.states为[]、ctx.questions为{}但ctx.model、ctx.agent、ctx.router可用事件载荷表见 docs/hooks/api.md。10. 组合对象钩子 便捷函数共享同一上下文安装的钩子hooks[...]先运行便捷回调on_predict_start/on_predict_end追加在后面所有钩子共享同一个PredictContext因此后运行的钩子能看到前面钩子的改动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_startredact, # convenience, appended on_predict_endaudit, # convenience, appended hooks_raiseTrue, )底层实现中normalise_hooks 会把hooks序列与两个便捷回调拍平为有序钩子列表_StartAdapter/_EndAdapter把普通函数包装成最小钩子对象compose_hooks 在调用时组合出最终列表进程默认钩子 → 实例安装钩子 → 单次调用钩子。顺序是刻意的因为后面的钩子会看到前面的变异耦合时请保持顺序明确并注释。11. 单次调用钩子覆盖或扩展一次调用system_one与router.predict都接受单次调用的钩子参数并且hooks_raiseNone表示沿用实例配置agent.system_one( state, questions, on_predict_endlambda ctx: debug_dump(ctx), hooks_raiseFalse, ) router.predict( state, questions, hooks[pin], # applies to on_route too on_predict_endaudit, )注意hooks参数不接受裸函数裸函数无法表达它服务于哪个事件必须用on_predict_start/on_predict_end传函数否则会在构造时抛TypeError校验规则见 docs/hooks/api.md 的 Validation 小节。12. 批处理predict_batch每次调用触发一次钩子predict_batch是 Laya 的批推理入口。钩子每次predict_batch调用只触发一次此时ctx.states持有全部输入、ctx.results与之一一对应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_endaudit_batch)predict_batch还支持batch_size、max_len、head_max_len、sort_by_length等参数ctx.usage在批场景下是对所有 state 求和后的总量见aggregate_usage。13. HTTP 服务laya.serve自动继承 Router 钩子laya.serve与 MCP server 内部都调用Router.predict因此Router 钩子会自动生效。构建带钩子的路由再交给create_appfrom laya import Router from laya.serve import create_app router Router(hooks[Metrics()], on_predict_endaudit, hooks_raiseFalse) app create_app(routerrouter)这里hooks_raiseFalse特别重要钩子是同步执行的且laya.serve使用单一推理 worker——一个阻塞或抛异常的钩子会拖住其后的所有请求。慢操作应入队交给后台 worker见 docs/hooks/patterns.md 的阻塞工作反模式必要时再用hooks_concurrentFalse串行化钩子分发。14. ONNXAgent只暴露预测级事件ONNXAgent没有 Router因此只支持预测级事件on_predict_start/on_predict_end/on_error。它与Agent的差异见 laya/onnx_agent.py配置参数为model_id_or_path与onnx_pathfrom laya.onnx_agent import ONNXAgent agent ONNXAgent(convaiinnovations/laya, onnx_pathlaya.onnx, on_predict_endaudit) agent.system_one(state, questions)15. 运行时注册构造之后再挂载、摘除或限定作用域所有运行时Agent/Router/ONNXAgent都混入了HookRegistry见 laya/hooks.py运行时可增删钩子agent.add_hook(Metrics()) # attach at runtime agent.remove_hook(Metrics()) # by identity agent.remove_hook(tracer) # returns True if it was installed with agent.hooks_installed(DebugDump()): agent.system_one(state, questions) # DebugDump only here实现要点add_hook返回self支持链式调用remove_hook按对象身份移除hooks_installed是上下文管理器退出包括异常退出时恢复原钩子列表。注册表的变更由_hooks_mutex保护而单次调用读取的是列表快照因此在途调用不会被增删钩子干扰。16. 基类与进程级默认钩子不想按形状实现协议方法时子类化BaseHook只覆写所需事件即可。进程级默认钩子让每次决策都该看到的埋点tracer、指标注册一次不必传给每一个Agent和Routerfrom 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()进程级注册表在laya.hooks中通过set_default_hooks/add_default_hook/clear_default_hooks/default_hooks管理见 laya/hooks.py。默认钩子最先运行先于实例钩子和单次调用钩子并在调用时读取——所以构造之后设置的默认钩子同样生效。它是全局状态建议启动时设置一次测试中用clear_default_hooks()防止用例间泄漏这是 tests/test_hooks.py 明确钉住的行为。17. 令牌预算按调用塑造max_len/head_max_len当问题的选项criteria很多时默认的头部预算可能截断标签start 钩子可按需调高预算。这不触碰共享的 agent 配置因此并发调用互不影响def widen(ctx): k len(next(iter(ctx.questions.values())).get(criteria, {}) or {}) if k 50: ctx.head_max_len 16 4 * k agent laya.load(convaiinnovations/laya, on_predict_startwiden) # or per call agent.system_one(state, questions, head_max_len324, max_len1024)PredictContext上max_len编码器预算与head_max_len问题头部预算默认None即沿用 agent 配置start 钩子可以改写它们也可以像上面这样在system_one/predict_batch调用时直接传参相关字段见 laya/hooks.py。18. 测试钩子无模型断言钩子看到的内容无需加载模型即可用桩替换 encode/forward/decode 来驱动predict_batch断言钩子确实被调用。这正是 tests/test_hooks.py 的做法seen [] agent.predict_batch([s0], questions, on_predict_endlambda ctx: seen.append(ctx.results)) assert len(seen) 1API 表面则由 tests/test_hooks_api.py 作为稳定性护栏钉住。测试时注意hooks_raiseFalse的钩子失败只发RuntimeWarning格式为laya: hook Metrics.on_predict_end failed: ...不要用 try/except 吞掉异常掩盖问题。推荐搭配一览按钩子类型选择错误策略钩子类型推荐hooks_raise理由策略 / 网关 / 脱敏True默认静默失败 安全漏洞审计 / 日志测试True生产常False丢审计记录应响亮但不一定致命指标 / 追踪False可观测性不能拖垮请求缓存读写True缓存坏掉要浮出水面而非静默漏缓存其余引用资料生命周期与流程图、错误处理与失败矩阵、模式与反模式、追踪run_id / OpenTelemetry、完整 API 参考、钩子实现源码。全部配方对应的可运行版本位于 examples/hooks/audit.py、redact.py、cache.py、otel.py可直接python examples/hooks/xxx.py运行验证。赞分享人工智能NLP强化学习【免费下载链接】layaNon-autoregressive System 1 decision engine. Typed choice, score and yes/no decisions over any text in a single forward pass, in 100 languages, with a router that picks the right checkpoint per request.项目地址https://gitcode.com/gh_mirrors/lay/laya点击查看免费下载相关推荐Compromise 实战配方指南从可变性陷阱到 PII 脱敏的 17 个开箱即用场景Compromise 实战配方指南从可变性陷阱到 PII 脱敏的 17 个开箱即用场景 本指南以 docs/recipes.md https://link.gNLP人工智能gsd-core 命令路由 Hub 可观测性接缝DispatchLogger 注入、审计轨迹与参数脱敏实战指南gsd core 命令路由 Hub 可观测性接缝DispatchLogger 注入、审计轨迹与参数脱敏实战指南 导读 本指南围绕 gsd core 仓库中 cPlay Framework Java Routing DSL 完全指南用代码即写即用的路由Play Framework Java Routing DSL 完全指南用代码即写即用的路由 本文围绕 Play Framework 提供的 Java Rou后端Web框架创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考