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.

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 and conditional workflows.

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
Compose as a workflow	Workflows
Run agents concurrently	Async execution

Developer Resources

• Quality review cookbook
• Workflows overview