技术·Agent From Zero:从一次模型调用到可靠 Agent · 第 4 篇·2026-09-15·18 分钟

第三篇:一次回答到底花了多少时间和钱

前两篇里,我们已经让程序能够调用模型,并把认证、网络和坏请求三类失败翻译成自己的错误类型。失败的调用有了分类,成功的调用却仍然只返回一句文本:这次调用用了多少 token、等了多久、花了多少钱,都没有留下记录。

没有这些数据,我们就无法回答:为什么今天变慢了?哪类问题最贵?上下文变长后,成本增长来自输入还是输出?某次优化究竟有没有效果?

后面可以把许多次调用汇总成趋势,但前提是每一次调用先留下数据。所以这一篇仍然不增加 Tool,也不实现 Agent Loop,只让 chat() 在返回答案的同时,返回一张最小"收据"。

文章对应代码仓库 agent-from-zero 的 v0.0.3-cost-latency-tracking。阅读时可以切换到这个 tag,对照 src/agent/llm.py 和 tests/test_llm.py。

只返回文本,哪些信息消失了

上一版 chat() 的出口只有这一行:

python
return response.choices[0].message.content

调用方拿到了最终答案,其他信息却在函数返回时全部丢失。

一次模型调用至少有三组值得保留的信息:

  • —结果:模型最终生成了什么;
  • —用量:输入与输出 token、缓存命中量,以及思考模式下的 reasoning token;
  • —性能:从发起调用到收到完整响应经过了多久。

如果这一层不保存它们,后面的 Agent Loop 即使连续调用模型十次,也只会留下十段文本。未来建设可观测性(Observability)时,自然也没有历史数据可供分析。

解决方法不是让 chat() 多打印几行日志,而是让这些数据成为返回值的一部分。

ChatResult 如何成为后续可观测性的起点
ChatResult 如何成为后续可观测性的起点

把一句文本换成一张结构化收据

我们先定义一个 ChatResult:

python
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: float

chat() 的返回值由字符串变成 ChatResult,调用方可以从同一个对象中读取答案和本次调用的各项指标:

python
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 是服务端随响应返回的用量统计:

python
usage = response.usage

tokens_in = usage.prompt_tokens
tokens_out = usage.completion_tokens

prompt_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 不是两个并列的数字。前者是后者的一部分:

text
completion_tokens / output tokens
├── reasoning tokens
└── 其余输出 token(在简单文本回复里主要对应最终回答)

例如,一次调用报告 100 个 completion_tokens,其中 70 个是 reasoning_tokens。即使应用程序只展示最终回答,成本和输出预算仍要按完整的 100 个 token 计算。

DeepSeek 分别提供内容和用量字段:

text
response.choices[0].message.reasoning_content
response.choices[0].message.content

response.usage.completion_tokens
response.usage.completion_tokens_details.reasoning_tokens

reasoning_content 是推理文本,reasoning_tokens 是它消耗的 token 数;一个用于读取内容,一个用于统计用量。非思考模式下,reasoning token 通常是 0,或者响应不提供这项明细。

当前 commit 没有显式启用思考模式,也没有保存 reasoning_tokens。因此,后面的五次真实调用只能比较 completion_tokens 总数,不能事后拆出推理部分。以后使用思考模型时,ChatResult 应该增加这一字段。

缓存命中和未命中:输入 token 不是一个价格

这里的缓存不是"直接保存上一次答案"。上下文缓存会保存模型已经处理过的输入前缀;后续请求如果复用了相同前缀,服务可以少做一部分重复计算,这部分输入就叫缓存命中。没有复用到的部分则是缓存未命中。

DeepSeek 的输入计费区分缓存命中与缓存未命中,响应会进一步提供:

python
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 响应里直接提供的字段,因此需要在本地测量:

python
started_at = time.perf_counter()

response = get_client().chat.completions.create(...)

latency_ms = (time.perf_counter() - started_at) * 1000

perf_counter() 返回的差值以秒为单位,乘以 1000 后得到毫秒。这里不用 time.time(),因为系统校时可能让墙上时间发生跳变;perf_counter() 是只会持续向前的单调时钟,更适合测量经过时间。

这里测到的是从发出请求到收到完整响应的总耗时,也包含 SDK 内部可能发生的重试和等待。它还没有区分首 token 延迟与后续生成时间:前者表示用户多久能看到第一个输出片段,需要通过流式响应单独测量。本篇先记录完整响应耗时。

成本为什么叫 `estimated_cost`

Chat Completions 响应会告诉我们 token 用量,却不会直接返回"本次消费了多少美元"。成本需要根据价格表自行计算。

思考模式下,reasoning token 也包含在输出总量里,因此成本公式仍使用完整的 completion_tokens,不能先减掉用户看不见的推理部分。

这一版代码保留了一份 DeepSeek Flash 非高峰时段的价格快照。快照表示"某个时间点观察到的价格",不是永久有效的当前价格:

python
# USD / 1M tokens,非高峰时段价格(2026-09-13 抓取)
PRICE_PER_MILLION_TOKENS_USD = {
    "input_cache_hit": 0.003,
    "input_cache_miss": 0.15,
    "output": 0.6,
}

因此计算公式是:

python
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():

python
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() 只负责测量并返回数据,不负责打印、存储或上报。命令行入口可以自行显示一行摘要:

python
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 连续调用,并记录结果。

text
Q1  12 字:询问 A100 库存
Q2  45 字:比较两款传感器
Q3 135 字:设计完整报价流程
Q4   2 字:你好
Q5  97 字:详细解释 Chat Completion API,要求不少于 300 字

真实结果(2026-09-13)如下:

问题输入 token输出 token延迟估算成本
Q1241302312 ms$0.000082
Q24174949 ms$0.000051
Q39116108820 ms$0.000980
Q418341122 ms$0.000023
Q5629685842 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,并不能保证命中。

本篇只记录这个现象。等课程进入上下文管理阶段,再用更长、更明确的重复前缀设计专门实验。

用测试替身验证数据流,而不是锁死模型表现

运行与本阶段对应的测试:

bash
pytest -v tests/test_llm.py

当前结果:

text
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. 1.给 Q3 的 system message 增加"回答控制在 100 字以内",重新记录 tokens_out、latency_ms 和成本,比较限制输出前后的差异。
  2. 2.连续发送完全相同、且包含较长共同前缀的问题,观察 cache_hit_tokens 是否出现非零值;不要只用一句很短的 system message 推断缓存机制。
  3. 3.为 _estimate_cost_usd() 增加调用时间参数,尝试支持高峰与非高峰价格。思考时区转换应该发生在计算函数内部,还是调用它之前。
  4. 4.使用支持思考模式的模型处理一个简单问题和一个复杂问题,同时记录 completion_tokens 与 reasoning_tokens。观察推理强度变化时,最终回答长度、总输出 token、延迟和成本是否一起变化。

回到问题:先有单次数据,才可能有可观测性

这一篇最重要的变化,不是多算了一个小数,而是改变了模型调用的输出契约:

一次模型调用不仅产生答案,也产生用量、延迟和成本证据。

ChatResult 只记录一次调用,还不是完整的追踪系统。它没有 trace ID、模型版本、首 token 延迟和 reasoning token 明细,也没有长期保存数据;但未来汇总指标所需的最小原材料,已经开始稳定产生。

至此,Phase 0 的三块地基已经搭好:程序能调用模型,能区分基础失败,也能记录每次调用付出的代价。

下一篇,我们会第一次离开"只会说话"的模型调用:先把本地库存和一份 Tool Schema 放到程序里,让模型拥有接触真实业务数据的可能。

参考资料

《Agent From Zero:从一次模型调用到可靠 Agent》合集 · 第 4 / 8 篇