From c9fa1d741b85efdddff357447c6efb088f24eae9 Mon Sep 17 00:00:00 2001 From: Parthiban Sivakumar Date: Fri, 4 Sep 2026 10:47:01 +0530 Subject: [PATCH 1/3] fix(crew): count each LLM instance once when summing usage metrics calculate_usage_metrics() added agent.llm.get_token_usage_summary() once per agent. Those counters are cumulative for the lifetime of the LLM instance and, as get_token_usage_summary documents, include calls issued by other agents sharing it. Agents sharing a single LLM therefore had that instance's totals added repeatedly, and crew.usage_metrics reported N times the real usage for N agents. Track contributing instances by object identity and skip repeats, with the same check on the manager agent's LLM. Identity rather than model name matters because separate LLM instances of the same model must still sum. Fixes #7259 Co-Authored-By: Claude Opus 5 --- lib/crewai/src/crewai/crew.py | 14 ++++++++- lib/crewai/tests/test_crew.py | 54 +++++++++++++++++++++++++++++++++++ 2 files changed, 67 insertions(+), 1 deletion(-) diff --git a/lib/crewai/src/crewai/crew.py b/lib/crewai/src/crewai/crew.py index 0f77b2d224..a77575f9a4 100644 --- a/lib/crewai/src/crewai/crew.py +++ b/lib/crewai/src/crewai/crew.py @@ -2205,8 +2205,16 @@ def calculate_usage_metrics(self) -> UsageMetrics: """Calculates and returns the usage metrics.""" total_usage_metrics = UsageMetrics() + # An LLM instance's counters are cumulative for its lifetime and + # include calls made by every agent sharing it, so each distinct + # instance must contribute to the total exactly once. + counted_llms: set[int] = set() + for agent in self.agents: if isinstance(agent.llm, BaseLLM): + if id(agent.llm) in counted_llms: + continue + counted_llms.add(id(agent.llm)) llm_usage = agent.llm.get_token_usage_summary() total_usage_metrics.add_usage_metrics(llm_usage) @@ -2220,7 +2228,11 @@ def calculate_usage_metrics(self) -> UsageMetrics: total_usage_metrics.add_usage_metrics(token_sum) if self.manager_agent: - if isinstance(self.manager_agent.llm, BaseLLM): + if ( + isinstance(self.manager_agent.llm, BaseLLM) + and id(self.manager_agent.llm) not in counted_llms + ): + counted_llms.add(id(self.manager_agent.llm)) llm_usage = self.manager_agent.llm.get_token_usage_summary() total_usage_metrics.add_usage_metrics(llm_usage) diff --git a/lib/crewai/tests/test_crew.py b/lib/crewai/tests/test_crew.py index 0195112cb9..c13b41d8a6 100644 --- a/lib/crewai/tests/test_crew.py +++ b/lib/crewai/tests/test_crew.py @@ -4989,3 +4989,57 @@ def test_memory_remember_receives_task_content(): assert "Researcher" in raw assert "Expected result:" in raw assert "Result:" in raw + + +def test_usage_metrics_counts_a_shared_llm_instance_once(): + """A single LLM instance shared by several agents must be counted once. + + ``get_token_usage_summary()`` returns totals cumulative for the lifetime of + the instance, including calls made by every agent sharing it, so adding it + per agent multiplied the reported usage by the number of agents. + """ + llm = LLM(model="gpt-4o") + llm.get_token_usage_summary = lambda: UsageMetrics( + total_tokens=100, prompt_tokens=80, completion_tokens=20, successful_requests=1 + ) + agents = [ + Agent(role=f"Role {i}", goal="goal", backstory="backstory", llm=llm) + for i in range(3) + ] + tasks = [ + Task(description=f"task {i}", expected_output="out", agent=agents[i]) + for i in range(3) + ] + + usage = Crew(agents=agents, tasks=tasks).calculate_usage_metrics() + + assert usage.total_tokens == 100 + assert usage.successful_requests == 1 + + +def test_usage_metrics_still_sums_distinct_llm_instances(): + """Separate LLM instances must still be summed, even for the same model.""" + + def make_llm() -> LLM: + llm = LLM(model="gpt-4o") + llm.get_token_usage_summary = lambda: UsageMetrics( + total_tokens=100, + prompt_tokens=80, + completion_tokens=20, + successful_requests=1, + ) + return llm + + agents = [ + Agent(role=f"Role {i}", goal="goal", backstory="backstory", llm=make_llm()) + for i in range(3) + ] + tasks = [ + Task(description=f"task {i}", expected_output="out", agent=agents[i]) + for i in range(3) + ] + + usage = Crew(agents=agents, tasks=tasks).calculate_usage_metrics() + + assert usage.total_tokens == 300 + assert usage.successful_requests == 3 From 4c30aa4f6ae62349c3cbec6c1f7f58c90be1096d Mon Sep 17 00:00:00 2001 From: Parthiban Sivakumar Date: Mon, 7 Sep 2026 10:22:59 +0530 Subject: [PATCH 2/3] fix(crew): record usage per LLM call instead of reading lifetime counters calculate_usage_metrics() aggregated get_token_usage_summary() per agent. Those counters are cumulative for the LLM instance and shared by every agent holding it, so a shared instance was counted once per agent and a second kickoff reported the first run's usage as well. Accumulate LLMCallCompletedEvent for the duration of a kickoff, scoped to the running crew, keyed by the agent that made the call. Calls made outside an agent (crew planning, guardrails) get their own bucket so they still reach the crew total. Recording at call time needs no before/after window, so agents running concurrently on one instance are attributed correctly. This mirrors Flow's _attach_usage_aggregation_listener. test_hierarchical_kickoff_usage_metrics_include_manager asserted against the removed internals via get_token_usage_summary and _token_process stubs; it now emits the LLM calls those stubs stood in for and asserts the same totals. Fixes #7259 Co-Authored-By: Claude Opus 5 --- lib/crewai/src/crewai/crew.py | 92 +++++++++---- lib/crewai/tests/test_crew.py | 99 +++++-------- lib/crewai/tests/test_crew_usage_metrics.py | 145 ++++++++++++++++++++ 3 files changed, 240 insertions(+), 96 deletions(-) create mode 100644 lib/crewai/tests/test_crew_usage_metrics.py diff --git a/lib/crewai/src/crewai/crew.py b/lib/crewai/src/crewai/crew.py index a77575f9a4..98f38d7d06 100644 --- a/lib/crewai/src/crewai/crew.py +++ b/lib/crewai/src/crewai/crew.py @@ -8,6 +8,7 @@ import json from pathlib import Path import re +import threading from typing import ( TYPE_CHECKING, Annotated, @@ -125,6 +126,7 @@ def get_supported_content_types(provider: str, api: str | None = None) -> list[s from crewai.types.streaming import CrewStreamingOutput from crewai.types.usage_metrics import UsageMetrics from crewai.utilities.constants import NOT_SPECIFIED, TRAINING_DATA_FILE +from crewai.utilities.crew.crew_context import get_crew_context from crewai.utilities.crew.models import CrewContext from crewai.utilities.env import get_env_context from crewai.utilities.evaluators.crew_evaluator_handler import CrewEvaluator @@ -226,6 +228,9 @@ class Crew(FlowTrackable, BaseModel): ) _kickoff_event_id: str | None = PrivateAttr(default=None) _execution_start_dispatched: bool = PrivateAttr(default=False) + _agent_usage: dict[str, UsageMetrics] = PrivateAttr(default_factory=dict) + _usage_lock: threading.Lock = PrivateAttr(default_factory=threading.Lock) + _usage_handler: Callable[[Any, Any], None] | None = PrivateAttr(default=None) _execution_end_dispatched: bool = PrivateAttr(default=False) name: str | None = Field(default="crew") @@ -1047,6 +1052,7 @@ def run_crew() -> None: execution_token = begin_execution() + self._attach_usage_listener() runtime_scope = crewai_event_bus._enter_runtime_scope() try: inputs = prepare_kickoff(self, inputs, input_files) @@ -1083,6 +1089,7 @@ def run_crew() -> None: # Safety net for the exception path; the success path already # drained in _create_crew_output before emitting completion. self._drain_memory_writes() + self._detach_usage_listener() clear_files(self.id) detach(token) end_execution(execution_token) @@ -1264,6 +1271,7 @@ async def run_crew() -> None: execution_token = begin_execution() + self._attach_usage_listener() runtime_scope = crewai_event_bus._enter_runtime_scope() try: inputs = prepare_kickoff(self, inputs, input_files) @@ -1300,6 +1308,7 @@ async def run_crew() -> None: # Safety net for the exception path; the success path already # drained in _create_crew_output before emitting completion. self._drain_memory_writes() + self._detach_usage_listener() clear_files(self.id) detach(token) end_execution(execution_token) @@ -2201,40 +2210,65 @@ def _finish_execution(self, final_string_output: str) -> None: if self.max_rpm: self._rpm_controller.stop_rpm_counter() - def calculate_usage_metrics(self) -> UsageMetrics: - """Calculates and returns the usage metrics.""" - total_usage_metrics = UsageMetrics() + def _attach_usage_listener(self) -> None: + """Accumulate per-agent token usage as each LLM call completes. - # An LLM instance's counters are cumulative for its lifetime and - # include calls made by every agent sharing it, so each distinct - # instance must contribute to the total exactly once. - counted_llms: set[int] = set() + Usage is recorded at call time rather than reconstructed from an LLM + instance's lifetime counters. Those counters are cumulative and shared + by every agent holding the instance, so reading them per agent both + multiplied usage across agents and carried earlier runs into later + ones. Recording each completed call instead needs no before/after + window, so agents running concurrently on a shared instance are still + attributed correctly. + """ + from crewai.events.types.llm_events import LLMCallCompletedEvent - for agent in self.agents: - if isinstance(agent.llm, BaseLLM): - if id(agent.llm) in counted_llms: - continue - counted_llms.add(id(agent.llm)) - llm_usage = agent.llm.get_token_usage_summary() + if self._usage_handler is not None: + return - total_usage_metrics.add_usage_metrics(llm_usage) - else: - if hasattr(agent, "_token_process"): - token_sum = agent._token_process.get_summary() - total_usage_metrics.add_usage_metrics(token_sum) + with self._usage_lock: + self._agent_usage = {} - if self.manager_agent and hasattr(self.manager_agent, "_token_process"): - token_sum = self.manager_agent._token_process.get_summary() - total_usage_metrics.add_usage_metrics(token_sum) + # Bind the accumulator in the closure so a handler still queued on the + # bus from an earlier kickoff writes into its own dict, not this one. + usage = self._agent_usage + lock = self._usage_lock + crew_id = str(self.id) - if self.manager_agent: - if ( - isinstance(self.manager_agent.llm, BaseLLM) - and id(self.manager_agent.llm) not in counted_llms - ): - counted_llms.add(id(self.manager_agent.llm)) - llm_usage = self.manager_agent.llm.get_token_usage_summary() - total_usage_metrics.add_usage_metrics(llm_usage) + def _accumulate(source: Any, event: LLMCallCompletedEvent) -> None: + context = get_crew_context() + if context is None or context.id != crew_id: + return + metrics = UsageMetrics.from_provider_dict(event.usage) + if metrics is None: + return + # Calls made outside an agent (crew planning, guardrails) still + # belong to the crew's total, so they get their own bucket. + key = getattr(event, "agent_id", None) or "__crew__" + with lock: + usage.setdefault(key, UsageMetrics()).add_usage_metrics(metrics) + + crewai_event_bus.on(LLMCallCompletedEvent)(_accumulate) + self._usage_handler = _accumulate + + def _detach_usage_listener(self) -> None: + """Stop accumulating usage once the kickoff has finished.""" + from crewai.events.types.llm_events import LLMCallCompletedEvent + + handler = self._usage_handler + if handler is None: + return + crewai_event_bus.off(LLMCallCompletedEvent, handler) + self._usage_handler = None + + def calculate_usage_metrics(self) -> UsageMetrics: + """Return the token usage recorded for the most recent kickoff.""" + total_usage_metrics = UsageMetrics() + + with self._usage_lock: + per_agent = list(self._agent_usage.values()) + for agent_usage in per_agent: + total_usage_metrics.add_usage_metrics(agent_usage) self.usage_metrics = total_usage_metrics return total_usage_metrics diff --git a/lib/crewai/tests/test_crew.py b/lib/crewai/tests/test_crew.py index c13b41d8a6..4e2b73c798 100644 --- a/lib/crewai/tests/test_crew.py +++ b/lib/crewai/tests/test_crew.py @@ -1813,6 +1813,10 @@ def test_agent_usage_metrics_are_captured_for_hierarchical_process(): def test_hierarchical_kickoff_usage_metrics_include_manager(researcher): """Ensure Crew.kickoff() sums UsageMetrics from both regular and manager agents.""" + from uuid import uuid4 + + from crewai.events.event_bus import crewai_event_bus + from crewai.events.types.llm_events import LLMCallCompletedEvent, LLMCallType manager = Agent( role="Manager", @@ -1834,12 +1838,24 @@ def test_hierarchical_kickoff_usage_metrics_include_manager(researcher): total_tokens=30, prompt_tokens=20, completion_tokens=10, successful_requests=1 ) - researcher.llm.get_token_usage_summary = MagicMock(return_value=researcher_metrics) - - # Mock the manager's _token_process since it uses the fallback path - manager._token_process = MagicMock( - get_summary=MagicMock(return_value=manager_metrics) - ) + def _emit(agent: Agent, metrics: UsageMetrics, calls: int) -> None: + """Emit the LLM calls the agent would have made for ``metrics``.""" + for _ in range(calls): + event = LLMCallCompletedEvent( + call_id=str(uuid4()), + model="gpt-4o", + response="ok", + call_type=LLMCallType.LLM_CALL, + usage={ + "prompt_tokens": metrics.prompt_tokens // calls, + "completion_tokens": metrics.completion_tokens // calls, + "total_tokens": metrics.total_tokens // calls, + }, + from_agent=agent, + ) + future = crewai_event_bus.emit(agent, event) + if future is not None: + future.result(timeout=5.0) crew = Crew( agents=[researcher], @@ -1848,14 +1864,17 @@ def test_hierarchical_kickoff_usage_metrics_include_manager(researcher): process=Process.hierarchical, ) - # We don't care about LLM output here; patch execute_sync to avoid network - with patch.object( - Task, - "execute_sync", - return_value=TaskOutput( + def _execute(*_args, **_kwargs) -> TaskOutput: + # Stand in for the LLM calls the agent and manager would make; usage is + # recorded from these events rather than from LLM lifetime counters. + _emit(researcher, researcher_metrics, researcher_metrics.successful_requests) + _emit(manager, manager_metrics, manager_metrics.successful_requests) + return TaskOutput( description="dummy", raw="Hello", agent=researcher.role, messages=[] - ), - ): + ) + + # We don't care about LLM output here; patch execute_sync to avoid network + with patch.object(Task, "execute_sync", side_effect=_execute): crew.kickoff() assert ( @@ -4989,57 +5008,3 @@ def test_memory_remember_receives_task_content(): assert "Researcher" in raw assert "Expected result:" in raw assert "Result:" in raw - - -def test_usage_metrics_counts_a_shared_llm_instance_once(): - """A single LLM instance shared by several agents must be counted once. - - ``get_token_usage_summary()`` returns totals cumulative for the lifetime of - the instance, including calls made by every agent sharing it, so adding it - per agent multiplied the reported usage by the number of agents. - """ - llm = LLM(model="gpt-4o") - llm.get_token_usage_summary = lambda: UsageMetrics( - total_tokens=100, prompt_tokens=80, completion_tokens=20, successful_requests=1 - ) - agents = [ - Agent(role=f"Role {i}", goal="goal", backstory="backstory", llm=llm) - for i in range(3) - ] - tasks = [ - Task(description=f"task {i}", expected_output="out", agent=agents[i]) - for i in range(3) - ] - - usage = Crew(agents=agents, tasks=tasks).calculate_usage_metrics() - - assert usage.total_tokens == 100 - assert usage.successful_requests == 1 - - -def test_usage_metrics_still_sums_distinct_llm_instances(): - """Separate LLM instances must still be summed, even for the same model.""" - - def make_llm() -> LLM: - llm = LLM(model="gpt-4o") - llm.get_token_usage_summary = lambda: UsageMetrics( - total_tokens=100, - prompt_tokens=80, - completion_tokens=20, - successful_requests=1, - ) - return llm - - agents = [ - Agent(role=f"Role {i}", goal="goal", backstory="backstory", llm=make_llm()) - for i in range(3) - ] - tasks = [ - Task(description=f"task {i}", expected_output="out", agent=agents[i]) - for i in range(3) - ] - - usage = Crew(agents=agents, tasks=tasks).calculate_usage_metrics() - - assert usage.total_tokens == 300 - assert usage.successful_requests == 3 diff --git a/lib/crewai/tests/test_crew_usage_metrics.py b/lib/crewai/tests/test_crew_usage_metrics.py new file mode 100644 index 0000000000..8a1183259a --- /dev/null +++ b/lib/crewai/tests/test_crew_usage_metrics.py @@ -0,0 +1,145 @@ +"""Tests for crew-level token usage aggregation. + +``crew.usage_metrics`` accumulates ``LLMCallCompletedEvent`` for the duration +of one kickoff, so usage reflects what that run actually consumed. Reading an +LLM instance's lifetime counters instead both multiplied usage across agents +sharing the instance and carried earlier runs into later ones. + +The aggregator is exercised through the real event bus with fabricated events +and explicit crew-context control; no live LLM provider is required. +""" + +from __future__ import annotations + +import contextvars +from typing import Any +from uuid import uuid4 + +from opentelemetry import baggage +from opentelemetry.context import attach, detach + +from crewai import Agent, Crew, Task +from crewai.events.event_bus import crewai_event_bus +from crewai.events.types.llm_events import LLMCallCompletedEvent, LLMCallType +from crewai.utilities.crew.models import CrewContext + + +def _emit_llm_call( + *, + crew_id: str | None, + total_tokens: int, + agent_id: str | None = None, +) -> None: + """Emit one fake ``LLMCallCompletedEvent`` under ``crew_id``'s context. + + Runs in a freshly-copied context so the crew context the bus snapshots at + emit time is exactly ``crew_id``, mirroring how ``LLM.call`` emits at + runtime from inside a kickoff. + """ + usage: dict[str, Any] = { + "prompt_tokens": total_tokens, + "completion_tokens": 0, + "total_tokens": total_tokens, + } + event = LLMCallCompletedEvent( + call_id=str(uuid4()), + model="gpt-4o-mini", + response="ok", + call_type=LLMCallType.LLM_CALL, + usage=usage, + ) + if agent_id is not None: + event.agent_id = agent_id + + ctx = contextvars.copy_context() + + def _emit() -> None: + token = None + if crew_id is not None: + token = attach( + baggage.set_baggage("crew_context", CrewContext(id=crew_id, key="k")) + ) + try: + future = crewai_event_bus.emit(object(), event) + if future is not None: + future.result(timeout=5.0) + finally: + if token is not None: + detach(token) + + ctx.run(_emit) + + +def _crew() -> Crew: + agent = Agent(role="Role", goal="goal", backstory="backstory", llm="gpt-4o") + task = Task(description="task", expected_output="out", agent=agent) + return Crew(agents=[agent], tasks=[task]) + + +def test_usage_sums_every_observed_call_once() -> None: + """Each completed call contributes exactly once, including agent-less ones.""" + crew = _crew() + crew_id = str(crew.id) + crew._attach_usage_listener() + try: + _emit_llm_call(crew_id=crew_id, total_tokens=100, agent_id="agent-a") + _emit_llm_call(crew_id=crew_id, total_tokens=50, agent_id="agent-b") + # e.g. crew planning, which runs outside any agent + _emit_llm_call(crew_id=crew_id, total_tokens=25) + finally: + crew._detach_usage_listener() + + assert crew.calculate_usage_metrics().total_tokens == 175 + + +def test_agents_sharing_an_llm_are_not_double_counted() -> None: + """Two agents on one LLM instance report the calls made, not a multiple. + + Lifetime counters on a shared instance previously produced N x usage for + N agents; per-call accumulation cannot. + """ + crew = _crew() + crew_id = str(crew.id) + crew._attach_usage_listener() + try: + _emit_llm_call(crew_id=crew_id, total_tokens=100, agent_id="agent-a") + _emit_llm_call(crew_id=crew_id, total_tokens=100, agent_id="agent-b") + finally: + crew._detach_usage_listener() + + assert crew.calculate_usage_metrics().total_tokens == 200 + + +def test_second_run_excludes_the_previous_run() -> None: + """A later kickoff reports only its own usage.""" + crew = _crew() + crew_id = str(crew.id) + + crew._attach_usage_listener() + try: + _emit_llm_call(crew_id=crew_id, total_tokens=100, agent_id="agent-a") + finally: + crew._detach_usage_listener() + assert crew.calculate_usage_metrics().total_tokens == 100 + + crew._attach_usage_listener() + try: + _emit_llm_call(crew_id=crew_id, total_tokens=50, agent_id="agent-a") + finally: + crew._detach_usage_listener() + + assert crew.calculate_usage_metrics().total_tokens == 50 + + +def test_calls_from_another_crew_are_ignored() -> None: + """Usage is scoped to the crew that is running.""" + crew = _crew() + crew._attach_usage_listener() + try: + _emit_llm_call(crew_id=str(crew.id), total_tokens=100, agent_id="agent-a") + _emit_llm_call(crew_id=str(uuid4()), total_tokens=999, agent_id="agent-z") + _emit_llm_call(crew_id=None, total_tokens=999, agent_id="agent-z") + finally: + crew._detach_usage_listener() + + assert crew.calculate_usage_metrics().total_tokens == 100 From 9cf2bb4e97b3b32c908c99817ce777eb0a928a70 Mon Sep 17 00:00:00 2001 From: Parthiban Sivakumar Date: Mon, 7 Sep 2026 10:36:25 +0530 Subject: [PATCH 3/3] fix(crew): measure usage per LLM instance across a run, not per agent Replaces the per-call event aggregation from the previous commit. CI showed why it could not work: _track_token_usage_internal only updates an LLM's counters and never emits LLMCallCompletedEvent, so any custom BaseLLM that overrides call() reported zero usage. test_usage_shape_parity covers exactly that case and failed with "assert 0 > 0". Snapshot each distinct LLM instance's counters at kickoff and report the growth across the run. De-duplicating by object identity keeps an instance shared by several agents from being counted once per agent, and taking the baseline per instance rather than per agent leaves no per-agent window to overlap when tasks run concurrently. Reading counters also keeps custom BaseLLM implementations working. test_hierarchical_kickoff_usage_metrics_include_manager keeps its original shape; its stubbed summaries now grow across the run so a delta is observable. Fixes #7259 Co-Authored-By: Claude Opus 5 --- lib/crewai/src/crewai/crew.py | 103 ++++------ lib/crewai/tests/test_crew.py | 46 ++--- lib/crewai/tests/test_crew_usage_metrics.py | 216 ++++++++------------ 3 files changed, 141 insertions(+), 224 deletions(-) diff --git a/lib/crewai/src/crewai/crew.py b/lib/crewai/src/crewai/crew.py index 98f38d7d06..381931c7bc 100644 --- a/lib/crewai/src/crewai/crew.py +++ b/lib/crewai/src/crewai/crew.py @@ -8,7 +8,6 @@ import json from pathlib import Path import re -import threading from typing import ( TYPE_CHECKING, Annotated, @@ -126,7 +125,6 @@ def get_supported_content_types(provider: str, api: str | None = None) -> list[s from crewai.types.streaming import CrewStreamingOutput from crewai.types.usage_metrics import UsageMetrics from crewai.utilities.constants import NOT_SPECIFIED, TRAINING_DATA_FILE -from crewai.utilities.crew.crew_context import get_crew_context from crewai.utilities.crew.models import CrewContext from crewai.utilities.env import get_env_context from crewai.utilities.evaluators.crew_evaluator_handler import CrewEvaluator @@ -228,9 +226,7 @@ class Crew(FlowTrackable, BaseModel): ) _kickoff_event_id: str | None = PrivateAttr(default=None) _execution_start_dispatched: bool = PrivateAttr(default=False) - _agent_usage: dict[str, UsageMetrics] = PrivateAttr(default_factory=dict) - _usage_lock: threading.Lock = PrivateAttr(default_factory=threading.Lock) - _usage_handler: Callable[[Any, Any], None] | None = PrivateAttr(default=None) + _usage_baselines: dict[int, UsageMetrics] = PrivateAttr(default_factory=dict) _execution_end_dispatched: bool = PrivateAttr(default=False) name: str | None = Field(default="crew") @@ -1052,7 +1048,7 @@ def run_crew() -> None: execution_token = begin_execution() - self._attach_usage_listener() + self._snapshot_usage_baselines() runtime_scope = crewai_event_bus._enter_runtime_scope() try: inputs = prepare_kickoff(self, inputs, input_files) @@ -1089,7 +1085,6 @@ def run_crew() -> None: # Safety net for the exception path; the success path already # drained in _create_crew_output before emitting completion. self._drain_memory_writes() - self._detach_usage_listener() clear_files(self.id) detach(token) end_execution(execution_token) @@ -1271,7 +1266,7 @@ async def run_crew() -> None: execution_token = begin_execution() - self._attach_usage_listener() + self._snapshot_usage_baselines() runtime_scope = crewai_event_bus._enter_runtime_scope() try: inputs = prepare_kickoff(self, inputs, input_files) @@ -1308,7 +1303,6 @@ async def run_crew() -> None: # Safety net for the exception path; the success path already # drained in _create_crew_output before emitting completion. self._drain_memory_writes() - self._detach_usage_listener() clear_files(self.id) detach(token) end_execution(execution_token) @@ -2210,65 +2204,50 @@ def _finish_execution(self, final_string_output: str) -> None: if self.max_rpm: self._rpm_controller.stop_rpm_counter() - def _attach_usage_listener(self) -> None: - """Accumulate per-agent token usage as each LLM call completes. + def _llm_instances(self) -> list[BaseLLM]: + """Return the distinct LLM instances this crew runs on. - Usage is recorded at call time rather than reconstructed from an LLM - instance's lifetime counters. Those counters are cumulative and shared - by every agent holding the instance, so reading them per agent both - multiplied usage across agents and carried earlier runs into later - ones. Recording each completed call instead needs no before/after - window, so agents running concurrently on a shared instance are still - attributed correctly. + De-duplicated by object identity: an instance shared by several agents + holds one set of counters, so it must be measured once. """ - from crewai.events.types.llm_events import LLMCallCompletedEvent - - if self._usage_handler is not None: - return - - with self._usage_lock: - self._agent_usage = {} - - # Bind the accumulator in the closure so a handler still queued on the - # bus from an earlier kickoff writes into its own dict, not this one. - usage = self._agent_usage - lock = self._usage_lock - crew_id = str(self.id) - - def _accumulate(source: Any, event: LLMCallCompletedEvent) -> None: - context = get_crew_context() - if context is None or context.id != crew_id: - return - metrics = UsageMetrics.from_provider_dict(event.usage) - if metrics is None: - return - # Calls made outside an agent (crew planning, guardrails) still - # belong to the crew's total, so they get their own bucket. - key = getattr(event, "agent_id", None) or "__crew__" - with lock: - usage.setdefault(key, UsageMetrics()).add_usage_metrics(metrics) - - crewai_event_bus.on(LLMCallCompletedEvent)(_accumulate) - self._usage_handler = _accumulate - - def _detach_usage_listener(self) -> None: - """Stop accumulating usage once the kickoff has finished.""" - from crewai.events.types.llm_events import LLMCallCompletedEvent - - handler = self._usage_handler - if handler is None: - return - crewai_event_bus.off(LLMCallCompletedEvent, handler) - self._usage_handler = None + instances: list[BaseLLM] = [] + seen: set[int] = set() + for agent in (*self.agents, self.manager_agent): + llm = getattr(agent, "llm", None) + if isinstance(llm, BaseLLM) and id(llm) not in seen: + seen.add(id(llm)) + instances.append(llm) + return instances + + def _snapshot_usage_baselines(self) -> None: + """Record each LLM instance's counters at the start of a kickoff. + + An instance's counters are cumulative for its lifetime, so usage for + one run is the difference between these baselines and the counters at + the end. Snapshotting per instance rather than per agent keeps a shared + instance from being counted once per agent, and needs no per-agent + window, so concurrent tasks on one instance stay correct. + """ + self._usage_baselines = { + id(llm): llm.get_token_usage_summary() for llm in self._llm_instances() + } def calculate_usage_metrics(self) -> UsageMetrics: - """Return the token usage recorded for the most recent kickoff.""" + """Return the token usage accrued during the most recent kickoff.""" total_usage_metrics = UsageMetrics() - with self._usage_lock: - per_agent = list(self._agent_usage.values()) - for agent_usage in per_agent: - total_usage_metrics.add_usage_metrics(agent_usage) + for llm in self._llm_instances(): + current = llm.get_token_usage_summary() + baseline = self._usage_baselines.get(id(llm)) + usage = current.delta_since(baseline) if baseline is not None else current + total_usage_metrics.add_usage_metrics(usage) + + for agent in (*self.agents, self.manager_agent): + if agent is None or isinstance(getattr(agent, "llm", None), BaseLLM): + continue + token_process = getattr(agent, "_token_process", None) + if token_process is not None: + total_usage_metrics.add_usage_metrics(token_process.get_summary()) self.usage_metrics = total_usage_metrics return total_usage_metrics diff --git a/lib/crewai/tests/test_crew.py b/lib/crewai/tests/test_crew.py index 4e2b73c798..42fdfb172f 100644 --- a/lib/crewai/tests/test_crew.py +++ b/lib/crewai/tests/test_crew.py @@ -1813,10 +1813,6 @@ def test_agent_usage_metrics_are_captured_for_hierarchical_process(): def test_hierarchical_kickoff_usage_metrics_include_manager(researcher): """Ensure Crew.kickoff() sums UsageMetrics from both regular and manager agents.""" - from uuid import uuid4 - - from crewai.events.event_bus import crewai_event_bus - from crewai.events.types.llm_events import LLMCallCompletedEvent, LLMCallType manager = Agent( role="Manager", @@ -1838,24 +1834,21 @@ def test_hierarchical_kickoff_usage_metrics_include_manager(researcher): total_tokens=30, prompt_tokens=20, completion_tokens=10, successful_requests=1 ) - def _emit(agent: Agent, metrics: UsageMetrics, calls: int) -> None: - """Emit the LLM calls the agent would have made for ``metrics``.""" - for _ in range(calls): - event = LLMCallCompletedEvent( - call_id=str(uuid4()), - model="gpt-4o", - response="ok", - call_type=LLMCallType.LLM_CALL, - usage={ - "prompt_tokens": metrics.prompt_tokens // calls, - "completion_tokens": metrics.completion_tokens // calls, - "total_tokens": metrics.total_tokens // calls, - }, - from_agent=agent, - ) - future = crewai_event_bus.emit(agent, event) - if future is not None: - future.result(timeout=5.0) + # Usage for a run is the growth of each LLM's counters across it, so the + # summaries read empty until the task runs and report totals afterwards. + consumed = {"done": False} + researcher.llm.get_token_usage_summary = ( + lambda: researcher_metrics if consumed["done"] else UsageMetrics() + ) + manager.llm.get_token_usage_summary = ( + lambda: manager_metrics if consumed["done"] else UsageMetrics() + ) + + def _execute(*_args, **_kwargs) -> TaskOutput: + consumed["done"] = True + return TaskOutput( + description="dummy", raw="Hello", agent=researcher.role, messages=[] + ) crew = Crew( agents=[researcher], @@ -1864,15 +1857,6 @@ def _emit(agent: Agent, metrics: UsageMetrics, calls: int) -> None: process=Process.hierarchical, ) - def _execute(*_args, **_kwargs) -> TaskOutput: - # Stand in for the LLM calls the agent and manager would make; usage is - # recorded from these events rather than from LLM lifetime counters. - _emit(researcher, researcher_metrics, researcher_metrics.successful_requests) - _emit(manager, manager_metrics, manager_metrics.successful_requests) - return TaskOutput( - description="dummy", raw="Hello", agent=researcher.role, messages=[] - ) - # We don't care about LLM output here; patch execute_sync to avoid network with patch.object(Task, "execute_sync", side_effect=_execute): crew.kickoff() diff --git a/lib/crewai/tests/test_crew_usage_metrics.py b/lib/crewai/tests/test_crew_usage_metrics.py index 8a1183259a..cd51555150 100644 --- a/lib/crewai/tests/test_crew_usage_metrics.py +++ b/lib/crewai/tests/test_crew_usage_metrics.py @@ -1,145 +1,99 @@ """Tests for crew-level token usage aggregation. -``crew.usage_metrics`` accumulates ``LLMCallCompletedEvent`` for the duration -of one kickoff, so usage reflects what that run actually consumed. Reading an -LLM instance's lifetime counters instead both multiplied usage across agents -sharing the instance and carried earlier runs into later ones. - -The aggregator is exercised through the real event bus with fabricated events -and explicit crew-context control; no live LLM provider is required. +``crew.usage_metrics`` reports what a kickoff consumed, measured as the growth +of each LLM instance's counters across the run. Reading those counters as +absolute totals instead both multiplied usage across agents sharing an +instance and carried earlier runs into later ones. + +Baselines are taken per distinct instance rather than per agent, so a shared +instance is measured once and no per-agent window exists to overlap when tasks +run concurrently. """ from __future__ import annotations -import contextvars -from typing import Any -from uuid import uuid4 - -from opentelemetry import baggage -from opentelemetry.context import attach, detach - -from crewai import Agent, Crew, Task -from crewai.events.event_bus import crewai_event_bus -from crewai.events.types.llm_events import LLMCallCompletedEvent, LLMCallType -from crewai.utilities.crew.models import CrewContext - - -def _emit_llm_call( - *, - crew_id: str | None, - total_tokens: int, - agent_id: str | None = None, -) -> None: - """Emit one fake ``LLMCallCompletedEvent`` under ``crew_id``'s context. - - Runs in a freshly-copied context so the crew context the bus snapshots at - emit time is exactly ``crew_id``, mirroring how ``LLM.call`` emits at - runtime from inside a kickoff. - """ - usage: dict[str, Any] = { - "prompt_tokens": total_tokens, - "completion_tokens": 0, - "total_tokens": total_tokens, - } - event = LLMCallCompletedEvent( - call_id=str(uuid4()), - model="gpt-4o-mini", - response="ok", - call_type=LLMCallType.LLM_CALL, - usage=usage, - ) - if agent_id is not None: - event.agent_id = agent_id - - ctx = contextvars.copy_context() - - def _emit() -> None: - token = None - if crew_id is not None: - token = attach( - baggage.set_baggage("crew_context", CrewContext(id=crew_id, key="k")) - ) - try: - future = crewai_event_bus.emit(object(), event) - if future is not None: - future.result(timeout=5.0) - finally: - if token is not None: - detach(token) - - ctx.run(_emit) - - -def _crew() -> Crew: - agent = Agent(role="Role", goal="goal", backstory="backstory", llm="gpt-4o") - task = Task(description="task", expected_output="out", agent=agent) - return Crew(agents=[agent], tasks=[task]) - - -def test_usage_sums_every_observed_call_once() -> None: - """Each completed call contributes exactly once, including agent-less ones.""" - crew = _crew() - crew_id = str(crew.id) - crew._attach_usage_listener() - try: - _emit_llm_call(crew_id=crew_id, total_tokens=100, agent_id="agent-a") - _emit_llm_call(crew_id=crew_id, total_tokens=50, agent_id="agent-b") - # e.g. crew planning, which runs outside any agent - _emit_llm_call(crew_id=crew_id, total_tokens=25) - finally: - crew._detach_usage_listener() - - assert crew.calculate_usage_metrics().total_tokens == 175 - - -def test_agents_sharing_an_llm_are_not_double_counted() -> None: - """Two agents on one LLM instance report the calls made, not a multiple. - - Lifetime counters on a shared instance previously produced N x usage for - N agents; per-call accumulation cannot. - """ - crew = _crew() - crew_id = str(crew.id) - crew._attach_usage_listener() - try: - _emit_llm_call(crew_id=crew_id, total_tokens=100, agent_id="agent-a") - _emit_llm_call(crew_id=crew_id, total_tokens=100, agent_id="agent-b") - finally: - crew._detach_usage_listener() - - assert crew.calculate_usage_metrics().total_tokens == 200 - - -def test_second_run_excludes_the_previous_run() -> None: - """A later kickoff reports only its own usage.""" - crew = _crew() - crew_id = str(crew.id) - - crew._attach_usage_listener() - try: - _emit_llm_call(crew_id=crew_id, total_tokens=100, agent_id="agent-a") - finally: - crew._detach_usage_listener() +from crewai import Agent, Crew, Task, LLM +from crewai.types.usage_metrics import UsageMetrics + + +class _Counter: + """Stands in for an LLM instance's cumulative lifetime counters.""" + + def __init__(self) -> None: + self.total = 0 + + def bind(self, llm: LLM) -> LLM: + llm.get_token_usage_summary = lambda: UsageMetrics( # type: ignore[method-assign] + total_tokens=self.total, + prompt_tokens=self.total, + successful_requests=1 if self.total else 0, + ) + return llm + + def consume(self, tokens: int) -> None: + self.total += tokens + + +def _crew(*llms: LLM) -> Crew: + agents = [ + Agent(role=f"Role {i}", goal="goal", backstory="backstory", llm=llm) + for i, llm in enumerate(llms) + ] + tasks = [ + Task(description=f"task {i}", expected_output="out", agent=agent) + for i, agent in enumerate(agents) + ] + return Crew(agents=agents, tasks=tasks) + + +def test_agents_sharing_one_llm_are_counted_once() -> None: + """Three agents on one instance report that instance's usage, not 3x it.""" + counter = _Counter() + llm = counter.bind(LLM(model="gpt-4o")) + crew = _crew(llm, llm, llm) + + crew._snapshot_usage_baselines() + counter.consume(100) + assert crew.calculate_usage_metrics().total_tokens == 100 - crew._attach_usage_listener() - try: - _emit_llm_call(crew_id=crew_id, total_tokens=50, agent_id="agent-a") - finally: - crew._detach_usage_listener() +def test_distinct_llm_instances_are_summed() -> None: + """Separate instances still add up, even for the same model.""" + first, second = _Counter(), _Counter() + crew = _crew(first.bind(LLM(model="gpt-4o")), second.bind(LLM(model="gpt-4o"))) + + crew._snapshot_usage_baselines() + first.consume(100) + second.consume(50) + + assert crew.calculate_usage_metrics().total_tokens == 150 + + +def test_a_later_run_excludes_the_previous_one() -> None: + """Usage is the growth across this run, not the instance's lifetime.""" + counter = _Counter() + crew = _crew(counter.bind(LLM(model="gpt-4o"))) + + crew._snapshot_usage_baselines() + counter.consume(100) + assert crew.calculate_usage_metrics().total_tokens == 100 + + crew._snapshot_usage_baselines() + counter.consume(50) assert crew.calculate_usage_metrics().total_tokens == 50 -def test_calls_from_another_crew_are_ignored() -> None: - """Usage is scoped to the crew that is running.""" - crew = _crew() - crew._attach_usage_listener() - try: - _emit_llm_call(crew_id=str(crew.id), total_tokens=100, agent_id="agent-a") - _emit_llm_call(crew_id=str(uuid4()), total_tokens=999, agent_id="agent-z") - _emit_llm_call(crew_id=None, total_tokens=999, agent_id="agent-z") - finally: - crew._detach_usage_listener() +def test_manager_sharing_an_agent_llm_is_counted_once() -> None: + """A manager on the same instance as its agents adds no extra usage.""" + counter = _Counter() + llm = counter.bind(LLM(model="gpt-4o")) + crew = _crew(llm) + crew.manager_agent = Agent( + role="Manager", goal="coordinate", backstory="backstory", llm=llm + ) + + crew._snapshot_usage_baselines() + counter.consume(100) assert crew.calculate_usage_metrics().total_tokens == 100