LLMFileAnalysisOperator & @task.llm_file_analysis¶
Use LLMFileAnalysisOperator
or the @task.llm_file_analysis decorator to analyze files from object storage
or local storage with a single prompt.
The operator resolves file_path through
ObjectStoragePath, reads supported
formats in a read-only manner, injects file metadata and normalized content into
the prompt, and optionally attaches images or PDFs as multimodal inputs.
See also
Basic Usage¶
Analyze a text-like file or prefix with one prompt:
@dag(tags=["example"])
def example_llm_file_analysis_basic():
LLMFileAnalysisOperator(
task_id="analyze_error_logs",
prompt="Find error patterns and correlate them with deployment timestamps.",
llm_conn_id="pydanticai_default",
file_path="s3://logs/app/2024-01-15/",
file_conn_id="aws_default",
)
Directory / Prefix Analysis¶
Use a directory or object-storage prefix when you want the operator to analyze
multiple files in one request. max_files bounds how many resolved files are
included in the request, while the size and text limits keep the request safe:
@dag(tags=["example"])
def example_llm_file_analysis_prefix():
LLMFileAnalysisOperator(
task_id="summarize_partitioned_logs",
prompt=(
"Summarize recurring errors across these partitioned log files and call out "
"which partition keys appear in the highest-severity findings."
),
llm_conn_id="pydanticai_default",
file_path="s3://logs/app/dt=2024-01-15/",
file_conn_id="aws_default",
max_files=10,
max_total_size_bytes=10 * 1024 * 1024,
max_text_chars=20_000,
)
Note
Prefix resolution enumerates objects under the supplied path and checks each
candidate to find files before max_files is applied. For very large
object-store prefixes, prefer a more specific path or a narrower prefix to
avoid expensive listing and stat calls.
Multimodal Analysis¶
Set multi_modal=True for PNG/JPG/PDF inputs so they are sent as binary
attachments to a vision-capable model:
@dag(tags=["example"])
def example_llm_file_analysis_multimodal():
LLMFileAnalysisOperator(
task_id="validate_dashboards",
prompt="Check charts for visual anomalies or stale data indicators.",
llm_conn_id="pydanticai_default",
file_path="s3://monitoring/dashboards/latest.png",
file_conn_id="aws_default",
multi_modal=True,
)
Structured Output¶
Set output_type to a Pydantic BaseModel when you want a typed response
back from the LLM instead of a plain string. The model instance is pushed to
XCom unchanged so downstream tasks can type-hint the class directly. The
declared output_type (and any BaseModel reachable from
Union/Optional/list shapes) is registered for deserialization by the
worker when it loads the Dag. Define the class at module scope and bind it to
an attribute matching its __name__: nested-in-function and dynamically-built
classes cannot be re-imported, so they are skipped at worker startup and fail to
deserialize at the consumer. Same-Dag downstream tasks need no configuration; the
UI XCom viewer renders the value
via the stringify path (no configuration needed). Cross-Dag xcom_pull
consumers still need the class qualname added to
[core] allowed_deserialization_classes (see the LLMOperator guide for
details).
# Pydantic output classes must be defined at module scope so they can be
# imported by name when downstream tasks deserialize the XCom payload.
class FileAnalysisSummary(BaseModel):
"""Structured output schema for the file-analysis examples."""
findings: list[str]
highest_severity: str
truncated_inputs: bool
@dag(tags=["example"])
def example_llm_file_analysis_structured():
LLMFileAnalysisOperator(
task_id="analyze_parquet_quality",
prompt=(
"Return the top data-quality findings from this Parquet dataset. "
"Include whether any inputs were truncated."
),
llm_conn_id="pydanticai_default",
file_path="s3://analytics/warehouse/customers/",
file_conn_id="aws_default",
output_type=FileAnalysisSummary,
sample_rows=5,
max_files=5,
)
TaskFlow Decorator¶
The @task.llm_file_analysis decorator wraps the operator. The function
returns the prompt string; file settings are passed to the decorator:
@dag(tags=["example"])
def example_llm_file_analysis_decorator():
@task.llm_file_analysis(
llm_conn_id="pydanticai_default",
file_path="s3://analytics/reports/quarterly.pdf",
file_conn_id="aws_default",
multi_modal=True,
)
def review_quarterly_report():
return "Extract the key revenue, risk, and compliance findings from this report."
review_quarterly_report()
Human-in-the-Loop Approval¶
Set require_approval=True to pause the task after the analysis and wait
for a human reviewer to approve the output before it is returned.
When allow_modifications=True, the reviewer can also edit the output:
@dag(tags=["example"])
def example_llm_file_analysis_approval():
from datetime import timedelta
LLMFileAnalysisOperator(
task_id="analyze_contract_with_approval",
prompt="Summarize the key obligations and flag any unusual termination clauses.",
llm_conn_id="pydanticai_default",
file_path="s3://legal/contracts/vendor-agreement.pdf",
file_conn_id="aws_default",
require_approval=True,
approval_timeout=timedelta(hours=1),
allow_modifications=True,
)
Parameters¶
prompt: The analysis request to send to the LLM (operator) or the return value of the decorated function (decorator).llm_conn_id: Airflow connection ID for the LLM provider.file_path: File or prefix to analyze.file_conn_id: Optional connection ID for the storage backend. Overrides a connection embedded infile_path.multi_modal: Allow PNG/JPG/PDF inputs as binary attachments. DefaultFalse.max_files: Maximum number of files included from a prefix. Extra files are omitted and noted in the prompt. Default20.max_file_size_bytes: Maximum size of any single input file. Default5 MiB.max_total_size_bytes: Maximum cumulative size across all resolved files. Default20 MiB.max_text_chars: Maximum normalized text context sent to the LLM after sampling and truncation. Default100000.sample_rows: Maximum number of sampled rows or records included for CSV, Parquet, and Avro inputs. This controls structural preview depth, whilemax_file_size_bytesandmax_total_size_bytesare byte-level read guards andmax_text_charsis the final prompt-text budget. Default10.model_id: Model identifier (e.g."openai:gpt-5"). Overrides the connection’s extra field.system_prompt: System-level instructions appended to the operator’s built-in read-only guidance.output_type: Expected output type (default:str). Set to a PydanticBaseModelfor structured output.agent_params: Additional keyword arguments passed to the pydantic-aiAgentconstructor (e.g.retries,model_settings).serialize_output: IfTrueandoutput_typeis a PydanticBaseModelsubclass, the model instance is dumped to adictviamodel_dump()before being pushed to XCom. DefaultFalse– the Pydantic instance flows through XCom unchanged. Set toTruewhen a downstream consumer needs the dict shape.
This operator also inherits LLMOperator’s HITL review parameters –
require_approval, approval_timeout, and allow_modifications – see
LLMOperator for details.
Supported Formats¶
Text-like:
.log,.txt,.md,.json,.csv,.parquet,.avroMultimodal:
.png,.jpg,.jpeg,.pdfwhenmulti_modal=Truegzip,bzip2, andxzcompressed text inputs are supported for.log,.json,.csv,.txt, and.md(.log.gz,.csv.bz2,.json.xz, …).bzip2andxzneed a Python interpreter built with thebz2andlzmamodules; otherwise those inputs raiseAirflowOptionalProviderFeatureException.Compression is not supported for
.parquet,.avro, image, or PDF inputs.For
bzip2andxz, concatenated streams are read in full, but any data after a point that does not start a valid stream (for examplexzStream Padding between streams, or trailing garbage) is silently ignored, so only the content up to that point is analyzed.gziprejects such files instead (BadGzipFile).
Parquet and Avro readers require their corresponding optional extras:
pip install apache-airflow-providers-common-ai[parquet]
pip install apache-airflow-providers-common-ai[avro]