diff --git a/py/src/braintrust/integrations/crewai/test_crewai.py b/py/src/braintrust/integrations/crewai/test_crewai.py index 86f4d4a7..50c798e5 100644 --- a/py/src/braintrust/integrations/crewai/test_crewai.py +++ b/py/src/braintrust/integrations/crewai/test_crewai.py @@ -71,11 +71,6 @@ def _reset_listener(): """Force a clean CrewAI listener for each test. The module-level ``_LISTENER`` singleton leaks between tests otherwise. - We deliberately do *not* touch LiteLLM patch markers here: stripping them - would confuse ``is_patched`` and let a subsequent ``patch_litellm()`` - double-wrap the module. The leaf-only token-metric tests below instead - monkeypatch ``is_litellm_patched`` directly when they need to pretend - LiteLLM is (or is not) patched. """ _reset_for_testing() yield @@ -250,8 +245,22 @@ def test_kickoff_llm_event_tree_parents_and_shape(memory_logger): kickoff = _build_kickoff_started() _emit(kickoff) + + # A real CrewAI LLM is needed as ``source`` so ``metadata.provider`` + # gets populated from ``LLM.provider``. + from crewai import LLM + from crewai.events.event_bus import crewai_event_bus + + llm_source = LLM(model="gpt-4o-mini") llm_started = _build_llm_started(parent_event_id=kickoff.event_id) - _emit(llm_started) + future = crewai_event_bus.emit(llm_source, llm_started) + if future is not None: + try: + future.result(timeout=5.0) + except Exception: + pass + _flush_event_bus(crewai_event_bus, timeout=5.0) + _emit(_build_llm_completed(llm_started, usage={"prompt_tokens": 3, "completion_tokens": 5, "total_tokens": 8})) _emit(_build_kickoff_completed(kickoff)) @@ -277,20 +286,33 @@ def test_kickoff_llm_event_tree_parents_and_shape(memory_logger): # Shape assertions. assert llm_span["input"]["messages"] == llm_started.messages assert llm_span["metadata"]["model"] == "gpt-4o-mini" + assert llm_span["metadata"]["provider"] == "openai" assert llm_span["metadata"]["call_id"] == "call-1" assert llm_span["output"] == "4" assert kickoff_span["output"] == "final answer" assert kickoff_span["input"] == kickoff.inputs -def test_llm_tokens_skipped_when_litellm_patched(memory_logger, monkeypatch): - """Leaf-only rule: LiteLLM patched -> no token metrics on crewai.llm.""" - # Pretend LiteLLM is patched without actually wrapping the module, so - # tests later in the suite that rely on a clean LiteLLM still see it. - monkeypatch.setattr( - "braintrust.integrations.crewai.tracing._litellm_owns_leaf_span", - lambda: True, - ) +def test_llm_tools_route_to_metadata_not_input(memory_logger): + """Tool definitions belong in ``metadata.tools`` per the spec, not in ``input``.""" + patch_crewai() + + tools = [ + {"type": "function", "function": {"name": "search", "description": "search the web"}}, + {"type": "function", "function": {"name": "sum", "description": "add two numbers"}}, + ] + started = _build_llm_started(tools=tools) + _emit(started) + _emit(_build_llm_completed(started)) + + span = memory_logger.pop()[0] + assert span["span_attributes"]["name"] == "crewai.llm" + assert span["metadata"]["tools"] == tools + assert "tools" not in span["input"], span["input"] + + +def test_llm_never_emits_token_metrics(memory_logger): + """crewai.llm is a wrapper span; the leaf provider owns tokens (see module docstring).""" patch_crewai() llm_started = _build_llm_started() @@ -308,31 +330,14 @@ def test_llm_tokens_skipped_when_litellm_patched(memory_logger, monkeypatch): metrics = by_name["crewai.llm"][0]["metrics"] assert "start" in metrics and "end" in metrics - for token_key in ("tokens", "prompt_tokens", "completion_tokens"): - assert token_key not in metrics, f"crewai.llm leaked {token_key}={metrics[token_key]} while litellm is patched" - - -def test_llm_tokens_emitted_when_litellm_not_patched(memory_logger, monkeypatch): - """Leaf: LiteLLM unpatched -> crewai.llm owns token metrics.""" - monkeypatch.setattr( - "braintrust.integrations.crewai.tracing._litellm_owns_leaf_span", - lambda: False, - ) - patch_crewai() - - llm_started = _build_llm_started() - _emit(llm_started) - _emit( - _build_llm_completed( - llm_started, - usage={"prompt_tokens": 11, "completion_tokens": 22, "total_tokens": 33}, - ) - ) - - metrics = memory_logger.pop()[0]["metrics"] - assert metrics["prompt_tokens"] == 11 - assert metrics["completion_tokens"] == 22 - assert metrics["tokens"] == 33 + for token_key in ( + "tokens", + "prompt_tokens", + "completion_tokens", + "prompt_cached_tokens", + "completion_reasoning_tokens", + ): + assert token_key not in metrics, f"crewai.llm leaked {token_key}={metrics.get(token_key)}" def test_llm_call_failed_logs_error(memory_logger): diff --git a/py/src/braintrust/integrations/crewai/tracing.py b/py/src/braintrust/integrations/crewai/tracing.py index baf4b046..48dff638 100644 --- a/py/src/braintrust/integrations/crewai/tracing.py +++ b/py/src/braintrust/integrations/crewai/tracing.py @@ -20,20 +20,21 @@ end events for the same scope can race across threads. A listener-level :class:`threading.Lock` guards the ``event_id -> Span`` map. -Token-metric rule (leaf-only): - -CrewAI delegates LLM calls to LiteLLM. When the LiteLLM integration is -also patched, the ``Completion`` span LiteLLM produces is the leaf span -and already owns token accounting. Emitting tokens on the enclosing -``crewai.llm`` span in that configuration would make trace-tree rollup -double-count the same tokens at every ancestor. - -We therefore follow the same pattern as the pydantic_ai integration -(see ``_wrapper_span_metrics`` in ``pydantic_ai/tracing.py``): -``crewai.llm`` emits timing + ``time_to_first_token`` unconditionally, and -token metrics only when LiteLLM is *not* patched. The check happens at -log time via the internal ``_is_litellm_patched`` helper on the LiteLLM -integration. +Token-metric rule (wrapper-only): + +CrewAI is always an orchestration layer: every LLM call it emits is +routed through an underlying provider SDK (LiteLLM, or a native +``openai`` / ``anthropic`` / ``bedrock`` / ``google-genai`` client). +Whichever of those the user has patched — commonly all of them via +``auto_instrument()`` — owns the leaf span with real token accounting. +If ``crewai.llm`` also logged tokens, trace-tree rollup would +double-count them at every ancestor. + +We therefore follow the same rule ``pydantic_ai`` uses for its wrapper +spans (see ``_wrapper_span_metrics`` in ``pydantic_ai/tracing.py``): +``crewai.llm`` emits timing + ``time_to_first_token`` only. Tokens live +on the leaf provider span. This is the "clear ownership" pattern the +sdk-integrations SKILL prefers over "if X is patched, skip" checks. """ # pylint: disable=import-error @@ -42,11 +43,7 @@ import time from typing import TYPE_CHECKING, Any -from braintrust.integrations.utils import ( - _normalize_chat_messages, - _parse_openai_usage_metrics, - _try_to_dict, -) +from braintrust.integrations.utils import _normalize_chat_messages from braintrust.logger import NOOP_SPAN, Span, current_span from braintrust.logger import start_span as _bt_start_span @@ -68,24 +65,6 @@ def start_span(*args, **kwargs): from crewai.events.event_bus import CrewAIEventsBus -# LiteLLM / LiteLLM-over-OpenAI usage field translation. Mirrors the -# mapping in ``braintrust.integrations.litellm.tracing`` so ``crewai.llm`` -# spans use identical metric names when LiteLLM is not the leaf. -_TOKEN_NAME_MAP: dict[str, str] = { - "total_tokens": "tokens", - "prompt_tokens": "prompt_tokens", - "completion_tokens": "completion_tokens", - "tokens": "tokens", - "input_tokens": "prompt_tokens", - "output_tokens": "completion_tokens", -} - -_TOKEN_PREFIX_MAP: dict[str, str] = { - "input": "prompt", - "output": "completion", -} - - # Provider parameters we want to surface on `crewai.llm` metadata. _LLM_CONFIG_FIELDS: tuple[str, ...] = ( "temperature", @@ -106,21 +85,6 @@ def start_span(*args, **kwargs): ) -def _litellm_owns_leaf_span() -> bool: - """Return True when LiteLLM's completion entry points have been patched. - - Late-imported so that importing the CrewAI integration never forces a - LiteLLM import. Errors are swallowed so a broken LiteLLM install - never prevents CrewAI tracing from running. - """ - try: - from braintrust.integrations.litellm import _is_litellm_patched - - return _is_litellm_patched() - except Exception: - return False - - # --------------------------------------------------------------------------- # Payload normalization helpers # --------------------------------------------------------------------------- @@ -137,7 +101,9 @@ def _agent_metadata(agent: Any) -> dict[str, Any]: meta[f"agent_{attr}"] = str(value) if attr == "id" else value llm = getattr(agent, "llm", None) if llm is not None: - meta["agent_llm"] = getattr(llm, "model", None) or str(llm) + model_name = getattr(llm, "model", None) + if model_name: + meta["agent_llm"] = model_name return meta @@ -198,32 +164,14 @@ def _llm_config_metadata(source: Any) -> dict[str, Any]: return meta -def _normalize_tools(tools: Any) -> Any: - """Normalize a list of CrewAI tool descriptors for logging.""" - if tools is None: - return None - if isinstance(tools, (list, tuple)): - out = [] - for tool in tools: - if isinstance(tool, dict): - out.append(tool) - else: - out.append(_try_to_dict(tool)) - return out - return _try_to_dict(tools) - - -def _normalize_output(value: Any) -> Any: - """Coerce provider-owned output objects into plain dicts/strings for logging.""" - if value is None: +def _provider_from_source(source: Any) -> str | None: + """Return the ``provider`` string off the emitting CrewAI ``LLM`` object.""" + if source is None: return None - if isinstance(value, (str, int, float, bool, dict, list)): - return value - coerced = _try_to_dict(value) - if coerced is value and not isinstance(value, (str, int, float, bool, dict, list)): - # Last-resort: render as a string rather than log a raw SDK object. - return str(value) - return coerced + provider = getattr(source, "provider", None) + if isinstance(provider, str) and provider: + return provider + return None # --------------------------------------------------------------------------- @@ -366,7 +314,7 @@ def on_crew_kickoff_started(source: Any, event: CrewKickoffStartedEvent) -> None @crewai_event_bus.on(CrewKickoffCompletedEvent) def on_crew_kickoff_completed(_source: Any, event: CrewKickoffCompletedEvent) -> None: - self._end_span(event, output=_normalize_output(getattr(event, "output", None))) + self._end_span(event, output=getattr(event, "output", None)) # Kickoff is the outermost scope; drop any orphan entries left over # when an inner end-event was never delivered so state does not grow # unbounded in long-running services. @@ -399,7 +347,7 @@ def on_task_started(source: Any, event: TaskStartedEvent) -> None: @crewai_event_bus.on(TaskCompletedEvent) def on_task_completed(_source: Any, event: TaskCompletedEvent) -> None: - self._end_span(event, output=_normalize_output(getattr(event, "output", None))) + self._end_span(event, output=getattr(event, "output", None)) @crewai_event_bus.on(TaskFailedEvent) def on_task_failed(_source: Any, event: TaskFailedEvent) -> None: @@ -418,7 +366,9 @@ def on_agent_started(_source: Any, event: AgentExecutionStartedEvent) -> None: **_task_metadata(task), **_causal_metadata(event), } - metadata["tools"] = _normalize_tools(getattr(event, "tools", None)) + agent_tools = getattr(event, "tools", None) + if agent_tools: + metadata["tools"] = agent_tools self._open_span( event, name="crewai.agent", @@ -447,18 +397,20 @@ def on_llm_started(source: Any, event: LLMCallStartedEvent) -> None: } if getattr(event, "model", None): metadata["model"] = event.model + provider = _provider_from_source(source) + if provider: + metadata["provider"] = provider if getattr(event, "call_id", None): metadata["call_id"] = event.call_id available_functions = getattr(event, "available_functions", None) if available_functions: metadata["available_functions"] = list(available_functions) - span_input: dict[str, Any] = { - "messages": getattr(event, "messages", None), - } tools = getattr(event, "tools", None) if tools: - span_input["tools"] = _normalize_tools(tools) - span_input["messages"] = _normalize_chat_messages(span_input["messages"]) + metadata["tools"] = tools + span_input: dict[str, Any] = { + "messages": _normalize_chat_messages(getattr(event, "messages", None)), + } self._open_span( event, name="crewai.llm", @@ -482,24 +434,10 @@ def on_llm_completed(_source: Any, event: LLMCallCompletedEvent) -> None: if getattr(event, "model", None): metadata["model"] = event.model - # Timing metrics are always safe. Token metrics are only emitted - # when LiteLLM is NOT patching the completion entry points, to - # avoid double-counting with the downstream ``Completion`` span. - extra_metrics: dict[str, Any] = {} - if not _litellm_owns_leaf_span(): - usage = getattr(event, "usage", None) - if usage is not None: - extra_metrics = _parse_openai_usage_metrics( - usage, - token_name_map=_TOKEN_NAME_MAP, - token_prefix_map=_TOKEN_PREFIX_MAP, - ) - self._end_span( event, - output=_normalize_output(getattr(event, "response", None)), + output=getattr(event, "response", None), metadata=metadata or None, - extra_metrics=extra_metrics, ) @crewai_event_bus.on(LLMCallFailedEvent) @@ -540,7 +478,7 @@ def on_tool_finished(_source: Any, event: ToolUsageFinishedEvent) -> None: extra_metadata["run_attempts"] = run_attempts self._end_span( event, - output=_normalize_output(getattr(event, "output", None)), + output=getattr(event, "output", None), metadata=extra_metadata or None, ) @@ -648,7 +586,6 @@ def _end_span( output: Any = None, error: Any = None, metadata: dict[str, Any] | None = None, - extra_metrics: dict[str, Any] | None = None, ) -> None: """Close the span opened by the matching start event. @@ -659,9 +596,7 @@ def _end_span( started_event_id = getattr(event, "started_event_id", None) or getattr(event, "event_id", None) if not started_event_id: return - self._end_span_by_event_id( - started_event_id, output=output, error=error, metadata=metadata, extra_metrics=extra_metrics - ) + self._end_span_by_event_id(started_event_id, output=output, error=error, metadata=metadata) def _end_span_by_event_id( @@ -671,7 +606,6 @@ def _end_span_by_event_id( output: Any = None, error: Any = None, metadata: dict[str, Any] | None = None, - extra_metrics: dict[str, Any] | None = None, ) -> None: with self._lock: span = self._spans.pop(started_event_id, None) @@ -684,8 +618,6 @@ def _end_span_by_event_id( metrics: dict[str, Any] = {"end": end_time} if first_token_time is not None and start_time is not None: metrics["time_to_first_token"] = first_token_time - start_time - if extra_metrics: - metrics.update(extra_metrics) log_payload: dict[str, Any] = {"metrics": metrics} if output is not None: