# 成本预测详解 > **核心问题**: 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 → 优化后再跑 ```