> ## Documentation Index
> Fetch the complete documentation index at: https://phidatainc.mintlify.site/llms.txt
> Use this file to discover all available pages before exploring further.

# 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](https://github.com/agno-agi/agno/blob/main/cookbook/data_labeling/_18_quality_review/basic.py) defines the schemas, agents, and executor functions used in this workflow composition:

```python theme={null}
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.

```python theme={null}
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`:

```python theme={null}
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:

```bash theme={null}
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](/use-cases/data-labeling/structured-extraction)          |
| Inspect workflow patterns | [Workflows](/workflows/overview)                                           |
| Run independent branches  | [Parallel workflows](/workflows/workflow-patterns/parallel-workflow)       |
| Gate a step on a result   | [Conditional workflows](/workflows/workflow-patterns/conditional-workflow) |

## Developer Resources

* [Quality review cookbook](https://github.com/agno-agi/agno/tree/main/cookbook/data_labeling/_18_quality_review)
* [Workflow reference](/reference/workflows/workflow)