diff --git a/lib/crewai/src/crewai/telemetry/tracing/handlers.py b/lib/crewai/src/crewai/telemetry/tracing/handlers.py index 405e1f7ff..2c7003502 100644 --- a/lib/crewai/src/crewai/telemetry/tracing/handlers.py +++ b/lib/crewai/src/crewai/telemetry/tracing/handlers.py @@ -111,6 +111,7 @@ from crewai.events.types.skill_events import ( SkillLoadedEvent, SkillUsedEvent, ) +from crewai.tasks.output_format import OutputFormat from crewai.telemetry.tracing import semantic_conventions from crewai.telemetry.tracing.context import ( PendingSpanEnd, @@ -569,6 +570,24 @@ def _serialize(value: Any) -> str: return json.dumps(to_serializable(value)) +def _task_output_format(task: Any, output: Any = None) -> str: + """The output format ``task`` declared, as the ``OutputFormat`` value string. + + A finished task's ``TaskOutput`` records the format it was produced under, + so that is preferred when given. Before completion, and on failure, the + format is read off the declaration (``output_json`` / ``output_pydantic``) + with the precedence ``Task._get_output_format`` applies. + """ + declared = getattr(output, "output_format", None) + if declared is not None: + return str(getattr(declared, "value", declared)) + if getattr(task, "output_json", None) is not None: + return OutputFormat.JSON.value + if getattr(task, "output_pydantic", None) is not None: + return OutputFormat.PYDANTIC.value + return OutputFormat.RAW.value + + def _set_span_attributes(span: Span, attributes: dict[str, Any]) -> None: for key, value in attributes.items(): if value is not None: @@ -904,6 +923,7 @@ def handle_task_started( name=task.name, description=task.description, expected_output=task.expected_output, + output_format=_task_output_format(task), ), **semantic_conventions.gen_ai( operation_name=semantic_conventions.GEN_AI_OP_EXECUTE_TASK, @@ -936,6 +956,7 @@ def handle_task_completed( span = ( ctx.active_spans.get(event.started_event_id) if event.started_event_id else None ) + output = event.output attrs: dict[str, Any] = { **semantic_conventions.crewai_span( @@ -948,11 +969,14 @@ def handle_task_completed( key=task.key, id=str(task.id), name=task.name, - output=task.output.raw if task.output else None, - ), - **semantic_conventions.gen_ai_io( - output_value=task.output.raw if task.output else None + output=output.raw, + output_format=_task_output_format(task, output), + # The declaration says what was asked for; these say what the run + # actually yielded, so a consumer can tell the two apart. + output_pydantic_produced=output.pydantic is not None, + output_json_produced=output.json_dict is not None, ), + **semantic_conventions.gen_ai_io(output_value=output.raw), } providers.emit_log( @@ -989,6 +1013,7 @@ def handle_task_failed( name=task.name, description=task.description, expected_output=task.expected_output, + output_format=_task_output_format(task), ), } @@ -1090,6 +1115,7 @@ def handle_agent_execution_started( tool_definitions=agent.tools, system_instructions=agent.backstory, ), + **semantic_conventions.gen_ai_io(input_value=event.task_prompt), **semantic_conventions.crewai_span(event_name=event.type, subject=agent.role), **semantic_conventions.crewai_agent(role=agent.role), } @@ -1141,6 +1167,7 @@ def handle_agent_execution_completed( tool_definitions=agent.tools, system_instructions=agent.backstory, ), + **semantic_conventions.gen_ai_io(output_value=event.output), **semantic_conventions.crewai_span(event_name=event.type, subject=agent.role), **semantic_conventions.crewai_agent(role=agent.role), } @@ -1398,6 +1425,7 @@ def handle_tool_usage_finished( event_name=event.type, subject=event.tool_name ), **semantic_conventions.crewai_agent(key=event.agent_key, role=event.agent_role), + **semantic_conventions.crewai_tool(from_cache=event.from_cache), **_tool_failure_attrs(failure), } diff --git a/lib/crewai/src/crewai/telemetry/tracing/semantic_conventions.py b/lib/crewai/src/crewai/telemetry/tracing/semantic_conventions.py index ec83d714b..36172d72d 100644 --- a/lib/crewai/src/crewai/telemetry/tracing/semantic_conventions.py +++ b/lib/crewai/src/crewai/telemetry/tracing/semantic_conventions.py @@ -23,6 +23,7 @@ Custom conventions (crewai.* namespace): crewai_policy() crewai.policy.* crewai_memory() crewai.memory.* crewai_knowledge() crewai.knowledge.* + crewai_tool() crewai.tool.* crewai_llm() crewai.llm.* crewai_mcp() crewai.mcp.* crewai_a2a() crewai.a2a.* @@ -58,6 +59,17 @@ MCP_TOOL_EXECUTION_DURATION_MS = "crewai.mcp.tool_execution_duration_ms" HUMAN_FEEDBACK_WAIT_DURATION_MS = "crewai.human_feedback.wait_duration_ms" HUMAN_FEEDBACK_REQUEST_ID = "crewai.human_feedback.request_id" +# Attribute keys that let a consumer of the spans reason about a run without +# reconstructing it from raw text. +TASK_OUTPUT_FORMAT = "crewai.task.output_format" +"""The output format the task declared: ``json``, ``pydantic`` or ``raw``.""" +TASK_OUTPUT_PYDANTIC_PRODUCED = "crewai.task.output_pydantic_produced" +"""Whether the task's output actually carries a parsed Pydantic object.""" +TASK_OUTPUT_JSON_PRODUCED = "crewai.task.output_json_produced" +"""Whether the task's output actually carries a parsed JSON dict.""" +TOOL_FROM_CACHE = "crewai.tool.from_cache" +"""Whether the tool result came from the tool cache rather than a live run.""" + GEN_AI_OP_INVOKE_WORKFLOW = "invoke_workflow" GEN_AI_OP_EXECUTE_METHOD = "execute_method" GEN_AI_OP_EXECUTE_TASK = "execute_task" @@ -360,6 +372,9 @@ def crewai_task( description: str | None = None, expected_output: str | None = None, output: str | None = None, + output_format: str | None = None, + output_pydantic_produced: bool | None = None, + output_json_produced: bool | None = None, ) -> dict[str, Any]: return _filter_none( { @@ -369,6 +384,9 @@ def crewai_task( "crewai.task.description": description, "crewai.task.expected_output": expected_output, "crewai.task.output": output, + TASK_OUTPUT_FORMAT: output_format, + TASK_OUTPUT_PYDANTIC_PRODUCED: output_pydantic_produced, + TASK_OUTPUT_JSON_PRODUCED: output_json_produced, } ) @@ -537,6 +555,23 @@ def crewai_knowledge( ) +def crewai_tool( + *, + from_cache: bool | None = None, +) -> dict[str, Any]: + """Attributes of a tool call that completed, whatever it returned. + + ``from_cache`` says whether the result was served from the tool cache; a + cached call never ran the tool, so its duration and result mean something + different to a consumer than a live call's. + """ + return _filter_none( + { + TOOL_FROM_CACHE: from_cache, + } + ) + + def crewai_tool_failure( *, message: str | None = None, diff --git a/lib/crewai/tests/telemetry/test_span_task_agent_tool_attributes.py b/lib/crewai/tests/telemetry/test_span_task_agent_tool_attributes.py new file mode 100644 index 000000000..14ef8da7b --- /dev/null +++ b/lib/crewai/tests/telemetry/test_span_task_agent_tool_attributes.py @@ -0,0 +1,297 @@ +"""The execute task, execute agent and call tool spans carry what a reader needs. + +A consumer of a run's spans could see a task's raw text but not the format it +declared, nor whether a Pydantic object or a JSON dict actually came out of it; +it could see an agent's goal, backstory and model but not the prompt the agent +was handed or the answer it gave; and it could see a tool's result but not +whether the tool ran or the cache answered. Each had to be reconstructed from +other spans, or could not be known at all. +""" + +from __future__ import annotations + +from datetime import datetime, timezone +import json + +from crewai import Agent, Crew, Task +from crewai.events.types.agent_events import ( + AgentExecutionCompletedEvent, + AgentExecutionStartedEvent, +) +from crewai.events.types.task_events import ( + TaskCompletedEvent, + TaskFailedEvent, + TaskStartedEvent, +) +from crewai.events.types.tool_usage_events import ( + ToolUsageFinishedEvent, + ToolUsageStartedEvent, +) +from crewai.tasks.output_format import OutputFormat +from crewai.tasks.task_output import TaskOutput +from crewai.telemetry.tracing import handlers +from crewai.telemetry.tracing.context import TelemetryExecutionContext +from opentelemetry.sdk.trace import TracerProvider +from opentelemetry.sdk.trace.export import SimpleSpanProcessor +from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter +from pydantic import BaseModel +import pytest + + +# Long and non-repeating, so a truncated or elided copy cannot compare equal. +# Under the default 32 KiB attribute cap, so it must arrive whole. +LONG_TEXT = "".join( + f"paragraph {i}: the quick brown fox jumps over the lazy dog\n" for i in range(400) +) +assert 20_000 < len(LONG_TEXT.encode("utf-8")) < 32 * 1024 + + +@pytest.fixture(autouse=True) +def enable_otel_sdk(monkeypatch: pytest.MonkeyPatch) -> None: + """The suite otherwise runs with OTEL_SDK_DISABLED, which makes every + assertion here pass vacuously against non-recording spans.""" + monkeypatch.delenv("OTEL_SDK_DISABLED", raising=False) + monkeypatch.delenv("CREWAI_DISABLE_TELEMETRY", raising=False) + monkeypatch.delenv("CREWAI_DISABLE_TRACKING", raising=False) + + +class _Providers: + """All a handler asks of its providers: a tracer and somewhere to log.""" + + def __init__(self, tracer) -> None: + self._tracer = tracer + + def get_tracer(self, name: str | None = None): + return self._tracer + + def emit_log(self, *args, **kwargs) -> None: + pass + + +@pytest.fixture +def pipeline(): + """Handler inputs whose spans land in memory, exported as each one ends.""" + exporter = InMemorySpanExporter() + provider = TracerProvider() + provider.add_span_processor(SimpleSpanProcessor(exporter)) + tracer = provider.get_tracer("test") + ctx = TelemetryExecutionContext( + kickoff_id="kickoff", automation_name="test", tracer=tracer + ) + # yield, not return: the frame keeps `provider` alive, and a collected + # provider shuts down its processor and silently loses spans. + yield _Providers(tracer), ctx, exporter + + +def _only_span(exporter: InMemorySpanExporter, name: str): + """The single exported span with ``name``, failing loudly if not unique.""" + matches = [s for s in exporter.get_finished_spans() if s.name == name] + assert len(matches) == 1, [s.name for s in exporter.get_finished_spans()] + return matches[0] + + +class Finding(BaseModel): + title: str + + +def _crewed_task(**declaration) -> Task: + """A task whose agent belongs to a crew, as the task handlers require.""" + agent = Agent(role="Researcher", goal="g", backstory="b") + task = Task(description="d", expected_output="e", agent=agent, **declaration) + crew = Crew(agents=[agent], tasks=[task]) + # A kickoff attaches the crew to its agents; these events skip the kickoff. + agent.crew = crew + return task + + +@pytest.mark.parametrize( + ("declaration", "produced", "expected"), + [ + ( + {"output_pydantic": Finding}, + {"pydantic": Finding(title="t"), "output_format": OutputFormat.PYDANTIC}, + ("pydantic", True, False), + ), + ( + {"output_json": Finding}, + {"json_dict": {"title": "t"}, "output_format": OutputFormat.JSON}, + ("json", False, True), + ), + ( + {}, + {"output_format": OutputFormat.RAW}, + ("raw", False, False), + ), + ], + ids=["pydantic", "json", "raw"], +) +def test_execute_task_span_says_what_was_declared_and_what_came_out( + pipeline, declaration, produced, expected +) -> None: + providers, ctx, exporter = pipeline + task = _crewed_task(**declaration) + output = TaskOutput( + description="d", raw='{"title": "t"}', agent="Researcher", **produced + ) + + started = TaskStartedEvent(context=None, task=task) + handlers.handle_task_started(providers, ctx, task, started) + handlers.handle_task_completed( + providers, + ctx, + task, + TaskCompletedEvent(output=output, task=task, started_event_id=started.event_id), + ) + + span = _only_span(exporter, "execute task") + output_format, pydantic_produced, json_produced = expected + assert span.attributes["crewai.task.output_format"] == output_format + assert span.attributes["crewai.task.output_pydantic_produced"] is pydantic_produced + assert span.attributes["crewai.task.output_json_produced"] is json_produced + # The raw text is still there, as before. + assert span.attributes["crewai.task.output"] == '{"title": "t"}' + + +def test_a_declared_pydantic_task_that_produced_none_is_told_apart(pipeline) -> None: + """The gap the two flags close: declared is not the same as produced.""" + providers, ctx, exporter = pipeline + task = _crewed_task(output_pydantic=Finding) + # The producer could not convert, so the output carries only raw text. + output = TaskOutput( + description="d", + raw="not json at all", + agent="Researcher", + output_format=OutputFormat.PYDANTIC, + ) + + started = TaskStartedEvent(context=None, task=task) + handlers.handle_task_started(providers, ctx, task, started) + handlers.handle_task_completed( + providers, + ctx, + task, + TaskCompletedEvent(output=output, task=task, started_event_id=started.event_id), + ) + + span = _only_span(exporter, "execute task") + assert span.attributes["crewai.task.output_format"] == "pydantic" + assert span.attributes["crewai.task.output_pydantic_produced"] is False + assert span.attributes["crewai.task.output_json_produced"] is False + + +def test_a_failed_task_still_says_what_it_declared(pipeline) -> None: + """There is no output to inspect, so the format comes from the declaration.""" + providers, ctx, exporter = pipeline + task = _crewed_task(output_pydantic=Finding) + + started = TaskStartedEvent(context=None, task=task) + handlers.handle_task_started(providers, ctx, task, started) + handlers.handle_task_failed( + providers, + ctx, + task, + TaskFailedEvent(error="boom", task=task, started_event_id=started.event_id), + ) + + span = _only_span(exporter, "execute task") + assert span.attributes["crewai.task.output_format"] == "pydantic" + assert "crewai.task.output_pydantic_produced" not in span.attributes + assert "crewai.task.output_json_produced" not in span.attributes + + +def _agent_span(pipeline, *, prompt: str, output: str): + providers, ctx, exporter = pipeline + agent = Agent(role="Researcher", goal="g", backstory="b") + + started = AgentExecutionStartedEvent( + agent=agent, task=None, tools=None, task_prompt=prompt + ) + handlers.handle_agent_execution_started(providers, ctx, agent, started) + # The completed handler waits (up to 5 s) for the agent's LLM-call count to + # settle; one recorded call is what a real run would have signalled. + handlers._record_agent_llm_call(ctx, str(agent.id)) + handlers.handle_agent_execution_completed( + providers, + ctx, + agent, + AgentExecutionCompletedEvent( + agent=agent, task=None, output=output, started_event_id=started.event_id + ), + ) + return _only_span(exporter, "execute agent") + + +def test_execute_agent_span_carries_the_exact_prompt_and_answer(pipeline) -> None: + span = _agent_span(pipeline, prompt=LONG_TEXT, output="The fox is quick.") + + # The same spec shape the task span already uses for its own text. + prompt = json.loads(span.attributes["gen_ai.input.messages"]) + assert prompt[0]["parts"][0]["content"] == LONG_TEXT + assert "gen_ai.input.messages.truncated" not in span.attributes + answer = json.loads(span.attributes["gen_ai.output.messages"]) + assert answer[0]["parts"][0]["content"] == "The fox is quick." + assert "gen_ai.output.messages.truncated" not in span.attributes + + +def test_the_agent_text_travels_under_the_standard_keys_only(pipeline) -> None: + """The prompt and the answer leave under the two keys the ``call llm`` span + already uses for its messages, and under no other. Whatever rule an + exporter or a redaction processor applies to LLM message content applies + to these unchanged; a copy under a ``crewai.agent.*`` key would escape a + rule that selects by key name.""" + span = _agent_span(pipeline, prompt="PROMPT-7597", output="ANSWER-7597") + + carrying = { + key + for key, value in span.attributes.items() + if "PROMPT-7597" in str(value) or "ANSWER-7597" in str(value) + } + assert carrying == {"gen_ai.input.messages", "gen_ai.output.messages"} + + +def test_an_over_cap_prompt_is_declared_truncated_never_silently_cut( + pipeline, monkeypatch: pytest.MonkeyPatch +) -> None: + cap = 2048 + monkeypatch.setenv("CREWAI_OTEL_MAX_ATTR_BYTES", str(cap)) + + span = _agent_span(pipeline, prompt=LONG_TEXT, output="short") + + payload = span.attributes["gen_ai.input.messages"] + assert span.attributes["gen_ai.input.messages.truncated"] is True + assert span.attributes["gen_ai.input.messages.original_size_bytes"] > cap + assert len(payload.encode("utf-8")) <= cap + # The answer fit, so it arrives whole and unmarked. + answer = json.loads(span.attributes["gen_ai.output.messages"]) + assert answer[0]["parts"][0]["content"] == "short" + assert "gen_ai.output.messages.truncated" not in span.attributes + + +@pytest.mark.parametrize("from_cache", [True, False], ids=["cached", "ran"]) +def test_call_tool_span_says_whether_the_cache_answered(pipeline, from_cache) -> None: + providers, ctx, exporter = pipeline + now = datetime.now(timezone.utc) + + started = ToolUsageStartedEvent( + tool_name="search", tool_args={"q": "fox"}, agent_key="k", agent_role="r" + ) + handlers.handle_tool_usage_started(providers, ctx, None, started) + handlers.handle_tool_usage_finished( + providers, + ctx, + None, + ToolUsageFinishedEvent( + tool_name="search", + tool_args={"q": "fox"}, + agent_key="k", + agent_role="r", + started_at=now, + finished_at=now, + output="found", + from_cache=from_cache, + started_event_id=started.event_id, + ), + ) + + span = _only_span(exporter, "call tool") + assert span.attributes["crewai.tool.from_cache"] is from_cache