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. The complete cookbook defines the schemas, agents, and executor functions used in this workflow composition:

from agno.db.sqlite import SqliteDb
from agno.workflow import Step, Workflow
from agno.workflow.condition import Condition
from agno.workflow.parallel import Parallel


label_a = Step(name="Labeler A", agent=labeler_a)
label_b = Step(name="Labeler B", agent=labeler_b)
review = Step(name="Reviewer", executor=run_reviewer)
adjudicate = Step(name="Adjudicator", executor=run_adjudicator)

workflow = Workflow(
    name="Quality review labeling",
    db=SqliteDb(db_file="tmp/labeling.db"),
    steps=[
        Parallel(label_a, label_b, name="Label"),
        review,
        Condition(
            name="Adjudicate",
            evaluator=has_disagreement,
            steps=[adjudicate],
        ),
    ],
)

Execution flow

Step	What it does
`Parallel`	Runs a Google labeler and an Anthropic labeler concurrently against the same input.
`Reviewer`	Reads both step outputs and returns a typed `DisagreementReport`.
`Condition`	Reads `needs_adjudication` from the reviewer output.
`Adjudicator`	Receives the original input, both labels, and the disagreement report. It runs only when the condition passes.
`SqliteDb`	Persists workflow runs in `tmp/labeling.db`.

The cookbook reviewer flags a field when both labelers return non-null, different values. Change the reviewer instructions if a null value and a populated value should also trigger adjudication.

Pass step outputs to the reviewer

StepInput.get_step_output() finds named steps inside the Parallel block. The reviewer executor uses those outputs to build its prompt.

from agno.workflow.types import StepInput, StepOutput


def run_reviewer(step_input: StepInput) -> StepOutput:
    a = step_input.get_step_output("Labeler A").content
    b = step_input.get_step_output("Labeler B").content
    prompt = (
        f"Labeler A:\n{a.model_dump_json(indent=2)}\n\n"
        f"Labeler B:\n{b.model_dump_json(indent=2)}"
    )
    report = reviewer.run(prompt).content
    return StepOutput(content=report)

The condition reads the reviewer output from previous_step_content:

def has_disagreement(step_input: StepInput) -> bool:
    report = step_input.previous_step_content
    return bool(report and getattr(report, "needs_adjudication", False))

Run the workflow

From the Agno repository root:

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

Test the workflow against a labeled validation set before using its output. Track reviewer decisions and final labels by prompt and model version so changes remain measurable.

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

Welcome

Get Started

Features

Use Cases

Quality review pipeline

Execution flow

Pass step outputs to the reviewer

Run the workflow

Next steps

Developer Resources

​Execution flow

​Pass step outputs to the reviewer

​Run the workflow

​Next steps

​Developer Resources

Execution flow

Pass step outputs to the reviewer

Run the workflow

Next steps

Developer Resources