agent-cost-analysis¶
第6章 · Agent 的评估 · 配套项目
chapter6/agent-cost-analysis
项目说明¶
实验 6-7：Agent 任务的端到端成本分析¶
配套《深入理解 AI Agent》第 6 章「实验 6-7 ★：Agent 任务的端到端成本分析」。
对一个典型的多轮 Agent 任务（客服退款）做全链路成本拆解，用自建的轻量 tracing / 可观测系统记录每次 LLM 调用的输入/输出/缓存 token、时延与成本：按步骤聚合出「哪一步最贵」，按成本构成拆出「未缓存输入 / 缓存输入 / 输出各占多少、工具返回注入了多少 token」，并给出单步成本分布（p50/p95/p99）；再做完整 2×2 A/B 对比，量化 KV-cache 复用 与 上下文压缩 两个杠杆各自及叠加后的真实成本差异。
- 默认模型
gpt-5.6-luna（当前廉价旗舰），通过 openai Python SDK 调用。首选OPENAI_API_KEY；未设置时自动回退到OPENROUTER_API_KEY（走 OpenRouter 兼容端点，gpt-*映射为openai/*）。由于gpt-5.x直连 OpenAI 需组织实名认证，只要存在OPENROUTER_API_KEY就优先走 OpenRouter。 - KV-cache 的节省是真实的：利用 OpenAI 的自动 prompt caching（前缀 ≥ 1024 token 且命中近期相同前缀时，
usage.prompt_tokens_details.cached_tokens > 0，这部分输入按缓存价 5 折计费）。 - 提供离线模式：不打模型，读入一份此前真实运行录下的 token 用量（
sample_trace.json，canned token counts），用可配置单价重新计算成本/成本构成/A/B 对比表——无需 API key 即可复现全部表格，也可一键换算到其它模型定价。
文件¶
| 文件 | 说明 |
|---|---|
config.py |
模型与价格：Pricing 单价对象 + 常见 OpenAI 模型单价预设，token→成本换算 |
tracer.py |
自建轻量 tracing：包裹每次 LLM 调用记录 token/缓存/时延/成本；成本构成拆解、单步成本分布、按步骤拆解表；支持从录制用量离线复算（from_records） |
agent.py |
多轮客服退款 Agent 任务；run_scenario(kv_cache, compress) 把两个开关正交组合成 2×2 场景，并用 tiktoken 估算「工具返回注入」token |
demo.py |
命令行入口（argparse）：在线跑真实模型 / 离线复算；选择 A/B 场景、模型单价、输出文件 |
sample_trace.json |
一次真实运行录下的四个场景逐步 token 用量（离线模式的输入，成本按当前单价重算） |
requirements.txt / env.example |
依赖与环境变量示例 |
运行¶
pip install -r requirements.txt
# 在线（真实调用模型，需要 key）：默认跑 A(朴素)+B(优化) 两组
export OPENAI_API_KEY=sk-... # 或 export OPENROUTER_API_KEY=sk-or-...（自动回退）
python demo.py
# 离线（无需 key）：用内置 canned trace 复算全部表格
python demo.py --offline --scenario all
在线模式会真实调用 OpenAI，--scenario all 约几十次 chat completion，运行一两分钟。
命令行参数（python demo.py --help）¶
| 参数 | 说明 |
|---|---|
--live / --offline |
在线真实调用（默认）/ 离线从 trace 文件复算（无需 key） |
--scenario NAME |
ab(默认=naive+both) / all(2×2 四组) / 逗号分隔子集 naive,kv,compress,both |
--trace FILE |
离线读取的 canned trace，默认 sample_trace.json |
--save-trace FILE |
在线跑时把真实 token 用量落盘，供之后 --offline 复算 |
--model NAME |
模型名（决定默认单价预设：gpt-4o-mini/gpt-4o/gpt-4.1-mini/gpt-4.1） |
--price-input/-cached/-output |
覆盖三档单价（每百万 token 美元） |
--no-warmup |
关闭 KV-cache 组的前缀预热（默认预热以稳定命中缓存） |
--output FILE |
把成本拆解结果（含成本构成/分布/逐步用量）写成 JSON |
不改任何参数直接
python demo.py，行为与之前一致：在线跑 A(朴素) 与 B(优化) 两组并打印拆解 + A/B 对比表。
A/B 四种策略（完整 2×2）¶
同一个 8 轮客服退款任务（查订单 → 查物流 → 查退款政策 → 查知识库 → 风控 → 发起退款 → 通知 → 关单），四组做的是同样的逻辑工作，只在上下文构造上不同——因此成本差异纯粹来自「是否 KV-cache 友好」与「是否压缩上下文」两个正交开关：
| 场景 | KV-cache | 压缩 | 上下文构造 |
|---|---|---|---|
naive A 朴素 |
✗ | ✗ | 每轮 system 前塞随机 session 头（破坏前缀）+ 历史工具返回原样带全 |
kv 仅缓存 |
✓ | ✗ | 稳定长前缀（system 逐字节不变）+ 历史不压缩 |
compress 仅压缩 |
✗ | ✓ | 前缀不稳定 + 仅最近 2 轮保留完整工具返回、更早压成一句话摘要 |
both B 优化 |
✓ | ✓ | 稳定长前缀 + 上下文压缩（两个杠杆叠加） |
为聚焦「输入侧」两个杠杆，四组都用
temperature=0且限制输出长度（max_tokens=160），让输出 token 成本近似为四组相等的固定项，避免模型生成长度的随机波动干扰对比。工具环境是「受控」的（工具返回内容预设，真实系统里来自订单/物流/知识库后端），但每一次 LLM 调用、每一份 token 用量、每一分成本都是真实打到 OpenAI 得到的，保证可复现。
真实运行输出（gpt-4o-mini）¶
以下为一次真实运行（python demo.py --scenario all）的输出，sample_trace.json 即由该次运行落盘、供 --offline 复现。
(a) 单次任务成本拆解：按步骤 + 按成本构成 + 分布¶
===== 成本拆解: A 朴素(无缓存/无压缩)（单次任务全链路拆解） =====
步骤 工具/动作 输入tok 缓存tok 工具tok 输出tok 时延(s) 成本($)
---------------------------------------------------------------------------------------
turn-1 query_order 1113 0 276 104 3.15 0.000229
turn-2 query_logistics 1807 0 829 99 2.09 0.000330
turn-3 check_refund_policy 2154 0 1046 139 2.69 0.000406
turn-4 query_knowledge_base 2564 0 1287 160 2.92 0.000481
turn-5 query_user_history 2863 0 1389 136 2.69 0.000511
turn-6 issue_refund 3123 0 1490 160 3.07 0.000564
turn-7 send_notification 3408 0 1579 160 3.09 0.000607
turn-8 close_ticket 3668 0 1648 160 2.50 0.000646
---------------------------------------------------------------------------------------
合计 20700 0 9544 1118 22.20 0.003776
最贵的一步 → turn-8 / close_ticket: $0.000646（占总成本 17.1%）
成本构成:
未缓存输入 20700 tok $0.003105 (82.2%)
缓存输入 0 tok $0.000000 (0.0%)
输出 1118 tok $0.000671 (17.8%)
其中「工具返回注入」累计输入 9544 tok （同一份工具返回在后续每轮被反复计费）
单步成本分布(n=8): 均值 $0.000472 p50 $0.000481 p95 $0.000646 p99 $0.000646
===== 成本拆解: B 优化(KV缓存+压缩)（单次任务全链路拆解） =====
步骤 工具/动作 输入tok 缓存tok 工具tok 输出tok 时延(s) 成本($)
---------------------------------------------------------------------------------------
turn-1 query_order 1056 1024 276 139 2.41 0.000165
turn-2 query_logistics 1781 0 829 112 2.18 0.000334
turn-3 check_refund_policy 2143 1024 1046 151 2.56 0.000335
turn-4 query_knowledge_base 2310 0 1052 160 3.03 0.000442
turn-5 query_user_history 2060 1024 635 160 2.48 0.000328
turn-6 issue_refund 2143 1024 551 160 2.71 0.000341
turn-7 send_notification 2188 1024 430 160 2.79 0.000347
turn-8 close_ticket 2354 1024 429 122 2.38 0.000349
---------------------------------------------------------------------------------------
合计 16035 6144 5248 1164 20.54 0.002643
最贵的一步 → turn-4 / query_knowledge_base: $0.000442（占总成本 16.7%）
成本构成:
未缓存输入 9891 tok $0.001484 (56.1%)
缓存输入 6144 tok $0.000461 (17.4%)
输出 1164 tok $0.000698 (26.4%)
其中「工具返回注入」累计输入 5248 tok （同一份工具返回在后续每轮被反复计费）
单步成本分布(n=8): 均值 $0.000330 p50 $0.000335 p95 $0.000442 p99 $0.000442
可以看到：朴素组 A 的输入 token 随轮次从 1113 一路涨到 3668（上下文累积效应，最后一步最贵），「工具返回注入」累计吃掉 9544 输入 token；优化组 B 的输入 token 被压缩策略压住（末轮 2354 而非 3668，工具注入累计降到 5248），且多数轮次持续命中 1024 缓存 token，缓存输入把整块费用打了 5 折。
(b) 完整 2×2 A/B 对比¶
===== A/B 成本对比（同一个 8 轮客服退款任务）=====
方案 总输入tok 缓存tok 缓存率 输出tok 总成本($) vs基线
------------------------------------------------------------------------------------------
A 朴素(无缓存/无压缩) 20700 0 0.0% 1118 0.003776 基线
KV 仅缓存(稳定前缀/不压缩) 20386 13568 66.6% 1112 0.002707 -28.3%
仅压缩(前缀不稳定/摘要) 16177 0 0.0% 1147 0.003115 -17.5%
B 优化(KV缓存+压缩) 16035 6144 38.3% 1164 0.002643 -30.0%
------------------------------------------------------------------------------------------
重点对比：A 朴素(无缓存/无压缩) → B 优化(KV缓存+压缩)
总 token: A=21818 → B=17199 减少 4619 (21.2%)
缓存 token: A=0 → B=6144 （B 靠稳定前缀命中缓存）
总成本: A=$0.003776 → B=$0.002643 降低 $0.001133 (30.0%)
成本倍率: A 是 B 的 1.43 倍
结论（读出两个杠杆各自与叠加的贡献）： - 仅 KV-cache：不改上下文长度，只靠稳定前缀让重复的系统提示/工具定义/历史轮次按缓存价计费，缓存率冲到 66.6%，端到端成本就降了 28.3%——本例中它是单个最有效的杠杆。 - 仅压缩：把旧轮次工具返回压成摘要，总输入 token 从 20700 降到 16177（约 −22%），端到端成本降 17.5%；但因为前缀不稳定，缓存率为 0。 - 两者叠加（B 优化）：输入 token 最少、又能命中缓存，端到端总成本降 30.0%（A 是 B 的 1.43 倍）。单看输入侧成本，降幅落在书中「KV Cache 可降低 30%-60% 输入 token 成本」的经验区间内。
注意 KV 与压缩并非简单相加：压缩缩短了历史，可被缓存的「历史轮次」也随之变少，所以 B 的缓存 token（6144）反而少于「仅 KV」（13568）。这正是评估的价值——两个优化叠加时要实测协同效应，而不是把各自的收益直接相加。
离线复算（无需 API key）¶
sample_trace.json 存的是上面这次真实运行的逐步 token 用量（实测值）；离线模式只做「按单价重算成本」这一步纯离线数学，因此无需 key 即可复现全部表格，还能一键换算到其它模型定价：
python demo.py --offline --scenario all # 用 gpt-4o-mini 单价复算
python demo.py --offline --scenario all --model gpt-4o # 同一份 token 用量，换 gpt-4o 单价重算
python demo.py --offline --price-input 0.20 --price-cached 0.10 --price-output 0.80
换 gpt-4o 单价后，同一批 token 的四组总成本等比放大（结论与占比不变，A=\(0.062930 → B=\)0.044047，仍是 −30.0%）——这说明成本优化的相对收益由 token 结构决定，与绝对单价无关。
注意事项¶
- 具体数字每次在线运行会有小幅波动：OpenAI 的 prompt cache 是尽力而为的（按约 128 token 的块缓存、约 5–10 分钟过期、偶尔未命中），因此某些轮次可能出现
cached_tokens=0（如上面 B 组的 turn-2/turn-4）。demo.py在正式计量前会先对 KV-cache 组跑一次「预热」把稳定前缀写入缓存，让命中更稳定；--no-warmup可关闭。 - 价格写在
config.py（PRICING_PRESETS），默认是 gpt-4o-mini 的公开单价（输入 $0.15 / 缓存 $0.075 / 输出 $0.60 每百万 token）。换模型：用--model（如--model gpt-4o）或--price-*直接覆盖单价即可；缓存命中要求稳定前缀 ≥ 1024 token，换更强模型不影响该机制。 - 「工具返回注入」token 用 tiktoken 按当前模型的编码器离线估算（统计每轮输入里工具返回文本占多少 token），用于回答书中「一次工具返回可能占 2000-5000 token，且在后续每轮被反复计费」这一放大因素。
- 凭据：首选
OPENAI_API_KEY；未设置时自动回退到OPENROUTER_API_KEY（走 OpenRouter，gpt-*映射为openai/*）。gpt-5.x直连需组织实名认证，故有OPENROUTER_API_KEY时优先走 OpenRouter。离线复算（--offline）无需任何 key。
源代码¶
agent.py¶
"""
一个多轮「客服退款 Agent」任务，用于成本分析（对应书 6.x 表6-4 的客服退款示例）。
为了让实验可复现、不依赖模型工具调用的随机性，这里用「受控工具环境」：
每一轮我们把上一步工具的返回结果喂给模型，由模型（真实 LLM 调用）决定下一步怎么做。
工具返回内容是预设好的（真实 API 里会是订单系统/物流系统的返回），
但每一次 LLM 调用、每一份 token 用量、每一分成本都是真实的。
本文件把「是否 KV-cache 友好」和「是否压缩上下文」两个开关正交拆开，
可组合出完整的 2×2 A/B（对应书中「对比启用/禁用 KV Cache、启用/禁用上下文压缩」）：
run_scenario(kv_cache=False, compress=False) —— A 朴素（前缀不稳定 + 不压缩）
run_scenario(kv_cache=True, compress=False) —— 仅 KV-cache（稳定长前缀，历史不压缩）
run_scenario(kv_cache=False, compress=True) —— 仅压缩（前缀不稳定，旧轮次摘要）
run_scenario(kv_cache=True, compress=True) —— B 优化（稳定前缀 + 压缩，两个杠杆叠加）
兼容旧接口：run_naive == (False, False)，run_optimized == (True, True)。
"""
import uuid
from functools import lru_cache
from config import MODEL, Pricing
from tracer import Tracer
# 最近保留几轮完整工具返回（更早的压成一句话摘要）。压缩关闭时视为无穷大。
KEEP_VERBOSE = 2
# 限制每轮输出长度：本实验聚焦「输入侧」的 KV-cache 与压缩两个杠杆，
# 把输出 token 控制在相近水平，可避免模型生成长度的随机波动干扰 A/B 成本对比。
MAX_OUTPUT_TOKENS = 160
# ---------------------------------------------------------------------------
# 一个「足够长且稳定」的系统提示 + 工具定义（> 1024 token），
# 这是 KV-cache 命中的关键：稳定的长前缀才会被 OpenAI 自动缓存。
# 内容是一个真实感的客服退款 Agent 的系统规范与工具手册。
# ---------------------------------------------------------------------------
STABLE_SYSTEM_PROMPT = """你是「云购商城」的高级客服 Agent，专门处理售后与退款事务。你必须严格遵循以下工作规范。
# 角色与目标
你的目标是在保障平台规则的前提下，高效、礼貌地帮助用户完成退款、退货、换货、物流查询等售后诉求。
你要主动澄清诉求、核对订单状态、判断是否符合退款政策，并在权限范围内执行操作。
# 可用工具手册（tool manual）
1. query_order(order_id): 查询订单详情。返回字段包括：order_id, status, item_name, sku, price,
quantity, pay_time, pay_channel, buyer_note, seller_note, is_prepaid, warehouse, promotion_tags。
2. query_logistics(order_id): 查询物流轨迹。返回字段：carrier, tracking_no, current_status,
last_scan_time, last_scan_location, estimated_delivery, full_trace(数组，含每个扫描节点)。
3. check_refund_policy(sku, reason): 查询该 SKU 在给定退款原因下的退款政策。返回字段：
refundable, need_return, restocking_fee_rate, refund_window_days, special_notes, approval_required。
4. query_user_history(user_id): 查询用户历史行为，用于风控。返回：total_orders, refund_count_90d,
dispute_count, risk_level, vip_tier, register_days。
5. issue_refund(order_id, amount, reason): 发起退款。返回：refund_id, status, expected_arrival,
channel, operator。仅当政策允许且金额不超过订单实付金额时才可调用。
6. send_notification(user_id, channel, template, params): 给用户发通知（sms/app/email）。
# 决策规范
- 先核对订单是否存在、状态是否允许退款（已发货未签收、已签收 7 天内、未发货均有不同处理路径）。
- 未发货：可直接全额退款，无需退货。
- 已发货未签收：需拦截物流或等待退回，退款在退货签收后发起。
- 已签收 7 天内且商品无质量问题：适用 7 天无理由，可能收取一定比例的手续费（restocking_fee_rate）。
- 商品质量问题：全额退款且不收手续费，需用户提供凭证。
- 涉及大额退款（> 500 元）或高风险用户（risk_level=high）需要人工审批（approval_required=true）。
- 每一步都要给出简短的中文推理，说明你「基于什么信息、决定下一步调用哪个工具或给出什么结论」。
# 输出要求
- 保持专业、简洁、有同理心。
- 每轮只推进一步，不要臆造工具尚未返回的数据。
- 最终解决时，明确告知用户退款金额、到账时间与后续动作。
# 合规与风控
- 不得泄露其它用户信息；不得承诺超出政策的赔付；金额与政策以工具返回为准。
- 对疑似欺诈（短期高频退款、异常物流轨迹）要保持谨慎并触发人工审批。
请始终遵守以上全部规范。"""
# ---------------------------------------------------------------------------
# 预设的多轮剧本：用户诉求 + 每一步工具返回（真实 API 里来自后端系统）。
# 工具返回故意写得比较「啰嗦」（大 JSON），以体现工具结果注入上下文的 token 成本。
# ---------------------------------------------------------------------------
USER_REQUEST = (
"你好，我上周买的蓝牙耳机（订单号 ORD20240517001）到货后一直连不上，"
"试了各种办法都没用，我想退货退款，怎么处理？"
)
# 每一轮：(逻辑步骤名, 关联工具, 该工具的"啰嗦"返回文本)
# 工具返回都写得比较大（真实的订单/物流/知识库返回往往几百到上千 token），
# 以体现「工具结果注入上下文后在后续每一轮被反复计费」这一放大因素。
_LOGISTICS_TRACE = ",".join(
'{"time":"2024-05-%02dT%02d:%02d","loc":"%s","desc":"%s","operator":"SF%04d","scan_type":"auto"}'
% (17 + i // 6, 6 + i, (i * 7) % 60, loc, desc, 1000 + i)
for i, (loc, desc) in enumerate([
("华东1仓", "包裹已揽收，称重0.42kg"), ("华东1仓分拣中心", "已分拣，发往上海转运"),
("上海转运中心", "到达转运中心"), ("上海转运中心", "已发出，运输中"),
("苏州中转场", "途经中转"), ("上海浦东集散点", "到达派送网点"),
("上海浦东集散点", "安排派送"), ("浦东xx营业点", "派送中，联系收件人"),
("浦东xx营业点", "首次派送未接通"), ("浦东xx营业点", "二次派送"),
("浦东xx营业点", "已签收，签收人：本人"),
])
)
TOOL_RESULTS = [
("turn-1", "query_order",
'{"order_id":"ORD20240517001","status":"SIGNED","item_name":"Acme 主动降噪蓝牙耳机 Pro",'
'"sku":"SKU-BT-9981","price":499.00,"quantity":1,"pay_time":"2024-05-17T10:22:31",'
'"pay_channel":"wechat_pay","buyer_note":"希望尽快发货，送人用","seller_note":"已核对库存",'
'"is_prepaid":true,"warehouse":"华东1仓","promotion_tags":["满300减30","会员日","新客礼"],'
'"actual_paid":469.00,"coupon_used":"CPN-30","points_earned":469,"invoice_requested":false,'
'"sign_time":"2024-05-19T14:03:11","after_sale_window_end":"2024-05-26T23:59:59",'
'"sub_items":[{"sku":"SKU-BT-9981","name":"耳机主体","qty":1},{"sku":"SKU-BT-9981-CASE","name":"充电盒","qty":1},{"sku":"SKU-BT-9981-TIP","name":"耳塞套装","qty":1}],'
'"address_hash":"a1b2c3d4","channel":"app","device":"iOS","order_source":"首页推荐位"}'),
("turn-2", "query_logistics",
'{"carrier":"顺丰速运","tracking_no":"SF1234567890123","current_status":"已签收",'
'"last_scan_time":"2024-05-19T14:03:11","last_scan_location":"上海市浦东新区xx营业点",'
'"estimated_delivery":"2024-05-19","weight_kg":0.42,"volume":"20x15x8cm","insured":true,'
'"full_trace":[' + _LOGISTICS_TRACE + ']}'),
("turn-3", "check_refund_policy",
'{"sku":"SKU-BT-9981","reason":"quality_issue_cannot_connect","refundable":true,'
'"need_return":true,"restocking_fee_rate":0.0,"refund_window_days":7,'
'"special_notes":"质量问题类退款免手续费；需用户回寄并由质检确认是否为质量问题；'
'若质检判定非质量问题（如人为损坏、私自拆修），将按原路退回商品且不予退款；'
'回寄运费由平台承担，用户需在系统中申请电子面单；退款在质检通过后 1 个工作日内发起；'
'3C 电子类目已激活/绑定账号的商品，需先解绑再回寄，否则质检不予通过。",'
'"approval_required":false,"category":"3C-电子","quality_claim_supported":true,'
'"return_label_provided":true,"qc_sla_days":2,"related_policy_ids":["P-3C-01","P-3C-07","P-QC-12"]}'),
("turn-4", "query_knowledge_base",
'{"query":"蓝牙耳机无法连接 排查","hits":['
'{"kb_id":"KB-1001","title":"蓝牙耳机无法连接的常见原因","content":"1.未进入配对模式；'
'2.手机蓝牙缓存异常需忘记设备重连；3.固件版本过低；4.电量过低；5.多设备抢占连接。"},'
'{"kb_id":"KB-1002","title":"Acme Pro 系列重置方法","content":"长按充电盒按键15秒至指示灯红白交替闪烁即完成重置，'
'随后在手机端删除旧配对记录重新搜索。若重置后仍无法搜索到设备，多为硬件故障，建议走质量问题退换。"},'
'{"kb_id":"KB-1003","title":"质量问题判定标准","content":"重置无效 + 换设备仍无法连接 + 无进液/外观损伤，'
'通常判定为质量问题，支持免费退换。"}],"suggested_action":"引导用户重置；若无效则判定质量问题走退款流程"}'),
("turn-5", "query_user_history",
'{"user_id":"U-88123","total_orders":37,"refund_count_90d":1,"dispute_count":0,'
'"risk_level":"low","vip_tier":"gold","register_days":1180,"payment_disputes":0,'
'"avg_order_value":312.5,"last_refund_reason":"尺码不合适","chargeback_count":0,'
'"complaint_count":0,"account_status":"normal","fraud_flags":[],"lifetime_value":11562.5}'),
("turn-6", "issue_refund",
'{"refund_id":"RF20240520777","status":"APPROVED","amount":469.00,'
'"expected_arrival":"1-3 个工作日","channel":"原路退回-微信","operator":"agent-bot",'
'"return_shipping":"平台承担","return_address":"华东1仓退货组","return_label":"SF-RET-998877",'
'"qc_required":true,"qc_deadline":"2024-05-27","refund_flow":"pending_return->qc->refund"}'),
("turn-7", "send_notification",
'{"user_id":"U-88123","channel":"app","template":"refund_approved",'
'"delivered":true,"message_id":"MSG-556677","sent_time":"2024-05-20T15:20:03",'
'"params":{"refund_id":"RF20240520777","amount":469.00,"return_label":"SF-RET-998877"},'
'"read_receipt":false,"fallback_sms_scheduled":true}'),
("turn-8", "close_ticket",
'{"ticket_id":"TK-20240520-3345","status":"resolved","resolution":"refund_after_return",'
'"csat_survey_sent":true,"handle_time_s":184,"escalated":false,"agent":"agent-bot",'
'"summary_logged":true,"tags":["退款","质量问题","3C","已闭环"]}'),
]
# 供「上下文压缩」策略使用的旧轮次一句话摘要（把啰嗦的工具返回压成要点）
TOOL_SUMMARIES = {
"turn-1": "[摘要] 订单 ORD20240517001：Acme降噪耳机Pro，实付469元，已于5/19签收，售后窗口至5/26。",
"turn-2": "[摘要] 物流：顺丰已签收（5/19 14:03，本人签收），11 个轨迹节点均正常无异常。",
"turn-3": "[摘要] 退款政策：质量问题可退、免手续费，需回寄质检，回寄运费平台承担，无需人工审批。",
"turn-4": "[摘要] 知识库：先引导重置耳机；重置无效即判定质量问题，支持免费退换。",
"turn-5": "[摘要] 用户风控：37单/90天仅1次退款/low风险/gold会员，信誉良好，无欺诈标记。",
"turn-6": "[摘要] 已发起退款 RF20240520777：469元原路退微信，需回寄质检，平台承担回寄运费。",
"turn-7": "[摘要] 已通过 app 通知用户退款已批准，附回寄面单。",
}
def _next_user_msg(tool_name: str, tool_result: str) -> str:
"""把工具返回包装成喂给模型的下一条 user 消息。"""
return (
f"[工具 {tool_name} 返回结果]\n{tool_result}\n\n"
f"请基于以上结果给出你的推理，并决定下一步动作。"
)
@lru_cache(maxsize=1)
def _encoder():
"""按当前模型取 tiktoken 编码器（离线可用），用于估算「工具返回注入」的 token。"""
import tiktoken
try:
return tiktoken.encoding_for_model(MODEL)
except Exception:
return tiktoken.get_encoding("cl100k_base")
def _ntok(text: str) -> int:
return len(_encoder().encode(text))
# ---------------------------------------------------------------------------
# 四种 A/B 场景的登记表：名字 + 两个开关。
# ---------------------------------------------------------------------------
SCENARIOS = {
"naive": ("A 朴素(无缓存/无压缩)", False, False),
"kv": ("KV 仅缓存(稳定前缀/不压缩)", True, False),
"compress": ("仅压缩(前缀不稳定/摘要)", False, True),
"both": ("B 优化(KV缓存+压缩)", True, True),
}
def build_messages(idx, step, tool, result, turns, kv_cache, compress):
"""构造第 idx 轮要发给模型的 messages，并返回本轮输入里「工具返回注入」的累计 token。
kv_cache=True → system 用逐字节稳定的长前缀（可被 OpenAI 自动缓存）；
kv_cache=False → 每轮在 system 最前面塞随机 session 头，破坏前缀一致性。
compress=True → 仅最近 KEEP_VERBOSE 轮保留完整工具返回，更早轮次压成一句话摘要。
"""
if kv_cache:
system = {"role": "system", "content": STABLE_SYSTEM_PROMPT}
else:
volatile_head = f"[会话追踪] session={uuid.uuid4()} 请求序号={uuid.uuid4()}\n\n"
system = {"role": "system", "content": volatile_head + STABLE_SYSTEM_PROMPT}
history = [{"role": "user", "content": USER_REQUEST}]
tool_ctx_tokens = 0
for j, (p_step, p_assistant, p_tool, p_result) in enumerate(turns):
history.append({"role": "assistant", "content": p_assistant})
if compress and idx - j > KEEP_VERBOSE:
compact = TOOL_SUMMARIES.get(p_step, f"[摘要] {p_tool} 已完成。")
history.append({"role": "user", "content": compact})
tool_ctx_tokens += _ntok(compact)
else:
history.append({"role": "user", "content": _next_user_msg(p_tool, p_result)})
tool_ctx_tokens += _ntok(p_result)
messages = [system] + history + [
{"role": "user", "content": _next_user_msg(tool, result)}
]
tool_ctx_tokens += _ntok(result) # 本轮新注入的工具返回
return messages, tool_ctx_tokens
def run_scenario(client, kv_cache: bool, compress: bool, name: str = None,
pricing: Pricing = None) -> Tracer:
"""跑一遍 8 轮客服退款任务，两个开关正交组合出 2×2 中的一格。
两组做的是同样的逻辑工作，只在上下文构造上不同，因此成本差异纯粹来自
KV-cache 复用与上下文压缩这两个输入侧杠杆。
"""
tracer = Tracer(client, name=name or f"kv={kv_cache},compress={compress}",
pricing=pricing)
turns = []
for idx, (step, tool, result) in enumerate(TOOL_RESULTS):
messages, tool_ctx = build_messages(
idx, step, tool, result, turns, kv_cache, compress)
resp = tracer.chat(step=step, tool=tool, tool_ctx_tokens=tool_ctx,
model=MODEL, messages=messages, temperature=0,
max_tokens=MAX_OUTPUT_TOKENS)
assistant_text = resp.choices[0].message.content or ""
turns.append((step, assistant_text, tool, result))
return tracer
def run_naive(client, pricing: Pricing = None) -> Tracer:
"""(a) 朴素做法：前缀不稳定 + 不压缩历史（KV-cache 命中不了、上下文疯长）。"""
return run_scenario(client, kv_cache=False, compress=False,
name=SCENARIOS["naive"][0], pricing=pricing)
def run_optimized(client, pricing: Pricing = None) -> Tracer:
"""(b) KV-cache 友好 + 上下文压缩：稳定长前缀命中缓存 + 旧轮次摘要。"""
return run_scenario(client, kv_cache=True, compress=True,
name=SCENARIOS["both"][0], pricing=pricing)
config.py¶
"""
全局配置：模型与价格。
价格换算成本时使用「每百万 token 单价（美元）」。
默认值取自 OpenAI gpt-4o-mini 的公开定价（2024-2025）：
- 输入 : $0.15 / 1M tokens
- 缓存命中输入 : $0.075 / 1M tokens （命中 prompt cache 的输入按 5 折计费）
- 输出 : $0.60 / 1M tokens
注意：
1. 默认模型为 gpt-5.6-luna（当前廉价旗舰）。首选凭据是 OPENAI_API_KEY；若未设置，
自动回退到 OPENROUTER_API_KEY 并把模型名映射成 OpenRouter id（gpt-* -> openai/*）。
由于 gpt-5.x 直连 OpenAI 需要组织实名认证，只要 OPENROUTER_API_KEY 存在就优先走
OpenRouter（见 make_client_and_model）。仍可用 COST_DEMO_MODEL / --model 切换任意模型。
2. OpenAI 的 prompt caching 是「自动」的：当请求前缀 >= 1024 token 且与近期请求
命中相同前缀时，usage.prompt_tokens_details.cached_tokens 会大于 0，
这部分 token 按缓存价（更便宜）计费。本项目正是用它来真实体现 KV-cache 的节省。
（OpenRouter 转发 OpenAI 时同样在 prompt_tokens_details.cached_tokens 回传缓存命中。）
"""
import os
from dataclasses import dataclass
# 使用的模型（默认当前廉价旗舰 gpt-5.6-luna；可用 COST_DEMO_MODEL / --model 覆盖）
MODEL = os.environ.get("COST_DEMO_MODEL", "gpt-5.6-luna")
# OpenRouter 回退：无 OPENAI_API_KEY 时用 OPENROUTER_API_KEY 走 OpenAI 兼容端点。
OPENROUTER_BASE_URL = "https://openrouter.ai/api/v1"
def _to_openrouter_model(model: str) -> str:
"""把模型名映射成 OpenRouter id：含 '/' 视为原生 id；gpt-* -> openai/*；
claude-* -> anthropic/claude-opus-4.8；其余回退到 openai/gpt-5.6-luna。"""
if "/" in model:
return model
if model.startswith("gpt-"):
return "openai/" + model
if model.startswith("claude-"):
return "anthropic/claude-opus-4.8"
return "openai/gpt-5.6-luna"
def make_client_and_model(model: str):
"""构造 OpenAI 兼容 client 并返回 (client, 实际调用的模型名)。
回退策略（universal OpenRouter fallback）：
- gpt-5.x 且存在 OPENROUTER_API_KEY -> 优先走 OpenRouter（直连需组织实名认证）；
- 否则有 OPENAI_API_KEY -> 直连 OpenAI，模型名不变；
- 否则有 OPENROUTER_API_KEY -> 走 OpenRouter，模型名按 _to_openrouter_model 映射；
- 两者皆无 -> 抛出清晰错误。
"""
from openai import OpenAI
primary = os.environ.get("OPENAI_API_KEY", "").strip()
orkey = os.environ.get("OPENROUTER_API_KEY", "").strip()
prefer_openrouter = bool(orkey) and model.startswith("gpt-5")
if not prefer_openrouter and primary:
return OpenAI(timeout=60.0, max_retries=2), model
if orkey:
return (
OpenAI(base_url=OPENROUTER_BASE_URL, api_key=orkey,
timeout=60.0, max_retries=2),
_to_openrouter_model(model),
)
if primary:
return OpenAI(timeout=60.0, max_retries=2), model
raise RuntimeError(
"缺少可用凭据：请设置 OPENAI_API_KEY（直连 OpenAI），或设置 "
"OPENROUTER_API_KEY（自动回退到 OpenRouter）；或改用 --offline 离线复算（无需 key）。"
)
# 每百万 token 的美元单价（默认 gpt-4o-mini）
PRICE_INPUT_PER_M = 0.15 # 普通输入
PRICE_CACHED_PER_M = 0.075 # 命中缓存的输入（gpt-4o-mini 缓存读取为输入价的 50%）
PRICE_OUTPUT_PER_M = 0.60 # 输出
@dataclass(frozen=True)
class Pricing:
"""一组每百万 token 的美元单价。"""
input_per_m: float
cached_per_m: float
output_per_m: float
def cost_usd(self, prompt_tokens: int, cached_tokens: int,
completion_tokens: int) -> float:
"""按 token 用量换算成本（美元）。
prompt_tokens : usage.prompt_tokens，包含了缓存命中的部分
cached_tokens : usage.prompt_tokens_details.cached_tokens，命中缓存的输入 token
completion_tokens: usage.completion_tokens
未命中缓存的输入 = prompt_tokens - cached_tokens，按普通输入价计费；
命中缓存的输入按缓存价计费。
"""
uncached_input = max(prompt_tokens - cached_tokens, 0)
return (
uncached_input / 1_000_000 * self.input_per_m
+ cached_tokens / 1_000_000 * self.cached_per_m
+ completion_tokens / 1_000_000 * self.output_per_m
)
# 常见 OpenAI 模型的公开单价预设（每百万 token，美元），方便 CLI 用 --model 一键切换。
# 换更强的模型不影响 KV-cache 机制（仍要求稳定前缀 >= 1024 token）。
PRICING_PRESETS = {
"gpt-4o-mini": Pricing(0.15, 0.075, 0.60),
"gpt-4o": Pricing(2.50, 1.25, 10.00),
"gpt-4.1-mini": Pricing(0.40, 0.10, 1.60),
"gpt-4.1": Pricing(2.00, 0.50, 8.00),
}
def default_pricing() -> Pricing:
"""返回默认模型（config 中 MODEL）的单价；未知模型回退到模块级 PRICE_* 默认值。"""
return PRICING_PRESETS.get(
MODEL, Pricing(PRICE_INPUT_PER_M, PRICE_CACHED_PER_M, PRICE_OUTPUT_PER_M)
)
def cost_usd(prompt_tokens: int, cached_tokens: int, completion_tokens: int,
pricing: "Pricing | None" = None) -> float:
"""按 token 用量换算成本（美元）。默认用模块级单价，可传入自定义 Pricing。"""
p = pricing or Pricing(PRICE_INPUT_PER_M, PRICE_CACHED_PER_M, PRICE_OUTPUT_PER_M)
return p.cost_usd(prompt_tokens, cached_tokens, completion_tokens)
demo.py¶
"""
实验 6-7：Agent 任务的端到端成本分析（可运行 demo + CLI）。
两种运行方式：
1) 在线（--live，默认）：真实调用模型（默认 gpt-5.6-luna），token 与 cached_tokens
取自 API 返回的 usage，成本按单价换算。需要 OPENAI_API_KEY 或 OPENROUTER_API_KEY
（无 OpenAI key 时自动回退到 OpenRouter；gpt-5.x 只要有 OpenRouter key 就优先走它）。
2) 离线（--offline）：不打模型，读入一份此前真实运行录下的 trace（canned token
counts），用可配置的单价重新计算成本、成本构成与 A/B 对比表。无需 API key。
无论哪种方式，都会产出两份交付：
(a) 单次任务的「按步骤 + 按成本构成」拆解（哪一步最贵、输入/缓存/输出各占多少）。
(b) A/B 对比表：朴素 vs 仅 KV-cache vs 仅压缩 vs 两者叠加（完整 2×2），
量化 总 token / 缓存 token / 缓存率 / 成本 / 相对基线的节省。
示例：
python demo.py # 在线，默认跑 A(朴素)+B(优化) 两组
python demo.py --scenario all # 在线，跑完整 2×2 四组
python demo.py --live --save-trace out.json # 在线跑并把真实用量落盘
python demo.py --offline # 离线，用内置 sample_trace.json 重算
python demo.py --offline --model gpt-4o # 离线，换 gpt-4o 单价重算同一份用量
python demo.py --offline --price-input 0.20 --price-cached 0.10 --price-output 0.80
"""
import argparse
import json
import os
import sys
import config
from config import PRICING_PRESETS, Pricing
DEFAULT_TRACE = os.path.join(os.path.dirname(__file__), "sample_trace.json")
SCENARIO_KEYS = ["naive", "kv", "compress", "both"]
def _pct(saved: float, base: float) -> str:
if base == 0:
return "0.0%"
return f"{saved / base * 100:.1f}%"
def build_pricing(args) -> Pricing:
"""根据 --model 预设 + --price-* 覆盖，构造本次计费用的单价。"""
base = PRICING_PRESETS.get(args.model)
if base is None:
base = config.default_pricing()
return Pricing(
input_per_m=args.price_input if args.price_input is not None else base.input_per_m,
cached_per_m=args.price_cached if args.price_cached is not None else base.cached_per_m,
output_per_m=args.price_output if args.price_output is not None else base.output_per_m,
)
def resolve_scenarios(arg: str):
"""把 --scenario 解析成有序去重的场景 key 列表。"""
if arg == "all":
return list(SCENARIO_KEYS)
if arg == "ab":
return ["naive", "both"]
keys, seen = [], set()
for k in arg.split(","):
k = k.strip()
if k and k not in seen:
keys.append(k)
seen.add(k)
return keys
# ---------------------------------------------------------------------------
# 采集：在线跑真实模型，或离线从 trace 文件读回
# ---------------------------------------------------------------------------
def collect_live(keys, pricing, warmup: bool):
import agent
if not (os.environ.get("OPENAI_API_KEY") or os.environ.get("OPENROUTER_API_KEY")):
print("未检测到 OPENAI_API_KEY 或 OPENROUTER_API_KEY，请先 export 其一 "
"（无 OpenAI key 时会自动回退到 OpenRouter），或改用 --offline（离线复算，无需 key）。",
file=sys.stderr)
sys.exit(1)
# 构造 client 并解析实际模型名（可能被回退映射成 OpenRouter id）。
client, resolved = config.make_client_and_model(config.MODEL)
if resolved != config.MODEL:
print(f">>> 已回退到 OpenRouter：模型 {config.MODEL} -> {resolved}")
config.MODEL = resolved
agent.MODEL = resolved
try:
agent._encoder.cache_clear()
except Exception:
pass
tracers = []
for k in keys:
name, kv, compress = agent.SCENARIOS[k]
# KV-cache 组先跑一次「预热」，把稳定前缀写入 OpenAI 的 prompt cache，
# 让正式计量时更稳定地命中 cached_tokens（真实系统里前缀早已是热的）。
if kv and warmup:
print(f">>> 预热 [{name}] 的稳定前缀（写入 prompt cache）...")
agent.run_scenario(client, kv, compress, name=name, pricing=pricing)
print(f">>> 正在运行 [{name}] {'(在线计量)' if kv else ''}...")
tr = agent.run_scenario(client, kv, compress, name=name, pricing=pricing)
tracers.append((k, tr))
return tracers
def collect_offline(keys, pricing, trace_path):
from tracer import Tracer
if not os.path.exists(trace_path):
print(f"找不到 trace 文件：{trace_path}", file=sys.stderr)
sys.exit(1)
with open(trace_path, "r", encoding="utf-8") as f:
data = json.load(f)
by_key = {s.get("key", s.get("name")): s for s in data.get("scenarios", [])}
print(f"离线模式：读入 {trace_path}")
print(f" 该 trace 采集自模型 = {data.get('model', '?')}，"
f"共 {len(by_key)} 个场景的真实录制用量（token 数为实测，成本按当前单价重算）。")
tracers = []
for k in keys:
sc = by_key.get(k)
if sc is None:
print(f" [跳过] trace 中没有场景 '{k}'（可用在线模式 --save-trace 补录）",
file=sys.stderr)
continue
spans = sc.get("spans")
if not spans:
print(f" [跳过] trace 中场景 '{k}' 缺少 spans 数据"
f"（可用在线模式 --save-trace 补录）", file=sys.stderr)
continue
tr = Tracer.from_records(spans, name=sc.get("name", k), pricing=pricing)
tracers.append((k, tr))
if not tracers:
print("trace 里没有任何被选中的场景，退出。", file=sys.stderr)
sys.exit(1)
return tracers
# ---------------------------------------------------------------------------
# 交付 (b)：A/B 对比表
# ---------------------------------------------------------------------------
def print_ab_table(tracers):
print("\n\n===== A/B 成本对比（同一个 8 轮客服退款任务）=====")
header = (f"{'方案':<26} {'总输入tok':>10} {'缓存tok':>10} {'缓存率':>8} "
f"{'输出tok':>8} {'总成本($)':>12} {'vs基线':>10}")
print(header)
print("-" * len(header))
base_cost = tracers[0][1].total_cost()
for _, tr in tracers:
pin = tr.total_prompt_tokens()
cac = tr.total_cached_tokens()
rate = f"{(cac / pin * 100):.1f}%" if pin else "0.0%"
cost = tr.total_cost()
vs = "基线" if abs(cost - base_cost) < 1e-12 else f"-{_pct(base_cost - cost, base_cost)}"
print(f"{tr.name:<26} {pin:>10} {cac:>10} {rate:>8} "
f"{tr.total_completion_tokens():>8} {cost:>12.6f} {vs:>10}")
print("-" * len(header))
# 用第一个（基线）和最后一个（通常是 both 优化）做重点量化
base_k, base = tracers[0]
best_k, best = tracers[-1]
if base_k != best_k:
tok_a = base.total_prompt_tokens() + base.total_completion_tokens()
tok_b = best.total_prompt_tokens() + best.total_completion_tokens()
cost_a, cost_b = base.total_cost(), best.total_cost()
print(f"\n重点对比：{base.name} → {best.name}")
print(f" 总 token: A={tok_a} → B={tok_b} "
f"减少 {tok_a - tok_b} ({_pct(tok_a - tok_b, tok_a)})")
print(f" 缓存 token: A={base.total_cached_tokens()} → "
f"B={best.total_cached_tokens()} （B 靠稳定前缀命中缓存）")
print(f" 总成本: A=${cost_a:.6f} → B=${cost_b:.6f} "
f"降低 ${cost_a - cost_b:.6f} ({_pct(cost_a - cost_b, cost_a)})")
if cost_b > 0:
print(f" 成本倍率: A 是 B 的 {cost_a / cost_b:.2f} 倍")
print("\n结论: 稳定长前缀让重复的系统提示/工具定义/历史轮次按缓存价计费，")
print(" 叠加上下文压缩控制上下文增长，二者共同显著降低了端到端成本。")
def dump_output(path, tracers, pricing, model):
out = {
"model": model,
"pricing": {"input": pricing.input_per_m, "cached": pricing.cached_per_m,
"output": pricing.output_per_m},
"scenarios": [],
}
import agent
for k, tr in tracers:
name = agent.SCENARIOS[k][0] if k in agent.SCENARIOS else tr.name
out["scenarios"].append({
"key": k, "name": name,
"total_cost": tr.total_cost(),
"component_costs": tr.component_costs(),
"cost_distribution": tr.cost_distribution(),
"spans": tr.to_records(),
})
with open(path, "w", encoding="utf-8") as f:
json.dump(out, f, ensure_ascii=False, indent=2)
print(f"\n已写出结果到 {path}")
def build_parser():
p = argparse.ArgumentParser(
prog="demo.py",
description="实验 6-7：Agent 任务端到端成本分析——对客服退款 Agent 做全链路成本拆解，"
"并对比 KV-cache / 上下文压缩两个杠杆的成本差异（完整 2×2 A/B）。",
epilog="示例：\n"
" python demo.py # 在线，默认跑 A(朴素)+B(优化)\n"
" python demo.py --scenario all # 在线，跑完整 2×2 四组\n"
" python demo.py --offline # 离线，用内置 canned trace 重算（无需 key）\n"
" python demo.py --offline --model gpt-4o # 换单价离线重算\n"
" python demo.py --live --save-trace out.json # 在线跑并落盘真实用量\n",
formatter_class=argparse.RawDescriptionHelpFormatter,
)
mode = p.add_mutually_exclusive_group()
mode.add_argument("--live", action="store_true",
help="在线模式（默认）：真实调用 OpenAI，需要 OPENAI_API_KEY。")
mode.add_argument("--offline", action="store_true",
help="离线模式：不打模型，从 trace 文件读真实录制的 token 用量并按单价重算成本。")
p.add_argument("--trace", metavar="FILE", default=DEFAULT_TRACE,
help=f"离线模式读取的 trace（canned token counts）文件，默认 {os.path.basename(DEFAULT_TRACE)}。")
p.add_argument("--save-trace", metavar="FILE", default=None,
help="在线模式下把本次真实 token 用量落盘为 trace 文件（供之后 --offline 复算）。")
p.add_argument("--scenario", metavar="NAME", default="ab",
help="选择要跑的 A/B 场景：ab(默认,=naive+both) / all(2×2 四组) / "
"或逗号分隔的子集 naive,kv,compress,both。")
p.add_argument("--model", metavar="NAME", default=config.MODEL,
help=f"模型名（决定默认单价预设，可选 {', '.join(PRICING_PRESETS)}），"
f"默认 {config.MODEL}。")
p.add_argument("--price-input", type=float, default=None,
help="覆盖输入单价（每百万 token 美元）。")
p.add_argument("--price-cached", type=float, default=None,
help="覆盖缓存命中输入单价（每百万 token 美元）。")
p.add_argument("--price-output", type=float, default=None,
help="覆盖输出单价（每百万 token 美元）。")
p.add_argument("--no-warmup", action="store_true",
help="在线模式下关闭 KV-cache 组的前缀预热（默认预热以稳定命中缓存）。")
p.add_argument("--output", metavar="FILE", default=None,
help="把成本拆解结果（含成本构成/分布/逐步用量）写成 JSON 文件。")
return p
def main():
args = build_parser().parse_args()
# 让 agent / tracer 使用选定模型
config.MODEL = args.model
try:
import agent
agent.MODEL = args.model
agent._encoder.cache_clear()
except Exception:
pass
pricing = build_pricing(args)
keys = resolve_scenarios(args.scenario)
bad = [k for k in keys if k not in SCENARIO_KEYS]
if bad:
print(f"未知场景 {bad}，可选：{SCENARIO_KEYS} / all / ab", file=sys.stderr)
sys.exit(2)
print(f"模型: {args.model}")
print(f"单价(每百万token): 输入 ${pricing.input_per_m} / 缓存输入 ${pricing.cached_per_m} "
f"/ 输出 ${pricing.output_per_m}")
if args.offline:
tracers = collect_offline(keys, pricing, args.trace)
else:
print("说明: OpenAI prompt caching 自动生效（前缀>=1024token 且近期命中相同前缀），")
print(" 命中的输入 token 出现在 usage.prompt_tokens_details.cached_tokens。")
tracers = collect_live(keys, pricing, warmup=not args.no_warmup)
if args.save_trace:
dump_output(args.save_trace, tracers, pricing, args.model)
# 交付 (a)：逐场景成本拆解
for _, tr in tracers:
tr.print_breakdown(title=f"{tr.name}（单次任务全链路拆解）")
# 交付 (b)：A/B 对比表
if len(tracers) >= 2:
print_ab_table(tracers)
if args.output:
dump_output(args.output, tracers, pricing, args.model)
if __name__ == "__main__":
main()
test_trace_offline.py¶
"""
Regression tests for offline trace parsing (实验 6-7 成本分析).
Covers two crash classes found in --offline mode:
- Tracer.from_records: trace JSON with explicit null token fields -> int(None) TypeError
- demo.collect_offline: scenario dict missing the optional "spans" key -> KeyError
"""
import json
import pytest
import config
import demo
from tracer import Tracer
def _span(**overrides):
span = {
"step": "turn-1",
"tool": "query_order",
"kind": "llm",
"prompt_tokens": 100,
"cached_tokens": 10,
"completion_tokens": 12,
"tool_ctx_tokens": 50,
"latency_s": 1.2,
}
span.update(overrides)
return span
def test_from_records_tolerates_null_fields():
"""Explicit JSON nulls in numeric fields are coerced, not int(None) TypeError."""
records = [_span(prompt_tokens=None, cached_tokens=None,
completion_tokens=None, tool_ctx_tokens=None, latency_s=None)]
tr = Tracer.from_records(records, pricing=config.default_pricing())
s = tr.spans[0]
assert s.prompt_tokens == 0
assert s.cached_tokens == 0
assert s.completion_tokens == 0
assert s.tool_ctx_tokens == -1 # null tool_ctx 视为「未知」
assert s.latency_s == 0.0
def test_from_records_tolerates_missing_fields():
"""Minimal span dicts (only step/tool) still parse."""
tr = Tracer.from_records([{"step": "turn-1", "tool": "query_order"}],
pricing=config.default_pricing())
assert tr.spans[0].prompt_tokens == 0
assert tr.spans[0].tool_ctx_tokens == -1
def test_from_records_keeps_real_values():
"""Normal values pass through unchanged (no coercion side effects)."""
tr = Tracer.from_records([_span()], pricing=config.default_pricing())
s = tr.spans[0]
assert (s.prompt_tokens, s.cached_tokens, s.completion_tokens) == (100, 10, 12)
assert s.tool_ctx_tokens == 50
assert s.latency_s == 1.2
def _write_trace(tmp_path, scenarios):
path = tmp_path / "trace.json"
path.write_text(json.dumps({"model": "gpt-5.6-luna", "scenarios": scenarios}),
encoding="utf-8")
return str(path)
def test_collect_offline_skips_scenario_without_spans(tmp_path, capsys):
"""A scenario missing 'spans' is skipped with a warning, not a KeyError crash."""
trace = _write_trace(tmp_path, [
{"key": "naive", "name": "A naive"}, # no spans -> skip
{"key": "both", "name": "B both", "spans": [_span()]}, # valid
])
tracers = demo.collect_offline(["naive", "both"], config.default_pricing(), trace)
assert [k for k, _ in tracers] == ["both"]
assert "缺少 spans" in capsys.readouterr().err
def test_collect_offline_exits_when_no_usable_scenario(tmp_path):
"""When every selected scenario lacks spans, exit cleanly like the empty-trace path."""
trace = _write_trace(tmp_path, [{"key": "naive", "name": "A naive"}])
with pytest.raises(SystemExit):
demo.collect_offline(["naive"], config.default_pricing(), trace)
if __name__ == "__main__":
pytest.main([__file__, "-v"])
tracer.py¶
"""
自建的轻量级 tracing / 可观测系统。
设计沿用分布式追踪的 span 树模型（见书 6.x「Agent 的可观测性」）：
- 一次 agent 任务 = 一条 Trace
- 每次 LLM 调用 / 工具调用 = 一个 Span
- Span 记录：所属步骤、类型、token 用量（prompt/completion/cached）、时延、成本
用法：
tracer = Tracer(client)
resp = tracer.chat(step="turn-1", tool="query_order",
model=..., messages=..., temperature=0)
tracer.print_breakdown() # 打印按步骤/工具聚合的成本拆解
离线复用（不打模型、只算成本）：
tracer = Tracer.from_records(records, pricing=..., name=...)
# records 里是此前真实运行录下的每一步 token 用量（canned token counts）
"""
import time
from dataclasses import dataclass, asdict, field
from typing import List, Optional
from config import Pricing, default_pricing
def _percentile(values: List[float], q: float) -> float:
"""最近秩（nearest-rank）百分位，避免引入 numpy 依赖。q 取 0~100。"""
if not values:
return 0.0
xs = sorted(values)
if len(xs) == 1:
return xs[0]
rank = max(1, min(len(xs), int(round(q / 100.0 * len(xs) + 0.5))))
return xs[rank - 1]
@dataclass
class Span:
"""一次被追踪的调用（这里主要是 LLM 调用）。"""
step: str # 逻辑步骤名，如 "turn-2"
tool: str # 该步骤关联的工具/动作名，用于归因“哪一步最贵”
kind: str = "llm" # span 类型：llm / tool
prompt_tokens: int = 0
cached_tokens: int = 0
completion_tokens: int = 0
# 该轮输入里「工具返回结果」占用的累计 token（同一份工具返回会在后续每轮被反复计费）。
# 由上层用 tokenizer 估算并填入；离线复用时从 records 读回。-1 表示未知。
tool_ctx_tokens: int = -1
latency_s: float = 0.0
cost_usd: float = 0.0
@property
def total_tokens(self) -> int:
return self.prompt_tokens + self.completion_tokens
@property
def uncached_prompt_tokens(self) -> int:
return max(self.prompt_tokens - self.cached_tokens, 0)
class Tracer:
"""包裹 OpenAI client，自动记录每次 LLM 调用的 usage / 时延 / 成本。"""
def __init__(self, client=None, name: str = "trace",
pricing: Optional[Pricing] = None):
self.client = client
self.name = name
self.pricing = pricing or default_pricing()
self.spans: List[Span] = []
# ---------- 采集 ----------
def chat(self, step: str, tool: str, tool_ctx_tokens: int = -1, **kwargs):
"""发起一次被追踪的 chat.completions 调用。
kwargs 原样透传给 openai client（model / messages / temperature 等）。
tool_ctx_tokens：本轮输入里工具返回结果占用的累计 token（可选，用于成本归因）。
返回原始的 OpenAI response 对象，方便上层取 content。
"""
t0 = time.time()
resp = self.client.chat.completions.create(**kwargs)
latency = time.time() - t0
usage = resp.usage
# cached_tokens 藏在 prompt_tokens_details 里，注意做防御式读取
cached = 0
details = getattr(usage, "prompt_tokens_details", None)
if details is not None:
cached = getattr(details, "cached_tokens", 0) or 0
span = Span(
step=step,
tool=tool,
kind="llm",
prompt_tokens=usage.prompt_tokens,
cached_tokens=cached,
completion_tokens=usage.completion_tokens,
tool_ctx_tokens=tool_ctx_tokens,
latency_s=latency,
cost_usd=self.pricing.cost_usd(
usage.prompt_tokens, cached, usage.completion_tokens),
)
self.spans.append(span)
return resp
# ---------- 离线复用（canned token counts → 重新计成本）----------
@classmethod
def from_records(cls, records: List[dict], name: str = "trace",
pricing: Optional[Pricing] = None) -> "Tracer":
"""用此前录下的 token 用量重建一条 trace，并按给定单价重算成本（不打模型）。"""
tr = cls(client=None, name=name, pricing=pricing)
for r in records:
span = Span(
step=r.get("step", ""),
tool=r.get("tool", ""),
kind=r.get("kind", "llm"),
# 防御：trace JSON 里字段缺失或显式为 null 时按 0（tool_ctx 按未知 -1）处理
prompt_tokens=int(r.get("prompt_tokens") or 0),
cached_tokens=int(r.get("cached_tokens") or 0),
completion_tokens=int(r.get("completion_tokens") or 0),
tool_ctx_tokens=int(r.get("tool_ctx_tokens") or -1),
latency_s=float(r.get("latency_s") or 0.0),
)
span.cost_usd = tr.pricing.cost_usd(
span.prompt_tokens, span.cached_tokens, span.completion_tokens)
tr.spans.append(span)
return tr
def to_records(self) -> List[dict]:
"""导出每一步的原始 token 用量（用于落盘成 canned trace，供离线复用）。"""
out = []
for s in self.spans:
d = asdict(s)
d.pop("cost_usd", None) # 成本由单价重算，不落盘固定值
out.append(d)
return out
# ---------- 聚合 ----------
def total_cost(self) -> float:
return sum(s.cost_usd for s in self.spans)
def total_prompt_tokens(self) -> int:
return sum(s.prompt_tokens for s in self.spans)
def total_cached_tokens(self) -> int:
return sum(s.cached_tokens for s in self.spans)
def total_completion_tokens(self) -> int:
return sum(s.completion_tokens for s in self.spans)
def total_uncached_prompt_tokens(self) -> int:
return sum(s.uncached_prompt_tokens for s in self.spans)
def total_tool_ctx_tokens(self) -> int:
return sum(s.tool_ctx_tokens for s in self.spans if s.tool_ctx_tokens >= 0)
def total_latency(self) -> float:
return sum(s.latency_s for s in self.spans)
def cache_rate(self) -> float:
pin = self.total_prompt_tokens()
return self.total_cached_tokens() / pin if pin else 0.0
def component_costs(self) -> dict:
"""把总成本拆成三个成本构成要素（对应书「成本的构成要素」）：
- 未缓存输入 / 缓存输入 / 输出
以及输入侧里「工具返回注入」token 占比（若已知）。"""
p = self.pricing
uncached_in = self.total_uncached_prompt_tokens()
cached_in = self.total_cached_tokens()
out = self.total_completion_tokens()
return {
"uncached_input_cost": uncached_in / 1_000_000 * p.input_per_m,
"cached_input_cost": cached_in / 1_000_000 * p.cached_per_m,
"output_cost": out / 1_000_000 * p.output_per_m,
"uncached_input_tokens": uncached_in,
"cached_input_tokens": cached_in,
"output_tokens": out,
"tool_ctx_tokens": self.total_tool_ctx_tokens(),
}
def cost_distribution(self) -> dict:
"""按步骤的单步成本分布（p50/p95/p99）。对应书「成本分布 p50/p95/p99」。"""
costs = [s.cost_usd for s in self.spans]
n = len(costs)
return {
"n": n,
"mean": (sum(costs) / n) if n else 0.0,
"p50": _percentile(costs, 50),
"p95": _percentile(costs, 95),
"p99": _percentile(costs, 99),
"max": max(costs) if costs else 0.0,
}
# ---------- 打印 ----------
def print_breakdown(self, title: Optional[str] = None):
"""打印一次 agent 任务的按步骤成本拆解表，并指出最贵的一步、成本构成与分布。"""
print()
print(f"===== 成本拆解: {title or self.name} =====")
header = (
f"{'步骤':<8} {'工具/动作':<20} {'输入tok':>8} {'缓存tok':>8} "
f"{'工具tok':>8} {'输出tok':>8} {'时延(s)':>8} {'成本($)':>12}"
)
print(header)
print("-" * len(header))
for s in self.spans:
tctx = s.tool_ctx_tokens if s.tool_ctx_tokens >= 0 else "-"
print(
f"{s.step:<8} {s.tool:<20} {s.prompt_tokens:>8} {s.cached_tokens:>8} "
f"{str(tctx):>8} {s.completion_tokens:>8} {s.latency_s:>8.2f} "
f"{s.cost_usd:>12.6f}"
)
print("-" * len(header))
tctx_total = self.total_tool_ctx_tokens() if any(
s.tool_ctx_tokens >= 0 for s in self.spans) else "-"
print(
f"{'合计':<8} {'':<20} {self.total_prompt_tokens():>8} "
f"{self.total_cached_tokens():>8} {str(tctx_total):>8} "
f"{self.total_completion_tokens():>8} "
f"{self.total_latency():>8.2f} {self.total_cost():>12.6f}"
)
# 归因：哪一步最贵
if self.spans:
worst = max(self.spans, key=lambda s: s.cost_usd)
total = self.total_cost()
share = worst.cost_usd / total * 100 if total else 0
print(
f"\n最贵的一步 → {worst.step} / {worst.tool}: "
f"${worst.cost_usd:.6f}（占总成本 {share:.1f}%）"
)
# 成本构成拆解（未缓存输入 / 缓存输入 / 输出）
comp = self.component_costs()
total = self.total_cost() or 1e-12
print("成本构成:")
print(f" 未缓存输入 {comp['uncached_input_tokens']:>8} tok "
f"${comp['uncached_input_cost']:.6f} ({comp['uncached_input_cost']/total*100:.1f}%)")
print(f" 缓存输入 {comp['cached_input_tokens']:>8} tok "
f"${comp['cached_input_cost']:.6f} ({comp['cached_input_cost']/total*100:.1f}%)")
print(f" 输出 {comp['output_tokens']:>8} tok "
f"${comp['output_cost']:.6f} ({comp['output_cost']/total*100:.1f}%)")
if comp["tool_ctx_tokens"] > 0:
print(f" 其中「工具返回注入」累计输入 {comp['tool_ctx_tokens']} tok "
f"（同一份工具返回在后续每轮被反复计费）")
# 单步成本分布
dist = self.cost_distribution()
print(f"单步成本分布(n={dist['n']}): 均值 ${dist['mean']:.6f} "
f"p50 ${dist['p50']:.6f} p95 ${dist['p95']:.6f} p99 ${dist['p99']:.6f}")
sample_trace.json¶
{
"model": "gpt-4o-mini",
"pricing": {
"input": 0.15,
"cached": 0.075,
"output": 0.6
},
"scenarios": [
{
"key": "naive",
"name": "A 朴素(无缓存/无压缩)",
"total_cost": 0.0037758,
"component_costs": {
"uncached_input_cost": 0.0031049999999999997,
"cached_input_cost": 0.0,
"output_cost": 0.0006708,
"uncached_input_tokens": 20700,
"cached_input_tokens": 0,
"output_tokens": 1118,
"tool_ctx_tokens": 9544
},
"cost_distribution": {
"n": 8,
"mean": 0.000471975,
"p50": 0.00048059999999999997,
"p95": 0.0006462,
"p99": 0.0006462,
"max": 0.0006462
},
"spans": [
{
"step": "turn-1",
"tool": "query_order",
"kind": "llm",
"prompt_tokens": 1113,
"cached_tokens": 0,
"completion_tokens": 104,
"tool_ctx_tokens": 276,
"latency_s": 3.1535720825195312
},
{
"step": "turn-2",
"tool": "query_logistics",
"kind": "llm",
"prompt_tokens": 1807,
"cached_tokens": 0,
"completion_tokens": 99,
"tool_ctx_tokens": 829,
"latency_s": 2.0884652137756348
},
{
"step": "turn-3",
"tool": "check_refund_policy",
"kind": "llm",
"prompt_tokens": 2154,
"cached_tokens": 0,
"completion_tokens": 139,
"tool_ctx_tokens": 1046,
"latency_s": 2.685513973236084
},
{
"step": "turn-4",
"tool": "query_knowledge_base",
"kind": "llm",
"prompt_tokens": 2564,
"cached_tokens": 0,
"completion_tokens": 160,
"tool_ctx_tokens": 1287,
"latency_s": 2.922238826751709
},
{
"step": "turn-5",
"tool": "query_user_history",
"kind": "llm",
"prompt_tokens": 2863,
"cached_tokens": 0,
"completion_tokens": 136,
"tool_ctx_tokens": 1389,
"latency_s": 2.6890718936920166
},
{
"step": "turn-6",
"tool": "issue_refund",
"kind": "llm",
"prompt_tokens": 3123,
"cached_tokens": 0,
"completion_tokens": 160,
"tool_ctx_tokens": 1490,
"latency_s": 3.073489189147949
},
{
"step": "turn-7",
"tool": "send_notification",
"kind": "llm",
"prompt_tokens": 3408,
"cached_tokens": 0,
"completion_tokens": 160,
"tool_ctx_tokens": 1579,
"latency_s": 3.0861823558807373
},
{
"step": "turn-8",
"tool": "close_ticket",
"kind": "llm",
"prompt_tokens": 3668,
"cached_tokens": 0,
"completion_tokens": 160,
"tool_ctx_tokens": 1648,
"latency_s": 2.504854917526245
}
]
},
{
"key": "kv",
"name": "KV 仅缓存(稳定前缀/不压缩)",
"total_cost": 0.0027075,
"component_costs": {
"uncached_input_cost": 0.0010227,
"cached_input_cost": 0.0010176,
"output_cost": 0.0006672,
"uncached_input_tokens": 6818,
"cached_input_tokens": 13568,
"output_tokens": 1112,
"tool_ctx_tokens": 9544
},
"cost_distribution": {
"n": 8,
"mean": 0.0003384375,
"p50": 0.00033465000000000003,
"p95": 0.00040815,
"p99": 0.00040815,
"max": 0.00040815
},
"spans": [
{
"step": "turn-1",
"tool": "query_order",
"kind": "llm",
"prompt_tokens": 1056,
"cached_tokens": 1024,
"completion_tokens": 121,
"tool_ctx_tokens": 276,
"latency_s": 2.1304330825805664
},
{
"step": "turn-2",
"tool": "query_logistics",
"kind": "llm",
"prompt_tokens": 1763,
"cached_tokens": 0,
"completion_tokens": 117,
"tool_ctx_tokens": 829,
"latency_s": 2.102344036102295
},
{
"step": "turn-3",
"tool": "check_refund_policy",
"kind": "llm",
"prompt_tokens": 2130,
"cached_tokens": 1024,
"completion_tokens": 136,
"tool_ctx_tokens": 1046,
"latency_s": 3.537381172180176
},
{
"step": "turn-4",
"tool": "query_knowledge_base",
"kind": "llm",
"prompt_tokens": 2540,
"cached_tokens": 1024,
"completion_tokens": 160,
"tool_ctx_tokens": 1287,
"latency_s": 3.393212080001831
},
{
"step": "turn-5",
"tool": "query_user_history",
"kind": "llm",
"prompt_tokens": 2834,
"cached_tokens": 2176,
"completion_tokens": 114,
"tool_ctx_tokens": 1389,
"latency_s": 1.906968116760254
},
{
"step": "turn-6",
"tool": "issue_refund",
"kind": "llm",
"prompt_tokens": 3081,
"cached_tokens": 2560,
"completion_tokens": 160,
"tool_ctx_tokens": 1490,
"latency_s": 2.7878849506378174
},
{
"step": "turn-7",
"tool": "send_notification",
"kind": "llm",
"prompt_tokens": 3361,
"cached_tokens": 2560,
"completion_tokens": 160,
"tool_ctx_tokens": 1579,
"latency_s": 2.825930118560791
},
{
"step": "turn-8",
"tool": "close_ticket",
"kind": "llm",
"prompt_tokens": 3621,
"cached_tokens": 3200,
"completion_tokens": 144,
"tool_ctx_tokens": 1648,
"latency_s": 2.7222893238067627
}
]
},
{
"key": "compress",
"name": "仅压缩(前缀不稳定/摘要)",
"total_cost": 0.00311475,
"component_costs": {
"uncached_input_cost": 0.00242655,
"cached_input_cost": 0.0,
"output_cost": 0.0006882,
"uncached_input_tokens": 16177,
"cached_input_tokens": 0,
"output_tokens": 1147,
"tool_ctx_tokens": 5248
},
"cost_distribution": {
"n": 8,
"mean": 0.00038934375,
"p50": 0.0004059,
"p95": 0.0004497,
"p99": 0.0004497,
"max": 0.0004497
},
"spans": [
{
"step": "turn-1",
"tool": "query_order",
"kind": "llm",
"prompt_tokens": 1112,
"cached_tokens": 0,
"completion_tokens": 127,
"tool_ctx_tokens": 276,
"latency_s": 2.223604917526245
},
{
"step": "turn-2",
"tool": "query_logistics",
"kind": "llm",
"prompt_tokens": 1829,
"cached_tokens": 0,
"completion_tokens": 91,
"tool_ctx_tokens": 829,
"latency_s": 2.916440963745117
},
{
"step": "turn-3",
"tool": "check_refund_policy",
"kind": "llm",
"prompt_tokens": 2164,
"cached_tokens": 0,
"completion_tokens": 129,
"tool_ctx_tokens": 1046,
"latency_s": 2.543590784072876
},
{
"step": "turn-4",
"tool": "query_knowledge_base",
"kind": "llm",
"prompt_tokens": 2310,
"cached_tokens": 0,
"completion_tokens": 160,
"tool_ctx_tokens": 1052,
"latency_s": 2.999290943145752
},
{
"step": "turn-5",
"tool": "query_user_history",
"kind": "llm",
"prompt_tokens": 2066,
"cached_tokens": 0,
"completion_tokens": 160,
"tool_ctx_tokens": 635,
"latency_s": 5.8334801197052
},
{
"step": "turn-6",
"tool": "issue_refund",
"kind": "llm",
"prompt_tokens": 2146,
"cached_tokens": 0,
"completion_tokens": 160,
"tool_ctx_tokens": 551,
"latency_s": 2.787147045135498
},
{
"step": "turn-7",
"tool": "send_notification",
"kind": "llm",
"prompt_tokens": 2192,
"cached_tokens": 0,
"completion_tokens": 160,
"tool_ctx_tokens": 430,
"latency_s": 3.470608949661255
},
{
"step": "turn-8",
"tool": "close_ticket",
"kind": "llm",
"prompt_tokens": 2358,
"cached_tokens": 0,
"completion_tokens": 160,
"tool_ctx_tokens": 429,
"latency_s": 2.5834107398986816
}
]
},
{
"key": "both",
"name": "B 优化(KV缓存+压缩)",
"total_cost": 0.00264285,
"component_costs": {
"uncached_input_cost": 0.00148365,
"cached_input_cost": 0.0004608,
"output_cost": 0.0006984000000000001,
"uncached_input_tokens": 9891,
"cached_input_tokens": 6144,
"output_tokens": 1164,
"tool_ctx_tokens": 5248
},
"cost_distribution": {
"n": 8,
"mean": 0.00033035625,
"p50": 0.00033525000000000004,
"p95": 0.00044249999999999997,
"p99": 0.00044249999999999997,
"max": 0.00044249999999999997
},
"spans": [
{
"step": "turn-1",
"tool": "query_order",
"kind": "llm",
"prompt_tokens": 1056,
"cached_tokens": 1024,
"completion_tokens": 139,
"tool_ctx_tokens": 276,
"latency_s": 2.405411958694458
},
{
"step": "turn-2",
"tool": "query_logistics",
"kind": "llm",
"prompt_tokens": 1781,
"cached_tokens": 0,
"completion_tokens": 112,
"tool_ctx_tokens": 829,
"latency_s": 2.1826939582824707
},
{
"step": "turn-3",
"tool": "check_refund_policy",
"kind": "llm",
"prompt_tokens": 2143,
"cached_tokens": 1024,
"completion_tokens": 151,
"tool_ctx_tokens": 1046,
"latency_s": 2.5647878646850586
},
{
"step": "turn-4",
"tool": "query_knowledge_base",
"kind": "llm",
"prompt_tokens": 2310,
"cached_tokens": 0,
"completion_tokens": 160,
"tool_ctx_tokens": 1052,
"latency_s": 3.0318493843078613
},
{
"step": "turn-5",
"tool": "query_user_history",
"kind": "llm",
"prompt_tokens": 2060,
"cached_tokens": 1024,
"completion_tokens": 160,
"tool_ctx_tokens": 635,
"latency_s": 2.475983142852783
},
{
"step": "turn-6",
"tool": "issue_refund",
"kind": "llm",
"prompt_tokens": 2143,
"cached_tokens": 1024,
"completion_tokens": 160,
"tool_ctx_tokens": 551,
"latency_s": 2.7105050086975098
},
{
"step": "turn-7",
"tool": "send_notification",
"kind": "llm",
"prompt_tokens": 2188,
"cached_tokens": 1024,
"completion_tokens": 160,
"tool_ctx_tokens": 430,
"latency_s": 2.7884562015533447
},
{
"step": "turn-8",
"tool": "close_ticket",
"kind": "llm",
"prompt_tokens": 2354,
"cached_tokens": 1024,
"completion_tokens": 122,
"tool_ctx_tokens": 429,
"latency_s": 2.3781700134277344
}
]
}
]
}