> ## Documentation Index
> Fetch the complete documentation index at: https://agno-v2-codex-docs-audit-20260719-0149.mintlify.site/llms.txt
> Use this file to discover all available pages before exploring further.

# Quality pipeline

> Two labelers from different providers, a reviewer that diffs them, and an adjudicator that resolves disagreement.

Run two labelers from different providers, compare their outputs, and adjudicate where they disagree. Persist the disagreement report when you need an audit trail.

```python theme={null}
from typing import List, Optional

from agno.agent import Agent
from agno.models.anthropic import Claude
from agno.models.google import Gemini
from pydantic import BaseModel, Field


class Contact(BaseModel):
    name: Optional[str] = None
    email: Optional[str] = None
    company: Optional[str] = None


class FieldDisagreement(BaseModel):
    field: str = Field(..., description="Top-level Contact field name")
    value_a: Optional[str] = None
    value_b: Optional[str] = None
    reason: str = Field(..., description="Why this field needs adjudication")


class DisagreementReport(BaseModel):
    disagreements: List[FieldDisagreement] = Field(default_factory=list)
    needs_adjudication: bool = Field(..., description="True if any field disagrees")


class FinalLabel(BaseModel):
    contact: Contact
    notes: Optional[str] = None


LABELER = "Extract contact info. Use exactly what the text shows. Null if missing."

labeler_a = Agent(model=Gemini(id="gemini-3.5-flash"), instructions=LABELER, output_schema=Contact)
labeler_b = Agent(model=Claude(id="claude-opus-4-7"), instructions=LABELER, output_schema=Contact)

reviewer = Agent(
    model=Claude(id="claude-opus-4-7"),
    instructions=(
        "Compare two labelers' Contact outputs field by field. A field "
        "needs adjudication whenever the values differ, including when one "
        "value is null. Set "
        "needs_adjudication=true if any field does."
    ),
    output_schema=DisagreementReport,
)

adjudicator = Agent(
    model=Claude(id="claude-opus-4-7"),
    instructions=(
        "Re-read the original text and resolve every reported "
        "disagreement. Return a FinalLabel with the correct values."
    ),
    output_schema=FinalLabel,
)


def label_with_quality_review(text: str) -> FinalLabel:
    a = labeler_a.run(text).content
    b = labeler_b.run(text).content

    report = reviewer.run(
        f"Labeler A:\n{a.model_dump_json()}\n\nLabeler B:\n{b.model_dump_json()}"
    ).content

    if not report.needs_adjudication:
        return FinalLabel(contact=a, notes="Labelers agreed.")

    return adjudicator.run(
        f"Original input:\n{text}\n\n"
        f"Labeler A:\n{a.model_dump_json()}\n\n"
        f"Labeler B:\n{b.model_dump_json()}\n\n"
        f"Reviewer report:\n{report.model_dump_json()}"
    ).content
```

## The flow

1. **Two labelers, two providers.** Provider disagreement identifies records that need review.
2. **Reviewer diffs them.** It emits one `FieldDisagreement` per conflicting field and a single `needs_adjudication` flag. Agreement short-circuits the expensive step.
3. **Adjudicator runs only on disagreement.** It re-reads the original input with both labels and the reviewer's report, then returns the final record.

Agreement does not prove that a label is correct. Measure this pipeline on a labeled validation set. Persisted disagreement rates by field, provider, and prompt version can identify areas for review.

## Production composition

The example above runs the agents sequentially so the pattern is readable. For a million-document job, wrap labelers in a `Parallel` step and gate the adjudicator behind a `Condition` in a `Workflow`. See [parallel workflows](/workflows/workflow-patterns/parallel-workflow) and [conditional workflows](/workflows/workflow-patterns/conditional-workflow).

## Production checklist

Agno gives you the orchestration primitives. These concerns are yours to add.

| Concern             | What to add                                                                                                                       |
| ------------------- | --------------------------------------------------------------------------------------------------------------------------------- |
| Rate limiting       | Wrap the agent call with a per-provider limiter, or front it with a gateway. Agno does not throttle outbound calls.               |
| Bounded concurrency | An `asyncio.Semaphore` around the batch fan-out.                                                                                  |
| Dead-letter queue   | Record failed item IDs and re-run them through a stricter pass.                                                                   |
| Idempotency         | A deterministic item ID and an output-store lookup before each run. Store results by item ID and prompt version.                  |
| Provider Batch APIs | Call provider batch endpoints directly when the workload needs them. Agno does not wrap these APIs.                               |
| Prompt versioning   | Track a `prompt_version` in run metadata so historical labels stay joinable.                                                      |
| Authoritative cost  | `RunMetrics.cost` is populated only when the provider returns it. Attach a token-rate table downstream if you need exact numbers. |

## Next steps

| Task                    | Guide                                                             |
| ----------------------- | ----------------------------------------------------------------- |
| Build the labelers      | [Data extraction](/use-cases/data-labeling/structured-extraction) |
| Compose as a workflow   | [Workflows](/workflows/overview)                                  |
| Run agents concurrently | [Async execution](/agents/running-agents)                         |

## Developer Resources

* [Quality review cookbook](https://github.com/agno-agi/agno/tree/v2.7.4/cookbook/data_labeling/_18_quality_review)
* [Workflows overview](/workflows/overview)