Quality review pipeline

Run two labelers concurrently, persist each workflow run, and adjudicate disagreements.

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

Developer Resources

• Quality review cookbook
• Workflow reference