Day 53 / 共 60 天 · 阶段8 LLM 与集成

token / 成本 / usage：算清每一次调用花了多少

Agent 一跑起来就在烧钱——每次 LLM 调用都消耗 token，token 就是钱。你 kickoff() 完想知道"这次跑了多少 token、缓存命中多少、一共几次请求"，靠的就是今天这套 usage 追踪。我们会看：统一的 UsageMetrics 数据模型、from_provider_dict 怎么把 OpenAI/Gemini/Anthropic 各不相同的字段名归一化、_track_token_usage_internal 怎么在每个 LLM 实例里累加、TokenProcess / TokenCalcHandler 两个累加器、以及 crew 级 calculate_usage_metrics 怎么把全队消耗汇总成一张总账。

📍 你在 60 天里的位置（阶段8 LLM 与集成 · 共 6 天）

D49 LLM 抽象→ D50 provider 适配→ D51 函数调用/结构化→ D52 上下文窗口→ D53 token/成本→ D54 遥测→ 阶段9 进阶

💡 先用一个类比兜住今天 usage 追踪就像手机话费账单。每打一个电话（每次 LLM 调用），运营商记一笔"时长/流量"。月底给你一张总账：打了多少分钟、用了多少流量、其中免费时段多少（缓存 token）。但麻烦在于——你可能同时用了移动、联通、电信（OpenAI/Gemini/Anthropic），每家账单格式还不一样。CrewAI 的活儿就是当那个把多家账单翻译成统一格式、再合并成一张总账的记账员。今天看它怎么记、怎么归一、怎么汇总。

L01

痛点：一次 kickoff 到底花了多少钱？

🤔 痛点你跑一个 3 个 Agent、5 个任务的 crew，中间 LLM 被调用了几十次。跑完你想知道：一共用了多少 token？花了多少钱？有多少 token 命中了缓存（省了钱）？可这些数据分散在每次调用的响应里，而且 OpenAI 叫 prompt_tokens、Gemini 叫 prompt_token_count、Anthropic 叫 input_tokens——名字都不一样。怎么把这些零散、异构的数据收集起来、算成一张能看的总账？

💡 一句话本质 CrewAI 定义一个统一的 UsageMetrics 模型作为"记账单位"，用 from_provider_dict 把各家五花八门的字段名归一化成它，然后靠 add_usage_metrics 一路累加：单次调用 → 累进 LLM 实例的 _token_usage → 最后 crew 级 calculate_usage_metrics 遍历所有 Agent 把各自的 usage 求和成总账。核心是"统一模型 + 归一化入口 + 逐级累加"三件套。

大白话先规定一张标准表格（有 prompt/completion/cached 等固定格子），不管哪家 API 回来的数据，都先翻译填进这张标准表；然后每次调用填一张、往实例的"小计"上加；最后 crew 把所有 Agent 的小计加成"总计"。你 kickoff() 后拿到的 crew.usage_metrics 就是这张总计表。

L02

UsageMetrics：统一的记账单位

一切的核心数据结构（types/usage_metrics.py:32）：

# types/usage_metrics.py:32
class UsageMetrics(BaseModel):
    """Track usage metrics for crew execution."""
    total_tokens: int = Field(default=0, description="Total number of tokens used.")
    prompt_tokens: int = Field(default=0, ...)           # 输入 token
    cached_prompt_tokens: int = Field(default=0, ...)    # 命中缓存的输入 token（省钱）
    completion_tokens: int = Field(default=0, ...)       # 输出 token
    reasoning_tokens: int = Field(default=0,             # o 系列 / Gemini 思考 token
        description="Number of reasoning/thinking tokens ...")
    cache_creation_tokens: int = Field(default=0,        # Anthropic 缓存写入 token
        description="Number of cache creation tokens (e.g. Anthropic cache writes).")
    successful_requests: int = Field(default=0, ...)     # 成功请求次数

    def add_usage_metrics(self, usage_metrics: Self) -> None:
        """Add usage metrics from another UsageMetrics object."""
        self.total_tokens += usage_metrics.total_tokens
        self.prompt_tokens += usage_metrics.prompt_tokens
        self.cached_prompt_tokens += usage_metrics.cached_prompt_tokens
        self.completion_tokens += usage_metrics.completion_tokens
        self.reasoning_tokens += usage_metrics.reasoning_tokens
        self.cache_creation_tokens += usage_metrics.cache_creation_tokens
        self.successful_requests += usage_metrics.successful_requests

全部 default=0每个字段默认 0，所以 UsageMetrics() 就是一张"空账单"，可作为累加的起点。

prompt vs completion输入 token 和输出 token 分开记——因为定价不同（输出通常更贵），分开才能准确算钱。

cached_prompt_tokens命中缓存的输入 token：这部分很便宜甚至免费，单独记才能体现省了多少。

reasoning / cache_creation为新型模型补的字段——o 系列的"思考"、Anthropic 的"缓存写入"都单独消耗，不记就漏账。

add_usage_metrics★累加的核心方法：把另一张账单的每个字段加到自己身上。整套汇总机制全靠它。

💡 为什么用一个 add_usage_metrics 方法而不是到处手写加法？因为累加发生在多个层级（单次→实例→Agent→crew）。如果每处都手写 a.prompt += b.prompt; a.completion += b.completion; ...，一旦新增字段（比如后来加的 reasoning_tokens），就得改一堆地方、极易漏。把"怎么加两张账单"收敛进这一个方法，新增字段只改这一处，所有调用点自动生效。这是 DRY（不重复自己）原则在数据聚合上的直接体现。

图注：七个整数字段，全默认 0；两张 UsageMetrics 相加即逐格相加，累加机制的最小单位。

L03

from_provider_dict：抹平各家字段名差异

最能体现"记账员翻译能力"的方法（types/usage_metrics.py:80）：

# types/usage_metrics.py:80
@classmethod
def from_provider_dict(cls, usage_data: dict | None) -> Self | None:
    """Normalize a provider's raw usage dict into a UsageMetrics.
    Accepts the full set of key aliases CrewAI providers emit:
    prompt_tokens / prompt_token_count (Gemini) / input_tokens (Anthropic)..."""
    if not usage_data:
        return None                                    # ★空输入返回 None，让调用方决定怎么办
    prompt_tokens = _first_int(usage_data,
        "prompt_tokens", "prompt_token_count", "input_tokens")   # ★三个别名任取其一
    completion_tokens = _first_int(usage_data,
        "completion_tokens", "candidates_token_count", "output_tokens")
    cached_prompt_tokens = _first_int(usage_data,
        "cached_tokens", "cached_prompt_tokens", "cache_read_input_tokens")
    if not cached_prompt_tokens:
        details = usage_data.get("prompt_tokens_details")        # OpenAI 把缓存嵌在子对象里
        if isinstance(details, dict):
            cached_prompt_tokens = _coerce_int(details.get("cached_tokens"))
    return cls(
        total_tokens=prompt_tokens + completion_tokens,
        prompt_tokens=prompt_tokens, completion_tokens=completion_tokens,
        cached_prompt_tokens=cached_prompt_tokens,
        reasoning_tokens=_coerce_int(usage_data.get("reasoning_tokens")),
        cache_creation_tokens=_coerce_int(usage_data.get("cache_creation_tokens")),
        successful_requests=1)                          # 每次归一化 = 一次成功请求

_first_int(三个别名)★核心：OpenAI 用 prompt_tokens、Gemini 用 prompt_token_count、Anthropic 用 input_tokens——按顺序试，哪个有值用哪个。一行抹平三家差异。

completion 同理输出 token 也有三个别名（candidates_token_count 是 Gemini 的说法），统一处理。

缓存 token 的嵌套兜底OpenAI 把缓存数藏在 prompt_tokens_details.cached_tokens 子对象里，顶层找不到就往里挖一层。

total = prompt + completion总数自己算，不信任 provider 传的 total（有的家不给或算法不一）。

successful_requests=1归一化成功 = 记一次成功请求。累加时这些 1 相加就是总请求数。

return None on empty★边界：没有 usage 数据就返回 None，把"这次要不要记账"的决定权交给调用方，不硬塞一张全 0 的账单。

💡 设计取舍①：为什么把归一化做成"多别名兜底"而不是每家一个解析函数？ 朴素做法：parse_openai_usage() / parse_gemini_usage() / parse_anthropic_usage() 各写一个，调用前先判断是哪家。问题：得先知道是哪家（provider 信息不一定跟着 usage 一起来），且新增一家就多一个函数 + 一处分支。源码做法：一个 from_provider_dict，靠"字段别名列表"吃下所有家——因为不同家的字段名几乎不冲突（OpenAI 不会有 input_tokens，Anthropic 不会有 prompt_tokens），按别名顺序试就能自动认出。用"数据形状的差异"替代"显式的类型判断"，加一家新 provider 往往只需在别名列表里加个名字。代价是万一两家真用了同名不同义的字段会出错，但实践中没发生。

L04

LLM 内部累加：每个实例记自己的小账

每个 LLM 实例把每次调用的 usage 累进自己的 _token_usage（llms/base_llm.py:948）：

# llms/base_llm.py:948
def _track_token_usage_internal(self, usage_data: dict[str, Any]) -> None:
    """Track token usage internally in the LLM instance."""
    metrics = UsageMetrics.from_provider_dict(usage_data)   # ① 先归一化（L03）
    if metrics is None:
        return                                              # 没数据就不记
    self._token_usage["prompt_tokens"]      += metrics.prompt_tokens      # ② 逐字段累加
    self._token_usage["completion_tokens"]  += metrics.completion_tokens
    self._token_usage["total_tokens"]       += metrics.total_tokens
    self._token_usage["successful_requests"]+= metrics.successful_requests
    self._token_usage["cached_prompt_tokens"]  += metrics.cached_prompt_tokens
    self._token_usage["reasoning_tokens"]       += metrics.reasoning_tokens
    self._token_usage["cache_creation_tokens"]  += metrics.cache_creation_tokens

def get_token_usage_summary(self) -> UsageMetrics:          # :966
    """Get summary of token usage for this LLM instance."""
    return UsageMetrics(**self._token_usage)                # ③ 打包成 UsageMetrics 交出去

先 from_provider_dict无论哪家 API 的原始 usage，先过 L03 归一化——保证累加的都是标准字段。

metrics is None → return没解析出东西就跳过，不污染累计值。

self._token_usage[...] +=★这个实例活着期间的累计小账。同一个 LLM 被调 10 次，就累加 10 次。

get_token_usage_summary对外交账的口子：把累计的 dict 包成一个 UsageMetrics 对象返回，供上层（crew）汇总。

还记得 Day 49 说 provider 版 call 里都会调 self._emit_call_completed_event 吗？token 追踪就是在完成事件前后触发 _track_token_usage_internal 的——每次调用成功，实例小账 +1 笔。这是"共性行为放基类"的又一例：所有 provider 共用同一套 token 记账逻辑。

L05

TokenProcess：给非 BaseLLM 路径用的累加器

还有一个更轻的累加器，服务 litellm 回调路径（agents/agent_builder/utilities/base_token_process.py）：

# base_token_process.py
class TokenProcess(BaseModel):
    """Track token usage during agent processing."""
    total_tokens: int = Field(default=0)
    prompt_tokens: int = Field(default=0)
    cached_prompt_tokens: int = Field(default=0)
    completion_tokens: int = Field(default=0)
    successful_requests: int = Field(default=0)

    def sum_prompt_tokens(self, tokens: int) -> None:
        self.prompt_tokens += tokens
        self.total_tokens += tokens              # ★加 prompt 时顺手加 total
    def sum_completion_tokens(self, tokens: int) -> None:
        self.completion_tokens += tokens
        self.total_tokens += tokens
    def sum_cached_prompt_tokens(self, tokens: int) -> None:
        self.cached_prompt_tokens += tokens      # 缓存不计入 total（避免重复）
    def sum_successful_requests(self, requests: int) -> None:
        self.successful_requests += requests
    def get_summary(self) -> UsageMetrics:
        return UsageMetrics(total_tokens=self.total_tokens, prompt_tokens=self.prompt_tokens,
            cached_prompt_tokens=self.cached_prompt_tokens,
            completion_tokens=self.completion_tokens, successful_requests=self.successful_requests)

sum_prompt_tokens 同时 += total加输入 token 时顺手把 total 也加上——省得最后再单独算总数。

cached 不进 total★注意：缓存 token 单独记，不累进 total。因为 cached 已经是 prompt 的一部分，重复加会翻倍。

get_summary → UsageMetrics和 BaseLLM.get_token_usage_summary 一样，最终都吐出统一的 UsageMetrics——两条记账路殊途同归。

💡 设计取舍②：为什么有 BaseLLM._token_usage 和 TokenProcess 两套记账？因为 CrewAI 有两条 LLM 调用路径（Day 49）：原生 provider 路径直接在 BaseLLM 实例里用 _track_token_usage_internal 累加；litellm 兜底路径则通过 litellm 的回调机制（L06 的 TokenCalcHandler）把数据喂给 TokenProcess。两条路的数据来源、时机不同，所以各有累加器。但关键是——它们最后都 get_summary()/get_token_usage_summary() 成同一个 UsageMetrics，汇总时（L07）用同一个 add_usage_metrics 无差别相加。入口分两套是被现实逼的，出口收敛成一套是设计的克制——差异只允许存在于必要处，能统一的地方绝不放任分裂。

L06

TokenCalcHandler：litellm 的回调钩子

litellm 路径靠这个回调把 usage 喂进 TokenProcess（utilities/token_counter_callback.py）：

# token_counter_callback.py
class TokenCalcHandler(BaseModel):
    token_cost_process: TokenProcess | None = Field(default=None)

    def log_success_event(self, kwargs, response_obj, start_time, end_time) -> None:
        """Log successful LLM API call and track token usage."""
        if self.token_cost_process is None:
            return
        with suppress_warnings():
            if isinstance(response_obj, dict) and "usage" in response_obj:
                usage = response_obj["usage"]
                if usage:
                    self.token_cost_process.sum_successful_requests(1)
                    if hasattr(usage, "prompt_tokens"):
                        self.token_cost_process.sum_prompt_tokens(usage.prompt_tokens)
                    if hasattr(usage, "completion_tokens"):
                        self.token_cost_process.sum_completion_tokens(usage.completion_tokens)
                    if (hasattr(usage, "prompt_tokens_details")
                            and usage.prompt_tokens_details
                            and usage.prompt_tokens_details.cached_tokens):
                        self.token_cost_process.sum_cached_prompt_tokens(
                            usage.prompt_tokens_details.cached_tokens)

log_success_event★litellm 约定的回调名：每次 API 调用成功，litellm 自动回调它，把响应对象递进来。你不用手动调。

持有 TokenProcess回调里拿到 usage，就调 token_cost_process.sum_* 累进那个累加器（L05）。

层层 hasattr 防御不同模型返回的 usage 结构不一，先 hasattr 确认字段存在再取——避免 AttributeError。

suppress_warnings()litellm 内部有时会打一堆警告，这里静默掉，保持日志干净。

大白话litellm 帮你调完模型，会主动"喊你一声"（回调 log_success_event），把账单塞给你。TokenCalcHandler 就是接这个电话的人：接到就把 token 数记进小本本（TokenProcess）。你只要在调用时把这个 handler 挂上去，剩下的自动完成。

L07

crew 级汇总：全队消耗合成一张总账

最后一步，crew 把所有 Agent 的小账加成总账（crew.py:2114）：

# crew.py:2114
def calculate_usage_metrics(self) -> UsageMetrics:
    """Calculates and returns the usage metrics."""
    total_usage_metrics = UsageMetrics()               # ① 一张空总账
    for agent in self.agents:
        if isinstance(agent.llm, BaseLLM):             # ★Day 49 的伏笔：判 BaseLLM 不判 LLM
            llm_usage = agent.llm.get_token_usage_summary()    # 原生路径的小账
            total_usage_metrics.add_usage_metrics(llm_usage)   # ② 累加
        else:
            if hasattr(agent, "_token_process"):
                token_sum = agent._token_process.get_summary() # litellm 路径的小账
                total_usage_metrics.add_usage_metrics(token_sum)
    if self.manager_agent and hasattr(self.manager_agent, "_token_process"):
        token_sum = self.manager_agent._token_process.get_summary()   # ③ 别忘了 manager
        total_usage_metrics.add_usage_metrics(token_sum)
    if self.manager_agent:
        if isinstance(self.manager_agent.llm, BaseLLM):
            total_usage_metrics.add_usage_metrics(self.manager_agent.llm.get_token_usage_summary())
    self.usage_metrics = total_usage_metrics           # ④ 存到 crew，供你读
    return total_usage_metrics

UsageMetrics() 空总账从全 0 开始，作为累加的初始值。

isinstance(agent.llm, BaseLLM)★呼应 Day 49 边界：因为工厂可能返回 AnthropicCompletion 等，判断要盯基类 BaseLLM，命中就走原生路径的小账。

else → _token_process不是 BaseLLM 的（litellm 路径），从 Agent 的 _token_process 取小账。两条路都收。

add_usage_metrics 反复用★L02 那个方法在这大放异彩：每个 Agent 的账单一路 add 进总账。所有累加最终收敛到这一个方法。

manager_agent 单独处理★边界：层级流程（Day 21）的 manager 也在烧 token，但它不在 self.agents 列表里，得单独加，否则漏账。

self.usage_metrics = ...结果存进 crew 字段。你 kickoff() 后 print(crew.usage_metrics) 读的就是它。

图注：两条记账路（原生实例 / litellm 回调）+ manager，最终都汇入 crew.usage_metrics。

L08

边界 + 今日小结

⚠️ 边界：usage 数据不一定"准"，别拿它当财务对账这套追踪有几个天生的不确定点：①不是每家 provider 都返回 usage——from_provider_dict 收到空就返回 None，那次调用的 token 就没记进去（漏账）；②缓存/思考 token 的字段各家支持程度不一，老模型可能压根不给 cached_tokens，这部分省钱信息就缺失；③流式调用的 usage 有时要等最后一个 chunk 才有，中途断开就丢。所以 crew.usage_metrics 适合用来大致估算、监控趋势、优化 prompt，但不该拿它跟 provider 的账单精确对账——真要对账以 provider 后台为准。教训：任何"从响应里顺手收集的指标"都带估算性质，用它做决策要留心它的盲区。

👶 小白：我 kickoff 完怎么看这次花了多少 token？

👨‍🏫 老师：result = crew.kickoff(); print(crew.usage_metrics) 就能看到一张 UsageMetrics：total_tokens / prompt_tokens / completion_tokens / cached_prompt_tokens / successful_requests 等。想估算钱，用 prompt_tokens 和 completion_tokens 分别乘对应模型的单价（输入输出价不同），再减去 cached_prompt_tokens 省下的部分。CrewAI 记的是 token，换算成钱要你自己按模型定价来。

                    🧠 今天你应该能回答
                    UsageMetrics 有哪些字段？为什么 prompt/completion/cached 要分开？
from_provider_dict 怎么用"别名列表"抹平各家字段名差异？
为什么有 BaseLLM._token_usage 和 TokenProcess 两套累加器？
TokenCalcHandler.log_success_event 是谁在什么时候调的？
calculate_usage_metrics 为什么要单独处理 manager_agent？
为什么 usage 数据不能拿来精确对账？

                

✋ 10 分钟动手

P=lib/crewai/src/crewai
sed -n '32,130p'  $P/types/usage_metrics.py            # UsageMetrics + from_provider_dict
sed -n '948,975p' $P/llms/base_llm.py                  # 实例内累加
cat $P/agents/agent_builder/utilities/base_token_process.py   # TokenProcess
sed -n '2114,2138p' $P/crew.py                         # crew 级汇总
# 亲眼看一次 kickoff 的 token 账单
python -c "
from crewai import Agent, Task, Crew
a=Agent(role='助手', goal='答题', backstory='你博学')
t=Task(description='用一句话解释什么是 token', expected_output='一句话', agent=a)
c=Crew(agents=[a], tasks=[t]); c.kickoff(); print(c.usage_metrics)
"

明日预告 · Day 54：token 是"花了多少"，那"框架内部到底发生了什么"呢？明天讲阶段8 收官——telemetry 遥测：Telemetry 单例怎么初始化 OpenTelemetry、怎么用环境变量一键关闭、_safe_telemetry_operation 怎么保证"遥测挂了也不影响你跑"、以及它到底收集/不收集什么（隐私边界）。

← Day 52 上下文窗口 Day 54 · telemetry 遥测 →