Quality review pipeline

Use a Workflow to run two labelers in parallel, review their outputs, and call an adjudicator when the reviewer finds a disagreement. This runnable adaptation of the quality-review cookbook checks every agent result before consuming it. An unsuccessful or unparseable stage fails the workflow instead of treating missing review fields as agreement.

Save the following program as quality_pipeline.py:

quality_pipeline.py

from typing import List, Optional, TypeVar

from agno.agent import Agent
from agno.db.sqlite import SqliteDb
from agno.run.base import RunStatus
from agno.workflow import Step, Workflow
from agno.workflow.condition import Condition
from agno.workflow.parallel import Parallel
from agno.workflow.types import HumanReview, StepInput, StepOutput
from pydantic import BaseModel, Field


class Contact(BaseModel):
    name: Optional[str] = None
    email: Optional[str] = None
    phone: Optional[str] = None
    company: Optional[str] = None
    title: 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_INSTRUCTIONS = """\
Extract contact information from the input. Use exactly what the text
shows. If a field is missing, leave it null. Do not guess.
"""

labeler_a = Agent(
    name="Labeler A",
    model="google:gemini-3.5-flash",
    instructions=LABELER_INSTRUCTIONS,
    output_schema=Contact,
)

labeler_b = Agent(
    name="Labeler B",
    model="anthropic:claude-opus-4-7",
    instructions=LABELER_INSTRUCTIONS,
    output_schema=Contact,
)

reviewer = Agent(
    name="Reviewer",
    model="anthropic:claude-opus-4-7",
    instructions="""\
You are given two labelers' Contact outputs. Compare them field by field.
A field needs adjudication when both labelers report non-null but
different values. Emit one FieldDisagreement per such field. Set
needs_adjudication=true if any field needs adjudication.
""",
    output_schema=DisagreementReport,
)

adjudicator = Agent(
    name="Adjudicator",
    model="anthropic:claude-opus-4-7",
    instructions="""\
Re-read the original input text and resolve every reported disagreement.
Return a FinalLabel.contact populated with the correct values for all
fields (use the agreed values for fields not in dispute).
""",
    output_schema=FinalLabel,
)

Schema = TypeVar("Schema", bound=BaseModel)


def checked_run(agent: Agent, prompt: str, schema: type[Schema]) -> Schema:
    run = agent.run(prompt)
    if run.status != RunStatus.completed or not isinstance(run.content, schema):
        raise RuntimeError(f"{agent.name} did not return a validated {schema.__name__}")
    return run.content


def checked_step(step_input: StepInput, name: str, schema: type[Schema]) -> Schema:
    output = step_input.get_step_output(name)
    if output is None or not output.success or not isinstance(output.content, schema):
        raise RuntimeError(f"{name} has no successful {schema.__name__} output")
    return output.content


def run_labeler_a(step_input: StepInput) -> StepOutput:
    return StepOutput(content=checked_run(labeler_a, str(step_input.input), Contact))


def run_labeler_b(step_input: StepInput) -> StepOutput:
    return StepOutput(content=checked_run(labeler_b, str(step_input.input), Contact))


def run_reviewer(step_input: StepInput) -> StepOutput:
    a = checked_step(step_input, "Labeler A", Contact)
    b = checked_step(step_input, "Labeler B", Contact)
    prompt = (
        f"Labeler A:\n{a.model_dump_json(indent=2)}\n\n"
        f"Labeler B:\n{b.model_dump_json(indent=2)}"
    )
    return StepOutput(content=checked_run(reviewer, prompt, DisagreementReport))


def has_disagreement(step_input: StepInput) -> bool:
    report = checked_step(step_input, "Reviewer", DisagreementReport)
    return report.needs_adjudication


def run_adjudicator(step_input: StepInput) -> StepOutput:
    a = checked_step(step_input, "Labeler A", Contact)
    b = checked_step(step_input, "Labeler B", Contact)
    report = checked_step(step_input, "Reviewer", DisagreementReport)
    prompt = (
        f"Original input:\n{step_input.input}\n\n"
        f"Labeler A:\n{a.model_dump_json(indent=2)}\n\n"
        f"Labeler B:\n{b.model_dump_json(indent=2)}\n\n"
        f"Reviewer report:\n{report.model_dump_json(indent=2)}"
    )
    return StepOutput(content=checked_run(adjudicator, prompt, FinalLabel))


workflow = Workflow(
    name="Quality review labeling",
    db=SqliteDb(db_file="tmp/labeling.db"),
    steps=[
        Parallel(
            Step(name="Labeler A", executor=run_labeler_a, human_review=HumanReview(on_error="fail")),
            Step(name="Labeler B", executor=run_labeler_b, human_review=HumanReview(on_error="fail")),
            name="Label",
        ),
        Step(name="Reviewer", executor=run_reviewer, human_review=HumanReview(on_error="fail")),
        Condition(
            name="Adjudicate",
            evaluator=has_disagreement,
            human_review=HumanReview(on_error="fail"),
            steps=[
                Step(
                    name="Adjudicator",
                    executor=run_adjudicator,
                    human_review=HumanReview(on_error="fail"),
                ),
            ],
        ),
    ],
)

if __name__ == "__main__":
    result = workflow.run("Liam Ortega is a Support Engineer at Meadow. Email: liam@meadow.io.")
    if result.status != RunStatus.completed:
        raise RuntimeError("Labeling failed; send the original input to review")
    print(result.content)

Execution flow

Step	What it does
`Parallel`	Runs the Google and Anthropic labelers concurrently, checking each `Contact`.
`Reviewer`	Reads both named outputs with `StepInput.get_step_output()` and checks the returned `DisagreementReport`.
`Condition`	Reads `needs_adjudication` from the checked reviewer output.
`Adjudicator`	Receives the original input, both labels, and the report; checks its `FinalLabel`.
`SqliteDb`	Persists workflow runs in `tmp/labeling.db`.

The explicit HumanReview(on_error="fail") settings stop execution when an executor raises. They do not add a human approval pause. The agreement path leaves a skipped-Condition message as the workflow's final content. Read the Reviewer step for its typed DisagreementReport and the two labeler steps for their Contact outputs. The adjudication path returns a FinalLabel. Consumers should inspect the result type and step outputs rather than assume every completed run ends with the same schema.

The reviewer is instructed to flag a field when both labelers return non-null, different values. Change its instructions if a null value and a populated value should also trigger adjudication. Schema checks validate structure; test the model's comparison and extraction quality against labeled inputs separately.

Run the workflow

uv venv .venv --python 3.12
source .venv/bin/activate
uv pip install "agno[google,anthropic,sqlite]"
export GOOGLE_API_KEY="..."
export ANTHROPIC_API_KEY="..."
python quality_pipeline.py

Test against a labeled validation set before using the output. Track reviewer decisions and final labels by prompt and model version. The linked cookbook currently uses unchecked executor results; use the complete adaptation above when you need the explicit failure behavior shown here.

Next steps

Task	Guide
Define the label schema	Data extraction
Inspect workflow patterns	Workflows
Run independent branches	Parallel workflows
Gate a step on a result	Conditional workflows

Quality review pipeline

Execution flow

Run the workflow

Next steps

Developer Resources

On this page