Human routing and eval
Route low-confidence fields for approval and track extraction accuracy against a golden set.
Document processing pipelines need a review path for low-confidence fields and a way to measure extraction quality over time. Add both to the same extraction agent.
Per-field confidence
Place the input scan at scan-low-quality.pdf. The later eval example needs golden/invoice-001.pdf; its batch fragment assumes your own golden_set records with path, id, and expected_description. Install uv pip install "agno[sqlite]" for the SQLite-backed writer and evals.
Wrap each field in a confidence carrier so a downstream check can decide what needs review. The schema is identical to the one in data labeling; the routing logic is the part that lives here.
output_schema requests structured output and Agno attempts to parse it. A failed or unparseable run can leave content as text. Check RunStatus.completed and the expected Pydantic type before reading fields, indexing, or writing downstream. These examples stop on failure; an application can instead send the original input to a review queue. Schema validation checks the shape and constraints, so verify extracted facts against the source separately.
Before running these examples, create and activate a Python environment, then install the provider and set its key:
uv venv .venv --python 3.12
source .venv/bin/activate
uv pip install "agno[openai]"
export OPENAI_API_KEY="..."Replace every https://example.com/... PDF URL with a real PDF URL accessible to the model provider, or use File(filepath="...") for a local PDF. Create the referenced local files before running the example. Extraction quality depends on the document and model; the output comments are illustrative.
from agno.run.base import RunStatus
from typing import Literal, Optional
from agno.agent import Agent
from agno.media import File
from agno.models.openai import OpenAIResponses
from pydantic import BaseModel
Confidence = Literal["high", "medium", "low"]
class ConfidentField(BaseModel):
value: Optional[str] = None
confidence: Confidence
class Invoice(BaseModel):
invoice_number: ConfidentField
vendor: ConfidentField
invoice_date: ConfidentField
total: ConfidentField
agent = Agent(
model=OpenAIResponses(id="gpt-5.5"),
instructions=(
"Extract invoice fields. For each field, report confidence: "
"high (explicit on the document), medium (inferred from structure), "
"low (guessed, partly obscured, or ambiguous). Be conservative."
),
output_schema=Invoice,
)
invoice_run = agent.run(
"Extract this invoice.",
files=[File(filepath="scan-low-quality.pdf")],
)
if invoice_run.status != RunStatus.completed or not isinstance(invoice_run.content, Invoice):
raise RuntimeError("No validated Invoice; send the input to review before continuing")
invoice = invoice_run.content
# Invoice(invoice_number=ConfidentField(value='1042', confidence='high'),
# vendor=ConfidentField(value='Acme Corp', confidence='high'),
# invoice_date=ConfidentField(value=None, confidence='low'),
# total=ConfidentField(value='1296.0', confidence='medium'))Route on low confidence
Walk the extracted fields, find values below the confidence threshold, and route the document in application code.
def low_confidence_fields(invoice: Invoice) -> list[str]:
return [
name
for name, field in invoice.model_dump().items()
if field.get("confidence") == "low"
]
flagged = low_confidence_fields(invoice)
if flagged:
send_to_human_queue(invoice, flagged)
else:
write_to_database(invoice)send_to_human_queue and write_to_database are application callbacks to implement for your systems. The example post_to_erp below is also a placeholder; replace its body with your ERP integration. Treat confidence as an agent-provided routing signal. Application code sets the threshold and chooses the action.
Gate the next action with requires_confirmation
Wrap a downstream action, such as a database write or ERP push, in a tool that requires approval. Every call pauses the run until a human confirms it. This places approval at the system boundary.
from agno.agent import Agent
from agno.db.sqlite import SqliteDb
from agno.models.openai import OpenAIResponses
from agno.tools import tool
@tool(requires_confirmation=True)
def post_to_erp(invoice_id: str, vendor: str, total: float) -> str:
"""Post an extracted invoice to the AP ledger."""
# ...real ERP call...
return f"Posted {invoice_id} for {vendor}: {total}"
db = SqliteDb(db_file="tmp/extraction.db")
writer = Agent(
model=OpenAIResponses(id="gpt-5.5"),
tools=[post_to_erp],
db=db,
instructions="Given a parsed invoice, post it to the ERP with post_to_erp.",
)
run = writer.run(
f"Post this invoice: {invoice.model_dump_json()}"
)
if run.is_paused:
for requirement in run.active_requirements:
if requirement.needs_confirmation:
# Surface this to a reviewer UI; here we approve directly.
print(f"Approve: {requirement.tool_execution.tool_name}")
requirement.confirm()
run = writer.continue_run(
run_id=run.run_id,
session_id=run.session_id,
requirements=run.requirements,
)The pause is persisted in db. A different process can reconstruct the same Agent configuration and continue with the stored run_id and session_id. See human approval for async variants and listing pending approvals from the database.
Accuracy against a golden set
A representative golden set compares extraction behavior across prompt or model changes.
from agno.agent import Agent
from agno.db.sqlite import SqliteDb
from agno.eval.accuracy import AccuracyEval
from agno.media import File
from agno.models.openai import OpenAIResponses
db = SqliteDb(db_file="tmp/extraction.db")
agent = Agent(
model=OpenAIResponses(id="gpt-5.5"),
instructions="Extract invoice fields. Null if missing.",
output_schema=Invoice,
)
run = agent.run(
"Extract this invoice.",
files=[File(filepath="golden/invoice-001.pdf")],
)
if run.status != RunStatus.completed or not isinstance(run.content, Invoice):
raise RuntimeError("No validated Invoice; record an extraction failure instead of scoring it")
evaluation = AccuracyEval(
db=db,
name="invoice-extraction-golden",
model=OpenAIResponses(id="gpt-5.5"),
input="Extract this invoice.",
expected_output=(
"Invoice number 1042, vendor Acme Corp, dated 2026-04-12, "
"total 1296.00 USD."
),
)
result = evaluation.run_with_output(
output=run.content.model_dump_json(), print_results=True
)
# AccuracyResult(avg_score=9.0, ...)
assert result is not None
print(result.avg_score)AccuracyEval.run() executes its configured agent but does not accept document files. Run document extraction first, then pass its serialized output to run_with_output. AccuracyEval uses a model judge to compare that output with expected_output; it is a semantic score, not an exact field-by-field metric.
results = []
for doc in golden_set:
run = agent.run("Extract this invoice.", files=[File(filepath=doc.path)])
if run.status != RunStatus.completed or not isinstance(run.content, Invoice):
raise RuntimeError(f"No validated Invoice for {doc.id}; record an extraction failure")
eval_ = AccuracyEval(
db=db,
name=f"invoice-{doc.id}",
model=OpenAIResponses(id="gpt-5.5"),
input="Extract this invoice.",
expected_output=doc.expected_description,
)
results.append(
eval_.run_with_output(
output=run.content.model_dump_json(), print_results=False, print_summary=False
)
)Passing db=db stores each evaluation result. Compare model-judge scores across repeated runs with a fixed judge configuration, and add deterministic field checks for invoice numbers, dates, totals, and line items. See the evals cookbook for database logging and the team variant.
Review controls
| Pattern | What it answers | When it fires |
|---|---|---|
| Confidence routing | "Which fields on this document need a human?" | Every run, per document |
| Approval-gated tools | "Should we let the agent take the next action?" | At a specific tool boundary |
| AccuracyEval over a golden set | "How does a model judge compare these outputs?" | After a prompt or model change, or on a schedule |
Confidence routing and approval-gated tools act on one document. AccuracyEval records a model-judge signal for comparing configurations.
Next steps
| Task | Guide |
|---|---|
| Schedule the eval to run nightly | Batch and durability |
| Approve from an external UI | Human approval |
| Add a two-labeler review step | Quality pipeline |