andre/crewAI
1
0
Fork 0
mirror of https://github.com/crewAIInc/crewAI.git synced 2026-07-22 07:15:10 +00:00
Code Issues Actions 6 Packages Projects Releases Wiki Activity

fix(agent): report per-call usage metrics on kickoff results (#6506)

Browse Source 
* fix(agent): report per-call usage metrics on kickoff results

Agent.kickoff() populated result.usage_metrics from the LLM instance's
lifetime token accumulator, so counts grew across calls and pooled
across agents sharing one LLM object — a second agent's first turn
appeared to cost the whole preceding session.

Snapshot the accumulator when a kickoff starts and report the delta on
the result (guardrail retries included), via the new
UsageMetrics.delta_since(). The LLM instance's cumulative counters are
untouched: get_token_usage_summary() keeps lifetime totals for
crew-level aggregation, and its docstring now states that scope
explicitly. Applies to both Agent and the deprecated LiteAgent, sync
and async paths.

Fixes EPD-177.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

* refactor(agent): drop lite_agent.py diff, add guardrail-retry usage test

Per review: LiteAgent's kickoff path is no longer used, so the per-call
usage snapshot only needs to live in agent/core.py — revert the
lite_agent.py changes entirely. This also removes the duplicated
_current_usage_summary helper and the instance-attr baseline CodeRabbit
flagged.

Add the requested guardrail-retry regression test: a guardrail that
rejects the first attempt and accepts the second must yield
usage_metrics covering both attempts (2x a single-attempt kickoff).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
João Moura
2026-07-10 18:17:45 -03:00
committed by GitHub
parent 7967b19057
commit a8b3ecb723
5 changed files with 262 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

71
lib/crewai/src/crewai/agent/core.py
View File

@@ -86,6 +86,7 @@ from crewai.skills.models import Skill as SkillModel
from crewai.state.checkpoint_config import CheckpointConfig, apply_checkpoint
from crewai.tools.agent_tools.agent_tools import AgentTools
from crewai.types.callback import SerializableCallable
from crewai.types.usage_metrics import UsageMetrics
from crewai.utilities.agent_utils import (
get_tool_names,
is_inside_event_loop,
@@ -1582,9 +1583,18 @@ class Agent(BaseAgent):
crewai_event_bus.emit(self, event=started_event)
self._kickoff_event_id = started_event.event_id
output = self._execute_and_build_output(executor, inputs, response_format)
usage_baseline = self._current_usage_summary()
output = self._execute_and_build_output(
executor, inputs, response_format, usage_baseline
)
return self._finalize_kickoff(
output, executor, inputs, response_format, messages, agent_info
output,
executor,
inputs,
response_format,
messages,
agent_info,
usage_baseline,
)
except Exception as e:
@@ -1598,6 +1608,7 @@ class Agent(BaseAgent):
response_format: type[Any] | None,
messages: str | list[LLMMessage],
agent_info: dict[str, Any],
usage_baseline: UsageMetrics | None = None,
) -> LiteAgentOutput:
"""Apply guardrails, save to memory, and emit completion event.
@@ -1608,6 +1619,8 @@ class Agent(BaseAgent):
response_format: Optional response format.
messages: The original messages.
agent_info: Agent metadata for events.
usage_baseline: Usage snapshot taken at kickoff start, so retries
report per-call usage relative to it.
Returns:
The finalized output.
@@ -1618,6 +1631,7 @@ class Agent(BaseAgent):
executor=executor,
inputs=inputs,
response_format=response_format,
usage_baseline=usage_baseline,
)
self._save_kickoff_to_memory(messages, output.raw)
@@ -1669,11 +1683,24 @@ class Agent(BaseAgent):
except Exception as e:
self._logger.log("error", f"Failed to save kickoff result to memory: {e}")
def _current_usage_summary(self) -> UsageMetrics:
"""Snapshot the cumulative usage counters backing this agent's LLM.
The counters live on the LLM instance (or the agent's token process
for non-BaseLLM models) and grow for the object's lifetime — across
calls and across agents sharing the instance. Per-call usage is the
delta between two snapshots.
"""
if isinstance(self.llm, BaseLLM):
return self.llm.get_token_usage_summary()
return self._token_process.get_summary()
def _build_output_from_result(
self,
result: dict[str, Any],
executor: AgentExecutor,
response_format: type[Any] | None = None,
usage_baseline: UsageMetrics | None = None,
) -> LiteAgentOutput:
"""Build a LiteAgentOutput from an executor result dict.
@@ -1683,6 +1710,9 @@ class Agent(BaseAgent):
result: The result dictionary from executor.invoke / invoke_async.
executor: The executor instance.
response_format: Optional response format.
usage_baseline: Usage snapshot taken at kickoff start. When given,
the output carries only this call's usage (the delta) instead
of the LLM instance's cumulative lifetime counters.
Returns:
LiteAgentOutput with raw output, formatted result, and metrics.
@@ -1727,10 +1757,9 @@ class Agent(BaseAgent):
else:
raw_output = str(output) if not isinstance(output, str) else output
if isinstance(self.llm, BaseLLM):
usage_metrics = self.llm.get_token_usage_summary()
else:
usage_metrics = self._token_process.get_summary()
usage_metrics = self._current_usage_summary()
if usage_baseline is not None:
usage_metrics = usage_metrics.delta_since(usage_baseline)
raw_str = (
raw_output
@@ -1759,20 +1788,26 @@ class Agent(BaseAgent):
executor: AgentExecutor,
inputs: dict[str, str],
response_format: type[Any] | None = None,
usage_baseline: UsageMetrics | None = None,
) -> LiteAgentOutput:
"""Execute the agent synchronously and build the output object."""
result = cast(dict[str, Any], executor.invoke(inputs))
return self._build_output_from_result(result, executor, response_format)
return self._build_output_from_result(
result, executor, response_format, usage_baseline
)
async def _execute_and_build_output_async(
self,
executor: AgentExecutor,
inputs: dict[str, str],
response_format: type[Any] | None = None,
usage_baseline: UsageMetrics | None = None,
) -> LiteAgentOutput:
"""Execute the agent asynchronously and build the output object."""
result = await executor.invoke_async(inputs)
return self._build_output_from_result(result, executor, response_format)
return self._build_output_from_result(
result, executor, response_format, usage_baseline
)
def _process_kickoff_guardrail(
self,
@@ -1781,6 +1816,7 @@ class Agent(BaseAgent):
inputs: dict[str, str],
response_format: type[Any] | None = None,
retry_count: int = 0,
usage_baseline: UsageMetrics | None = None,
) -> LiteAgentOutput:
"""Process guardrail for kickoff execution with retry logic.
@@ -1790,6 +1826,9 @@ class Agent(BaseAgent):
inputs: Input dictionary for re-execution.
response_format: Optional response format.
retry_count: Current retry count.
usage_baseline: Usage snapshot taken at kickoff start, so a
retried output reports the whole call's usage, not just the
last attempt's.
Returns:
Validated/updated output.
@@ -1827,7 +1866,9 @@ class Agent(BaseAgent):
role="user",
)
output = self._execute_and_build_output(executor, inputs, response_format)
output = self._execute_and_build_output(
executor, inputs, response_format, usage_baseline
)
return self._process_kickoff_guardrail(
output=output,
@@ -1835,6 +1876,7 @@ class Agent(BaseAgent):
inputs=inputs,
response_format=response_format,
retry_count=retry_count + 1,
usage_baseline=usage_baseline,
)
if guardrail_result.result is not None:
@@ -1897,11 +1939,18 @@ class Agent(BaseAgent):
crewai_event_bus.emit(self, event=started_event)
self._kickoff_event_id = started_event.event_id
usage_baseline = self._current_usage_summary()
output = await self._execute_and_build_output_async(
executor, inputs, response_format
executor, inputs, response_format, usage_baseline
)
return self._finalize_kickoff(
output, executor, inputs, response_format, messages, agent_info
output,
executor,
inputs,
response_format,
messages,
agent_info,
usage_baseline,
)
except Exception as e:

6
lib/crewai/src/crewai/lite_agent_output.py
View File

@@ -38,7 +38,11 @@ class LiteAgentOutput(BaseModel):
)
agent_role: str = Field(description="Role of the agent that produced this output")
usage_metrics: dict[str, Any] | None = Field(
description="Token usage metrics for this execution", default=None
description=(
"Token usage metrics for this kickoff call only (guardrail "
"retries included), not the LLM instance's cumulative totals"
),
default=None,
)
messages: list[LLMMessage] = Field(
description="Messages of the agent", default_factory=list

9
lib/crewai/src/crewai/llms/base_llm.py
View File

@@ -966,8 +966,15 @@ class BaseLLM(BaseModel, ABC):
def get_token_usage_summary(self) -> UsageMetrics:
"""Get summary of token usage for this LLM instance.
The counters are cumulative for the lifetime of this instance: they
grow across every call made through it, including calls issued by
different agents sharing the instance. For usage scoped to a single
call, snapshot before and after and use
``UsageMetrics.delta_since`` (agent kickoff results already report
per-call usage this way).
Returns:
Dictionary with token usage totals
UsageMetrics with this instance's lifetime token usage totals.
"""
return UsageMetrics(**self._token_usage)

32
lib/crewai/src/crewai/types/usage_metrics.py
View File

@@ -76,6 +76,38 @@ class UsageMetrics(BaseModel):
self.cache_creation_tokens += usage_metrics.cache_creation_tokens
self.successful_requests += usage_metrics.successful_requests
def delta_since(self, baseline: Self) -> Self:
"""Return the per-call usage accrued since ``baseline`` was captured.
Both objects must come from the same monotonically increasing
accumulator (e.g. an LLM instance's lifetime counters). Differences
are clamped at zero so a reset accumulator can't produce negative
usage.
Args:
baseline: A snapshot of the same accumulator taken earlier.
Returns:
A new UsageMetrics with the field-wise difference.
"""
return type(self)(
total_tokens=max(0, self.total_tokens - baseline.total_tokens),
prompt_tokens=max(0, self.prompt_tokens - baseline.prompt_tokens),
cached_prompt_tokens=max(
0, self.cached_prompt_tokens - baseline.cached_prompt_tokens
),
completion_tokens=max(
0, self.completion_tokens - baseline.completion_tokens
),
reasoning_tokens=max(0, self.reasoning_tokens - baseline.reasoning_tokens),
cache_creation_tokens=max(
0, self.cache_creation_tokens - baseline.cache_creation_tokens
),
successful_requests=max(
0, self.successful_requests - baseline.successful_requests
),
)
@classmethod
def from_provider_dict(cls, usage_data: dict[str, Any] | None) -> Self | None:
"""Normalize a provider's raw usage dict into a ``UsageMetrics``.

157
lib/crewai/tests/agents/test_lite_agent.py
View File

@@ -1138,3 +1138,160 @@ def test_lite_agent_memory_instance_recall_and_save_called():
mock_memory.remember_many.assert_called_once_with(
["Fact one.", "Fact two."], agent_role="Test"
)
class _FixedUsageLLM(BaseLLM):
"""Offline BaseLLM that records fixed usage (100/10 tokens) per call."""
def __init__(self):
super().__init__(model="fixed-usage-model")
def call(
self,
messages,
tools=None,
callbacks=None,
available_functions=None,
from_task=None,
from_agent=None,
response_model=None,
) -> str:
self._track_token_usage_internal(
{"prompt_tokens": 100, "completion_tokens": 10, "total_tokens": 110}
)
return "Thought: I know the answer.\nFinal Answer: fake answer"
def supports_function_calling(self) -> bool:
return False
def supports_stop_words(self) -> bool:
return False
def get_context_window_size(self) -> int:
return 4096
class TestKickoffUsageMetricsArePerCall:
"""Regression tests for EPD-177: kickoff results used to expose the LLM
instance's cumulative lifetime counters, so counts accumulated across
calls and pooled across agents sharing one LLM object.
"""
def _make_agent(self, role: str, llm: BaseLLM) -> Agent:
return Agent(
role=role,
goal="Answer questions.",
backstory="Test agent.",
llm=llm,
verbose=False,
)
def test_agents_sharing_one_llm_report_per_call_usage(self):
shared = _FixedUsageLLM()
r1 = self._make_agent("agent one", shared).kickoff("question one")
r2 = self._make_agent("agent two", shared).kickoff("question two")
assert r1.usage_metrics is not None
assert r1.usage_metrics["prompt_tokens"] > 0
# The second agent's call must not include the first agent's tokens.
assert r2.usage_metrics == r1.usage_metrics
# The shared LLM instance still exposes cumulative lifetime totals.
lifetime = shared.get_token_usage_summary()
assert lifetime.prompt_tokens == (
r1.usage_metrics["prompt_tokens"] + r2.usage_metrics["prompt_tokens"]
)
assert lifetime.successful_requests == (
r1.usage_metrics["successful_requests"]
+ r2.usage_metrics["successful_requests"]
)
def test_repeated_kickoffs_on_same_agent_report_per_call_usage(self):
agent = self._make_agent("agent", _FixedUsageLLM())
r1 = agent.kickoff("question one")
r2 = agent.kickoff("question two")
assert r1.usage_metrics is not None
assert r1.usage_metrics["prompt_tokens"] > 0
assert r2.usage_metrics == r1.usage_metrics
@pytest.mark.asyncio
async def test_async_kickoff_reports_per_call_usage(self):
shared = _FixedUsageLLM()
r1 = await self._make_agent("agent one", shared).kickoff_async("question one")
r2 = await self._make_agent("agent two", shared).kickoff_async("question two")
assert r1.usage_metrics is not None
assert r1.usage_metrics["prompt_tokens"] > 0
assert r2.usage_metrics == r1.usage_metrics
def test_guardrail_retry_usage_includes_all_attempts(self):
"""A guardrail retry re-invokes the LLM within the same kickoff, so
the result must report the whole call's usage — every attempt — not
just the last one."""
baseline = (
self._make_agent("baseline", _FixedUsageLLM())
.kickoff("question one")
.usage_metrics
)
attempts: list[str] = []
def flaky_guardrail(output):
attempts.append(output.raw)
if len(attempts) == 1:
return (False, "Please try again.")
return (True, output.raw)
agent = Agent(
role="agent",
goal="Answer questions.",
backstory="Test agent.",
llm=_FixedUsageLLM(),
guardrail=flaky_guardrail,
verbose=False,
)
result = agent.kickoff("question one")
assert len(attempts) == 2
assert result.usage_metrics["successful_requests"] == (
2 * baseline["successful_requests"]
)
assert result.usage_metrics["prompt_tokens"] == 2 * baseline["prompt_tokens"]
assert result.usage_metrics["total_tokens"] == 2 * baseline["total_tokens"]
class TestUsageMetricsDeltaSince:
def test_field_wise_difference(self):
baseline = UsageMetrics(
total_tokens=110,
prompt_tokens=100,
completion_tokens=10,
successful_requests=1,
)
current = UsageMetrics(
total_tokens=330,
prompt_tokens=300,
completion_tokens=30,
cached_prompt_tokens=5,
reasoning_tokens=7,
cache_creation_tokens=3,
successful_requests=3,
)
delta = current.delta_since(baseline)
assert delta == UsageMetrics(
total_tokens=220,
prompt_tokens=200,
completion_tokens=20,
cached_prompt_tokens=5,
reasoning_tokens=7,
cache_creation_tokens=3,
successful_requests=2,
)
def test_clamps_negative_differences_to_zero(self):
baseline = UsageMetrics(total_tokens=100, prompt_tokens=90, successful_requests=2)
delta = UsageMetrics().delta_since(baseline)
assert delta == UsageMetrics()