Tracing

any_agent.tracing.trace.AgentTrace

Bases: BaseModel

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

Source code in src/any_agent/tracing/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.trace.AgentSpan] that form the trace.
    """

    final_output: str | None = None
    """Contains the final output message returned by the agent.
    """

    model_config = ConfigDict(arbitrary_types_allowed=True)

    def get_total_cost(self) -> TotalTokenUseAndCost:
        """Return the current total cost and token usage statistics."""
        counts: list[CountInfo] = []
        costs: list[CostInfo] = []
        for span in self.spans:
            if span.attributes and "cost_prompt" in span.attributes:
                count = CountInfo(
                    token_count_prompt=span.attributes["llm.token_count.prompt"],
                    token_count_completion=span.attributes[
                        "llm.token_count.completion"
                    ],
                )
                cost = CostInfo(
                    cost_prompt=span.attributes["cost_prompt"],
                    cost_completion=span.attributes["cost_completion"],
                )
                counts.append(count)
                costs.append(cost)

        total_cost = sum(cost.cost_prompt + cost.cost_completion for cost in costs)
        total_tokens = sum(
            count.token_count_prompt + count.token_count_completion for count in counts
        )
        total_token_count_prompt = sum(count.token_count_prompt for count in counts)
        total_token_count_completion = sum(
            count.token_count_completion for count in counts
        )
        total_cost_prompt = sum(cost.cost_prompt for cost in costs)
        total_cost_completion = sum(cost.cost_completion for cost in costs)
        return TotalTokenUseAndCost(
            total_cost=total_cost,
            total_tokens=total_tokens,
            total_token_count_prompt=total_token_count_prompt,
            total_token_count_completion=total_token_count_completion,
            total_cost_prompt=total_cost_prompt,
            total_cost_completion=total_cost_completion,
        )

final_output = 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.

get_total_cost()

Return the current total cost and token usage statistics.

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

def get_total_cost(self) -> TotalTokenUseAndCost:
    """Return the current total cost and token usage statistics."""
    counts: list[CountInfo] = []
    costs: list[CostInfo] = []
    for span in self.spans:
        if span.attributes and "cost_prompt" in span.attributes:
            count = CountInfo(
                token_count_prompt=span.attributes["llm.token_count.prompt"],
                token_count_completion=span.attributes[
                    "llm.token_count.completion"
                ],
            )
            cost = CostInfo(
                cost_prompt=span.attributes["cost_prompt"],
                cost_completion=span.attributes["cost_completion"],
            )
            counts.append(count)
            costs.append(cost)

    total_cost = sum(cost.cost_prompt + cost.cost_completion for cost in costs)
    total_tokens = sum(
        count.token_count_prompt + count.token_count_completion for count in counts
    )
    total_token_count_prompt = sum(count.token_count_prompt for count in counts)
    total_token_count_completion = sum(
        count.token_count_completion for count in counts
    )
    total_cost_prompt = sum(cost.cost_prompt for cost in costs)
    total_cost_completion = sum(cost.cost_completion for cost in costs)
    return TotalTokenUseAndCost(
        total_cost=total_cost,
        total_tokens=total_tokens,
        total_token_count_prompt=total_token_count_prompt,
        total_token_count_completion=total_token_count_completion,
        total_cost_prompt=total_cost_prompt,
        total_cost_completion=total_cost_completion,
    )

any_agent.tracing.trace.AgentSpan

Bases: BaseModel

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

Source code in src/any_agent/tracing/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_readable_span(cls, readable_span: "ReadableSpan") -> "AgentSpan":
        """Create an AgentSpan from a ReadableSpan."""
        return cls(
            name=readable_span.name,
            kind=SpanKind.from_otel(readable_span.kind),
            parent=SpanContext.from_otel(readable_span.parent),
            start_time=readable_span.start_time,
            end_time=readable_span.end_time,
            status=Status.from_otel(readable_span.status),
            context=SpanContext.from_otel(readable_span.context),
            attributes=dict(readable_span.attributes)
            if readable_span.attributes
            else {},
            links=[Link.from_otel(link) for link in readable_span.links],
            events=[Event.from_otel(event) for event in readable_span.events],
            resource=Resource.from_otel(readable_span.resource),
        )

    def add_cost_info(self) -> None:
        """Extend attributes with `TokenUseAndCost`."""
        cost_info = extract_token_use_and_cost(self.attributes)
        if cost_info:
            self.set_attributes(cost_info.model_dump(exclude_none=True))

    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

add_cost_info()

Extend attributes with TokenUseAndCost.

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

def add_cost_info(self) -> None:
    """Extend attributes with `TokenUseAndCost`."""
    cost_info = extract_token_use_and_cost(self.attributes)
    if cost_info:
        self.set_attributes(cost_info.model_dump(exclude_none=True))

from_readable_span(readable_span) classmethod

Create an AgentSpan from a ReadableSpan.

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

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

set_attributes(attributes)

Set attributes for the span.

Source code in src/any_agent/tracing/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