lambdagent/docs/patterns.md

协作模式库 (Pattern Library)

Pattern 是位于 Skill（原子能力）和 Orchestration（业务编排）之间的可复用协作结构。 每个 Pattern 是一个二阶 Lambda 项：接受 Agent（Term），返回 Agent（Term）。 Pattern 自身不包含业务逻辑——它定义的是"多个 Agent 如何协作"。

---

一、六大协作模式总览

模式	中文名	λA 构造	用途
review	审查模式	Loop + Guard + Compose	生产 → 审查 → 通过或返工
fan_out_merge	扇出合并模式	Par + Compose	多路并行 → 汇总
pipeline	流水线模式	Compose (链式)	顺序多阶段处理
escalation	逐级升级模式	If (嵌套)	简单→复杂逐级尝试
map_reduce	分而治之模式	Tool + Par + Compose	拆分 → 并行处理 → 合并
debate	对抗辩论模式	Loop + Compose	正方 vs 反方 → 裁决

---

二、审查模式 (Review)

定义

生产者产出内容，审查者检验质量，不通过则返工，直到审查通过或达到最大轮数。

生产者 ──→ 审查者 ──→ 通过? ──Yes──→ 输出
                        │
                        No
                        │
                        ↓
                      返工（回到生产者）

λA 形式表征

review(producer, reviewer, n) =
  fix_n(λself. λx.
    let r = producer(x) in
    let v = reviewer(r) in
    if "APPROVED" ∈ v then r
    else self(r ⊕ v)
  )

使用的构造：Loop (Y 组合子) + Guard (输出验证) + Compose (管道)

类型签名 (Paper III)

review :
  (A →^ε₁ B)        -- producer: 输入 A 产出 B
× (B →^ε₂ B)        -- reviewer: 检查 B，输出含 APPROVED 或修改意见
× ℕ                  -- max_rounds
→ (A →^{(ε₁·ε₂)ⁿ} {x:B | P(x)})
                     -- 结果: 带精化类型的 B（审查通过保证）

成本分级

g_review = (g_producer · g_reviewer)ⁿ
         = ((p₁·p₂)ⁿ, n·(t₁+t₂), n·(l₁+l₂), n·(m₁+m₂))

最坏情况: n 轮都不通过，成本 = n × (生产 + 审查) 的单轮成本

代数性质

• 满足循环展开律 (Paper II 定理 39)：review(f, g, n) ≡ If(P∘g∘f, g∘f, f >> review(f, g, n-1))
• 注意：Guard 不满足分配律 (Paper II 命题 42)，所以 review(f>>g, r) ≠ f >> review(g, r)

用途

• 代码审查（编写 → 审查 → 修改）
• 论文修改（写作 → 评审 → 修订）
• 翻译校对（翻译 → 校对 → 修正）
• 数据清洗（提取 → 验证 → 修正）

YAML 配置

orchestration:
  pattern: review
  producer:
    skill: content-writer
  reviewer:
    skill: content-reviewer
  maxRounds: 3
  approvalKeyword: "APPROVED"

Python 代码

from lambdagent.patterns import review_pattern
from lambdagent.primitives import Lam

writer = Lam("writer", "写一篇关于 AI 安全的技术博客")
editor = Lam("editor", "审查文章质量。通过则回复 APPROVED，否则给出修改意见")

pipeline = review_pattern(writer, editor, max_rounds=3)
result = pipeline("AI 安全最新进展")

---

三、扇出合并模式 (Fan-Out Merge)

定义

将同一个任务分发给多个 Agent 并行处理，收集所有结果后交给合并者综合输出。

            ┌─→ Agent A ─→ 结果 A ─┐
输入 ──→ 分发 ─→ Agent B ─→ 结果 B ──→ 合并者 ──→ 最终输出
            └─→ Agent C ─→ 结果 C ─┘

λA 形式表征

fan_out_merge(agents, merger) =
  λx. merger(⟨a₁(x), a₂(x), ..., aₙ(x)⟩)

使用的构造：Par (并行对) + Compose (管道)

类型签名

fan_out_merge :
  List(A →^εᵢ Bᵢ)                -- agents: 各自独立处理
× (B₁×...×Bₙ →^ε_m C)           -- merger: 接收所有结果，综合输出
→ (A →^{(ε₁∥...∥εₙ)·ε_m} C)     -- 效果: 并行 ∥ 然后串行 ·

成本分级

g_fan_out_merge = (g₁ ∥ ... ∥ gₙ) · g_merger
               = (∏pᵢ · p_m,  Σtᵢ + t_m,  max(lᵢ) + l_m,  Σmᵢ + m_m)

延迟 = max(各分支延迟) + 合并延迟   ← 并行优势: 延迟取 max 而非 sum
成本 = 所有分支成本之和 + 合并成本   ← 成本不省（全部都要跑）

代数性质

• 满足 Pair 对称律 (Paper II 定理 41)：Agent 的执行顺序不影响结果
• 前提条件：store independence (Paper II 命题 30)——各 Agent 写不同的存储键

用途

• 多角度分析（安全分析 + 性能分析 + 代码质量 → 综合报告）
• 多源搜索（Google + Arxiv + GitHub → 汇总）
• 专家委员会（3 个专家独立评估 → 综合意见）
• 多语言翻译（同时翻译为英/日/韩 → 汇总）

YAML 配置

orchestration:
  pattern: fan_out_merge
  fan_out:
    - skill: web-research
      input: "Research: {topic}"
    - skill: competitor-analysis
      input: "Analyze competitors for: {topic}"
    - skill: audience-profiling
      input: "Profile target audience for: {topic}"
  merger:
    type: simple
    systemPrompt: "综合以上三份分析报告，输出一份完整的内容策略。"

Python 代码

from lambdagent.patterns import fan_out_merge
from lambdagent.primitives import Lam

analysis = fan_out_merge(
    agents=[
        Lam("security", "分析代码的安全风险"),
        Lam("performance", "分析代码的性能瓶颈"),
        Lam("quality", "分析代码质量和可维护性"),
    ],
    merger=Lam("synthesizer", "综合三份分析，输出完整的代码审查报告"),
)
result = analysis("def transfer(amount, to_account): ...")

---

四、流水线模式 (Pipeline)

定义

多个阶段按固定顺序串行执行，每个阶段的输出是下一个阶段的输入。

输入 ──→ 阶段 1 ──→ 阶段 2 ──→ 阶段 3 ──→ 输出

λA 形式表征

pipeline(f₁, f₂, ..., fₙ) = f₁ >> f₂ >> ... >> fₙ
                           = λx. fₙ(... f₂(f₁(x)))

使用的构造：Compose (函数组合)

类型签名

pipeline :
  (τ₁ →^ε₁ τ₂) × (τ₂ →^ε₂ τ₃) × ... × (τₙ₋₁ →^εₙ τₙ)
→ (τ₁ →^{ε₁·ε₂·...·εₙ} τₙ)

关键: T-Compose 规则要求 output(fᵢ) <: input(fᵢ₊₁)

成本分级

g_pipeline = g₁ · g₂ · ... · gₙ
           = (∏pᵢ, Σtᵢ, Σlᵢ, Σmᵢ)

成本 = 所有阶段之和（串行累加）

代数性质

• 结合律 (Paper II 定理 36)：(f >> g) >> h ≡ f >> (g >> h) —— 可以任意重新分组
• 单位律 (Paper II 定理 37, 38)：Id >> f ≡ f ≡ f >> Id —— 可以安全插入/移除恒等变换
• 路由分配律 (Paper II 定理 40)：Route(c, {lᵢ: fᵢ}) >> g ≡ Route(c, {lᵢ: fᵢ >> g}) —— 后处理可下推到分支内

用途

• 翻译 → 润色 → 校对
• 数据提取 → 清洗 → 分析 → 报告
• 代码生成 → 测试 → 优化

YAML 配置

type: chain
chain:
  steps:
    - skill: translator
    - skill: polisher
    - skill: proofreader

Python 代码

from lambdagent.patterns import pipeline_pattern
from lambdagent.primitives import Lam

pipe = pipeline_pattern(
    Lam("translate", "将中文翻译为英文"),
    Lam("polish", "润色英文文本，使其更地道"),
    Lam("proofread", "校对语法和拼写错误"),
)
result = pipe("人工智能正在改变世界")

---

五、逐级升级模式 (Escalation)

定义

从最简单的 Agent 开始尝试，搞不定就交给更强的 Agent，逐级升级直到解决。

输入 ──→ L1 (简单) ──→ 搞定? ──Yes──→ 输出
                         │
                         No (ESCALATE)
                         │
                         ↓
          L2 (中等) ──→ 搞定? ──Yes──→ 输出
                         │
                         No
                         ↓
          L3 (专家) ──→ 输出（最终兜底）

λA 形式表征

escalation(a₁, a₂, ..., aₙ) =
  λx. if ¬esc(a₁(x)) then a₁(x)
      else if ¬esc(a₂(x)) then a₂(x)
      else ... else aₙ(x)

其中 esc(r) = "ESCALATE" ∈ r

使用的构造：If (嵌套条件分支)

类型签名

escalation :
  List(A →^εᵢ B)        -- agents: 按能力递增排列
→ (A →^{ε₁⊔...⊔εₙ} B)  -- 效果上确界: 最坏情况跑到最后一个

成本分级

g_escalation = g₁ ⊔ g₂ ⊔ ... ⊔ gₙ
             = (min(pᵢ), max(tᵢ), max(lᵢ), max(mᵢ))

最坏情况: 前 n-1 个都升级，最后一个兜底
最好情况: 第 1 个就解决，成本 = g₁
平均情况: 取决于问题难度分布

代数性质

• escalation([a]) ≡ a —— 单元素退化为自身
• escalation([a, b]) ≡ If(¬esc∘a, a, b) —— 两元素等价于单层 If
• 不满足交换律：Agent 顺序影响成本（应把便宜的放前面）

用途

• 客服分级：L1 自动回复 → L2 人工客服 → L3 专家
• 模型分级：Haiku（快/便宜）→ Sonnet（中等）→ Opus（强/贵）
• 解决方案：模板匹配 → 规则引擎 → LLM 推理

YAML 配置

orchestration:
  pattern: escalation
  agents:
    - skill: quick-answer         # L1: 快速模板匹配
    - skill: detailed-analysis    # L2: 深入分析
    - skill: expert-consultation  # L3: 专家级推理
  escalationKeyword: "ESCALATE"

Python 代码

from lambdagent.patterns import escalation_pattern
from lambdagent.primitives import Lam

support = escalation_pattern(
    agents=[
        Lam("L1", "尝试用简单规则回答。如果无法回答，输出 ESCALATE"),
        Lam("L2", "详细分析问题并回答。如果超出能力，输出 ESCALATE"),
        Lam("L3", "你是领域专家，给出权威回答"),
    ],
    escalation_keyword="ESCALATE",
)
result = support("如何优化 PostgreSQL 的慢查询？")

---

六、分而治之模式 (Map-Reduce)

定义

将大任务拆分为小块，并行处理每个小块，最后合并所有结果。

输入 ──→ 拆分器 ──→ [块1, 块2, ..., 块n]
                        │
               ┌────────┼────────┐
               ↓        ↓        ↓
            处理器    处理器    处理器     ← 并行
               │        │        │
               ↓        ↓        ↓
            [结果1, 结果2, ..., 结果n]
                        │
                        ↓
                     合并器 ──→ 最终输出

λA 形式表征

map_reduce(splitter, mapper, reducer) =
  λx. let chunks = splitter(x) in
      let results = ⟨mapper(c₁), ..., mapper(cₙ)⟩ in
      reducer(results)

使用的构造：Tool (拆分) + Par (并行映射) + Compose (合并)

类型签名

map_reduce :
  (A →^ε_s List(C))          -- splitter: 拆分为块
× (C →^ε_m R)                -- mapper: 处理每个块
× (List(R) →^ε_r B)          -- reducer: 合并结果
→ (A →^{ε_s · (ε_m)∥ⁿ · ε_r} B)

成本分级

g_map_reduce = g_split · (g_mapper)∥ⁿ · g_reduce
             = (p_s·(p_m)ⁿ·p_r,  t_s+n·t_m+t_r,  l_s+max(l_m)+l_r,  m_s+n·m_m+m_r)

延迟: 拆分 + max(各块处理) + 合并  ← 并行优势
成本: 拆分 + n × 单块处理 + 合并    ← 成本随块数线性增长

用途

• 大文档分析（按章节拆分 → 逐章分析 → 综合）
• 多文件代码审查（拆分文件 → 逐文件审查 → 汇总）
• 批量数据处理（拆分数据集 → 并行处理 → 合并结果）

YAML 配置

orchestration:
  pattern: map_reduce
  splitter:
    type: simple
    systemPrompt: "将输入文档按章节拆分，用 --- 分隔"
  mapper:
    skill: chapter-analyzer
  reducer:
    type: simple
    systemPrompt: "综合所有章节的分析结果，输出完整报告"

Python 代码

from lambdagent.patterns import map_reduce_pattern
from lambdagent.primitives import Lam, Tool

doc_analyzer = map_reduce_pattern(
    splitter=Tool("split_chapters", lambda doc: doc.split("\n## ")),
    mapper=Lam("analyze", "分析这一章的关键论点和数据支撑"),
    reducer=Lam("merge", "综合所有章节分析，生成完整的文档审查报告"),
)
result = doc_analyzer(long_document)

---

七、对抗辩论模式 (Debate)

定义

正方和反方围绕一个论题展开多轮辩论，评委在每轮后裁决，直到评委给出最终结论。

           ┌──────────────────────────────┐
           │                              │
输入 ──→ 正方论证 ──→ 反方反驳 ──→ 评委裁决 ──→ 最终? ──Yes──→ 输出
                                            │
                                            No (继续辩论)
                                            │
                                            └──→ 下一轮

λA 形式表征

debate(pro, con, judge, n) =
  fix_n(λself. λx.
    let p = pro(x) in
    let c = con(p) in
    let j = judge("正方: " ⊕ p ⊕ "\n反方: " ⊕ c) in
    if "FINAL" ∈ j then j else self(x ⊕ j)
  )

使用的构造：Loop (Y 组合子) + Compose (正方 → 反方 → 评委)

类型签名

debate :
  (A →^ε₁ B)        -- proponent: 正方论证
× (B →^ε₂ B)        -- opponent: 反方反驳
× (B →^ε₃ B)        -- judge: 评委裁决
× ℕ                  -- max_rounds
→ (A →^{(ε₁·ε₂·ε₃)ⁿ} B)

成本分级

g_debate = (g_pro · g_con · g_judge)ⁿ
         = ((p₁·p₂·p₃)ⁿ, n·(t₁+t₂+t₃), n·(l₁+l₂+l₃), n·(m₁+m₂+m₃))

每轮成本 = 正方 + 反方 + 评委 的三次 LLM 调用

代数性质

• 辩论不满足对称律：交换正反方会改变结果（论证逻辑不同）
• 满足循环展开律：第 n 轮 = 一轮辩论 + 第 n-1 轮的递归

用途

• 技术决策（"应该用微服务还是单体？"）
• 投资分析（看多 vs 看空 → 裁决）
• 风险评估（乐观预测 vs 悲观预测 → 综合判断）
• 方案对比（方案 A 支持者 vs 方案 B 支持者 → 评委选择）

YAML 配置

orchestration:
  pattern: debate
  proponent:
    type: simple
    systemPrompt: "你是方案 A 的支持者，论证其优势"
  opponent:
    type: simple
    systemPrompt: "你是方案 A 的反对者，指出其缺陷"
  judge:
    type: simple
    systemPrompt: "综合正反方观点。如已有明确结论，以 FINAL 开头给出裁决"
  maxRounds: 3

Python 代码

from lambdagent.patterns import debate_pattern
from lambdagent.primitives import Lam

decision = debate_pattern(
    proponent=Lam("pro", "论证为什么应该采用微服务架构"),
    opponent=Lam("con", "指出微服务架构在当前场景的风险和问题"),
    judge=Lam("judge",
        "综合正反方观点，给出裁决。"
        "如果已有明确结论，以 FINAL 开头输出最终建议"),
    max_rounds=3,
)
result = decision("我们的电商平台要从单体迁移到微服务吗？日活 10 万，团队 5 人")

---

八、模式选择决策树

你的任务需要...

多个 Agent 对同一输入独立工作?
  ├── Yes → 需要合并结果?
  │         ├── Yes → 扇出合并模式 (fan_out_merge)
  │         └── No  → 直接用 Par
  └── No

多个阶段按顺序处理?
  ├── Yes → 有质量审查需求?
  │         ├── Yes → 审查模式 (review)
  │         └── No  → 流水线模式 (pipeline)
  └── No

大输入需要拆分处理?
  ├── Yes → 分而治之模式 (map_reduce)
  └── No

需要从多个方案中选择?
  ├── Yes → 需要对抗论证?
  │         ├── Yes → 对抗辩论模式 (debate)
  │         └── No  → 逐级升级模式 (escalation)
  └── No

简单任务?
  └── 直接用 Lam 或 Tool

---

九、模式组合

模式可以嵌套组合，构建更复杂的协作流程：

# 审查模式 + 扇出合并模式 = 多角度分析后审查
analysis = fan_out_merge(
    agents=[security_analyst, perf_analyst, quality_analyst],
    merger=synthesizer,
)
reviewed_analysis = review_pattern(analysis, reviewer, max_rounds=2)

# 流水线模式 + 逐级升级模式 = 先快速尝试，不行再深入
quick_then_deep = pipeline_pattern(
    escalation_pattern([quick_model, deep_model]),
    formatter,
)

# 分而治之 + 审查 = 分块处理后审查质量
reviewed_map_reduce = review_pattern(
    producer=map_reduce_pattern(splitter, mapper, reducer),
    reviewer=quality_checker,
    max_rounds=2,
)

λA 的组合律保证这些嵌套是类型安全的——T-Compose 在每个组合边界检查类型兼容性。

---

十、与 Claude Code 混合编排

在混合模式下，Claude Code 作为规划者选择模式和 Skill 组合，PaaS 用本地模型执行：

Claude Code:
  "这个任务适合 fan_out_merge 模式，
   扇出 3 个 skill: web-research + code-analysis + doc-review，
   合并后用 review 模式审查 2 轮"

     ↓ 调用 MCP tool

lambdagentpaas:
  1. 从 Skill Registry 查找 3 个 skill
  2. 用 fan_out_merge pattern 组合
  3. 外层包 review pattern
  4. lint + type_check + cost_estimate
  5. CEK Engine + 本地 LLM 执行

Claude Code 只做一次规划调用（$0.03），PaaS 用本地模型执行所有子任务（$0）。