# 成本预测详解

> **核心问题**: Agent 跑一次要花多少钱？——在花钱之前就知道答案。
> **理论基础**: Paper III §4.3 分级类型 (Graded Types)
> **代码实现**: `lambdagent/cost_grade.py`

---

## 一、原理：分级类型

每个 Agent（Lambda 项）被赋予一个**成本分级** `g = (p, p_fatal, t, l, m)`：

| 分量 | 含义 | 单位 |
|------|------|------|
| **p** | 端到端成功概率 | [0, 1] |
| **p_fatal** | 单步致命失败率 | [0, 1] |
| **t** | token 消耗的**上界** | 整数 |
| **l** | 延迟的**上界** | 秒 |
| **m** | 金钱成本的**上界** | 美元 |

**核心保证**: 实际成本 ≤ 预测上界。

### 概率模型：区分致命失败与非致命探索

传统做法是把每步成功率连乘（`p_total = p^n`），但这**过度悲观**：

```
传统模型: 10 步 ReAct, 每步 95% → 0.95^10 = 59.87%
  问题: 把"LLM 想错方向"也算成失败
  实际: 想错方向是正常探索, Agent 会自我纠正

修正模型: 区分两种"失败"
  致命失败 (p_fatal = 1%): API 500/解析异常/工具崩溃 → 管道无法继续
  非致命探索: 想错方向/搜索没结果 → Agent 自我纠正 → 不算失败

  10 步: (1 - 0.01)^10 = 0.99^10 = 90.44%

  传统模型: 59.87% → 过度悲观, 开发者不敢写长循环
  修正模型: 90.44% → 符合直觉, 10 步探索很合理
```

---

## 二、组合规则

### 串行 `f >> g`

两个 Agent 依次执行：

```
g_serial = g1 · g2

  p:       p1 × p2          两个都不出致命错误
  t:       t1 + t2          Token 累加
  l:       l1 + l2          延迟累加（依次等待）
  m:       m1 + m2          成本累加
  p_fatal: 合并致命率       1 - (1-f1)(1-f2)
```

### 并行 `Pair(f, g)`

两个 Agent 同时执行：

```
g_parallel = g1 ∥ g2

  p:       p1 × p2          两个都不出致命错误
  t:       t1 + t2          Token 累加（并行不省 token）
  l:       max(l1, l2)      延迟取最慢（并行省时间!）
  m:       m1 + m2          成本累加（并行不省钱）
```

### 迭代 `Loop(body, cond, n)` — 修正版

```
g_iterate = g^n (修正版)

  p:       (1 - p_fatal)^n   只对致命失败连乘（非致命探索不降概率）
  t:       n × t             Token 最坏 n 次
  l:       n × l             延迟最坏 n 次
  m:       n × m             成本最坏 n 次
```

**为什么这样改**: ReAct 10 步不是"10 次独立实验都要成功"，而是"10 步探索中不出致命错误"。Agent 在第 3 步想错方向，第 4 步自我纠正——这不是失败，是正常工作流程。

### Guard `Guard(agent, P, k)`

```
g_guard = (至少一次通过验证, k×t, k×l, k×m)

  p:       1 - (1-p_useful)^k   至少一次通过验证器 P
  t/l/m:   k × 单次              最坏重试 k 次
```

Guard 的概率模型不变——重试就是为了"通过验证"，每次都是独立尝试。

### 条件 `If(cond, then, else)`

```
g_if = g_cond · max(g_then, g_else)   取最坏分支
```

### 路由 `Route(classifier, {routes})`

```
g_route = g_classifier · max(g_route_i)   取最贵分支
```

---

## 三、模型成本参数

| 模型 | tokens/次 | 延迟/次 | 价格/1K tokens | 致命率/次 |
|------|----------|---------|---------------|----------|
| claude-sonnet-4 | 800 | 2.0s | $0.003 | 1% |
| claude-opus-4 | 1200 | 5.0s | $0.015 | 1% |
| claude-haiku-4.5 | 500 | 0.5s | $0.00025 | 1% |
| gpt-4o | 800 | 1.5s | $0.0025 | 1% |
| qwen3-max | 600 | 1.0s | $0.001 | 1% |

单次 Lam 调用的分级：

```
g_sonnet = (p=0.99, t=800, l=2.0s, m=$0.0024, p_fatal=0.01)

含义: 
  99% 概率不出致命错误
  最多用 800 tokens, 花 $0.0024
  致命失败率 1% (API 500/超时)
```

Tool 调用：

```
g_tool = (p=0.98, t=0, l=0.1s, m=$0.00, p_fatal=0.02)

含义:
  98% 概率不出致命错误
  不花 token 不花钱
  致命失败率 2% (文件不存在/权限不足)
```

---

## 四、演示案例

### 案例: 海事法规安全审查管道

```yaml
stage1: 3 路并行 ReAct 扫描 (Sonnet × 10 步 × 3 路)
stage2: 深度分析 (Opus × 15 步)
stage3: 报告生成 (Opus 单次) + 审查 (Sonnet × 8 步, Guard 重试 2 次)
```

### 逐步计算

**第 1 步: 单个 Scanner (Sonnet ReAct 10 步)**

```
g_sonnet = (p=0.99, t=800, l=2.0s, m=$0.0024, p_fatal=0.01)

g_scanner = g_sonnet^10 (迭代规则, 修正版)
  p:       (1-0.01)^10 = 0.99^10 = 0.9044   ← 10 步不出致命错误 90.4%
  t:       10 × 800 = 8,000
  l:       10 × 2.0 = 20.0s
  m:       10 × 0.0024 = $0.024
```

**第 2 步: 阶段 1 — 3 路并行扫描**

```
g_stage1 = g_scanner ∥ g_scanner ∥ g_scanner (并行规则)
  p:       0.9044^3 = 0.7397               ← 三个都不出致命错误 74.0%
  t:       8000 × 3 = 24,000
  l:       max(20, 20, 20) = 20.0s          ← 并行! 不是 60s
  m:       $0.024 × 3 = $0.072
```

**第 3 步: 阶段 2 — Opus 深度分析 15 步**

```
g_opus = (p=0.99, t=1200, l=5.0s, m=$0.018, p_fatal=0.01)

g_stage2 = g_opus^15
  p:       0.99^15 = 0.8601                ← 15 步不出致命错误 86.0%
  t:       15 × 1200 = 18,000
  l:       15 × 5.0 = 75.0s
  m:       15 × 0.018 = $0.270
```

**第 4 步: 阶段 3 — 报告 + 审查**

```
g_writer = g_opus = (0.99, 1200, 5.0s, $0.018)

g_react_review = g_sonnet^8
  p:       0.99^8 = 0.9227
  t:       8 × 800 = 6,400
  l:       16.0s
  m:       $0.0192

g_reviewer = guard(g_react_review, retries=2)   ← k=3 次尝试
  p:       1-(1-0.9227)^3 = 0.9995             ← 3 次至少一次通过: 99.95%
  t:       3 × 6400 = 19,200
  l:       3 × 16.0 = 48.0s
  m:       3 × 0.0192 = $0.0576

g_stage3 = g_writer · g_reviewer (串行)
  p:       0.99 × 0.9995 = 0.9895
  t:       1200 + 19200 = 20,400
  l:       5.0 + 48.0 = 53.0s
  m:       $0.018 + $0.0576 = $0.0756
```

**第 5 步: 总管道 = stage1 · stage2 · stage3**

```
g_total = g_stage1 · g_stage2 · g_stage3
  p:       0.7397 × 0.8601 × 0.9895 = 0.6295    ← 62.95%
  t:       24000 + 18000 + 20400 = 62,400
  l:       20.0 + 75.0 + 53.0 = 148.0s
  m:       $0.072 + $0.270 + $0.0756 = $0.4176
```

### 修正前 vs 修正后

```
              修正前 (旧模型)        修正后 (新模型)
             p=0.95^n 每步连乘     p=(1-0.01)^n 只算致命
────────    ──────────────       ──────────────────
Scanner     59.87%                90.44%
Stage 1     21.46%                73.97%
Stage 2     46.33%                86.01%
Stage 3     91.38%                98.95%
Total       9.08%                 62.95%

成本: 完全相同 (t, l, m 不受影响)
  t = 62,400 tokens
  l = 148.0s
  m = $0.4176
```

**概率从 9% → 63%**——同一个管道，同一个成本，但成功率预估合理了 7 倍。

---

## 五、概率衰减表

```
传统模型 (p=0.95^n):                   修正模型 (p=(1-0.01)^n):

步数  概率    感觉                    步数  概率    感觉
───  ─────  ──────                  ───  ─────  ──────
1    95.0%  正常                    1    99.0%  正常
5    77.4%  还行                    5    95.1%  很好
10   59.9%  有风险                  10   90.4%  正常
15   46.3%  不可接受                15   86.0%  正常
20   35.8%  失败概率大              20   81.8%  可以
50   7.7%   几乎必定失败            50   60.5%  有风险
100  0.6%   不可能成功              100  36.6%  需要注意

传统模型在 15 步就崩到 46% → 开发者不敢写长循环
修正模型在 50 步还有 60% → 合理, 长任务是可行的
```

---

## 六、代码演示

```python
from lambdagent.primitives import Lam, Compose, Loop, Pair, Tool
from lambdagent.extensions import Par, Guard
from lambdagent.cost_grade import estimate_cost, format_cost_estimate

# ═══ 构建管道 ═══

scanner = Loop(
    body=Lam("scan", "扫描法规", model="claude-sonnet-4-20250514"),
    condition=lambda r, s: "DONE" in str(r),
    max_steps=10,
)
stage1 = Par(scanner, scanner, scanner)

stage2 = Loop(
    body=Lam("analyze", "深度分析", model="claude-opus-4-20250514"),
    condition=lambda r, s: "DONE" in str(r),
    max_steps=15,
)

writer = Lam("write", "生成报告", model="claude-opus-4-20250514")
reviewer = Guard(
    Loop(
        body=Lam("review", "审查报告", model="claude-sonnet-4-20250514"),
        condition=lambda r, s: "APPROVED" in str(r),
        max_steps=8,
    ),
    validator=lambda r: "APPROVED" in str(r),
    retry=2,
)
stage3 = Compose(writer, reviewer)

pipeline = Compose(Compose(stage1, stage2), stage3)

# ═══ 编译时成本预测 ═══

grade = estimate_cost(pipeline)
print(format_cost_estimate(grade))

# ═══ 逐阶段分解 ═══

g1 = estimate_cost(stage1)
g2 = estimate_cost(stage2)
g3 = estimate_cost(stage3)

print(f"Stage 1 (并行扫描):  p={g1.probability:.1%}, ${g1.money:.4f}, {g1.latency:.0f}s")
print(f"Stage 2 (深度分析):  p={g2.probability:.1%}, ${g2.money:.4f}, {g2.latency:.0f}s")
print(f"Stage 3 (报告审查):  p={g3.probability:.1%}, ${g3.money:.4f}, {g3.latency:.0f}s")
print(f"Total:              p={grade.probability:.1%}, ${grade.money:.4f}, {grade.latency:.0f}s")

# ═══ 运行后交叉验证 ═══

from lambdagent.cost_grade import validate_cost

validation = validate_cost(grade, actual_tokens=45000, actual_money=0.35)
print(f"Valid: {validation['valid']}")    # True (45000 < 62400)
print(f"Alert: {validation['alert']}")    # None (在上界内)

# 异常检测:
validation2 = validate_cost(grade, actual_tokens=130000, actual_money=0.90)
print(f"Alert: {validation2['alert']}")
# → COST_ANOMALY: Actual tokens (130000) are 2.1x the predicted upper bound
```

---

## 七、与 Managed Agents 的成本对比

同一个管道在不同平台运行：

| 成本项 | lambdagentpaas (本地) | Managed Agents (云端) |
|--------|----------------------|----------------------|
| 成本预测 | **编译时给出上界** | 没有, 跑完才知道 |
| LLM token 费 | $0 (本地 Ollama) | $0.4176 (API 定价) |
| Runtime 费 | $0 | $0.0033 ($0.08/h × 148s) |
| 失败重跑费 | $0 (本地不花钱) | ~37% 概率失败 × $0.42 |
| **10 次总费** | **$0** | **~$5.7** |

---

## 八、成本预测的局限性

| 局限 | 原因 | 影响 |
|------|------|------|
| **成本上界可能保守** | 假设所有循环跑满 maxSteps | 实际通常 50-80% |
| **不预测实际步数** | 不知道 LLM 何时 terminate | 只能用 maxSteps |
| **模型参数是经验值** | tokens_per_call 是估算 | ±20% |
| **不考虑 prompt cache** | Anthropic 有缓存优惠 | 实际可能便宜 40% |
| **致命率是固定值** | p_fatal=1% 是经验值 | 不同 API/网络条件差异大 |

**尽管有局限，成本上界仍然有效**——实际花费不会超过预测。宁可高估也不低估。

---

## 九、核心洞察

```
成本 (t, l, m) = 准确的最坏上界
  → 不管 Agent 怎么探索, 跑满 maxSteps 就是最多花这些钱
  → 这个保证是数学证明的, 不是估计

概率 (p) = 区分致命/非致命的修正模型
  → 致命失败 (API 错误): 每步 1%, 连乘后 n 步 ≈ (1-0.01)^n
  → 非致命探索 (想错方向): 不降低概率, Agent 会自我纠正
  → 步数越多, 成本越高, 但成功率衰减远比传统模型慢

成本预测的价值:
  不是"精确到分", 而是"在花钱之前发现设计问题"
  传统: 跑 10 次对比 → 每次 $0.42 → $4.20 试错成本
  lambdagentpaas: 编译时预测 → $0 → 优化后再跑
```
