pydantic_ai.usage

RequestUsage dataclass

Bases: UsageBase

LLM usage associated with a single request.

This is an implementation of genai_prices.types.AbstractUsage so it can be used to calculate the price of the request using genai-prices.

Source code in pydantic_ai_slim/pydantic_ai/usage.py

@dataclass(repr=False, kw_only=True)
class RequestUsage(UsageBase):
    """LLM usage associated with a single request.

    This is an implementation of `genai_prices.types.AbstractUsage` so it can be used to calculate the price of the
    request using [genai-prices](https://github.com/pydantic/genai-prices).
    """

    @property
    def requests(self):
        return 1

    def incr(self, incr_usage: RequestUsage) -> None:
        """Increment the usage in place.

        Args:
            incr_usage: The usage to increment by.
        """
        return _incr_usage_tokens(self, incr_usage)

    def __add__(self, other: RequestUsage) -> RequestUsage:
        """Add two RequestUsages together.

        This is provided so it's trivial to sum usage information from multiple parts of a response.

        **WARNING:** this CANNOT be used to sum multiple requests without breaking some pricing calculations.
        """
        new_usage = copy(self)
        new_usage.incr(other)
        return new_usage

    @classmethod
    def extract(
        cls,
        data: Any,
        *,
        provider: str,
        provider_url: str,
        provider_fallback: str,
        api_flavor: str = 'default',
        details: dict[str, Any] | None = None,
    ) -> RequestUsage:
        """Extract usage information from the response data using genai-prices.

        Args:
            data: The response data from the model API.
            provider: The actual provider ID
            provider_url: The provider base_url
            provider_fallback: The fallback provider ID to use if the actual provider is not found in genai-prices.
                For example, an OpenAI model should set this to "openai" in case it has an obscure provider ID.
            api_flavor: The API flavor to use when extracting usage information,
                e.g. 'chat' or 'responses' for OpenAI.
            details: Becomes the `details` field on the returned `RequestUsage` for convenience.
        """
        details = details or {}
        for provider_id, provider_api_url in [(None, provider_url), (provider, None), (provider_fallback, None)]:
            try:
                provider_obj = get_snapshot().find_provider(None, provider_id, provider_api_url)
                _model_ref, extracted_usage = provider_obj.extract_usage(data, api_flavor=api_flavor)
                return cls(**{k: v for k, v in extracted_usage.__dict__.items() if v is not None}, details=details)
            except Exception:
                pass
        return cls(details=details)

incr

incr(incr_usage: RequestUsage) -> None

Increment the usage in place.

Parameters:

Name	Type	Description	Default
incr_usage	RequestUsage	The usage to increment by.	required

Source code in pydantic_ai_slim/pydantic_ai/usage.py

def incr(self, incr_usage: RequestUsage) -> None:
    """Increment the usage in place.

    Args:
        incr_usage: The usage to increment by.
    """
    return _incr_usage_tokens(self, incr_usage)

__add__

__add__(other: RequestUsage) -> RequestUsage

Add two RequestUsages together.

This is provided so it's trivial to sum usage information from multiple parts of a response.

WARNING: this CANNOT be used to sum multiple requests without breaking some pricing calculations.

Source code in pydantic_ai_slim/pydantic_ai/usage.py

def __add__(self, other: RequestUsage) -> RequestUsage:
    """Add two RequestUsages together.

    This is provided so it's trivial to sum usage information from multiple parts of a response.

    **WARNING:** this CANNOT be used to sum multiple requests without breaking some pricing calculations.
    """
    new_usage = copy(self)
    new_usage.incr(other)
    return new_usage

extract classmethod

extract(
    data: Any,
    *,
    provider: str,
    provider_url: str,
    provider_fallback: str,
    api_flavor: str = "default",
    details: dict[str, Any] | None = None
) -> RequestUsage

Extract usage information from the response data using genai-prices.

Parameters:

Name	Type	Description	Default
data	Any	The response data from the model API.	required
provider	str	The actual provider ID	required
provider_url	str	The provider base_url	required
provider_fallback	str	The fallback provider ID to use if the actual provider is not found in genai-prices. For example, an OpenAI model should set this to "openai" in case it has an obscure provider ID.	required
api_flavor	str	The API flavor to use when extracting usage information, e.g. 'chat' or 'responses' for OpenAI.	'default'
details	dict[str, Any] | None	Becomes the details field on the returned RequestUsage for convenience.	None

Source code in pydantic_ai_slim/pydantic_ai/usage.py

@classmethod
def extract(
    cls,
    data: Any,
    *,
    provider: str,
    provider_url: str,
    provider_fallback: str,
    api_flavor: str = 'default',
    details: dict[str, Any] | None = None,
) -> RequestUsage:
    """Extract usage information from the response data using genai-prices.

    Args:
        data: The response data from the model API.
        provider: The actual provider ID
        provider_url: The provider base_url
        provider_fallback: The fallback provider ID to use if the actual provider is not found in genai-prices.
            For example, an OpenAI model should set this to "openai" in case it has an obscure provider ID.
        api_flavor: The API flavor to use when extracting usage information,
            e.g. 'chat' or 'responses' for OpenAI.
        details: Becomes the `details` field on the returned `RequestUsage` for convenience.
    """
    details = details or {}
    for provider_id, provider_api_url in [(None, provider_url), (provider, None), (provider_fallback, None)]:
        try:
            provider_obj = get_snapshot().find_provider(None, provider_id, provider_api_url)
            _model_ref, extracted_usage = provider_obj.extract_usage(data, api_flavor=api_flavor)
            return cls(**{k: v for k, v in extracted_usage.__dict__.items() if v is not None}, details=details)
        except Exception:
            pass
    return cls(details=details)

RunUsage dataclass

Bases: UsageBase

LLM usage associated with an agent run.

Responsibility for calculating request usage is on the model; Pydantic AI simply sums the usage information across requests.

Source code in pydantic_ai_slim/pydantic_ai/usage.py

@dataclass(repr=False, kw_only=True)
class RunUsage(UsageBase):
    """LLM usage associated with an agent run.

    Responsibility for calculating request usage is on the model; Pydantic AI simply sums the usage information across requests.
    """

    requests: int = 0
    """Number of requests made to the LLM API."""

    tool_calls: int = 0
    """Number of successful tool calls executed during the run."""

    input_tokens: int = 0
    """Total number of input/prompt tokens."""

    cache_write_tokens: int = 0
    """Total number of tokens written to the cache."""

    cache_read_tokens: int = 0
    """Total number of tokens read from the cache."""

    input_audio_tokens: int = 0
    """Total number of audio input tokens."""

    cache_audio_read_tokens: int = 0
    """Total number of audio tokens read from the cache."""

    output_tokens: int = 0
    """Total number of output/completion tokens."""

    details: dict[str, int] = dataclasses.field(default_factory=dict)
    """Any extra details returned by the model."""

    def incr(self, incr_usage: RunUsage | RequestUsage) -> None:
        """Increment the usage in place.

        Args:
            incr_usage: The usage to increment by.
        """
        if isinstance(incr_usage, RunUsage):
            self.requests += incr_usage.requests
            self.tool_calls += incr_usage.tool_calls
        return _incr_usage_tokens(self, incr_usage)

    def __add__(self, other: RunUsage | RequestUsage) -> RunUsage:
        """Add two RunUsages together.

        This is provided so it's trivial to sum usage information from multiple runs.
        """
        new_usage = copy(self)
        new_usage.incr(other)
        return new_usage

requests class-attribute instance-attribute

requests: int = 0

Number of requests made to the LLM API.

tool_calls class-attribute instance-attribute

tool_calls: int = 0

Number of successful tool calls executed during the run.

input_tokens class-attribute instance-attribute

input_tokens: int = 0

Total number of input/prompt tokens.

cache_write_tokens class-attribute instance-attribute

cache_write_tokens: int = 0

Total number of tokens written to the cache.

cache_read_tokens class-attribute instance-attribute

cache_read_tokens: int = 0

Total number of tokens read from the cache.

input_audio_tokens class-attribute instance-attribute

input_audio_tokens: int = 0

Total number of audio input tokens.

cache_audio_read_tokens class-attribute instance-attribute

cache_audio_read_tokens: int = 0

Total number of audio tokens read from the cache.

output_tokens class-attribute instance-attribute

output_tokens: int = 0

Total number of output/completion tokens.

details class-attribute instance-attribute

details: dict[str, int] = field(default_factory=dict)

Any extra details returned by the model.

incr

incr(incr_usage: RunUsage | RequestUsage) -> None

Increment the usage in place.

Parameters:

Name	Type	Description	Default
incr_usage	RunUsage | RequestUsage	The usage to increment by.	required

Source code in pydantic_ai_slim/pydantic_ai/usage.py

def incr(self, incr_usage: RunUsage | RequestUsage) -> None:
    """Increment the usage in place.

    Args:
        incr_usage: The usage to increment by.
    """
    if isinstance(incr_usage, RunUsage):
        self.requests += incr_usage.requests
        self.tool_calls += incr_usage.tool_calls
    return _incr_usage_tokens(self, incr_usage)

__add__

__add__(other: RunUsage | RequestUsage) -> RunUsage

Add two RunUsages together.

This is provided so it's trivial to sum usage information from multiple runs.

Source code in pydantic_ai_slim/pydantic_ai/usage.py

def __add__(self, other: RunUsage | RequestUsage) -> RunUsage:
    """Add two RunUsages together.

    This is provided so it's trivial to sum usage information from multiple runs.
    """
    new_usage = copy(self)
    new_usage.incr(other)
    return new_usage

Usage dataclass deprecated

Bases: RunUsage

Deprecated

Usage is deprecated, use RunUsage instead

Deprecated alias for RunUsage.

Source code in pydantic_ai_slim/pydantic_ai/usage.py

@dataclass(repr=False, kw_only=True)
@deprecated('`Usage` is deprecated, use `RunUsage` instead')
class Usage(RunUsage):
    """Deprecated alias for `RunUsage`."""

UsageLimits dataclass

Limits on model usage.

The request count is tracked by pydantic_ai, and the request limit is checked before each request to the model. Token counts are provided in responses from the model, and the token limits are checked after each response.

Each of the limits can be set to None to disable that limit.

Source code in pydantic_ai_slim/pydantic_ai/usage.py

@dataclass(repr=False, kw_only=True)
class UsageLimits:
    """Limits on model usage.

    The request count is tracked by pydantic_ai, and the request limit is checked before each request to the model.
    Token counts are provided in responses from the model, and the token limits are checked after each response.

    Each of the limits can be set to `None` to disable that limit.
    """

    request_limit: int | None = 50
    """The maximum number of requests allowed to the model."""
    tool_calls_limit: int | None = None
    """The maximum number of successful tool calls allowed to be executed."""
    input_tokens_limit: int | None = None
    """The maximum number of input/prompt tokens allowed."""
    output_tokens_limit: int | None = None
    """The maximum number of output/response tokens allowed."""
    total_tokens_limit: int | None = None
    """The maximum number of tokens allowed in requests and responses combined."""
    count_tokens_before_request: bool = False
    """If True, perform a token counting pass before sending the request to the model,
    to enforce `request_tokens_limit` ahead of time. This may incur additional overhead
    (from calling the model's `count_tokens` API before making the actual request) and is disabled by default."""

    @property
    @deprecated('`request_tokens_limit` is deprecated, use `input_tokens_limit` instead')
    def request_tokens_limit(self) -> int | None:
        return self.input_tokens_limit

    @property
    @deprecated('`response_tokens_limit` is deprecated, use `output_tokens_limit` instead')
    def response_tokens_limit(self) -> int | None:
        return self.output_tokens_limit

    @overload
    def __init__(
        self,
        *,
        request_limit: int | None = 50,
        tool_calls_limit: int | None = None,
        input_tokens_limit: int | None = None,
        output_tokens_limit: int | None = None,
        total_tokens_limit: int | None = None,
        count_tokens_before_request: bool = False,
    ) -> None:
        self.request_limit = request_limit
        self.tool_calls_limit = tool_calls_limit
        self.input_tokens_limit = input_tokens_limit
        self.output_tokens_limit = output_tokens_limit
        self.total_tokens_limit = total_tokens_limit
        self.count_tokens_before_request = count_tokens_before_request

    @overload
    @deprecated(
        'Use `input_tokens_limit` instead of `request_tokens_limit` and `output_tokens_limit` and `total_tokens_limit`'
    )
    def __init__(
        self,
        *,
        request_limit: int | None = 50,
        tool_calls_limit: int | None = None,
        request_tokens_limit: int | None = None,
        response_tokens_limit: int | None = None,
        total_tokens_limit: int | None = None,
        count_tokens_before_request: bool = False,
    ) -> None:
        self.request_limit = request_limit
        self.tool_calls_limit = tool_calls_limit
        self.input_tokens_limit = request_tokens_limit
        self.output_tokens_limit = response_tokens_limit
        self.total_tokens_limit = total_tokens_limit
        self.count_tokens_before_request = count_tokens_before_request

    def __init__(
        self,
        *,
        request_limit: int | None = 50,
        tool_calls_limit: int | None = None,
        input_tokens_limit: int | None = None,
        output_tokens_limit: int | None = None,
        total_tokens_limit: int | None = None,
        count_tokens_before_request: bool = False,
        # deprecated:
        request_tokens_limit: int | None = None,
        response_tokens_limit: int | None = None,
    ):
        self.request_limit = request_limit
        self.tool_calls_limit = tool_calls_limit
        self.input_tokens_limit = input_tokens_limit or request_tokens_limit
        self.output_tokens_limit = output_tokens_limit or response_tokens_limit
        self.total_tokens_limit = total_tokens_limit
        self.count_tokens_before_request = count_tokens_before_request

    def has_token_limits(self) -> bool:
        """Returns `True` if this instance places any limits on token counts.

        If this returns `False`, the `check_tokens` method will never raise an error.

        This is useful because if we have token limits, we need to check them after receiving each streamed message.
        If there are no limits, we can skip that processing in the streaming response iterator.
        """
        return any(
            limit is not None for limit in (self.input_tokens_limit, self.output_tokens_limit, self.total_tokens_limit)
        )

    def check_before_request(self, usage: RunUsage) -> None:
        """Raises a `UsageLimitExceeded` exception if the next request would exceed any of the limits."""
        request_limit = self.request_limit
        if request_limit is not None and usage.requests >= request_limit:
            raise UsageLimitExceeded(f'The next request would exceed the request_limit of {request_limit}')

        input_tokens = usage.input_tokens
        if self.input_tokens_limit is not None and input_tokens > self.input_tokens_limit:
            raise UsageLimitExceeded(
                f'The next request would exceed the input_tokens_limit of {self.input_tokens_limit} ({input_tokens=})'
            )

        total_tokens = usage.total_tokens
        if self.total_tokens_limit is not None and total_tokens > self.total_tokens_limit:
            raise UsageLimitExceeded(  # pragma: lax no cover
                f'The next request would exceed the total_tokens_limit of {self.total_tokens_limit} ({total_tokens=})'
            )

    def check_tokens(self, usage: RunUsage) -> None:
        """Raises a `UsageLimitExceeded` exception if the usage exceeds any of the token limits."""
        input_tokens = usage.input_tokens
        if self.input_tokens_limit is not None and input_tokens > self.input_tokens_limit:
            raise UsageLimitExceeded(f'Exceeded the input_tokens_limit of {self.input_tokens_limit} ({input_tokens=})')

        output_tokens = usage.output_tokens
        if self.output_tokens_limit is not None and output_tokens > self.output_tokens_limit:
            raise UsageLimitExceeded(
                f'Exceeded the output_tokens_limit of {self.output_tokens_limit} ({output_tokens=})'
            )

        total_tokens = usage.total_tokens
        if self.total_tokens_limit is not None and total_tokens > self.total_tokens_limit:
            raise UsageLimitExceeded(f'Exceeded the total_tokens_limit of {self.total_tokens_limit} ({total_tokens=})')

    def check_before_tool_call(self, projected_usage: RunUsage) -> None:
        """Raises a `UsageLimitExceeded` exception if the next tool call(s) would exceed the tool call limit."""
        tool_calls_limit = self.tool_calls_limit
        tool_calls = projected_usage.tool_calls
        if tool_calls_limit is not None and tool_calls > tool_calls_limit:
            raise UsageLimitExceeded(
                f'The next tool call(s) would exceed the tool_calls_limit of {tool_calls_limit} ({tool_calls=}).'
            )

    __repr__ = _utils.dataclasses_no_defaults_repr

request_limit class-attribute instance-attribute

request_limit: int | None = request_limit

The maximum number of requests allowed to the model.

tool_calls_limit class-attribute instance-attribute

tool_calls_limit: int | None = tool_calls_limit

The maximum number of successful tool calls allowed to be executed.

input_tokens_limit class-attribute instance-attribute

input_tokens_limit: int | None = (
    input_tokens_limit or request_tokens_limit
)

The maximum number of input/prompt tokens allowed.

output_tokens_limit class-attribute instance-attribute

output_tokens_limit: int | None = (
    output_tokens_limit or response_tokens_limit
)

The maximum number of output/response tokens allowed.

total_tokens_limit class-attribute instance-attribute

total_tokens_limit: int | None = total_tokens_limit

The maximum number of tokens allowed in requests and responses combined.

count_tokens_before_request class-attribute instance-attribute

count_tokens_before_request: bool = (
    count_tokens_before_request
)

If True, perform a token counting pass before sending the request to the model, to enforce request_tokens_limit ahead of time. This may incur additional overhead (from calling the model's count_tokens API before making the actual request) and is disabled by default.

has_token_limits

has_token_limits() -> bool

Returns True if this instance places any limits on token counts.

If this returns False, the check_tokens method will never raise an error.

This is useful because if we have token limits, we need to check them after receiving each streamed message. If there are no limits, we can skip that processing in the streaming response iterator.

Source code in pydantic_ai_slim/pydantic_ai/usage.py

def has_token_limits(self) -> bool:
    """Returns `True` if this instance places any limits on token counts.

    If this returns `False`, the `check_tokens` method will never raise an error.

    This is useful because if we have token limits, we need to check them after receiving each streamed message.
    If there are no limits, we can skip that processing in the streaming response iterator.
    """
    return any(
        limit is not None for limit in (self.input_tokens_limit, self.output_tokens_limit, self.total_tokens_limit)
    )

check_before_request

check_before_request(usage: RunUsage) -> None

Raises a UsageLimitExceeded exception if the next request would exceed any of the limits.

Source code in pydantic_ai_slim/pydantic_ai/usage.py

def check_before_request(self, usage: RunUsage) -> None:
    """Raises a `UsageLimitExceeded` exception if the next request would exceed any of the limits."""
    request_limit = self.request_limit
    if request_limit is not None and usage.requests >= request_limit:
        raise UsageLimitExceeded(f'The next request would exceed the request_limit of {request_limit}')

    input_tokens = usage.input_tokens
    if self.input_tokens_limit is not None and input_tokens > self.input_tokens_limit:
        raise UsageLimitExceeded(
            f'The next request would exceed the input_tokens_limit of {self.input_tokens_limit} ({input_tokens=})'
        )

    total_tokens = usage.total_tokens
    if self.total_tokens_limit is not None and total_tokens > self.total_tokens_limit:
        raise UsageLimitExceeded(  # pragma: lax no cover
            f'The next request would exceed the total_tokens_limit of {self.total_tokens_limit} ({total_tokens=})'
        )

check_tokens

check_tokens(usage: RunUsage) -> None

Raises a UsageLimitExceeded exception if the usage exceeds any of the token limits.

Source code in pydantic_ai_slim/pydantic_ai/usage.py

def check_tokens(self, usage: RunUsage) -> None:
    """Raises a `UsageLimitExceeded` exception if the usage exceeds any of the token limits."""
    input_tokens = usage.input_tokens
    if self.input_tokens_limit is not None and input_tokens > self.input_tokens_limit:
        raise UsageLimitExceeded(f'Exceeded the input_tokens_limit of {self.input_tokens_limit} ({input_tokens=})')

    output_tokens = usage.output_tokens
    if self.output_tokens_limit is not None and output_tokens > self.output_tokens_limit:
        raise UsageLimitExceeded(
            f'Exceeded the output_tokens_limit of {self.output_tokens_limit} ({output_tokens=})'
        )

    total_tokens = usage.total_tokens
    if self.total_tokens_limit is not None and total_tokens > self.total_tokens_limit:
        raise UsageLimitExceeded(f'Exceeded the total_tokens_limit of {self.total_tokens_limit} ({total_tokens=})')

check_before_tool_call

check_before_tool_call(projected_usage: RunUsage) -> None

Raises a UsageLimitExceeded exception if the next tool call(s) would exceed the tool call limit.

Source code in pydantic_ai_slim/pydantic_ai/usage.py

def check_before_tool_call(self, projected_usage: RunUsage) -> None:
    """Raises a `UsageLimitExceeded` exception if the next tool call(s) would exceed the tool call limit."""
    tool_calls_limit = self.tool_calls_limit
    tool_calls = projected_usage.tool_calls
    if tool_calls_limit is not None and tool_calls > tool_calls_limit:
        raise UsageLimitExceeded(
            f'The next tool call(s) would exceed the tool_calls_limit of {tool_calls_limit} ({tool_calls=}).'
        )