# 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 直接编码为现有原语的组合，**今天就能实现**：

```python
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
```

**使用示例**：

```python
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 的形式体系内完全可推理：

```python
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 行**：

```python
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 为例）：

```python
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
```

**实际使用**：

```python
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 也触发
```

### 唯一的危险：重复触发

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

```python
# ❌ 错误：三层都在记录同一个 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** | 业务层 | 一次性输入预处理、快速原型调试打印 | 内联写在业务中 |

---

## 五、并存示例

```python
# ── 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 扩展、执行时观察者三个层次，正交互补，推荐并存使用。