Structured output and XCom

Set output_type to a Pydantic BaseModel subclass. The LLM is instructed to return structured data, and the model instance is pushed to XCom unchanged so downstream tasks can type-hint the class directly (def downstream(result: MyModel)) and use attribute access (result.field). Inside a list of types ([MyModel, int]) the model is pushed as a dict instead.

The declared output_type (and any BaseModel reachable from Union/Optional/list[...] types) is registered for XCom deserialization by the worker when it loads the Dag, before any task runs – so no edit to [core] allowed_deserialization_classes is needed. The Pydantic class must be defined at module scope and bound to an attribute matching its __name__; classes nested inside a function or @dag-decorated body, parameterized generics, and dynamically-built classes whose __name__ does not match the attribute they are bound to cannot be re-imported, so they are skipped with a warning at worker startup and the value fails to deserialize at the consumer. A list of types (output_type=[MyModel, int]) registers no class, so the operator pushes the model as a dict even when serialize_output is False.

airflow/providers/common/ai/example_dags/example_llm.py[source]

# Pydantic output classes must be defined at module scope so they survive
# XCom serialization (their qualname is used to re-import them downstream).
class Entities(BaseModel):
    """Named entities extracted from a text."""

    names: list[str]
    locations: list[str]

airflow/providers/common/ai/example_dags/example_llm.py[source]

@dag(tags=["example"])
def example_llm_operator_structured():
    LLMOperator(
        task_id="extract_entities",
        prompt="Extract all named entities from the article.",
        llm_conn_id="pydanticai_default",
        system_prompt="Extract named entities.",
        output_type=Entities,
    )

Registration covers downstream tasks in the same Dag: every worker walks the loaded Dag’s tasks at startup and registers each declared class, so it also works for mapped producers (.expand(...)) and for workers that load Dags from a cache that bypasses operator construction.

The worker finds no class to register when output_type is a list of output types such as [Summary, int], an output marker such as ToolOutput(Summary), an output function, a TypeAliasType or a NewType. For these output types, LLMOperator, @task.llm, LLMFileAnalysisOperator, AgentOperator and @task.agent push each model in the output as a JSON-compatible dict, including models inside a list, tuple or dict. A model is pushed as a dict even if its class is listed in [core] allowed_deserialization_classes, or if another task in the same Dag declares the class as its output_type. A downstream task therefore reads result["field"], not result.field.

In a JSON-compatible dict, values such as datetime, date, UUID and Decimal arrive as strings, tuples and sets arrive as lists, and Enum members arrive as their values.

A model at the top level is still dumped with plain model_dump() when serialize_output=True is set or on Airflow versions before 3.3. Either condition is enough. A plain Enum member (one that does not also subclass str or int), an HttpUrl or a Path in that model then stays a Python object. The default XCom backend cannot serialize these objects, so the XCom push fails with TypeError.

Upgrading from common.ai 1.0.0 to 1.1.0 can change the type that a downstream task receives:

  • On Airflow 3.3 and later, common.ai 1.0.0 pushed each model in the output as a model instance when output_type was one of the output types listed above. It dumped a top-level model with plain model_dump() only when serialize_output=True was set or in HITL review. A downstream task loaded the instance when the class was listed in [core] allowed_deserialization_classes or when another task in the same Dag declared the class as its output_type. That downstream task now gets a dict, and result.field raises AttributeError.

  • In HITL review on Airflow 3.3 and later without serialize_output=True, common.ai 1.0.0 dumped a top-level model of these output types with plain model_dump() after the reviewer approved the model. Values such as datetime, date, timedelta, UUID and Decimal in that model now arrive as strings, for example "PT1M30S" for a timedelta. Tuples and sets in that model now arrive as lists.

  • Before Airflow 3.3, common.ai 1.0.0 pushed a model inside a list, tuple or dict as a model instance, for every output_type. A downstream task that loaded the instance because the class was listed in [core] allowed_deserialization_classes now gets a dict.

The Airflow UI’s XCom viewer renders Pydantic instances via the stringify path, which produces a representation like my_module.MyModel@version=1(field=value,...) without consulting the allow-list. It is not pretty (no field-by-field rendering today), but the value shows up; no configuration is required.

The remaining gap is cross-Dag xcom_pull – a task in a different Dag that pulls this XCom only parses its own Dag file, not the producer’s, so the class is not auto-registered. Add the class qualified name to [core] allowed_deserialization_classes (or a glob that matches it) to make that pattern work.

When a model arrives as a dict

LLMOperator, @task.llm, LLMFileAnalysisOperator, AgentOperator and @task.agent decide in the same way whether a model is pushed as an instance or dumped. Set serialize_output=True when a downstream consumer needs the dict shape (for example to forward it to a system that expects JSON-style payloads).

  • With serialize_output=True, an output that is itself a model is dumped with model_dump(), so field types such as date and Decimal are kept.

  • A model inside a list, tuple, set or dict is not dumped when the worker registers its class (output_type=list[MyModel] on Airflow 3.3 and later). The XCom serializer pushes it as an instance, with or without serialize_output, unless the output also holds a model whose class the worker does not register (see the next bullets).

  • On Airflow 3.3 and later, a model whose class the worker cannot register is dumped to JSON values when it is itself the output and serialize_output is False. With serialize_output=True it is dumped with model_dump() as the first bullet describes. The worker registers the classes it finds by reading output_type and finds none in a list of types such as output_type=[MyModel, int].

  • The operator dumps the whole output with to_jsonable_python, using field names rather than aliases (also for a model that sets serialize_by_alias=True), as soon as it holds such a model in a list, tuple, set or dict, or as a set member or a dict key. A date or Decimal next to the model becomes a string, a set becomes a list, and a model used as a dict key becomes its string form.

  • Before Airflow 3.3 the worker registers no class. A model that is itself the output is then dumped with model_dump() whether or not serialize_output is set, and an output that holds a model in a list, tuple, set or dict is dumped as the previous bullet describes, for every output_type.

In a dump to JSON values, HttpUrl, Path and Enum fields become plain JSON values. A model that is itself the output with serialize_output=True keeps its Python field values instead, so an HttpUrl, a Path or a plain Enum member in it stays an object that the default XCom backend cannot serialize.

Was this entry helpful?