airflow.providers.common.ai.utils.logging¶

Logging utilities for pydantic-ai agent runs.

Functions¶

`log_run_summary`(logger, result, *[, usage])	Log model name, token usage, and tool call sequence from an agent run.
`log_run_usage`(logger, usage, *, outcome)	Log token usage/cost for a run that never produced an `AgentRunResult` (the failure path).
`format_usage_for_xcom`(usage)	Build the XCom `usage` payload -- shared by the success and failure paths.
`wrap_toolsets_for_logging`(toolsets, logger)	Wrap each toolset in a LoggingToolset.

airflow.providers.common.ai.utils.logging.log_run_summary(logger, result, *, usage=None)[source]¶

Log model name, token usage, and tool call sequence from an agent run.

Parameters:: usage (pydantic_ai.usage.RunUsage | None) – Usage to log instead of result.usage – e.g. this attempt’s delta when result.usage would report the cross-attempt cumulative total instead.

airflow.providers.common.ai.utils.logging.log_run_usage(logger, usage, *, outcome)[source]¶

Log token usage/cost for a run that never produced an AgentRunResult (the failure path).

airflow.providers.common.ai.utils.logging.format_usage_for_xcom(usage)[source]¶

Build the XCom usage payload – shared by the success and failure paths.

airflow.providers.common.ai.utils.logging.wrap_toolsets_for_logging(toolsets, logger)[source]¶

Wrap each toolset in a LoggingToolset.