Weekly status report with a hallucination check¶

A program manager wants a weekly status on a set of proposals: promised, landed, still open. The evidence is spread across a wiki and a repository, and people act on the report, so it has to be right. This Dag gathers the evidence, has the model assess each proposal, synthesizes a report, then has a second model call judge it against the evidence and a plain-Python step apply only the flagged corrections. A person reviews the result.

Airflow maps gathering and assessment per proposal, keeps every intermediate in XCom, and makes the correction step deterministic so the model cannot rewrite the report while fixing it.

The example tracks Airflow Improvement Proposals against Confluence and GitHub, both public, and solves the job a second way with one autonomous agent for comparison.

What this demonstrates¶

Single prompts: LLMOperator and @task.llm – mapped LLMOperator calls with structured output for the per-proposal analysis, one bounded by UsageLimits for the synthesis, and one whose only job is validation.
Structured output and XCom – ValidationResult lists each claim with a verdict, so the correction step has something exact to act on.
Human in the Loop (HITL) Operators – ApprovalOperator before the report goes out.
Agent Skills: AgentSkillsToolset – the agent variant loads a SKILL.md bundle that tells it how to assess progress.

Run it¶

Install the provider. The agent variant also needs the skills extra:
```
pip install "apache-airflow-providers-common-ai[openai,skills]"
```
Optionally set GITHUB_TOKEN in the environment. Without it the Dag paces itself to the unauthenticated rate limit and takes longer.

Trigger either Dag. Both take an aip_numbers param:

airflow dags test example_aip_progress_tracker
airflow dags test example_aip_progress_tracker_skills

The validate_report XCom shows the disputed claims and apply_validation shows what changed. The run pauses at review_report; answer from Required Actions in the UI (see the note on What you can build).

The Dag¶

The validation output type and the step that applies it:

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

class ClaimValidation(BaseModel):
    """A single claim from the report checked against evidence."""

    claim: str
    grounded: bool
    evidence_found: str
    correction: str


class ValidationResult(BaseModel):
    """Result of the AI hallucination validation step."""

    overall_verdict: Literal["pass", "pass_with_warnings", "fail"]
    ungrounded_claims: list[ClaimValidation]
    hallucination_risk: Literal["low", "medium", "high"]

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

    validate = LLMOperator(
        task_id="validate_report",
        llm_conn_id=LLM_CONN_ID,
        system_prompt=VALIDATION_SYSTEM_PROMPT,
        prompt="""\
Verify the following synthesized report against the raw per-AIP evidence.
Flag any claims not grounded in the evidence.

=== SYNTHESIZED REPORT ===
{{ ti.xcom_pull(task_ids='synthesize_report') }}

=== RAW PER-AIP EVIDENCE ===
{{ ti.xcom_pull(task_ids='format_report') }}""",
        output_type=ValidationResult,
        serialize_output=True,
        usage_limits=UsageLimits(
            request_limit=5,
            input_tokens_limit=30_000,
            output_tokens_limit=8_000,
        ),
        agent_params={"model_settings": {"temperature": 0}},
    )

The mapped analysis that feeds it:

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

    analyses = LLMOperator.partial(
        task_id="analyze_aip",
        llm_conn_id=LLM_CONN_ID,
        system_prompt=ANALYSIS_SYSTEM_PROMPT,
        output_type=AIPStatus,
        serialize_output=True,
        agent_params={"model_settings": {"temperature": 0}},
    ).expand(prompt=prompts)

The full Dag, evidence gathering included, is example_aip_progress_tracker in example_aip_progress_tracker.py. It is long because the evidence gathering is real.

The agent variant¶

The same job as one operator call. The agent reads a skill that explains how to assess a proposal, gets the wiki and repository as tool functions, and decides its own order of work:

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

    report = AgentOperator(
        task_id="track_aip_progress",
        llm_conn_id=LLM_CONN_ID,
        system_prompt=AGENT_SYSTEM_PROMPT,
        prompt=prompt,
        toolsets=[
            AgentSkillsToolset(sources=[SKILLS_DIR]),
            aip_toolset,
        ],
        agent_params={"model_settings": {"temperature": 0}},
        usage_limits=UsageLimits(
            request_limit=30,
            input_tokens_limit=200_000,
            output_tokens_limit=16_000,
        ),
    )

Use the pipeline when every step must be auditable. Use the agent when a fixed task graph would only re-encode what the model can work out from the skill.

Adapting it¶

Replace the Confluence and GitHub fetchers with your own sources: a project tracker, a design-doc folder, a deployment log. Everything from analyze_aip down is source-agnostic.
Keep the validation step even if you drop the rest. It is cheap and catches the failures that make people stop trusting generated reports.
Put it on schedule="@weekly" and send the approved report to a channel from a task after review_report.