lambdagent/docs/hook-system.md

Hook 系统在 Lambdagent 体系下的实现分析

基于 Paper10 操作语义框架（src/lambdagent/）的设计分析 日期：2026-04-01

---

一、Hook 的形式化本质

"Hook" 在操作语义层面是在特定 CEK 转移规则触发的时刻插入额外计算。

更精确地，一个 Hook 是一个 Term 变换器 $H$：

$$H : \text{Term} \to \text{Term}$$

它将原始 Agent $a$ 包装成 $H(a)$，使得在 $a$ 的 β-规约的某个时刻（之前、之后、或条件判断中），$H$ 注入的计算被执行。

关键发现：这个定义与 Lambdagent 的现有原语完全对齐——

Hook 类型	对应 Lambdagent 原语	CEK 触发规则
Pre-hook（执行前拦截）	Compose(Tool(pre_fn), agent)	C-Comp + C-Tool
Post-hook（执行后处理）	Compose(agent, Tool(post_fn))	C-CompRet + C-Tool
Guard-hook（条件拦截/阻断）	Guard(agent, validator, k)	C-Guard + C-GuardOK/Retry/Fail
Event-hook（观察但不干预）	CEK Machine 的 Transition trace	所有规则

---

二、三种实现策略

Layer 1：Term-Level Hook（纯 Lambda 演算，零基础设施改动）

Hook 直接编码为现有原语的组合，今天就能实现：

from lambdagent.primitives import Lam, Compose, Tool
from lambdagent.extensions import Guard

# Pre-hook: 执行前校验/记录
def pre_hook(fn):
    """装饰器：在 agent 执行前运行 fn（不修改输入）"""
    def wrapper(agent):
        side_effect = Tool("pre_hook", lambda x: (fn(x), x)[1])  # pass-through
        return Compose(side_effect, agent)
    return wrapper

# Post-hook: 执行后处理/记录
def post_hook(fn):
    """装饰器：在 agent 执行后运行 fn（可修改输出）"""
    def wrapper(agent):
        transformer = Tool("post_hook", fn)
        return Compose(agent, transformer)
    return wrapper

# Blocking-hook: 条件阻断（等价于 Guard）
def blocking_hook(predicate, retry=0):
    """装饰器：输出不满足 predicate 时重试或阻断"""
    def wrapper(agent):
        return Guard(agent, predicate, k=retry)
    return wrapper

使用示例：

import logging

@pre_hook(lambda x: logging.info(f"[PreHook] input={x}"))
@post_hook(lambda x: x.upper())
@blocking_hook(lambda x: len(x) < 500, retry=2)
def my_agent():
    return Lam("writer", "Write a short summary:", "gpt-4o-mini")

# 展开后等价于：
# Guard(
#   Compose(
#     Tool(log_fn),
#     Compose(Lam("writer",...), Tool(uppercase))
#   ),
#   predicate=lambda x: len(x) < 500,
#   k=2
# )

优点：符合 Paper10 的操作语义——Hook 就是 Lambda 项，可被形式推理。 局限：无法在 Tool/LLM 调用的内部（C-Lam 规则触发的瞬间）插入逻辑。

---

Layer 2：HookTerm（混合方案，与形式语义完全兼容）

将 Hook 建模为一个新的 Term，具有自己的小步规则，在 Paper10 的形式体系内完全可推理：

class HookTerm(Term):
    """
    Hook 作为一等 Lambda 项

    小步规则（新增 E-Hook*）：
      E-HookPre:  ⟨Hook(a,pre,post) v, E, K, σ, c⟩
                  → ⟨a v, E, HookK(post, v) :: K, σ, c⟩

      E-HookPost: ⟨v, E, HookK(post, orig) :: K, σ, c⟩
                  → ⟨post(v), E, K, σ, c⟩
    """
    def __init__(
        self,
        agent: Term,
        pre:  Callable | None = None,   # pre(input) → None (side effect only)
        post: Callable | None = None,   # post(output) → output (可变换输出)
        event_filter: set[str] | None = None,  # {'llm', 'tool', 'guard', ...}
    ):
        super().__init__(f"Hook({agent._name})")
        self.agent = agent
        self.pre  = pre  or (lambda x: x)
        self.post = post or (lambda x: x)
        self.event_filter = event_filter

    def apply(self, input, ctx=None):
        ctx = ctx or Context()
        self.pre(input)                   # E-HookPre fires
        result = self.agent.apply(input, ctx)
        return self.post(result)          # E-HookPost fires

对应两条新的小步规则（可写入 Paper10 §4）：

E-HookPre:
  ⟨Hook(a, pre, post) v, E, K, σ, c⟩
  —→τ  ⟨a v, E, HookK(post, v) :: K, σ, c⟩
  (前提: pre(v) 已求值)

E-HookPost:
  ⟨v', E, HookK(post, orig) :: K, σ, c⟩
  —→τ  ⟨post(v'), E, K, σ, c⟩

---

Layer 3：Machine-Level Hook（Observer 模式，扩展 CEK Machine）

在 CEK Machine 的精确位置注入回调，只需改动 cek_machine.py 约 30 行：

from dataclasses import dataclass, field
from typing import Callable, List

@dataclass
class HookRegistry:
    """
    Hook 注册表：在 CEK 转移规则触发点插入回调

    事件类型（对应 CEK 规则）：
      pre_llm        — C-Lam 触发前（LLM 调用前）
      post_llm       — C-Lam 完成后（LLM 响应后，可修改输出）
      pre_tool       — C-Tool 触发前
      post_tool      — C-Tool 完成后
      on_guard_ok    — C-GuardOK
      on_guard_retry — C-GuardRetry
      on_guard_fail  — C-GuardFail
      on_loop_unfold — C-LoopUnfold
      on_step        — 每次 CEK 转移后
      on_halt        — 机器终止时
    """
    pre_llm:        List[Callable] = field(default_factory=list)
    post_llm:       List[Callable] = field(default_factory=list)
    pre_tool:       List[Callable] = field(default_factory=list)
    post_tool:      List[Callable] = field(default_factory=list)
    on_guard_ok:    List[Callable] = field(default_factory=list)
    on_guard_retry: List[Callable] = field(default_factory=list)
    on_guard_fail:  List[Callable] = field(default_factory=list)
    on_loop_unfold: List[Callable] = field(default_factory=list)
    on_step:        List[Callable] = field(default_factory=list)
    on_halt:        List[Callable] = field(default_factory=list)

    def register(self, event: str, fn: Callable):
        getattr(self, event).append(fn)

    def fire(self, event: str, **kwargs):
        for fn in getattr(self, event, []):
            fn(**kwargs)

_dispatch_app 中插入钩子点（以 C-Lam 为例）：

def _dispatch_app(self, op, arg):
    s = self.state

    if isinstance(op, Lam):
        # ★ Pre-LLM Hook
        self.hooks.fire("pre_llm", term=op, input=arg, state=s)

        result = op.apply(arg, ctx)

        # ★ Post-LLM Hook（hook 可通过 dict 修改输出）
        hook_result = {"value": result}
        self.hooks.fire("post_llm", term=op, input=arg,
                        output=hook_result, cost=cost_llm, state=s)
        result = hook_result["value"]

        s.control = result
        return "C-Lam", label

实际使用：

registry = HookRegistry()

# 成本审计
registry.register("post_llm", lambda term, input, output, cost, state:
    print(f"[Audit] {term._name}: {cost.tokens} tokens"))

# PII 过滤
import re
def pii_filter(term, input, output, cost, state):
    output["value"] = re.sub(r'\b\d{11}\b', '[PHONE]', str(output["value"]))
registry.register("post_llm", pii_filter)

# Guard 失败报警
registry.register("on_guard_fail", lambda **kw:
    send_alert(f"Guard failed: {kw['term']._name}"))

machine = AgentCEKMachine(hooks=registry)
result = machine.run(my_agent, "input")

---

三、三层可以并存（不是三选一）

正交性

三层的作用域完全正交，分别作用于不同的对象：

Layer 3: Machine-Level Hook
│  作用域：CEK Machine 的所有转移，全局、跨切面
│  对象：执行（execution）
│
└─► 观察并包含 ↓
    Layer 2: HookTerm
    │  作用域：特定 Term 的包装，局部、可复用
    │  对象：程序结构（term structure）
    │
    └─► 展开为 ↓
        Layer 1: Compose + Tool + Guard
           作用域：单次、内联的变换
           对象：Lambda 项的代数组合

Layer 1/2 是"程序长什么样"，Layer 3 是"程序运行时发生了什么"。

Layer 3 "看到" Layer 2 的内部

当 Layer 2 的 HookTerm 执行时，它内部的 pre_fn 和 post_fn 在 CEK Machine 里触发 C-Tool 规则，Layer 3 的 pre_tool/post_tool 钩子会同样观察到它们：

执行 HookTerm(Lam("writer"), pre=log, post=upper) 时：

CEK 序列：
  C-Hook
  → C-Tool(log)      ← Layer 3 的 pre_tool hook 也触发
  → C-Lam(writer)    ← Layer 3 的 pre_llm/post_llm 触发
  → C-Tool(upper)    ← Layer 3 的 post_tool hook 也触发

唯一的危险：重复触发

如果三层对同一事件做同一件事，会触发多次：

# ❌ 错误：三层都在记录同一个 LLM 的输出
agent = Compose(Lam("writer", ...), Tool("log1", log_fn))   # Layer 1
agent = HookTerm(agent, post=log_fn)                         # Layer 2 再记录
registry.register("post_llm", lambda **kw: log_fn(...))     # Layer 3 再记录

解决方法：不同层负责不同关切。

---

四、推荐的职责划分

层级	定位	典型用途	特点
Layer 3	基础设施层	全局 token 审计、限流、OpenTelemetry、异常上报	对业务代码透明
Layer 2	策略层	PII 过滤、安全沙盒、领域特定 retry 策略	作为库主动引用
Layer 1	业务层	一次性输入预处理、快速原型调试打印	内联写在业务中

---

五、并存示例

# ── Layer 3: 全局基础设施（系统级，一次性配置）──
registry = HookRegistry()
registry.register("post_llm", cost_auditor)      # 审计所有 LLM 调用
registry.register("on_guard_fail", pager_duty)   # 任何 Guard 失败就报警
machine = AgentCEKMachine(hooks=registry)

# ── Layer 2: 领域策略（库级，按需引用）──
class MedicalHook(HookTerm):
    """医疗场景专用：PII 过滤 + HIPAA 合规日志"""
    def __init__(self, agent):
        super().__init__(
            agent,
            pre=lambda x: verify_patient_consent(x),
            post=lambda x: strip_pii(x),
        )

# ── Layer 1: 业务逻辑（业务级，内联）──
diagnosis_agent = MedicalHook(           # Layer 2 包装
    Compose(
        Tool("fetch_ehr", get_records),  # Layer 1 inline
        Lam("diagnoser", MEDICAL_PROMPT),
        Tool("format_icd", icd_encoder), # Layer 1 inline
    )
)

# Layer 3 的 registry 自动观察整个执行过程
result = machine.run(diagnosis_agent, patient_id)

执行时每一层做自己该做的事：

Layer 3 看到：fetch_ehr → verify_consent → diagnoser_llm → strip_pii → format_icd
              └─ cost_auditor 汇总所有 LLM token ─────────────────────────────────┘

Layer 2 做了：verify_consent（pre）和 strip_pii（post）

Layer 1 做了：fetch_ehr 和 format_icd 的数据流组合

---

六、与 Claude Code Hook 系统的映射

Claude Code 的 Hook 事件类型与 Lambdagent CEK 规则的精确对应：

Claude Code Hook	Lambdagent 对应	CEK 触发点
PreToolUse	HookTerm(tool, pre=fn)	C-Tool 之前
PostToolUse	HookTerm(tool, post=fn)	C-ToolRet 之后
Stop	Guard(agent, λv.False, 0)	C-GuardFail
Notification	Tool(notify_fn) in trace	on_step callback
SubagentStop	Loop bound check	C-LoopBound

---

七、形式化角度：代数效果的三层

这个结构在 PL 理论里对应代数效果（Algebraic Effects）的处理器栈：

Effect handlers (Layer 3)   ←  全局 handler，捕获所有 perform
        ↕
Effect operations (Layer 2) ←  局部 perform，可复用
        ↕
Pure terms (Layer 1)        ←  无效果的纯变换

三层天然并存，因为它们描述的是同一个计算的不同视角，不是同一个问题的三个备选解法：

• 选一层 → 只有一个视角
• 三层并存 → 完整的可观测性（Layer 3）+ 可复用性（Layer 2）+ 灵活性（Layer 1）

---

八、结论

Hook 系统不是 Lambdagent 体系之外的东西，它本身就是 Lambda 演算的 Compose（函数组合）和 Guard（依赖类型验证）的特例。

Claude Code 的 Hook 系统在形式上等价于在 CEK Machine 的 YIELD 点（LLM/Tool 调用前后）插入额外的 β-规约步骤。三层实现策略分别对应 Lambda 演算代数、一等 Term 扩展、执行时观察者三个层次，正交互补，推荐并存使用。