diff --git a/lib/crewai/src/crewai/agent/core.py b/lib/crewai/src/crewai/agent/core.py index 5751f3a9a..4f4c2368c 100644 --- a/lib/crewai/src/crewai/agent/core.py +++ b/lib/crewai/src/crewai/agent/core.py @@ -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: diff --git a/lib/crewai/src/crewai/lite_agent_output.py b/lib/crewai/src/crewai/lite_agent_output.py index 1ac79d422..bd17f5855 100644 --- a/lib/crewai/src/crewai/lite_agent_output.py +++ b/lib/crewai/src/crewai/lite_agent_output.py @@ -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 diff --git a/lib/crewai/src/crewai/llms/base_llm.py b/lib/crewai/src/crewai/llms/base_llm.py index d679bf670..ece84fde0 100644 --- a/lib/crewai/src/crewai/llms/base_llm.py +++ b/lib/crewai/src/crewai/llms/base_llm.py @@ -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) diff --git a/lib/crewai/src/crewai/types/usage_metrics.py b/lib/crewai/src/crewai/types/usage_metrics.py index aa1597229..891795d8a 100644 --- a/lib/crewai/src/crewai/types/usage_metrics.py +++ b/lib/crewai/src/crewai/types/usage_metrics.py @@ -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``. diff --git a/lib/crewai/tests/agents/test_lite_agent.py b/lib/crewai/tests/agents/test_lite_agent.py index a4f404c85..db3686a6a 100644 --- a/lib/crewai/tests/agents/test_lite_agent.py +++ b/lib/crewai/tests/agents/test_lite_agent.py @@ -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()