前两篇里,我们已经让程序能够调用模型,并把认证、网络和坏请求三类失败翻译成自己的错误类型。失败的调用有了分类,成功的调用却仍然只返回一句文本:这次调用用了多少 token、等了多久、花了多少钱,都没有留下记录。
没有这些数据,我们就无法回答:为什么今天变慢了?哪类问题最贵?上下文变长后,成本增长来自输入还是输出?某次优化究竟有没有效果?
后面可以把许多次调用汇总成趋势,但前提是每一次调用先留下数据。所以这一篇仍然不增加 Tool,也不实现 Agent Loop,只让 chat() 在返回答案的同时,返回一张最小"收据"。
文章对应代码仓库 agent-from-zero 的 v0.0.3-cost-latency-tracking。阅读时可以切换到这个 tag,对照 src/agent/llm.py 和 tests/test_llm.py。
只返回文本,哪些信息消失了
上一版 chat() 的出口只有这一行:
return response.choices[0].message.content调用方拿到了最终答案,其他信息却在函数返回时全部丢失。
一次模型调用至少有三组值得保留的信息:
- —结果:模型最终生成了什么;
- —用量:输入与输出 token、缓存命中量,以及思考模式下的 reasoning token;
- —性能:从发起调用到收到完整响应经过了多久。
如果这一层不保存它们,后面的 Agent Loop 即使连续调用模型十次,也只会留下十段文本。未来建设可观测性(Observability)时,自然也没有历史数据可供分析。
解决方法不是让 chat() 多打印几行日志,而是让这些数据成为返回值的一部分。
把一句文本换成一张结构化收据
我们先定义一个 ChatResult:
from dataclasses import dataclass
@dataclass
class ChatResult:
text: str
tokens_in: int
tokens_out: int
cache_hit_tokens: int
cache_miss_tokens: int
latency_ms: float
estimated_cost_usd: floatchat() 的返回值由字符串变成 ChatResult,调用方可以从同一个对象中读取答案和本次调用的各项指标:
result = chat("A100 还有多少库存?")
print(result.text)
print(result.tokens_in)
print(result.latency_ms)
print(result.estimated_cost_usd)token、延迟和成本由此成为调用结果的一部分,不会在函数返回时丢失。这仍是一张"最小收据":它记录了输出 token 总数,却没有单独记录思考模式下的 reasoning token。
`text`:调用真正交付的答案
text 仍然来自 response.choices[0].message.content。原来把 chat() 的返回值直接当字符串使用的代码,需要改为读取 result.text。
`tokens_in` 和 `tokens_out`:使用了多少模型计算量
token 数不由程序自己估算,而是直接读取服务端响应里的 usage。usage 是服务端随响应返回的用量统计:
usage = response.usage
tokens_in = usage.prompt_tokens
tokens_out = usage.completion_tokensprompt_tokens 是本次请求输入给模型的 token 总数,包含 system message 和 user message;completion_tokens 是模型为这次 completion 生成的输出 token 总数,当前代码把它保存为 tokens_out。不同 API 也可能把同一个总量命名为 output_tokens,只是字段名不同。
token 不等于字符数或单词数,具体切分规则由模型决定。这里直接保存服务端报告的用量,不在程序里自行估算。
reasoning token 和最终回答 token 有什么区别
在非思考模式下,API 只返回最终回答。启用思考模式(thinking mode)后,模型会先生成用于分析问题的推理内容,再生成最终回答。可以把它理解成"先在草稿纸上推演,再把结论写到答题纸上":
- —
reasoning_content是推理过程; - —
content是最终交付给用户的回答。
reasoning token 和 output token 不是两个并列的数字。前者是后者的一部分:
completion_tokens / output tokens
├── reasoning tokens
└── 其余输出 token(在简单文本回复里主要对应最终回答)例如,一次调用报告 100 个 completion_tokens,其中 70 个是 reasoning_tokens。即使应用程序只展示最终回答,成本和输出预算仍要按完整的 100 个 token 计算。
DeepSeek 分别提供内容和用量字段:
response.choices[0].message.reasoning_content
response.choices[0].message.content
response.usage.completion_tokens
response.usage.completion_tokens_details.reasoning_tokensreasoning_content 是推理文本,reasoning_tokens 是它消耗的 token 数;一个用于读取内容,一个用于统计用量。非思考模式下,reasoning token 通常是 0,或者响应不提供这项明细。
当前 commit 没有显式启用思考模式,也没有保存 reasoning_tokens。因此,后面的五次真实调用只能比较 completion_tokens 总数,不能事后拆出推理部分。以后使用思考模型时,ChatResult 应该增加这一字段。
缓存命中和未命中:输入 token 不是一个价格
这里的缓存不是"直接保存上一次答案"。上下文缓存会保存模型已经处理过的输入前缀;后续请求如果复用了相同前缀,服务可以少做一部分重复计算,这部分输入就叫缓存命中。没有复用到的部分则是缓存未命中。
DeepSeek 的输入计费区分缓存命中与缓存未命中,响应会进一步提供:
cache_hit_tokens = usage.prompt_cache_hit_tokens or 0
cache_miss_tokens = usage.prompt_cache_miss_tokens or 0使用 or 0 是因为这些字段可能为空;当前结果结构统一用整数表示,所以没有报告命中时按 0 处理。
为什么不只保存 prompt_tokens?因为相同数量的输入 token,缓存命中与未命中的价格不同。把两者过早合并,成本计算就会丢掉必要信息。
用单调时钟测量真正经过的时间
延迟不是 API 响应里直接提供的字段,因此需要在本地测量:
started_at = time.perf_counter()
response = get_client().chat.completions.create(...)
latency_ms = (time.perf_counter() - started_at) * 1000perf_counter() 返回的差值以秒为单位,乘以 1000 后得到毫秒。这里不用 time.time(),因为系统校时可能让墙上时间发生跳变;perf_counter() 是只会持续向前的单调时钟,更适合测量经过时间。
这里测到的是从发出请求到收到完整响应的总耗时,也包含 SDK 内部可能发生的重试和等待。它还没有区分首 token 延迟与后续生成时间:前者表示用户多久能看到第一个输出片段,需要通过流式响应单独测量。本篇先记录完整响应耗时。
成本为什么叫 `estimated_cost`
Chat Completions 响应会告诉我们 token 用量,却不会直接返回"本次消费了多少美元"。成本需要根据价格表自行计算。
思考模式下,reasoning token 也包含在输出总量里,因此成本公式仍使用完整的 completion_tokens,不能先减掉用户看不见的推理部分。
这一版代码保留了一份 DeepSeek Flash 非高峰时段的价格快照。快照表示"某个时间点观察到的价格",不是永久有效的当前价格:
# USD / 1M tokens,非高峰时段价格(2026-09-13 抓取)
PRICE_PER_MILLION_TOKENS_USD = {
"input_cache_hit": 0.003,
"input_cache_miss": 0.15,
"output": 0.6,
}因此计算公式是:
def _estimate_cost_usd(
cache_hit_tokens: int,
cache_miss_tokens: int,
tokens_out: int,
) -> float:
return (
cache_hit_tokens / 1_000_000
* PRICE_PER_MILLION_TOKENS_USD["input_cache_hit"]
+ cache_miss_tokens / 1_000_000
* PRICE_PER_MILLION_TOKENS_USD["input_cache_miss"]
+ tokens_out / 1_000_000
* PRICE_PER_MILLION_TOKENS_USD["output"]
)名字中的 estimated 不能省略。这个数字根据价格快照计算,并不是供应商账单:
- —官方价格可能调整;
- —账号可能有折扣或其他结算规则;
- —调用时段和模型别名都可能改变实际单价。
实验当天,Flash 高峰时段单价是非高峰的两倍,当前 commit 没有根据调用时间切换价格,因此高峰期会低估成本。读者复用代码时,应重新核对价格,而不是把教程中的历史数字当成长期配置。
模型映射也会变化。代码请求的是 deepseek-chat,真实响应的 response.model 当时是 deepseek-flash,所以价格按 Flash 计算;服务商以后可能改变这个别名的实际指向。更完整的记录应该同时保存响应模型和价格表版本,当前 ChatResult 还没有这两个字段。
让 `chat()` 返回完整结果,但保持安静
把前面的数据收集放回 chat():
def chat(
user_message: str,
system: str = DEFAULT_SYSTEM,
temperature: float = 0.0,
) -> ChatResult:
started_at = time.perf_counter()
try:
response = get_client().chat.completions.create(...)
except AuthenticationError as error:
raise AuthenticationFailure(str(error)) from error
except APIConnectionError as error:
raise NetworkFailure(str(error)) from error
except BadRequestError as error:
raise BadRequestFailure(str(error)) from error
latency_ms = (time.perf_counter() - started_at) * 1000
usage = response.usage
cache_hit_tokens = usage.prompt_cache_hit_tokens or 0
cache_miss_tokens = usage.prompt_cache_miss_tokens or 0
tokens_out = usage.completion_tokens
return ChatResult(
text=response.choices[0].message.content,
tokens_in=usage.prompt_tokens,
tokens_out=tokens_out,
cache_hit_tokens=cache_hit_tokens,
cache_miss_tokens=cache_miss_tokens,
latency_ms=latency_ms,
estimated_cost_usd=_estimate_cost_usd(
cache_hit_tokens,
cache_miss_tokens,
tokens_out,
),
)chat() 只负责测量并返回数据,不负责打印、存储或上报。命令行入口可以自行显示一行摘要:
result = chat(question)
print(result.text)
print(
f"[chat] tokens_in={result.tokens_in} "
f"tokens_out={result.tokens_out} "
f"latency_ms={result.latency_ms:.0f} "
f"cost_usd={result.estimated_cost_usd:.6f}"
)把"测量"和"展示"分开,可以避免未来 Agent Loop 每调用一次模型,底层函数就擅自写入一行日志。
连续跑五次,直觉哪里错了
代码能返回指标,不代表我们已经理解了指标。我们准备了五个长度和要求不同的问题,对真实的 deepseek-chat 连续调用,并记录结果。
Q1 12 字:询问 A100 库存
Q2 45 字:比较两款传感器
Q3 135 字:设计完整报价流程
Q4 2 字:你好
Q5 97 字:详细解释 Chat Completion API,要求不少于 300 字真实结果(2026-09-13)如下:
| 问题 | 输入 token | 输出 token | 延迟 | 估算成本 |
|---|---|---|---|---|
| Q1 | 24 | 130 | 2312 ms | $0.000082 |
| Q2 | 41 | 74 | 949 ms | $0.000051 |
| Q3 | 91 | 1610 | 8820 ms | $0.000980 |
| Q4 | 18 | 34 | 1122 ms | $0.000023 |
| Q5 | 62 | 968 | 5842 ms | $0.000590 |
表格中的"输出 token"是服务端报告的完整 completion_tokens。输入 token 大致随问题长度增长,但延迟和成本没有表现出同样的顺序:Q3 只有 91 个输入 token,却生成了 1610 个输出 token,耗时 8820 ms,也是五次里最贵的一次;要求长回答的 Q5 也出现了相同方向的变化。
这组数据支持一个比"问题越长越贵"更准确的观察:
在这五次实验中,模型生成的输出量比输入问题的长度更能解释成本和延迟差异。
五个样本不能证明普遍的因果规律,网络负载和模型服务状态也会影响延迟。但这个结果已经给出下一步实验方向:限制输出长度,可能比只压缩用户问题更直接。
为什么五次缓存命中都是零
五次请求共享同一条 DEFAULT_SYSTEM,但 cache_hit_tokens 全部为 0。这不等于"缓存没有生效":DeepSeek 的缓存要求后续请求复用已保存的输入前缀,而且采用 best-effort(尽力而为)策略;重复一条很短的 system message,并不能保证命中。
本篇只记录这个现象。等课程进入上下文管理阶段,再用更长、更明确的重复前缀设计专门实验。
用测试替身验证数据流,而不是锁死模型表现
运行与本阶段对应的测试:
pytest -v tests/test_llm.py当前结果:
tests/test_llm.py::test_chat_returns_the_model_text PASSED
tests/test_llm.py::test_chat_sends_system_and_user_messages PASSED
tests/test_llm.py::test_chat_accepts_custom_system_and_temperature PASSED
tests/test_llm.py::test_get_client_reads_api_key_from_env PASSED
tests/test_llm.py::TestChatResult::test_token_counts_come_from_response_usage PASSED
tests/test_llm.py::TestChatResult::test_cache_hit_and_miss_tokens_are_reported_separately PASSED
tests/test_llm.py::TestChatResult::test_latency_is_measured_and_non_negative PASSED
tests/test_llm.py::TestChatResult::test_estimated_cost_matches_the_pricing_table PASSED
tests/test_llm.py::TestFailureClassification::test_authentication_error_is_reclassified PASSED
tests/test_llm.py::TestFailureClassification::test_connection_error_is_reclassified_as_network_failure PASSED
tests/test_llm.py::TestFailureClassification::test_timeout_error_is_also_reclassified_as_network_failure PASSED
tests/test_llm.py::TestFailureClassification::test_bad_request_error_is_reclassified PASSED
tests/test_llm.py::TestFailureClassification::test_all_three_failures_share_a_common_base_class PASSED
13 passed in 0.45s新增的四个测试只验证我们控制得了的部分:
- —token 数是否忠实读取
response.usage; - —cache hit 与 cache miss 是否分开保存;
- —延迟是否为真实测量出的非负数;
- —成本是否严格遵守当前价格表和计算公式。
测试没有断言"一条长问题必须比短问题慢",也没有锁死某次真实调用的输出 token。模型如何回答、服务端当时有多忙,都不是这段代码能控制的行为。
留给你的四个实验
- 1.给 Q3 的 system message 增加"回答控制在 100 字以内",重新记录
tokens_out、latency_ms和成本,比较限制输出前后的差异。 - 2.连续发送完全相同、且包含较长共同前缀的问题,观察
cache_hit_tokens是否出现非零值;不要只用一句很短的 system message 推断缓存机制。 - 3.为
_estimate_cost_usd()增加调用时间参数,尝试支持高峰与非高峰价格。思考时区转换应该发生在计算函数内部,还是调用它之前。 - 4.使用支持思考模式的模型处理一个简单问题和一个复杂问题,同时记录
completion_tokens与reasoning_tokens。观察推理强度变化时,最终回答长度、总输出 token、延迟和成本是否一起变化。
回到问题:先有单次数据,才可能有可观测性
这一篇最重要的变化,不是多算了一个小数,而是改变了模型调用的输出契约:
一次模型调用不仅产生答案,也产生用量、延迟和成本证据。
ChatResult 只记录一次调用,还不是完整的追踪系统。它没有 trace ID、模型版本、首 token 延迟和 reasoning token 明细,也没有长期保存数据;但未来汇总指标所需的最小原材料,已经开始稳定产生。
至此,Phase 0 的三块地基已经搭好:程序能调用模型,能区分基础失败,也能记录每次调用付出的代价。
下一篇,我们会第一次离开"只会说话"的模型调用:先把本地库存和一份 Tool Schema 放到程序里,让模型拥有接触真实业务数据的可能。