pydantic_ai.result

ResultData module-attribute

ResultData = TypeVar('ResultData')

Type variable for the result data of a run.

RunResult dataclass

Bases: _BaseRunResult[ResultData]

Result of a non-streamed run.

all_messages

all_messages() -> list[Message]

Return the history of messages.

all_messages_json

all_messages_json() -> bytes

Return all messages from all_messages as JSON bytes.

new_messages

new_messages() -> list[Message]

Return new messages associated with this run.

System prompts and any messages from older runs are excluded.

new_messages_json

new_messages_json() -> bytes

Return new messages from new_messages as JSON bytes.

data instance-attribute

data: ResultData

Data from the final response in the run.

cost

cost() -> Cost

Return the cost of the whole run.

StreamedRunResult dataclass

Bases: _BaseRunResult[ResultData], Generic[AgentDeps, ResultData]

Result of a streamed run that returns structured data via a tool call.

all_messages

all_messages() -> list[Message]

Return the history of messages.

all_messages_json

all_messages_json() -> bytes

Return all messages from all_messages as JSON bytes.

new_messages

new_messages() -> list[Message]

Return new messages associated with this run.

System prompts and any messages from older runs are excluded.

new_messages_json

new_messages_json() -> bytes

Return new messages from new_messages as JSON bytes.

cost_so_far instance-attribute

cost_so_far: Cost

Cost of the run up until the last request.

is_complete class-attribute instance-attribute

is_complete: bool = False

Whether the stream has all been received.

This is set to True when one of stream, stream_text, stream_structured or get_data completes.

stream async

stream(
    *,
    text_delta: bool = False,
    debounce_by: float | None = 0.1
) -> AsyncIterator[ResultData]

Stream the response as an async iterable.

Result validators are called on each iteration, if text_delta=False (the default) or for structured responses.

Note

Result validators will NOT be called on the text result if text_delta=True.

The pydantic validator for structured data will be called in partial mode on each iteration.

Parameters:

Name	Type	Description	Default
text_delta	bool	if True, yield each chunk of text as it is received, if False (default), yield the full text up to the current point.	False
debounce_by	float | None	by how much (if at all) to debounce/group the response chunks by. None means no debouncing. Debouncing is particularly important for long structured responses to reduce the overhead of performing validation as each token is received.	0.1

Returns:

Type	Description
AsyncIterator[ResultData]	An async iterable of the response data.

stream_text async

stream_text(
    *,
    text_delta: bool = False,
    debounce_by: float | None = 0.1
) -> AsyncIterator[str]

Stream the text result as an async iterable.

Note

This method will fail if the response is structured, e.g. if is_structured returns True.

Parameters:

Name	Type	Description	Default
text_delta	bool	if True, yield each chunk of text as it is received, if False (default), yield the full text up to the current point.	False
debounce_by	float | None	by how much (if at all) to debounce/group the response chunks by. None means no debouncing. Debouncing is particularly important for long structured responses to reduce the overhead of performing validation as each token is received.	0.1

stream_structured async

stream_structured(
    *, debounce_by: float | None = 0.1
) -> AsyncIterator[tuple[ModelStructuredResponse, bool]]

Stream the response as an async iterable of Structured LLM Messages.

Note

This method will fail if the response is text, e.g. if is_structured returns False.

Parameters:

Name	Type	Description	Default
debounce_by	float | None	by how much (if at all) to debounce/group the response chunks by. None means no debouncing. Debouncing is particularly important for long structured responses to reduce the overhead of performing validation as each token is received.	0.1

Returns:

Type	Description
AsyncIterator[tuple[ModelStructuredResponse, bool]]	An async iterable of the structured response message and whether that is the last message.

get_data async

get_data() -> ResultData

Stream the whole response, validate and return it.

is_structured property

is_structured: bool

Return whether the stream response contains structured data (as opposed to text).

cost

cost() -> Cost

Return the cost of the whole run.

Note

This won't return the full cost until the stream is finished.

timestamp

timestamp() -> datetime

Get the timestamp of the response.

validate_structured_result async

validate_structured_result(
    message: ModelStructuredResponse,
    *,
    allow_partial: bool = False
) -> ResultData

Validate a structured result message.

Cost dataclass

Cost of a request or run.

Responsibility for calculating costs is on the model used, PydanticAI simply sums the cost of requests.

You'll need to look up the documentation of the model you're using to convent "token count" costs to monetary costs.

request_tokens class-attribute instance-attribute

request_tokens: int | None = None

Tokens used in processing the request.

response_tokens class-attribute instance-attribute

response_tokens: int | None = None

Tokens used in generating the response.

total_tokens class-attribute instance-attribute

total_tokens: int | None = None

Total tokens used in the whole run, should generally be equal to request_tokens + response_tokens.

details class-attribute instance-attribute

details: dict[str, int] | None = None

Any extra details returned by the model.

__add__

__add__(other: Cost) -> Cost

Add two costs together.

This is provided so it's trivial to sum costs from multiple requests and runs.