成本预测详解

核心问题: 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% (文件不存在/权限不足)

四、演示案例

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

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% → 合理, 长任务是可行的

六、代码演示

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 → 优化后再跑

cost-prediction-explained.md 11 KB 永久链接 文件历史 原始文件