Tracing

any_agent.tracing.agent_trace.AgentTrace

Bases: BaseModel

A trace that can be exported to JSON or printed to the console.

Source code in src/any_agent/tracing/agent_trace.py

class AgentTrace(BaseModel):
    """A trace that can be exported to JSON or printed to the console."""

    spans: list[AgentSpan] = Field(default_factory=list)
    """A list of [`AgentSpan`][any_agent.tracing.agent_trace.AgentSpan] that form the trace.
    """

    final_output: str | BaseModel | None = Field(default=None)
    """Contains the final output message returned by the agent.
    """

    model_config = ConfigDict(arbitrary_types_allowed=True)

    @field_serializer("final_output")
    def serialize_final_output(self, value: str | BaseModel | None) -> Any:
        """Serialize the final_output and handle any BaseModel subclass."""
        if value is None:
            return None
        if isinstance(value, str):
            return value
        if isinstance(value, BaseModel):
            # This will properly serialize any BaseModel subclass
            return value.model_dump()
        return value

    def _invalidate_tokens_and_cost_cache(self) -> None:
        """Clear the cached tokens_and_cost property if it exists."""
        if "tokens" in self.__dict__:
            del self.tokens
        if "cost" in self.__dict__:
            del self.cost

    def add_span(self, span: AgentSpan | Span) -> None:
        """Add an AgentSpan to the trace and clear the tokens_and_cost cache if present."""
        if not isinstance(span, AgentSpan):
            span = AgentSpan.from_otel(span)
        self.spans.append(span)
        self._invalidate_tokens_and_cost_cache()

    def add_spans(self, spans: list[AgentSpan]) -> None:
        """Add a list of AgentSpans to the trace and clear the tokens_and_cost cache if present."""
        self.spans.extend(spans)
        self._invalidate_tokens_and_cost_cache()

    def spans_to_messages(self) -> list[AgentMessage]:
        """Convert spans to standard message format.

        Returns:
            List of message dicts with 'role' and 'content' keys.

        """
        messages: list[AgentMessage] = []

        # Process spans in chronological order (excluding the final invoke_agent span)
        # Filter out any agent invocation spans
        filtered_spans: list[AgentSpan] = []
        for span in self.spans:
            if not span.is_agent_invocation():
                filtered_spans.append(span)

        for span in filtered_spans:
            if span.is_llm_call():
                # Extract input messages from the span
                input_messages = span.get_input_messages()
                if input_messages:
                    for msg in input_messages:
                        if not any(
                            existing.role == msg.role
                            and existing.content == msg.content
                            for existing in messages
                        ):
                            messages.append(msg)

                # Add the assistant's response
                output_content = span.get_output_content()
                if output_content:
                    # Avoid duplicate assistant messages
                    if not (
                        messages
                        and messages[-1].role == "assistant"
                        and messages[-1].content == output_content
                    ):
                        messages.append(
                            AgentMessage(role="assistant", content=output_content)
                        )

            elif span.is_tool_execution():
                # For tool executions, include the result in the conversation
                output_content = span.get_output_content()
                if output_content:
                    tool_name = span.attributes["gen_ai.tool.name"]
                    tool_args = span.attributes["gen_ai.tool.args"]
                    messages.append(
                        AgentMessage(
                            role="assistant",
                            content=f"[Tool {tool_name} executed: {output_content} with args: {tool_args}]",
                        )
                    )

        return messages

    @property
    def duration(self) -> timedelta:
        """Duration of the parent `invoke_agent` span as a datetime.timedelta object.

        The duration is computed from the span's start and end time (in nanoseconds).

        Raises ValueError if:
            - There are no spans.
            - The invoke_agent span is not the last span.
            - Any of the start/end times are missing.
        """
        if not self.spans:
            msg = "No spans found in trace"
            raise ValueError(msg)
        span = self.spans[-1]
        if not span.is_agent_invocation():
            msg = "Last span is not `invoke_agent`"
            raise ValueError(msg)
        if span.start_time is not None and span.end_time is not None:
            duration_ns = span.end_time - span.start_time
            return timedelta(seconds=duration_ns / 1_000_000_000)
        msg = "Start or end time is missing for the `invoke_agent` span"
        raise ValueError(msg)

    @cached_property
    def tokens(self) -> TokenInfo:
        """The current total token count for this trace. Cached after first computation."""
        sum_input_tokens = 0
        sum_output_tokens = 0
        for span in self.spans:
            if span.is_llm_call():
                sum_input_tokens += span.attributes.get("gen_ai.usage.input_tokens", 0)
                sum_output_tokens += span.attributes.get(
                    "gen_ai.usage.output_tokens", 0
                )
        return TokenInfo(input_tokens=sum_input_tokens, output_tokens=sum_output_tokens)

    @cached_property
    def cost(self) -> CostInfo:
        """The current total cost for this trace. Cached after first computation."""
        sum_input_cost = 0.0
        sum_output_cost = 0.0
        for span in self.spans:
            if span.is_llm_call():
                cost_info = compute_cost_info(span.attributes)
                if cost_info:
                    sum_input_cost += cost_info.input_cost
                    sum_output_cost += cost_info.output_cost
        return CostInfo(input_cost=sum_input_cost, output_cost=sum_output_cost)

cost cached property

The current total cost for this trace. Cached after first computation.

duration property

Duration of the parent invoke_agent span as a datetime.timedelta object.

The duration is computed from the span's start and end time (in nanoseconds).

Raises ValueError if

• There are no spans.
• The invoke_agent span is not the last span.
• Any of the start/end times are missing.

final_output = Field(default=None) class-attribute instance-attribute

Contains the final output message returned by the agent.

spans = Field(default_factory=list) class-attribute instance-attribute

A list of AgentSpan that form the trace.

tokens cached property

The current total token count for this trace. Cached after first computation.

add_span(span)

Add an AgentSpan to the trace and clear the tokens_and_cost cache if present.

Source code in src/any_agent/tracing/agent_trace.py

def add_span(self, span: AgentSpan | Span) -> None:
    """Add an AgentSpan to the trace and clear the tokens_and_cost cache if present."""
    if not isinstance(span, AgentSpan):
        span = AgentSpan.from_otel(span)
    self.spans.append(span)
    self._invalidate_tokens_and_cost_cache()

add_spans(spans)

Add a list of AgentSpans to the trace and clear the tokens_and_cost cache if present.

Source code in src/any_agent/tracing/agent_trace.py

def add_spans(self, spans: list[AgentSpan]) -> None:
    """Add a list of AgentSpans to the trace and clear the tokens_and_cost cache if present."""
    self.spans.extend(spans)
    self._invalidate_tokens_and_cost_cache()

serialize_final_output(value)

Serialize the final_output and handle any BaseModel subclass.

Source code in src/any_agent/tracing/agent_trace.py

@field_serializer("final_output")
def serialize_final_output(self, value: str | BaseModel | None) -> Any:
    """Serialize the final_output and handle any BaseModel subclass."""
    if value is None:
        return None
    if isinstance(value, str):
        return value
    if isinstance(value, BaseModel):
        # This will properly serialize any BaseModel subclass
        return value.model_dump()
    return value

spans_to_messages()

Convert spans to standard message format.

Returns:

Type	Description
list[AgentMessage]	List of message dicts with 'role' and 'content' keys.

Source code in src/any_agent/tracing/agent_trace.py

def spans_to_messages(self) -> list[AgentMessage]:
    """Convert spans to standard message format.

    Returns:
        List of message dicts with 'role' and 'content' keys.

    """
    messages: list[AgentMessage] = []

    # Process spans in chronological order (excluding the final invoke_agent span)
    # Filter out any agent invocation spans
    filtered_spans: list[AgentSpan] = []
    for span in self.spans:
        if not span.is_agent_invocation():
            filtered_spans.append(span)

    for span in filtered_spans:
        if span.is_llm_call():
            # Extract input messages from the span
            input_messages = span.get_input_messages()
            if input_messages:
                for msg in input_messages:
                    if not any(
                        existing.role == msg.role
                        and existing.content == msg.content
                        for existing in messages
                    ):
                        messages.append(msg)

            # Add the assistant's response
            output_content = span.get_output_content()
            if output_content:
                # Avoid duplicate assistant messages
                if not (
                    messages
                    and messages[-1].role == "assistant"
                    and messages[-1].content == output_content
                ):
                    messages.append(
                        AgentMessage(role="assistant", content=output_content)
                    )

        elif span.is_tool_execution():
            # For tool executions, include the result in the conversation
            output_content = span.get_output_content()
            if output_content:
                tool_name = span.attributes["gen_ai.tool.name"]
                tool_args = span.attributes["gen_ai.tool.args"]
                messages.append(
                    AgentMessage(
                        role="assistant",
                        content=f"[Tool {tool_name} executed: {output_content} with args: {tool_args}]",
                    )
                )

    return messages

any_agent.tracing.agent_trace.AgentSpan

Bases: BaseModel

A span that can be exported to JSON or printed to the console.

Source code in src/any_agent/tracing/agent_trace.py

class AgentSpan(BaseModel):
    """A span that can be exported to JSON or printed to the console."""

    name: str
    kind: SpanKind
    parent: SpanContext | None = None
    start_time: int | None = None
    end_time: int | None = None
    status: Status
    context: SpanContext
    attributes: dict[str, Any]
    links: list[Link]
    events: list[Event]
    resource: Resource

    model_config = ConfigDict(arbitrary_types_allowed=False)

    @classmethod
    def from_otel(cls, otel_span: Span) -> AgentSpan:
        """Create an AgentSpan from an OTEL Span."""
        return cls(
            name=otel_span.name,
            kind=SpanKind.from_otel(otel_span.kind),
            parent=SpanContext.from_otel(otel_span.parent),
            start_time=otel_span.start_time,
            end_time=otel_span.end_time,
            status=Status.from_otel(otel_span.status),
            context=SpanContext.from_otel(otel_span.context),
            attributes=dict(otel_span.attributes) if otel_span.attributes else {},
            links=[Link.from_otel(link) for link in otel_span.links],
            events=[Event.from_otel(event) for event in otel_span.events],
            resource=Resource.from_otel(otel_span.resource),
        )

    def to_readable_span(self) -> ReadableSpan:
        """Create an ReadableSpan from the AgentSpan."""
        return ReadableSpan(
            name=self.name,
            kind=self.kind,
            parent=self.parent,
            start_time=self.start_time,
            end_time=self.end_time,
            status=self.status,
            context=self.context,
            attributes=self.attributes,
            links=self.links,
            events=self.events,
            resource=self.resource,
        )

    def set_attributes(self, attributes: Mapping[str, AttributeValue]) -> None:
        """Set attributes for the span."""
        for key, value in attributes.items():
            if key in self.attributes:
                logger.warning("Overwriting attribute %s with %s", key, value)
            self.attributes[key] = value

    def is_agent_invocation(self) -> bool:
        """Check whether this span is an agent invocation (the very first span)."""
        return self.attributes.get("gen_ai.operation.name") == "invoke_agent"

    def is_llm_call(self) -> bool:
        """Check whether this span is a call to an LLM."""
        return self.attributes.get("gen_ai.operation.name") == "call_llm"

    def is_tool_execution(self) -> bool:
        """Check whether this span is an execution of a tool."""
        return self.attributes.get("gen_ai.operation.name") == "execute_tool"

    def get_input_messages(self) -> list[AgentMessage] | None:
        """Extract input messages from an LLM call span.

        Returns:
            List of message dicts with 'role' and 'content' keys, or None if not available.

        """
        if not self.is_llm_call():
            msg = "Span is not an LLM call"
            raise ValueError(msg)

        messages_json = self.attributes.get("gen_ai.input.messages")
        if not messages_json:
            logger.debug("No input messages found in span")
            return None

        try:
            parsed_messages = json.loads(messages_json)
            # Ensure it's a list of dicts
        except (json.JSONDecodeError, TypeError) as e:
            msg = "Failed to parse input messages from span"
            logger.error(msg)
            raise ValueError(msg) from e
        if not isinstance(parsed_messages, list):
            msg = "Input messages are not a list of messages"
            raise ValueError(msg)
        return [AgentMessage.model_validate(msg) for msg in parsed_messages]

    def get_output_content(self) -> str | None:
        """Extract output content from an LLM call or tool execution span.

        Returns:
            The output content as a string, or None if not available.

        """
        if not self.is_llm_call() and not self.is_tool_execution():
            msg = "Span is not an LLM call or tool execution"
            raise ValueError(msg)

        output = self.attributes.get("gen_ai.output")
        if not output:
            logger.debug("No output found in span")
            return None
        return str(output)

from_otel(otel_span) classmethod

Create an AgentSpan from an OTEL Span.

Source code in src/any_agent/tracing/agent_trace.py

@classmethod
def from_otel(cls, otel_span: Span) -> AgentSpan:
    """Create an AgentSpan from an OTEL Span."""
    return cls(
        name=otel_span.name,
        kind=SpanKind.from_otel(otel_span.kind),
        parent=SpanContext.from_otel(otel_span.parent),
        start_time=otel_span.start_time,
        end_time=otel_span.end_time,
        status=Status.from_otel(otel_span.status),
        context=SpanContext.from_otel(otel_span.context),
        attributes=dict(otel_span.attributes) if otel_span.attributes else {},
        links=[Link.from_otel(link) for link in otel_span.links],
        events=[Event.from_otel(event) for event in otel_span.events],
        resource=Resource.from_otel(otel_span.resource),
    )

get_input_messages()

Extract input messages from an LLM call span.

Returns:

Type	Description
list[AgentMessage] | None	List of message dicts with 'role' and 'content' keys, or None if not available.

Source code in src/any_agent/tracing/agent_trace.py

def get_input_messages(self) -> list[AgentMessage] | None:
    """Extract input messages from an LLM call span.

    Returns:
        List of message dicts with 'role' and 'content' keys, or None if not available.

    """
    if not self.is_llm_call():
        msg = "Span is not an LLM call"
        raise ValueError(msg)

    messages_json = self.attributes.get("gen_ai.input.messages")
    if not messages_json:
        logger.debug("No input messages found in span")
        return None

    try:
        parsed_messages = json.loads(messages_json)
        # Ensure it's a list of dicts
    except (json.JSONDecodeError, TypeError) as e:
        msg = "Failed to parse input messages from span"
        logger.error(msg)
        raise ValueError(msg) from e
    if not isinstance(parsed_messages, list):
        msg = "Input messages are not a list of messages"
        raise ValueError(msg)
    return [AgentMessage.model_validate(msg) for msg in parsed_messages]

get_output_content()

Extract output content from an LLM call or tool execution span.

Returns:

Type	Description
str | None	The output content as a string, or None if not available.

Source code in src/any_agent/tracing/agent_trace.py

def get_output_content(self) -> str | None:
    """Extract output content from an LLM call or tool execution span.

    Returns:
        The output content as a string, or None if not available.

    """
    if not self.is_llm_call() and not self.is_tool_execution():
        msg = "Span is not an LLM call or tool execution"
        raise ValueError(msg)

    output = self.attributes.get("gen_ai.output")
    if not output:
        logger.debug("No output found in span")
        return None
    return str(output)

is_agent_invocation()

Check whether this span is an agent invocation (the very first span).

Source code in src/any_agent/tracing/agent_trace.py

def is_agent_invocation(self) -> bool:
    """Check whether this span is an agent invocation (the very first span)."""
    return self.attributes.get("gen_ai.operation.name") == "invoke_agent"

is_llm_call()

Check whether this span is a call to an LLM.

Source code in src/any_agent/tracing/agent_trace.py

def is_llm_call(self) -> bool:
    """Check whether this span is a call to an LLM."""
    return self.attributes.get("gen_ai.operation.name") == "call_llm"

is_tool_execution()

Check whether this span is an execution of a tool.

Source code in src/any_agent/tracing/agent_trace.py

def is_tool_execution(self) -> bool:
    """Check whether this span is an execution of a tool."""
    return self.attributes.get("gen_ai.operation.name") == "execute_tool"

set_attributes(attributes)

Set attributes for the span.

Source code in src/any_agent/tracing/agent_trace.py

def set_attributes(self, attributes: Mapping[str, AttributeValue]) -> None:
    """Set attributes for the span."""
    for key, value in attributes.items():
        if key in self.attributes:
            logger.warning("Overwriting attribute %s with %s", key, value)
        self.attributes[key] = value

to_readable_span()

Create an ReadableSpan from the AgentSpan.

Source code in src/any_agent/tracing/agent_trace.py

def to_readable_span(self) -> ReadableSpan:
    """Create an ReadableSpan from the AgentSpan."""
    return ReadableSpan(
        name=self.name,
        kind=self.kind,
        parent=self.parent,
        start_time=self.start_time,
        end_time=self.end_time,
        status=self.status,
        context=self.context,
        attributes=self.attributes,
        links=self.links,
        events=self.events,
        resource=self.resource,
    )

any_agent.tracing.disable_console_traces()

Disable printing traces to the console.

Source code in src/any_agent/tracing/__init__.py

def disable_console_traces() -> None:
    """Disable printing traces to the console."""
    with TRACE_PROVIDER._active_span_processor._lock:
        TRACE_PROVIDER._active_span_processor._span_processors = tuple(
            p
            for p in TRACE_PROVIDER._active_span_processor._span_processors
            if not isinstance(getattr(p, "span_exporter", None), _ConsoleExporter)
        )

any_agent.tracing.enable_console_traces()

Enable printing traces to the console.

Source code in src/any_agent/tracing/__init__.py

def enable_console_traces() -> None:
    """Enable printing traces to the console."""
    has_console_exporter = any(
        isinstance(getattr(p, "span_exporter", None), _ConsoleExporter)
        for p in TRACE_PROVIDER._active_span_processor._span_processors
    )
    if not has_console_exporter:
        TRACE_PROVIDER.add_span_processor(SimpleSpanProcessor(_ConsoleExporter()))