模式與反模式

鉤子是一條小小的擴充套件縫，用好用壞都很容易。這一頁收集在生產裡站得住的形狀，以及會咬人的那些。

• 模式
  • 審計日誌
  • PII 脫敏
  • 快取
  • 指標
  • 防護欄
  • 置信度門控
  • 路由覆蓋
  • 模型生命週期
  • 多租戶上下文
  • 組合
  • 限定作用域的埋點
  • 程序級埋點
  • token 預算的塑造
• 反模式
  • 阻塞的工作
  • 從 end 鉤子丟擲以控制流程
  • 沒有鎖的共享可變狀態
  • 靜默失敗
  • 留住上下文
  • 脫敏太晚
  • 在逐呼叫鉤子裡做逐問題邏輯
  • 遞迴 predict
  • hooks= 裡的普通可呼叫物件
  • 在 end 鉤子裡假定結果存在
  • 依賴順序的鉤子

模式

審計日誌

記錄每一個決策，並記下足以重建它的東西：狀態、問題、答案、模型、路由決策、usage 和延遲。

import json

def audit(ctx):
    for state, result in zip(ctx.states, ctx.results or []):
        json.dump({
            "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),
        }, sys.stdout)
        sys.stdout.write("\n")

laya.load("convaiinnovations/laya", on_predict_end=audit)

一個鉤子每次呼叫觸發一次，而一次 predict_batch 呼叫把每個狀態都裝在裡面，所以記錄是按決策寫 的：ctx.states 和 ctx.results 按下標對齊。ctx.usage 和 ctx.elapsed_ms 是整次呼叫的合計； 每個結果帶它自己的 usage。

如果丟一行日誌絕不能導致請求失敗，就讓它寬鬆：hooks_raise=False。如果審計軌跡是一項合規要求， 就讓它嚴格。

PII 脫敏

脫敏必須發生在 on_predict_start 裡、tokenization 之前，否則模型已經見過那些資料了。

import re
EMAIL = re.compile(r"\b[\w.+-]+@[\w-]+\.[\w.-]+\b")

def redact(ctx):
    ctx.states = [
        EMAIL.sub("[email]", s) if isinstance(s, str) else s
        for s in ctx.states
    ]

laya.load("convaiinnovations/laya", on_predict_start=redact)

脫敏鉤子是一條策略鉤子：保持 hooks_raise=True，因為一個靜默壞掉的脫敏器就是一次資料洩漏。

快取

一個 start 鉤子查快取並呼叫 ctx.skip(...)；一個 end 鉤子填充它。命中時前向傳播被跳過。

import hashlib, json

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

laya.load("convaiinnovations/laya", on_predict_start=read, on_predict_end=write)

鉤子每次呼叫觸發一次，所以在 predict_batch 上只按一個狀態做 key 是不夠的：ctx.skip() 會替換 這次呼叫本會返回的每一個結果。併發服務時用一把鎖守住快取。在 Router 上，被快取的載荷仍然會拿到 一個 routing 鍵，所以返回形狀不變。

同一對鉤子也能通過單個 LangChain 節點的 hooks= 參數用在它上面，這是在不改變 該 agent 其他每個呼叫方所見內容的前提下，快取圖裡某一步熱點的辦法。

指標

從 ctx.model、ctx.usage 和 ctx.elapsed_ms 得到計數器和直方圖。讓它寬鬆。

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)

laya.load("convaiinnovations/laya", on_predict_end=metrics, hooks_raise=False)

防護欄

一條策略鉤子丟擲以攔下一個請求。hooks_raise=True（預設）讓攔截抵達呼叫方；on_error 和 on_predict_end 仍然執行，於是審計軌跡記下它。

class Blocked(Exception):
    pass

def guard(ctx):
    if any("ssn" in str(state).lower() for state in ctx.states):
        raise Blocked("possible PII in state")

laya.load("convaiinnovations/laya", on_predict_start=guard)

拿狀態形狀去測它。一個只讀 ctx.states[0] 的防護欄會攔下一次單呼叫，卻讓一次 predict_batch 呼叫把剩下的每一個狀態都送進前向傳播。

置信度門控

一個 end 鉤子把低置信度的答案改寫成安全的回退值，或者為下游邏輯給它加標註。這是一次結果改動，不是 一次拒絕。

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

laya.load("convaiinnovations/laya", on_predict_end=gate)

通過 ctx.results 來改動，它為這次呼叫的每個狀態儲存一個 dict：只門控第一個，就會把其他每一個 低置信度的答案不加標註地發出去。

路由覆蓋

on_route 可以替換 ctx.decision，為某一類流量釘住一個 checkpoint。

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(hooks=[pin])

模型生命週期

on_load 和 on_evict 觀察 checkpoint。用它們做預熱日誌、記憶體核算或驅逐告警。它們執行在 Router 鎖之外，所以鉤子可以回撥進 Router。

class Lifecycle:
    def on_load(self, ctx):
        print("loaded", ctx.model)

    def on_evict(self, ctx):
        print("evicted", ctx.model)

Router(hooks=[Lifecycle()])

多租戶上下文

通過在鉤子閉包裡捕獲它，或者從一個 context-local 裡讀它，來把租戶 id 串起來。不要在鉤子物件上存逐 請求的狀態，除非上鎖。

def make_audit(tenant):
    def audit(ctx):
        ship(tenant, ctx.run_id, ctx.results)
    return audit

agent = laya.load("convaiinnovations/laya", on_predict_end=make_audit("acme"))

組合

幾種不同種類的鉤子自然地組合；已安裝的鉤子按順序先執行。

agent = laya.load(
    "convaiinnovations/laya",
    hooks=[Metrics(), Guardrail()],     # metrics first, then policy
    on_predict_start=redact,            # convenience callables appended after hooks
    hooks_raise=True,                   # policy failures are fatal
)

讓這個順序刻意且記錄在案，因為更晚的鉤子會看到更早那個的改動。

限定作用域的埋點

只給需要它的程式碼掛上一個 tracer 或除錯鉤子，而不是重建整個 agent。hooks_installed 在退出時恢復 之前的列表，即便程式碼塊拋異常也一樣。

with agent.hooks_installed(DebugDump()):
    agent.system_one(state, questions)   # DebugDump only here

add_hook/remove_hook 不用程式碼塊也能做同樣的事，適合一個活得和程序一樣久的 tracer。

程序級埋點

一個每個決策都該看到的 tracer 或指標鉤子可以註冊一次，而不是傳給每個 Agent 和 Router。預設值 在例項鉤子和逐呼叫鉤子之前執行。

from laya import BaseHook, hooks

class Metrics(BaseHook):
    def on_predict_end(self, ctx):
        record(ctx.model, ctx.elapsed_ms)

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

這是全域性狀態，所以要有意識地限定它：在啟動時設定一次，並在測試裡 clear_default_hooks()，這樣 一個測試就不會把鉤子洩漏進下一個。

token 預算的塑造

一個 start 鉤子可以為一次呼叫抬高 token 預算，例如當一個問題的選項很多、預設的 head 預算會讓標籤 塌掉時。有四個細節決定這個鉤子是有幫助，還是悄悄把這次呼叫變得更糟：

• 一個 start 鉤子的 ctx.head_max_len 替換這次呼叫的預算。在它之前生效的是呼叫方自己的逐呼叫 值，或者 ctx.agent.cfg 裡的 checkpoint 預設值 —— 所以要拿它做比較：寫一個光禿禿的數字可能 會拉低呼叫方已經設好的預算。
• 一次呼叫回答它攜帶的每一個問題，所以按其中最寬的那個定尺寸，而不是按碰巧最先出現的那個。
• 一旦選項裝不下 head，laya/common.py 給每個選項 max(4, (head_max_len - 16) // k) 個 token。 因此 16 + 4 * k 正好落在那條下限上：每個標籤仍然被削減到它和其他標籤共享的那些 token，這 正是這個鉤子寫來避免的塌陷。16 + 8 * k 讓它們保持可區分。
• 狀態拿到 max_len - head_max_len - 8 個 token，所以拓寬的 head 必須連帶拓寬 max_len，否則 狀態就丟了自己的視窗。

def widen_for_high_cardinality(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                          # 8 tokens per label, not the core's floor of 4
    if need > head:                            # only ever widen, never lower
        ctx.head_max_len = need
        ctx.max_len = max(window, need + 8 + 64)   # 8 reserved, then room for the state

agent = laya.load("convaiinnovations/laya", on_predict_start=widen_for_high_cardinality)

這不碰共享的 agent 配置，所以併發的呼叫不受影響。同樣的旋鈕也可以逐呼叫使用： agent.system_one(state, questions, head_max_len=512, max_len=1024)。

拓寬不是免費的：更長的視窗意味著更大的張量，而 checkpoint 是在 512（laya）和 1,024 個 token 上訓練的。超過之後，用 predict_shortlist 收窄候選比硬撐預算更好。

反模式

阻塞的工作

鉤子執行在呼叫執行緒上，而 laya.serve 用一個單獨的推理 worker。一個 sleep、等一次網路往返或呼叫 input() 的鉤子會把它後面的每個請求都堵住。

# bad: blocks the whole server
def audit(ctx):
    requests.post("https://slow.example/decisions", json=..., timeout=30)

# better: enqueue, let a background worker ship it
def audit(ctx):
    QUEUE.put_nowait(record(ctx))

如果你非要做慢工作，設定 hooks_concurrent=False 至少讓鉤子本身不重疊，並讓 laya.serve 跑在 一個佇列後面。

從 end 鉤子丟擲以控制流程

on_predict_end 在推理之後執行。在那裡丟擲會丟掉一個已經算出來的結果，而且在成功路徑上還會浮現給 呼叫方。要攔截，用 start 鉤子，在付推理代價之前；要改答案，改寫 ctx.results。

沒有鎖的共享可變狀態

同一個鉤子例項在很多執行緒上執行。self.counter += 1 會競爭。

# bad
class Count:
    def __init__(self): self.n = 0
    def on_predict_end(self, ctx): self.n += 1

# good
import threading
class Count:
    def __init__(self):
        self.n = 0
        self._lock = threading.Lock()
    def on_predict_end(self, ctx):
        with self._lock:
            self.n += 1

靜默失敗

hooks_raise=False 每次失敗警告一次，但一個自己捕獲一切的鉤子把真正的問題藏了起來。

# bad: no one will ever know the audit trail stopped
def audit(ctx):
    try:
        ship(record(ctx))
    except Exception:
        pass

如果一個鉤子是可選的，讓 hooks_raise=False 處理它，並盯著那些警告。如果不是，讓它拋。

留住上下文

一個把 ctx 追加進列表的鉤子會讓整個狀態、問題、結果和 agent 都活著。

# bad: unbounded memory growth
SEEN = []
def audit(ctx):
    SEEN.append(ctx)

# good: keep only what you need
SEEN = []
def audit(ctx):
    SEEN.append((ctx.run_id, ctx.model, ctx.elapsed_ms))

脫敏太晚

到了 on_predict_end，模型已經把狀態 tokenize 過了。請在 on_predict_start 裡脫敏。

在逐呼叫鉤子裡做逐問題邏輯

每次呼叫只有一個 PredictContext，一次前向傳播回答每一個問題。沒有逐問題的事件。在 on_predict_end 裡遍歷答案，也遍歷狀態：在一個批次上，一個上下文攜帶這次呼叫的每個狀態。

def flag(ctx):
    for result in ctx.results or []:
        for qid, answer in result["answers"].items():
            if answer.get("confidence", 1.0) < 0.5:
                alert(qid, ctx.run_id)

遞迴 predict

一個呼叫 agent.predict/system_one 的鉤子會再次執行鉤子。沒有深度保護的話它會遞迴下去。

# bad
def enrich(ctx):
    ctx.results = [agent.predict(ctx.states[0], EXTRA_QUESTIONS)]

# good: guard, or use a separate agent with no hooks
def enrich(ctx):
    if getattr(ctx, "_enriched", False):
        return
    ctx._enriched = True
    ctx.results = [enricher.predict(state, EXTRA_QUESTIONS) for state in ctx.states]

hooks= 裡的普通可呼叫物件

hooks= 接受鉤子物件；一個裸的可呼叫物件說不清它是為什麼事件，所以被拒絕。用 on_predict_start= / on_predict_end=。

# bad: TypeError
laya.load("convaiinnovations/laya", hooks=[lambda ctx: None])

# good
laya.load("convaiinnovations/laya", on_predict_end=lambda ctx: None)

在 end 鉤子裡假定結果存在

失敗路徑上 ctx.results 是 None，除非某個 start 鉤子設過它。總要檢查。

def audit(ctx):
    if ctx.results is None:
        log_failure(ctx.run_id, ctx.error)
        return
    log_success(ctx.run_id, ctx.results)

依賴順序的鉤子

一個從另一個鉤子讀取改動的鉤子，除非順序被釘住，否則很脆弱。已安裝的鉤子按列表順序執行，然後是 便捷可呼叫物件；把任何耦合記錄下來，或者把耦合的鉤子合併成一個物件。

另見

• 錯誤：這些反模式背後幾個的失敗矩陣。
• 示例：上面這些模式的更完整版本。