cookbook: data labeling primitives (18 workflows) (#7976)

## Summary

Adds a new `cookbook/data_labeling/` directory built around **one
cookbook per core data-labeling primitive**. Each subfolder is a
self-contained example with `README.md`, `basic.py`, optional
task-meaningful variants, and a `TEST_LOG.md` per the cookbook
convention.

The 18 workflows, grouped by family:

| Family | Cookbooks |
|---|---|
| Text | `text_classification`, `text_multilabel_classification`,
`text_extraction`, `text_span_labeling`, `text_pairwise_preference` |
| Image | `image_classification`, `image_extraction`,
`image_extraction_to_vectordb`, `image_bounding_boxes` |
| Audio | `audio_classification`, `audio_transcription`,
`audio_extraction` |
| Video | `video_classification`, `video_extraction` |
| Document | `document_classification`, `document_extraction` |
| Composed | `llm_as_judge`, `quality_review` |

Strategic rationale (why labeling, why a primitives-per-cookbook shape)
and the implementation plan live in
`specs/agno/cookbooks/data-labeling/` (private specs repo).

The branch also includes a small refresh of `cookbook/99_docs/`
(`sorting_hat.py` / `workbench.py` scratch examples) — unrelated to the
data-labeling work but bundled in this PR by request.

## Type of change

- [ ] Bug fix
- [x] New feature
- [ ] Breaking change
- [x] Improvement
- [ ] Model update
- [ ] Other:

## Checklist

- [x] Code complies with style guidelines (`cookbook/STYLE_GUIDE.md`)
- [x] Ran format/validation scripts (`./scripts/format.sh` and
`./scripts/validate.sh`)
- [x] Self-review completed
- [x] Documentation updated — per-cookbook READMEs explain purpose,
when-to-use, run commands
- [x] Examples and guides: this PR _is_ examples
- [x] Tested in clean environment — every cookbook run end-to-end; see
each `TEST_LOG.md`
- [ ] Tests added/updated — N/A; cookbook content

### Duplicate and AI-Generated PR Check

- [x] Searched existing open PRs; nothing else addresses this
- [x] No similar PR exists
- [x] This PR was largely AI-generated (Claude Code) under direction;
code was reviewed and end-to-end tested before commit

## Additional Notes

### Conventions

- `OpenAIResponses(id="gpt-5.5")` is the default for text / image /
document workloads
- `Gemini(id="gemini-3-flash-preview")` for audio + video (native
modality input)
- `Claude(id="claude-sonnet-4-5")` for the second labeler in
`quality_review/` (provider diversity)
- Inline Pydantic schemas; no shared `schemas.py` per cookbook
(forkability beats DRY here)
- Variants are task-meaningful (e.g. `with_confidence.py`,
`multilabel.py`), not provider swaps

### Two test-driven fixes worth flagging

- `image_bounding_boxes/with_confidence.py` — gpt-5.5 produced
degenerate (all-zero or all-one) boxes without per-field x/y/w/h
`description=` strings on the Pydantic model. Added descriptions +
expanded coordinate guidance in the instructions; now produces tight
boxes.
- `document_extraction/with_confidence.py` — OpenAI's strict
structured-output mode rejects `Field(..., description=...)` on a field
whose type is a referenced sub-model. Dropped the field-level
description; explanatory text moved to a code comment.

### Out of scope / deliberate omissions

- No batch driver, rate limiter, DLQ, or warehouse export — these are
production patterns and don't belong in primitive cookbooks. The earlier
`pipeline.py` / `run_batch.py` / `export.py` files are removed; if those
patterns get cookbooks later they live under `cookbook/05_production/`
(TBD).
- No labeling UI / Label Studio integration.
- No RLHF training loop (we show the _data shape_ in
`text_pairwise_preference`, not the trainer).
- `image_extraction_to_vectordb/basic.py` requires `lancedb` (not in
`agno[demo]` extras). README documents the install. Worth either folding
into the demo extras or shipping a `requirements.in` for that folder —
happy to do as a follow-up.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

Co-authored-by: Claude Opus 4.7 (1M context) <[email protected]>
This commit is contained in:
Ashpreet
2026-05-19 22:12:19 +01:00
committed by GitHub
co-authored by Claude Opus 4.7
parent 3a0aaaba66
commit bf88d39df7
92 changed files with 4263 additions and 233 deletions
+86 -130
View File
@@ -1,20 +1,14 @@
# This file was autogenerated by uv via the following command:
# ./generate_requirements.sh
agno==2.5.3
agno==2.6.7
# via -r cookbook/00_quickstart/requirements.in
aiohappyeyeballs==2.6.1
# via aiohttp
aiohttp==3.13.3
# via google-genai
aiosignal==1.4.0
# via aiohttp
annotated-doc==0.0.4
# via
# fastapi
# typer
annotated-types==0.7.0
# via pydantic
anyio==4.12.1
anyio==4.13.0
# via
# google-genai
# httpx
@@ -22,20 +16,17 @@ anyio==4.12.1
# sse-starlette
# starlette
# watchfiles
attrs==25.4.0
attrs==26.1.0
# via
# aiohttp
# jsonschema
# referencing
backoff==2.2.1
# via posthog
bcrypt==5.0.0
# via chromadb
beautifulsoup4==4.14.3
# via yfinance
build==1.4.0
build==1.5.0
# via chromadb
certifi==2026.1.4
certifi==2026.4.22
# via
# curl-cffi
# httpcore
@@ -47,28 +38,26 @@ cffi==2.0.0
# via
# cryptography
# curl-cffi
charset-normalizer==3.4.4
charset-normalizer==3.4.7
# via requests
chromadb==1.5.1
chromadb==1.5.9
# via -r cookbook/00_quickstart/requirements.in
click==8.3.1
click==8.4.0
# via
# rich-toolkit
# typer
# uvicorn
cryptography==46.0.5
cryptography==48.0.0
# via
# google-auth
# pyjwt
curl-cffi==0.13.0
curl-cffi==0.15.0
# via yfinance
distro==1.9.0
# via
# google-genai
# posthog
# via google-genai
dnspython==2.8.0
# via email-validator
docstring-parser==0.17.0
docstring-parser==0.18.0
# via agno
durationpy==0.10
# via kubernetes
@@ -76,37 +65,35 @@ email-validator==2.3.0
# via
# fastapi
# pydantic
fastapi==0.129.0
fastapi==0.136.1
# via -r cookbook/00_quickstart/requirements.in
fastapi-cli==0.0.23
fastapi-cli==0.0.24
# via fastapi
fastapi-cloud-cli==0.13.0
fastapi-cloud-cli==0.17.1
# via fastapi-cli
fastar==0.8.0
# via fastapi-cloud-cli
filelock==3.24.3
fastar==0.11.0
# via
# fastapi
# fastapi-cloud-cli
filelock==3.29.0
# via huggingface-hub
flatbuffers==25.12.19
# via onnxruntime
frozendict==2.4.7
# via yfinance
frozenlist==1.8.0
# via
# aiohttp
# aiosignal
fsspec==2026.2.0
fsspec==2026.4.0
# via huggingface-hub
gitdb==4.0.12
# via gitpython
gitpython==3.1.46
gitpython==3.1.50
# via agno
google-auth==2.48.0
google-auth==2.53.0
# via google-genai
google-genai==1.64.0
google-genai==2.4.0
# via -r cookbook/00_quickstart/requirements.in
googleapis-common-protos==1.72.0
googleapis-common-protos==1.75.0
# via opentelemetry-exporter-otlp-proto-grpc
grpcio==1.78.0
grpcio==1.80.0
# via
# chromadb
# opentelemetry-exporter-otlp-proto-grpc
@@ -117,7 +104,7 @@ h11==0.16.0
# uvicorn
h2==4.3.0
# via httpx
hf-xet==1.2.0
hf-xet==1.5.0
# via huggingface-hub
hpack==4.1.0
# via h2
@@ -136,20 +123,17 @@ httpx==0.28.1
# mcp
httpx-sse==0.4.3
# via mcp
huggingface-hub==1.4.1
huggingface-hub==1.15.0
# via tokenizers
hyperframe==6.1.0
# via h2
idna==3.11
idna==3.15
# via
# anyio
# email-validator
# httpx
# requests
# yarl
importlib-metadata==8.7.1
# via opentelemetry-api
importlib-resources==6.5.2
importlib-resources==7.1.0
# via chromadb
jinja2==3.1.6
# via fastapi
@@ -161,27 +145,21 @@ jsonschema-specifications==2025.9.1
# via jsonschema
kubernetes==35.0.0
# via chromadb
markdown-it-py==4.0.0
markdown-it-py==4.2.0
# via rich
markupsafe==3.0.3
# via jinja2
mcp==1.26.0
mcp==1.27.1
# via -r cookbook/00_quickstart/requirements.in
mdurl==0.1.2
# via markdown-it-py
mmh3==5.2.0
mmh3==5.2.1
# via chromadb
mpmath==1.3.0
# via sympy
multidict==6.7.1
# via
# aiohttp
# yarl
multitasking==0.0.12
multitasking==0.0.13
# via yfinance
nest-asyncio==1.6.0
# via -r cookbook/00_quickstart/requirements.in
numpy==2.4.2
numpy==2.4.6
# via
# chromadb
# onnxruntime
@@ -190,17 +168,17 @@ numpy==2.4.2
# yfinance
oauthlib==3.3.1
# via requests-oauthlib
onnxruntime==1.24.2
onnxruntime==1.26.0
# via chromadb
openinference-instrumentation==0.1.44
openinference-instrumentation==0.1.51
# via openinference-instrumentation-agno
openinference-instrumentation-agno==0.1.28
openinference-instrumentation-agno==0.1.34
# via -r cookbook/00_quickstart/requirements.in
openinference-semantic-conventions==0.1.26
openinference-semantic-conventions==0.1.29
# via
# openinference-instrumentation
# openinference-instrumentation-agno
opentelemetry-api==1.39.1
opentelemetry-api==1.42.0
# via
# -r cookbook/00_quickstart/requirements.in
# chromadb
@@ -210,75 +188,67 @@ opentelemetry-api==1.39.1
# opentelemetry-instrumentation
# opentelemetry-sdk
# opentelemetry-semantic-conventions
opentelemetry-exporter-otlp-proto-common==1.39.1
opentelemetry-exporter-otlp-proto-common==1.42.0
# via opentelemetry-exporter-otlp-proto-grpc
opentelemetry-exporter-otlp-proto-grpc==1.39.1
opentelemetry-exporter-otlp-proto-grpc==1.42.0
# via chromadb
opentelemetry-instrumentation==0.60b1
opentelemetry-instrumentation==0.63b0
# via openinference-instrumentation-agno
opentelemetry-proto==1.39.1
opentelemetry-proto==1.42.0
# via
# opentelemetry-exporter-otlp-proto-common
# opentelemetry-exporter-otlp-proto-grpc
opentelemetry-sdk==1.39.1
opentelemetry-sdk==1.42.0
# via
# -r cookbook/00_quickstart/requirements.in
# chromadb
# openinference-instrumentation
# opentelemetry-exporter-otlp-proto-grpc
opentelemetry-semantic-conventions==0.60b1
opentelemetry-semantic-conventions==0.63b0
# via
# openinference-instrumentation-agno
# opentelemetry-instrumentation
# opentelemetry-sdk
orjson==3.11.7
orjson==3.11.9
# via chromadb
overrides==7.7.0
# via chromadb
packaging==26.0
packaging==26.2
# via
# agno
# build
# huggingface-hub
# onnxruntime
# opentelemetry-instrumentation
pandas==3.0.1
pandas==3.0.3
# via yfinance
peewee==3.19.0
peewee==4.0.5
# via yfinance
pgvector==0.4.2
# via -r cookbook/00_quickstart/requirements.in
pillow==12.1.1
pillow==12.2.0
# via -r cookbook/00_quickstart/requirements.in
platformdirs==4.9.2
platformdirs==4.9.6
# via yfinance
posthog==5.4.0
# via chromadb
propcache==0.4.1
# via
# aiohttp
# yarl
protobuf==6.33.5
protobuf==6.33.6
# via
# googleapis-common-protos
# onnxruntime
# opentelemetry-proto
# yfinance
psycopg==3.3.3
psycopg==3.3.4
# via -r cookbook/00_quickstart/requirements.in
psycopg-binary==3.3.3
psycopg-binary==3.3.4
# via psycopg
pyasn1==0.6.2
# via
# pyasn1-modules
# rsa
pyasn1==0.6.3
# via pyasn1-modules
pyasn1-modules==0.4.2
# via google-auth
pybase64==1.4.3
# via chromadb
pycparser==3.0
# via cffi
pydantic==2.12.5
pydantic==2.13.4
# via
# agno
# chromadb
@@ -288,18 +258,19 @@ pydantic==2.12.5
# mcp
# pydantic-extra-types
# pydantic-settings
pydantic-core==2.41.5
pydantic-core==2.46.4
# via pydantic
pydantic-extra-types==2.11.0
pydantic-extra-types==2.11.1
# via fastapi
pydantic-settings==2.13.1
pydantic-settings==2.14.1
# via
# agno
# chromadb
# fastapi
# mcp
pygments==2.19.2
pygments==2.20.0
# via rich
pyjwt==2.11.0
pyjwt==2.12.1
# via mcp
pypika==0.51.1
# via chromadb
@@ -309,18 +280,17 @@ python-dateutil==2.9.0.post0
# via
# kubernetes
# pandas
# posthog
python-dotenv==1.2.1
python-dotenv==1.2.2
# via
# agno
# pydantic-settings
# uvicorn
python-multipart==0.0.22
python-multipart==0.0.29
# via
# agno
# fastapi
# mcp
pytz==2025.2
pytz==2026.2
# via yfinance
pyyaml==6.0.3
# via
@@ -333,23 +303,23 @@ referencing==0.37.0
# via
# jsonschema
# jsonschema-specifications
requests==2.32.5
requests==2.34.2
# via
# google-auth
# google-genai
# kubernetes
# posthog
# requests-oauthlib
# yfinance
requests-oauthlib==2.0.0
# via kubernetes
rich==14.3.3
rich==15.0.0
# via
# agno
# chromadb
# curl-cffi
# rich-toolkit
# typer
rich-toolkit==0.19.4
rich-toolkit==0.19.9
# via
# fastapi-cli
# fastapi-cloud-cli
@@ -359,59 +329,49 @@ rpds-py==0.30.0
# via
# jsonschema
# referencing
rsa==4.9.1
# via google-auth
sentry-sdk==2.53.0
sentry-sdk==2.60.0
# via fastapi-cloud-cli
shellingham==1.5.4
# via
# huggingface-hub
# typer
# via typer
six==1.17.0
# via
# kubernetes
# posthog
# python-dateutil
smmap==5.0.2
smmap==5.0.3
# via gitdb
sniffio==1.3.1
# via google-genai
soupsieve==2.8.3
# via beautifulsoup4
sqlalchemy==2.0.46
sqlalchemy==2.0.49
# via -r cookbook/00_quickstart/requirements.in
sse-starlette==3.2.0
sse-starlette==3.4.4
# via mcp
starlette==0.52.1
starlette==1.0.0
# via
# fastapi
# mcp
# sse-starlette
sympy==1.14.0
# via onnxruntime
tenacity==9.1.4
# via
# chromadb
# google-genai
tokenizers==0.22.2
tokenizers==0.23.1
# via chromadb
tqdm==4.67.3
# via
# chromadb
# huggingface-hub
typer==0.24.0
typer==0.25.1
# via
# agno
# chromadb
# fastapi-cli
# fastapi-cloud-cli
# typer-slim
typer-slim==0.24.0
# via huggingface-hub
# huggingface-hub
typing-extensions==4.15.0
# via
# agno
# aiosignal
# anyio
# beautifulsoup4
# chromadb
@@ -440,12 +400,12 @@ typing-inspection==0.4.2
# mcp
# pydantic
# pydantic-settings
urllib3==2.6.3
urllib3==2.7.0
# via
# kubernetes
# requests
# sentry-sdk
uvicorn==0.41.0
uvicorn==0.47.0
# via
# chromadb
# fastapi
@@ -454,23 +414,19 @@ uvicorn==0.41.0
# mcp
uvloop==0.22.1
# via uvicorn
watchfiles==1.1.1
watchfiles==1.2.0
# via uvicorn
websocket-client==1.9.0
# via kubernetes
websockets==15.0.1
websockets==16.0
# via
# google-genai
# uvicorn
# yfinance
wrapt==1.17.3
wrapt==2.1.2
# via
# openinference-instrumentation
# openinference-instrumentation-agno
# opentelemetry-instrumentation
yarl==1.22.0
# via aiohttp
yfinance==1.2.0
yfinance==1.3.0
# via -r cookbook/00_quickstart/requirements.in
zipp==3.23.0
# via importlib-metadata
-18
View File
@@ -1,18 +0,0 @@
from agno.agent import Agent
from agno.db.sqlite import SqliteDb
from agno.os import AgentOS
from agno.tools.workspace import Workspace
workbench = Agent(
name="Workbench",
model="openai:gpt-5.4",
db=SqliteDb(db_file="agno.db"), # session storage
tools=[Workspace(".")], # read/write/edit/shell in this directory
enable_agentic_memory=True, # remembers across sessions
add_history_to_context=True, # include past runs
num_history_runs=3, # last 3 conversations
)
# Serve via AgentOS → streaming, auth, session isolation, API endpoints
agent_os = AgentOS(agents=[workbench], tracing=True)
app = agent_os.get_app()
View File
-23
View File
@@ -1,23 +0,0 @@
from agno.agent import Agent
from agno.db.sqlite import SqliteDb
from agno.os import AgentOS
from agno.tools.workspace import Workspace
agent = Agent(
name="Agno Agent",
model="openai:gpt-5.4",
tools=[
Workspace(
root=".",
allowed=["read", "list", "search"],
confirm=["write", "edit", "delete", "shell"],
)
],
)
agent_os = AgentOS(
agents=[agent],
tracing=True,
db=SqliteDb(db_file="tmp/agentos.db"),
)
app = agent_os.get_app()
-17
View File
@@ -1,17 +0,0 @@
from agno.agents.claude import ClaudeAgent
from agno.db.sqlite import SqliteDb
from agno.os import AgentOS
agent = ClaudeAgent(
name="Claude Agent",
model="claude-opus-4-7",
allowed_tools=["Read", "Bash"],
permission_mode="acceptEdits",
)
agent_os = AgentOS(
agents=[agent],
tracing=True,
db=SqliteDb(db_file="tmp/agentos.db"),
)
app = agent_os.get_app()
-18
View File
@@ -1,18 +0,0 @@
import dspy
from agno.agents.dspy import DSPyAgent
from agno.db.sqlite import SqliteDb
from agno.os import AgentOS
dspy.configure(lm=dspy.LM("openai/gpt-5.4"))
agent = DSPyAgent(
name="DSPy Assistant",
program=dspy.ChainOfThought("question -> answer"),
)
agent_os = AgentOS(
agents=[agent],
tracing=True,
db=SqliteDb(db_file="tmp/agentos.db"),
)
app = agent_os.get_app()
-27
View File
@@ -1,27 +0,0 @@
from agno.agents.langgraph import LangGraphAgent
from agno.db.sqlite import SqliteDb
from agno.os import AgentOS
from langchain_openai import ChatOpenAI
from langgraph.graph import MessagesState, StateGraph
def chatbot(state: MessagesState):
return {"messages": [ChatOpenAI(model="gpt-5.4").invoke(state["messages"])]}
graph = StateGraph(MessagesState)
graph.add_node("chatbot", chatbot)
graph.set_entry_point("chatbot")
compiled = graph.compile()
agent = LangGraphAgent(
name="LangGraph Chatbot",
graph=compiled,
)
agent_os = AgentOS(
agents=[agent],
tracing=True,
db=SqliteDb(db_file="tmp/agentos.db"),
)
app = agent_os.get_app()
+20
View File
@@ -0,0 +1,20 @@
from pathlib import Path
from agno.agent import Agent
from agno.tools.workspace import Workspace
folder = Path(__file__).parent
sorting_hat = Agent(
name="Sorting Hat",
model="openai:gpt-5.5",
tools=[Workspace(root=str(folder), allowed=["read", "list", "search", "shell"])],
instructions=(
"Walk the folder, figure out what's there, and propose a clean organization. "
"Decide the categories yourself. Use shell commands when they help (e.g. `file`, "
"`pdftotext`). Return a tidy summary, a category breakdown, and a folder tree."
),
markdown=True,
)
sorting_hat.print_response(f"Inventory and organize {folder}", stream=True)
+18
View File
@@ -0,0 +1,18 @@
from agno.agent import Agent
from agno.db.sqlite import SqliteDb
from agno.os import AgentOS
from agno.tools.workspace import Workspace
workbench = Agent(
name="Workbench",
model="openai:gpt-5.5",
db=SqliteDb(db_file="workbench.db"), # session storage
tools=[Workspace(".")], # operate in this directory
enable_agentic_memory=True, # remembers across sessions
add_history_to_context=True, # add past runs to context
num_history_runs=3, # last 3 runs
)
# Serve via AgentOS, get streaming, auth, session isolation, API endpoints
agent_os = AgentOS(agents=[workbench], tracing=True)
app = agent_os.get_app()
+83
View File
@@ -0,0 +1,83 @@
# Data labeling
End-to-end examples for data classification and labeling using agents.
Each subfolder holds examples for one theme, containing a `basic.py` that runs end-to-end, plus variants that add task-meaningful options on top.
Workflows are organized by modality (text, image, audio, video, document) and output shape (classify, extract, rank, span-label). Two further patterns (`_17_llm_as_judge`, `_18_quality_review`) compose on top of any of these.
Start with [`_01_text_classification/basic.py`](_01_text_classification/basic.py). Every other cookbook mirrors its structure.
## Layout
````
cookbook/data_labeling/
├── README.md
├── <workflow>/
│ ├── README.md
│ ├── basic.py # smallest readable example
│ ├── <variant>.py # one file per task-meaningful variant
│ ├── schemas.py # shared Pydantic types, if any
│ ├── data/ # sample inputs or dataset pointers
│ └── TEST_LOG.md # run log per the cookbook convention
└── ...
````
## Workflows
### Text
- [`_01_text_classification/`](_01_text_classification/): assign one of N labels (sentiment, intent, topic).
- [`_02_text_multilabel_classification/`](_02_text_multilabel_classification/): assign any subset of N tags, optionally hierarchical.
- [`_03_text_extraction/`](_03_text_extraction/): text into a typed Pydantic object (entities, fields, nested structures).
- [`_04_text_span_labeling/`](_04_text_span_labeling/): mark character or token spans (NER, PII detection, claim and evidence highlighting).
- [`_05_text_pairwise_preference/`](_05_text_pairwise_preference/): rank A vs B against a rubric (RLHF data shape).
### Image
- [`_06_image_classification/`](_06_image_classification/): single or multi-label per image.
- [`_07_image_extraction/`](_07_image_extraction/): image into a typed object (attributes, OCR fields, captions).
- [`_09_image_extraction_to_vectordb/`](_09_image_extraction_to_vectordb/): extract, embed, and store for similarity search.
- [`_08_image_bounding_boxes/`](_08_image_bounding_boxes/): region detection with `(x, y, w, h)` per object.
### Audio
- [`_10_audio_classification/`](_10_audio_classification/): clip-level labels (language, speaker, emotion, genre).
- [`_11_audio_transcription/`](_11_audio_transcription/): speech-to-text with optional diarization and timestamps.
- [`_12_audio_extraction/`](_12_audio_extraction/): call or meeting recording into a typed object (action items, attendees, decisions).
### Video
- [`_13_video_classification/`](_13_video_classification/): clip-level labels.
- [`_14_video_extraction/`](_14_video_extraction/): events, scene descriptions, action timestamps.
### Document
- [`_15_document_classification/`](_15_document_classification/): invoice, receipt, contract, spec sheet.
- [`_16_document_extraction/`](_16_document_extraction/): multipage PDF into a typed object, with line items where relevant.
### Composed patterns
These layer on top of any modality.
- [`_17_llm_as_judge/`](_17_llm_as_judge/): score outputs against a rubric. The same machinery as labeling, repurposed for evals.
- [`_18_quality_review/`](_18_quality_review/): labeler, reviewer, adjudicator pipeline applied on top of an extraction primitive.
## Running a cookbook
From the agno repo root, create and activate the demo venv:
```bash
./scripts/demo_setup.sh
```
```bash
source .venvs/demo/bin/activate
```
```bash
python cookbook/data_labeling/_01_text_classification/basic.py
```
Each subfolder's `README.md` documents its inputs, the model it expects, and any extra dependencies.
| Variable | Used by |
|---|---|
| `OPENAI_API_KEY` | Default for text and most extraction cookbooks |
| `ANTHROPIC_API_KEY` | Image and document cookbooks where Claude is the picked model |
| `GOOGLE_API_KEY` | Audio and video cookbooks (Gemini) |
The per-cookbook README calls out which model it uses and why.
@@ -0,0 +1,37 @@
# Text Classification
Assign one of a fixed set of labels to a piece of text — the simplest data
labeling primitive. Input is a string; output is a label from a closed set.
## Files
- `basic.py` — text → single label.
- `with_confidence.py` — adds self-reported confidence per prediction. Use
when you need to route low-confidence cases to a human or a stronger
model.
- `with_rationale.py` — adds a short rationale string explaining why this
label was chosen. Useful for auditability and as training data.
## When to use
When the output is one of a fixed, exhaustive set of labels:
- Sentiment: positive / negative / neutral
- Intent: refund / complaint / question / praise
- Topic: sports / politics / tech / health
- Quality bucket: good / mediocre / poor
If multiple labels can apply at once, use
[`_02_text_multilabel_classification/`](../_02_text_multilabel_classification/).
If the output is structured (entities, fields), use
[`_03_text_extraction/`](../_03_text_extraction/).
## Run
```bash
python cookbook/data_labeling/_01_text_classification/basic.py
python cookbook/data_labeling/_01_text_classification/with_confidence.py
python cookbook/data_labeling/_01_text_classification/with_rationale.py
```
Requires `OPENAI_API_KEY`.
@@ -0,0 +1,33 @@
# Test Log - _01_text_classification
Tested 2026-05-17 against `gpt-5.5` (OpenAIResponses), agno 2.6.6.
### basic.py
**Status:** PASS
**Description:** Sentiment classification (positive/negative/neutral) over three product reviews.
**Result:** All three samples classified as expected.
---
### with_confidence.py
**Status:** PASS
**Description:** Same task as basic.py with an extra `confidence` field on the output.
**Result:** Classifications correct; confidence tracks ambiguity (e.g. "It's fine I guess." -> neutral, medium).
---
### with_rationale.py
**Status:** PASS
**Description:** Same task with a free-text rationale alongside each label.
**Result:** Rationales correctly cite the deciding phrases in each input.
---
@@ -0,0 +1,49 @@
"""
Text Classification - Basic
===========================
Assign one of a fixed set of labels to a piece of text. The simplest data
labeling primitive: input is a string, output is a label from a closed set.
This example classifies short product reviews into sentiment classes.
"""
from typing import Literal
from agno.agent import Agent, RunOutput # noqa
from agno.models.openai import OpenAIResponses
from pydantic import BaseModel, Field
from rich.pretty import pprint # noqa
# ---------------------------------------------------------------------------
# Schema
# ---------------------------------------------------------------------------
class Classification(BaseModel):
label: Literal["positive", "negative", "neutral"] = Field(
..., description="The assigned sentiment label"
)
# ---------------------------------------------------------------------------
# Create Agent
# ---------------------------------------------------------------------------
agent = Agent(
model=OpenAIResponses(id="gpt-5.5"),
instructions="You classify product reviews by sentiment.",
output_schema=Classification,
)
# ---------------------------------------------------------------------------
# Run Agent
# ---------------------------------------------------------------------------
if __name__ == "__main__":
samples = [
"I love this product, fantastic quality and fast shipping.",
"Broken on arrival, total waste of money.",
"It works as described, nothing special.",
]
for text in samples:
run: RunOutput = agent.run(text)
pprint({"input": text, "result": run.content})
@@ -0,0 +1,61 @@
"""
Text Classification - With Confidence
=====================================
Adds a per-prediction confidence field. Use when downstream consumers need
to route low-confidence labels to a human queue or to a stronger model.
"""
from typing import Literal
from agno.agent import Agent, RunOutput # noqa
from agno.models.openai import OpenAIResponses
from pydantic import BaseModel, Field
from rich.pretty import pprint # noqa
# ---------------------------------------------------------------------------
# Schema
# ---------------------------------------------------------------------------
class Classification(BaseModel):
label: Literal["positive", "negative", "neutral"] = Field(
..., description="The assigned sentiment label"
)
confidence: Literal["high", "medium", "low"] = Field(
..., description="Self-reported confidence in the label"
)
# ---------------------------------------------------------------------------
# Agent Instructions
# ---------------------------------------------------------------------------
instructions = """\
Classify the sentiment of the input text. Report a confidence level:
- high - the sentiment is clear and unambiguous
- medium - the sentiment is mostly clear but with some hedging or mixed signals
- low - the text is sarcastic, ambiguous, or off-topic
"""
# ---------------------------------------------------------------------------
# Create Agent
# ---------------------------------------------------------------------------
agent = Agent(
model=OpenAIResponses(id="gpt-5.5"),
instructions=instructions,
output_schema=Classification,
)
# ---------------------------------------------------------------------------
# Run Agent
# ---------------------------------------------------------------------------
if __name__ == "__main__":
samples = [
"Best purchase of my life, life-changing!",
"It's fine I guess.",
"Yeah right, this thing is 'amazing'.",
]
for text in samples:
run: RunOutput = agent.run(text)
pprint({"input": text, "result": run.content})
@@ -0,0 +1,58 @@
"""
Text Classification - With Rationale
====================================
Adds a short rationale explaining the label. Useful for auditability and
for training datasets where the reasoning trace is itself an artifact.
"""
from typing import Literal
from agno.agent import Agent, RunOutput # noqa
from agno.models.openai import OpenAIResponses
from pydantic import BaseModel, Field
from rich.pretty import pprint # noqa
# ---------------------------------------------------------------------------
# Schema
# ---------------------------------------------------------------------------
class Classification(BaseModel):
label: Literal["positive", "negative", "neutral"] = Field(
..., description="The assigned sentiment label"
)
rationale: str = Field(
..., description="One sentence explaining why this label was chosen"
)
# ---------------------------------------------------------------------------
# Agent Instructions
# ---------------------------------------------------------------------------
instructions = """\
Classify the sentiment of the input text. Quote or paraphrase the specific
words that drove your decision in the rationale.
"""
# ---------------------------------------------------------------------------
# Create Agent
# ---------------------------------------------------------------------------
agent = Agent(
model=OpenAIResponses(id="gpt-5.5"),
instructions=instructions,
output_schema=Classification,
)
# ---------------------------------------------------------------------------
# Run Agent
# ---------------------------------------------------------------------------
if __name__ == "__main__":
samples = [
"Shipping was fast but the product itself fell apart in a week.",
"Better than expected, will buy again.",
]
for text in samples:
run: RunOutput = agent.run(text)
pprint({"input": text, "result": run.content})
@@ -0,0 +1,29 @@
# Text Multilabel Classification
Assign any subset of N tags to a piece of text. Unlike single-label
classification, multiple tags can apply to the same input.
## Files
- `basic.py` — text → set of tags from a flat label space.
- `with_confidence.py` — adds per-tag confidence.
- `hierarchical.py` — parent/child tags from a taxonomy.
## When to use
- Restaurant reviews tagged by `[food, service, value, atmosphere, cleanliness]`.
- Customer support tickets tagged by `[bug, feature_request, billing, account]`.
- News articles tagged by topic taxonomies (`sports/football`, `tech/ai`).
If exactly one label applies, use
[`_01_text_classification/`](../_01_text_classification/) instead.
## Run
```bash
python cookbook/data_labeling/_02_text_multilabel_classification/basic.py
python cookbook/data_labeling/_02_text_multilabel_classification/with_confidence.py
python cookbook/data_labeling/_02_text_multilabel_classification/hierarchical.py
```
Requires `OPENAI_API_KEY`.
@@ -0,0 +1,33 @@
# Test Log - _02_text_multilabel_classification
Tested 2026-05-17 against `gpt-5.5` (OpenAIResponses), agno 2.6.6.
### basic.py
**Status:** PASS
**Description:** Tag restaurant reviews with any subset of {food, service, value, atmosphere, cleanliness}.
**Result:** Tags match the aspects mentioned in each review.
---
### with_confidence.py
**Status:** PASS
**Description:** Same task with per-tag confidence.
**Result:** Reasonable tag/confidence pairs; empty list returned when no aspects are addressed.
---
### hierarchical.py
**Status:** PASS
**Description:** Two-level (parent/child) tagging on news snippets.
**Result:** Tags correctly nested (e.g. business -> markets, tech -> ai chips).
---
@@ -0,0 +1,62 @@
"""
Text Multilabel Classification - Basic
======================================
Assign any subset of a fixed tag set to a piece of text. Multiple tags may
apply to the same input.
This example tags restaurant reviews by which aspects the reviewer commented
on.
"""
from typing import List, Literal
from agno.agent import Agent, RunOutput # noqa
from agno.models.openai import OpenAIResponses
from pydantic import BaseModel, Field
from rich.pretty import pprint # noqa
Aspect = Literal["food", "service", "value", "atmosphere", "cleanliness"]
# ---------------------------------------------------------------------------
# Schema
# ---------------------------------------------------------------------------
class Tagging(BaseModel):
tags: List[Aspect] = Field(
..., description="All aspects the reviewer commented on; empty if none"
)
# ---------------------------------------------------------------------------
# Agent Instructions
# ---------------------------------------------------------------------------
instructions = """\
Tag the review with every aspect the reviewer commented on. Include an
aspect only when the text actually addresses it. An aspect can be mentioned
positively or negatively - both count.
"""
# ---------------------------------------------------------------------------
# Create Agent
# ---------------------------------------------------------------------------
agent = Agent(
model=OpenAIResponses(id="gpt-5.5"),
instructions=instructions,
output_schema=Tagging,
)
# ---------------------------------------------------------------------------
# Run Agent
# ---------------------------------------------------------------------------
if __name__ == "__main__":
samples = [
"Pasta was excellent and our server was attentive. A bit pricey but worth it.",
"Place was filthy. Floors sticky, bathroom unusable.",
"Came for the vibes, stayed for the cocktails. The space is gorgeous.",
]
for text in samples:
run: RunOutput = agent.run(text)
pprint({"input": text, "result": run.content})
@@ -0,0 +1,72 @@
"""
Text Multilabel Classification - Hierarchical
=============================================
Tags drawn from a two-level taxonomy: a parent category and a child within
that category. Useful when the label space is large and naturally nested
(news topics, product catalogs, support categories).
"""
from typing import List, Literal
from agno.agent import Agent, RunOutput # noqa
from agno.models.openai import OpenAIResponses
from pydantic import BaseModel, Field
from rich.pretty import pprint # noqa
ParentTopic = Literal["sports", "politics", "tech", "business", "health"]
# ---------------------------------------------------------------------------
# Schema
# ---------------------------------------------------------------------------
class HierarchicalTag(BaseModel):
parent: ParentTopic
child: str = Field(
...,
description=(
"Specific subtopic within the parent. Examples: "
"sports -> football | basketball | tennis; "
"tech -> ai | hardware | security; "
"business -> markets | startups | regulation."
),
)
class Tagging(BaseModel):
tags: List[HierarchicalTag]
# ---------------------------------------------------------------------------
# Agent Instructions
# ---------------------------------------------------------------------------
instructions = """\
Tag the news article with all parent/child pairs it covers. The child must
be a meaningful subtopic of the parent, and should reflect what the article
is actually about - not every entity mentioned in passing.
"""
# ---------------------------------------------------------------------------
# Create Agent
# ---------------------------------------------------------------------------
agent = Agent(
model=OpenAIResponses(id="gpt-5.5"),
instructions=instructions,
output_schema=Tagging,
)
# ---------------------------------------------------------------------------
# Run Agent
# ---------------------------------------------------------------------------
if __name__ == "__main__":
samples = [
"The Fed held rates steady as markets reacted to a surprise jobs report. "
"Tech stocks led the rally, with AI chipmakers up 4 percent.",
"Manchester United fired their head coach after a third consecutive loss. "
"The board is reportedly courting a replacement from Spain.",
]
for text in samples:
run: RunOutput = agent.run(text)
pprint({"input": text, "result": run.content})
@@ -0,0 +1,67 @@
"""
Text Multilabel Classification - With Confidence
================================================
Adds per-tag confidence. Useful for downstream routing: low-confidence tags
can be flagged for human review while high-confidence ones land straight in
the training set.
"""
from typing import List, Literal
from agno.agent import Agent, RunOutput # noqa
from agno.models.openai import OpenAIResponses
from pydantic import BaseModel, Field
from rich.pretty import pprint # noqa
Aspect = Literal["food", "service", "value", "atmosphere", "cleanliness"]
# ---------------------------------------------------------------------------
# Schema
# ---------------------------------------------------------------------------
class Tag(BaseModel):
aspect: Aspect
confidence: Literal["high", "medium", "low"] = Field(
..., description="Confidence that the review actually addresses this aspect"
)
class Tagging(BaseModel):
tags: List[Tag]
# ---------------------------------------------------------------------------
# Agent Instructions
# ---------------------------------------------------------------------------
instructions = """\
Tag the review with every aspect the reviewer addresses. For each tag,
report confidence:
- high - explicit, unambiguous mention
- medium - implicit or partial mention
- low - inferred, hedged, or could be the reviewer just venting
"""
# ---------------------------------------------------------------------------
# Create Agent
# ---------------------------------------------------------------------------
agent = Agent(
model=OpenAIResponses(id="gpt-5.5"),
instructions=instructions,
output_schema=Tagging,
)
# ---------------------------------------------------------------------------
# Run Agent
# ---------------------------------------------------------------------------
if __name__ == "__main__":
samples = [
"Pasta was excellent and the server brought refills without asking.",
"Not sure I'd come back. Something was off.",
"Cocktails were $22. The room is loud. Food was fine.",
]
for text in samples:
run: RunOutput = agent.run(text)
pprint({"input": text, "result": run.content})
@@ -0,0 +1,34 @@
# Text Extraction
Extract typed structured data from free-form text. The output is a Pydantic
object whose schema you control. The most common labeling shape in
production today.
## Files
- `basic.py` — text → flat typed object (single record).
- `with_confidence.py` — adds per-field confidence using a shared
`ConfidentField` wrapper.
- `nested.py` — extract a list of nested sub-objects (action items, line
items, attendees, etc.).
## When to use
- Pull contact info out of an email signature.
- Extract action items from a meeting transcript.
- Lift fields from unstructured user input into a database row.
If you only need a single label, use
[`_01_text_classification/`](../_01_text_classification/). If you need character
positions of mentioned entities, see
[`_04_text_span_labeling/`](../_04_text_span_labeling/).
## Run
```bash
python cookbook/data_labeling/_03_text_extraction/basic.py
python cookbook/data_labeling/_03_text_extraction/with_confidence.py
python cookbook/data_labeling/_03_text_extraction/nested.py
```
Requires `OPENAI_API_KEY`.
@@ -0,0 +1,33 @@
# Test Log - _03_text_extraction
Tested 2026-05-17 against `gpt-5.5` (OpenAIResponses), agno 2.6.6.
### basic.py
**Status:** PASS
**Description:** Extract `Contact` (name, email, phone, company, title) from email-signature-style text.
**Result:** All fields extracted verbatim; missing fields left null.
---
### with_confidence.py
**Status:** PASS
**Description:** Same task with `ConfidentField` wrapping each value.
**Result:** Confidences are sensible: high for explicit fields, medium for inferred ones (e.g. "@mike" as a name).
---
### nested.py
**Status:** PASS
**Description:** Extract a `Meeting` containing a list of nested `ActionItem`s from a meeting transcript.
**Result:** Three action items extracted with correct owners and due dates.
---
@@ -0,0 +1,61 @@
"""
Text Extraction - Basic
=======================
Extract typed structured data from free-form text. The output is a Pydantic
object whose schema you control.
This example extracts contact info from an email signature.
"""
from typing import Optional
from agno.agent import Agent, RunOutput # noqa
from agno.models.openai import OpenAIResponses
from pydantic import BaseModel, Field
from rich.pretty import pprint # noqa
# ---------------------------------------------------------------------------
# Schema
# ---------------------------------------------------------------------------
class Contact(BaseModel):
name: Optional[str] = Field(None, description="Full name as written")
email: Optional[str] = Field(None, description="Email address")
phone: Optional[str] = Field(None, description="Phone number, raw format")
company: Optional[str] = Field(None, description="Company or organization")
title: Optional[str] = Field(None, description="Job title")
# ---------------------------------------------------------------------------
# Agent Instructions
# ---------------------------------------------------------------------------
instructions = """\
Extract contact information from the input. Use exactly what the text shows
- do not normalize or reformat. If a field is missing, leave it null. Do
not guess.
"""
# ---------------------------------------------------------------------------
# Create Agent
# ---------------------------------------------------------------------------
agent = Agent(
model=OpenAIResponses(id="gpt-5.5"),
instructions=instructions,
output_schema=Contact,
)
# ---------------------------------------------------------------------------
# Run Agent
# ---------------------------------------------------------------------------
if __name__ == "__main__":
samples = [
"Hi - Sarah Johnson, VP of Marketing at Acme Corp. "
"[email protected] / +1-555-0102.",
"regards, Mike ([email protected])",
]
for text in samples:
run: RunOutput = agent.run(text)
pprint({"input": text, "result": run.content})
@@ -0,0 +1,65 @@
"""
Text Extraction - Nested
========================
Extract a list of nested sub-objects from text. The same shape used for
line items, meeting attendees, action items, citations, etc.
This example extracts action items from a meeting transcript.
"""
from typing import List, Optional
from agno.agent import Agent, RunOutput # noqa
from agno.models.openai import OpenAIResponses
from pydantic import BaseModel, Field
from rich.pretty import pprint # noqa
# ---------------------------------------------------------------------------
# Schema
# ---------------------------------------------------------------------------
class ActionItem(BaseModel):
owner: str = Field(..., description="Person responsible, as named in the meeting")
description: str = Field(..., description="What they committed to do")
due_date: Optional[str] = Field(None, description="ISO yyyy-mm-dd if mentioned")
class Meeting(BaseModel):
action_items: List[ActionItem]
# ---------------------------------------------------------------------------
# Agent Instructions
# ---------------------------------------------------------------------------
instructions = """\
Extract action items from the meeting transcript. An action item is a
commitment a named person made during the meeting. Only include items that
are clearly assigned to a specific person; ignore vague group asks. If a
due date is not mentioned, leave it null.
"""
# ---------------------------------------------------------------------------
# Create Agent
# ---------------------------------------------------------------------------
agent = Agent(
model=OpenAIResponses(id="gpt-5.5"),
instructions=instructions,
output_schema=Meeting,
)
# ---------------------------------------------------------------------------
# Run Agent
# ---------------------------------------------------------------------------
if __name__ == "__main__":
transcript = """\
Mike: I'll send out the updated roadmap by Friday.
Sarah: Great. And I'll set up the kickoff with the design team next week.
Jess: We should probably get budget approval at some point.
Mike: Yeah. Let me draft the budget memo by end of next week so we can
send it to finance.
"""
run: RunOutput = agent.run(transcript)
pprint(run.content)
@@ -0,0 +1,70 @@
"""
Text Extraction - With Confidence
=================================
Adds per-field confidence using a shared `ConfidentField` wrapper. Use when
downstream consumers need to route low-confidence fields to a human queue
or a stronger model.
"""
from typing import Literal, Optional
from agno.agent import Agent, RunOutput # noqa
from agno.models.openai import OpenAIResponses
from pydantic import BaseModel, Field
from rich.pretty import pprint # noqa
# ---------------------------------------------------------------------------
# Schema
# ---------------------------------------------------------------------------
class ConfidentField(BaseModel):
value: Optional[str] = None
confidence: Literal["high", "medium", "low"] = Field(
..., description="Confidence in the extracted value"
)
class Contact(BaseModel):
name: ConfidentField
email: ConfidentField
phone: ConfidentField
company: ConfidentField
title: ConfidentField
# ---------------------------------------------------------------------------
# Agent Instructions
# ---------------------------------------------------------------------------
instructions = """\
Extract contact information from the input. For each field:
- value: what the text shows; null if the field is missing
- confidence: high if explicit and unambiguous;
medium if implied or partially formatted;
low if guessed or ambiguous
Use exactly what the text shows. Do not normalize or paraphrase.
"""
# ---------------------------------------------------------------------------
# Create Agent
# ---------------------------------------------------------------------------
agent = Agent(
model=OpenAIResponses(id="gpt-5.5"),
instructions=instructions,
output_schema=Contact,
)
# ---------------------------------------------------------------------------
# Run Agent
# ---------------------------------------------------------------------------
if __name__ == "__main__":
samples = [
"Sarah Johnson, VP of Marketing at Acme Corp. [email protected] / +1-555-0102.",
"ping @mike on the eng team",
]
for text in samples:
run: RunOutput = agent.run(text)
pprint({"input": text, "result": run.content})
@@ -0,0 +1,30 @@
# Text Span Labeling
Find labeled substrings within a text. The output is a list of `(text,
label)` pairs; character offsets are computed in post-processing using
`text.find()`. Asking the LLM to count characters is unreliable - the right
shape is to have the model return the literal substring and let Python
locate it.
## Files
- `basic.py` — entity span detection: PERSON, ORG, LOCATION, DATE.
- `pii_redaction.py` — PII span detection plus a simple redact step.
## When to use
- NER on customer support tickets to populate a contact graph.
- PII detection before storing user input.
- Highlighting claims or evidence in a longer document.
If you just want a typed object rather than spans, use
[`_03_text_extraction/`](../_03_text_extraction/).
## Run
```bash
python cookbook/data_labeling/_04_text_span_labeling/basic.py
python cookbook/data_labeling/_04_text_span_labeling/pii_redaction.py
```
Requires `OPENAI_API_KEY`.
@@ -0,0 +1,23 @@
# Test Log - _04_text_span_labeling
Tested 2026-05-17 against `gpt-5.5` (OpenAIResponses), agno 2.6.6.
### basic.py
**Status:** PASS
**Description:** NER over a short sentence; agent returns exact substrings, Python computes offsets.
**Result:** Five entities found (DATE, PERSON, ORG, LOCATION, ORG) with correct offsets.
---
### pii_redaction.py
**Status:** PASS
**Description:** Detect PII spans (name, phone, email, credit card) and produce a redacted string.
**Result:** All four PII items detected; redacted text replaces each span with its tag.
---
@@ -0,0 +1,72 @@
"""
Text Span Labeling - Basic
==========================
Detect labeled substrings (entities) within a text. The model emits the
exact substring plus its type; offsets are computed in post-processing.
Asking the LLM to count characters is unreliable. Returning the literal
substring and locating it in Python is the robust pattern.
"""
from typing import List, Literal
from agno.agent import Agent, RunOutput # noqa
from agno.models.openai import OpenAIResponses
from pydantic import BaseModel, Field
from rich.pretty import pprint # noqa
# ---------------------------------------------------------------------------
# Schema
# ---------------------------------------------------------------------------
class Entity(BaseModel):
text: str = Field(..., description="Exact substring from the input")
label: Literal["PERSON", "ORG", "LOCATION", "DATE"] = Field(
..., description="Entity type"
)
class Entities(BaseModel):
entities: List[Entity]
# ---------------------------------------------------------------------------
# Agent Instructions
# ---------------------------------------------------------------------------
instructions = """\
Extract all named entities from the input. For each entity, return the
exact substring as it appears in the text (case and punctuation preserved)
along with its label. Do not paraphrase or normalize. Do not include
pronouns or generic references.
"""
# ---------------------------------------------------------------------------
# Create Agent
# ---------------------------------------------------------------------------
agent = Agent(
model=OpenAIResponses(id="gpt-5.5"),
instructions=instructions,
output_schema=Entities,
)
# ---------------------------------------------------------------------------
# Run Agent
# ---------------------------------------------------------------------------
def with_positions(text: str, entities: List[Entity]):
"""Find each entity's first occurrence offset; useful for downstream tagging."""
for e in entities:
start = text.find(e.text)
end = start + len(e.text) if start >= 0 else None
yield {"label": e.label, "text": e.text, "start": start, "end": end}
if __name__ == "__main__":
text = (
"On March 3rd, Sarah Johnson left Acme Corp to join a startup based in "
"Berlin called Lumen Labs."
)
run: RunOutput = agent.run(text)
pprint(list(with_positions(text, run.content.entities)))
@@ -0,0 +1,73 @@
"""
Text Span Labeling - PII Redaction
==================================
Detect personally identifiable information and produce a redacted version
of the input. Same pattern as `basic.py`: model returns the exact substring
and its PII type; Python does the find-and-replace.
"""
from typing import List, Literal
from agno.agent import Agent, RunOutput # noqa
from agno.models.openai import OpenAIResponses
from pydantic import BaseModel, Field
from rich.pretty import pprint # noqa
# ---------------------------------------------------------------------------
# Schema
# ---------------------------------------------------------------------------
class PIIItem(BaseModel):
text: str = Field(..., description="Exact substring containing the PII")
type: Literal["email", "phone", "ssn", "credit_card", "person_name"]
class PIIDetection(BaseModel):
items: List[PIIItem]
# ---------------------------------------------------------------------------
# Agent Instructions
# ---------------------------------------------------------------------------
instructions = """\
Detect every span of personally identifiable information. For each item,
return the exact substring (preserving case and formatting) and its type.
Do not redact in your output - emit the raw text. Redaction happens in
post-processing.
Cover: email addresses, phone numbers, SSNs, credit card numbers, and full
person names. Skip generic references like "the customer".
"""
# ---------------------------------------------------------------------------
# Create Agent
# ---------------------------------------------------------------------------
agent = Agent(
model=OpenAIResponses(id="gpt-5.5"),
instructions=instructions,
output_schema=PIIDetection,
)
# ---------------------------------------------------------------------------
# Run Agent
# ---------------------------------------------------------------------------
def redact(text: str, items: List[PIIItem]) -> str:
"""Replace each detected span with a token of the form [TYPE]."""
out = text
for item in items:
out = out.replace(item.text, f"[{item.type.upper()}]")
return out
if __name__ == "__main__":
text = (
"Customer Jane Doe called from 415-555-0199 about her order. "
"She asked us to email [email protected] with the receipt. "
"Card on file ends 4242 4242 4242 4242."
)
run: RunOutput = agent.run(text)
detection = run.content
pprint({"detected": detection.items, "redacted": redact(text, detection.items)})
@@ -0,0 +1,30 @@
# Text Pairwise Preference
Given a prompt and two candidate responses, decide which is better. This is
the data shape used for RLHF/DPO preference datasets.
## Files
- `basic.py` — pick the winner with no further structure.
- `with_rubric.py` — pick based on an explicit rubric supplied in the
instructions.
- `with_rationale.py` — winner plus a one-sentence explanation.
## When to use
- Building a preference dataset to fine-tune a reward model.
- Comparing two model versions on a held-out prompt set.
- Bake-offs between prompts.
If you want a single score against a rubric rather than a pairwise
comparison, use [`_17_llm_as_judge/`](../_17_llm_as_judge/).
## Run
```bash
python cookbook/data_labeling/_05_text_pairwise_preference/basic.py
python cookbook/data_labeling/_05_text_pairwise_preference/with_rubric.py
python cookbook/data_labeling/_05_text_pairwise_preference/with_rationale.py
```
Requires `OPENAI_API_KEY`.
@@ -0,0 +1,33 @@
# Test Log - _05_text_pairwise_preference
Tested 2026-05-17 against `gpt-5.5` (OpenAIResponses), agno 2.6.6.
### basic.py
**Status:** PASS
**Description:** Pick winner between two responses to the same prompt; output A/B/tie.
**Result:** Picks the more informative response (A) over the trivial one (B).
---
### with_rationale.py
**Status:** PASS
**Description:** Same task with a free-text rationale for the choice.
**Result:** Winner and rationale agree; rationale cites the discriminating quality.
---
### with_rubric.py
**Status:** PASS
**Description:** Same task driven by an explicit rubric in the instructions.
**Result:** Picks the more accurate/complete response per the rubric.
---
@@ -0,0 +1,63 @@
"""
Text Pairwise Preference - Basic
================================
Given a prompt and two responses, pick the better one. The output is the
data shape used to train reward models (RLHF) or do DPO fine-tuning.
"""
from typing import Literal
from agno.agent import Agent, RunOutput # noqa
from agno.models.openai import OpenAIResponses
from pydantic import BaseModel, Field
from rich.pretty import pprint # noqa
# ---------------------------------------------------------------------------
# Schema
# ---------------------------------------------------------------------------
class Preference(BaseModel):
winner: Literal["A", "B", "tie"] = Field(
..., description="Which response is better, or 'tie' if equally good"
)
# ---------------------------------------------------------------------------
# Agent Instructions
# ---------------------------------------------------------------------------
instructions = """\
You are evaluating two candidate responses to the same prompt. Decide which
response better answers the prompt. Return 'A', 'B', or 'tie'. Use 'tie'
only when the two are genuinely indistinguishable in quality.
"""
# ---------------------------------------------------------------------------
# Create Agent
# ---------------------------------------------------------------------------
agent = Agent(
model=OpenAIResponses(id="gpt-5.5"),
instructions=instructions,
output_schema=Preference,
)
# ---------------------------------------------------------------------------
# Run Agent
# ---------------------------------------------------------------------------
def build_input(prompt: str, response_a: str, response_b: str) -> str:
return (
f"Prompt:\n{prompt}\n\nResponse A:\n{response_a}\n\nResponse B:\n{response_b}"
)
if __name__ == "__main__":
prompt = "Explain why the sky is blue, in one sentence."
a = (
"Sunlight scatters off air molecules, and shorter (blue) wavelengths "
"scatter more than longer ones, so we see blue from every direction."
)
b = "Because of physics."
run: RunOutput = agent.run(build_input(prompt, a, b))
pprint({"A": a, "B": b, "result": run.content})
@@ -0,0 +1,63 @@
"""
Text Pairwise Preference - With Rationale
=========================================
Winner plus a short rationale. The rationale is itself useful training
data: it can be distilled into a reward model or used to explain a model's
own preferences to a human reviewer.
"""
from typing import Literal
from agno.agent import Agent, RunOutput # noqa
from agno.models.openai import OpenAIResponses
from pydantic import BaseModel, Field
from rich.pretty import pprint # noqa
# ---------------------------------------------------------------------------
# Schema
# ---------------------------------------------------------------------------
class Preference(BaseModel):
winner: Literal["A", "B", "tie"] = Field(
..., description="Which response is better"
)
rationale: str = Field(..., description="One sentence explaining the decision")
# ---------------------------------------------------------------------------
# Agent Instructions
# ---------------------------------------------------------------------------
instructions = """\
You are evaluating two candidate responses to the same prompt. Decide
which response is better and provide a one-sentence rationale that names
the specific quality that drove the decision (correctness, completeness,
clarity, tone, etc.).
"""
# ---------------------------------------------------------------------------
# Create Agent
# ---------------------------------------------------------------------------
agent = Agent(
model=OpenAIResponses(id="gpt-5.5"),
instructions=instructions,
output_schema=Preference,
)
# ---------------------------------------------------------------------------
# Run Agent
# ---------------------------------------------------------------------------
def build_input(prompt: str, response_a: str, response_b: str) -> str:
return (
f"Prompt:\n{prompt}\n\nResponse A:\n{response_a}\n\nResponse B:\n{response_b}"
)
if __name__ == "__main__":
prompt = "Suggest a name for a side project that helps people sleep better."
a = "How about 'Drift'? Short, evocative, and the .app domain is probably free."
b = "SleepHelperProMaxXL"
run: RunOutput = agent.run(build_input(prompt, a, b))
pprint({"A": a, "B": b, "result": run.content})
@@ -0,0 +1,73 @@
"""
Text Pairwise Preference - With Rubric
======================================
Same shape as `basic.py` but the judgement is grounded in an explicit
rubric supplied in the instructions. Use this when you need consistency
across many graders or many runs.
"""
from typing import Literal
from agno.agent import Agent, RunOutput # noqa
from agno.models.openai import OpenAIResponses
from pydantic import BaseModel, Field
from rich.pretty import pprint # noqa
# ---------------------------------------------------------------------------
# Schema
# ---------------------------------------------------------------------------
class Preference(BaseModel):
winner: Literal["A", "B", "tie"] = Field(..., description="Winner per rubric")
# ---------------------------------------------------------------------------
# Agent Instructions
# ---------------------------------------------------------------------------
instructions = """\
You are evaluating two candidate responses to the same prompt. Apply the
rubric below in order. Earlier criteria dominate later ones.
Rubric:
1. Factual correctness - is everything stated true?
2. Completeness - does it actually answer the prompt?
3. Clarity - is it easy to understand for the target reader?
4. Concision - is anything in it wasted?
Return 'A', 'B', or 'tie'. Use 'tie' only when the two responses are
indistinguishable on the rubric above.
"""
# ---------------------------------------------------------------------------
# Create Agent
# ---------------------------------------------------------------------------
agent = Agent(
model=OpenAIResponses(id="gpt-5.5"),
instructions=instructions,
output_schema=Preference,
)
# ---------------------------------------------------------------------------
# Run Agent
# ---------------------------------------------------------------------------
def build_input(prompt: str, response_a: str, response_b: str) -> str:
return (
f"Prompt:\n{prompt}\n\nResponse A:\n{response_a}\n\nResponse B:\n{response_b}"
)
if __name__ == "__main__":
prompt = "How do I cancel my subscription?"
a = (
"Go to Settings > Subscription and click 'Cancel'. The cancellation "
"takes effect at the end of your current billing period."
)
b = (
"You can cancel anytime. Just dig around in the app, you'll find it. "
"Maybe try the help docs."
)
run: RunOutput = agent.run(build_input(prompt, a, b))
pprint({"A": a, "B": b, "result": run.content})
@@ -0,0 +1,28 @@
# Image Classification
Assign a label to an image. Same shape as text classification - input is an
image, output is one or more labels from a closed set.
## Files
- `basic.py` — single label per image.
- `multilabel.py` — any subset of N tags per image.
## When to use
- Routing user-uploaded photos by content type.
- Pre-tagging a media library before manual cleanup.
- Quality / NSFW gates before ingest.
If you want to extract structured fields rather than labels (color, brand,
text on the image), use [`_07_image_extraction/`](../_07_image_extraction/).
## Run
```bash
python cookbook/data_labeling/_06_image_classification/basic.py
python cookbook/data_labeling/_06_image_classification/multilabel.py
```
Requires `OPENAI_API_KEY`. The samples use stable Wikimedia URLs - swap
your own image URLs or local paths in the `Image(...)` call.
@@ -0,0 +1,23 @@
# Test Log - _06_image_classification
Tested 2026-05-17 against `gpt-5.5` (OpenAIResponses), agno 2.6.6. Inputs are public Wikimedia URLs.
### basic.py
**Status:** PASS
**Description:** Single-label classification over a cat photo and a dog collage.
**Result:** Both classified correctly.
---
### multilabel.py
**Status:** PASS
**Description:** Multi-label tagging of scene attributes per image.
**Result:** Tag sets are coherent with the photo content.
---
@@ -0,0 +1,47 @@
"""
Image Classification - Basic
============================
Assign a single label from a closed set to an image. Same primitive as text
classification with image input.
"""
from typing import Literal
from agno.agent import Agent, RunOutput # noqa
from agno.media import Image
from agno.models.openai import OpenAIResponses
from pydantic import BaseModel, Field
from rich.pretty import pprint # noqa
# ---------------------------------------------------------------------------
# Schema
# ---------------------------------------------------------------------------
class Classification(BaseModel):
label: Literal["dog", "cat", "bird", "fish", "other"] = Field(
..., description="What kind of animal is in the image"
)
# ---------------------------------------------------------------------------
# Create Agent
# ---------------------------------------------------------------------------
agent = Agent(
model=OpenAIResponses(id="gpt-5.5"),
instructions="You classify images by animal type.",
output_schema=Classification,
)
# ---------------------------------------------------------------------------
# Run Agent
# ---------------------------------------------------------------------------
if __name__ == "__main__":
samples = [
"https://upload.wikimedia.org/wikipedia/commons/4/4d/Cat_November_2010-1a.jpg",
"https://upload.wikimedia.org/wikipedia/commons/d/d9/Collage_of_Nine_Dogs.jpg",
]
for url in samples:
run: RunOutput = agent.run("Classify this image.", images=[Image(url=url)])
pprint({"url": url, "result": run.content})
@@ -0,0 +1,68 @@
"""
Image Classification - Multilabel
=================================
Assign any subset of N tags to an image. Useful for scene tagging and
content categorization.
"""
from typing import List, Literal
from agno.agent import Agent, RunOutput # noqa
from agno.media import Image
from agno.models.openai import OpenAIResponses
from pydantic import BaseModel, Field
from rich.pretty import pprint # noqa
SceneTag = Literal[
"outdoor",
"indoor",
"daytime",
"nighttime",
"people",
"vehicle",
"nature",
"architecture",
]
# ---------------------------------------------------------------------------
# Schema
# ---------------------------------------------------------------------------
class Tagging(BaseModel):
tags: List[SceneTag] = Field(
..., description="All scene tags that apply; empty if none"
)
# ---------------------------------------------------------------------------
# Agent Instructions
# ---------------------------------------------------------------------------
instructions = """\
Tag the image with every scene attribute that clearly applies. Include a
tag only if it is unambiguously present in the image - skip tags that are
inferred or implied.
"""
# ---------------------------------------------------------------------------
# Create Agent
# ---------------------------------------------------------------------------
agent = Agent(
model=OpenAIResponses(id="gpt-5.5"),
instructions=instructions,
output_schema=Tagging,
)
# ---------------------------------------------------------------------------
# Run Agent
# ---------------------------------------------------------------------------
if __name__ == "__main__":
samples = [
"https://upload.wikimedia.org/wikipedia/commons/0/0c/GoldenGateBridge-001.jpg",
"https://upload.wikimedia.org/wikipedia/commons/a/a8/Tour_Eiffel_Wikimedia_Commons.jpg",
]
for url in samples:
run: RunOutput = agent.run("Tag this image.", images=[Image(url=url)])
pprint({"url": url, "result": run.content})
@@ -0,0 +1,32 @@
# Image Extraction
Image → typed Pydantic object. Same shape as text extraction with image
input: descriptive attributes, OCR'd fields, structured metadata.
## Files
- `basic.py` — image → typed scene attributes.
- `with_confidence.py` — adds per-field confidence.
- `ocr_fields.py` — extract text-heavy fields from an image (sign, receipt,
product label).
## When to use
- Auto-cataloging product photos (color, style, type).
- Pre-filling form fields from a photo (receipt, business card).
- Generating searchable metadata for a media archive (see also
[`_09_image_extraction_to_vectordb/`](../_09_image_extraction_to_vectordb/)).
If you only need a label, use
[`_06_image_classification/`](../_06_image_classification/). If you need pixel
regions, use [`_08_image_bounding_boxes/`](../_08_image_bounding_boxes/).
## Run
```bash
python cookbook/data_labeling/_07_image_extraction/basic.py
python cookbook/data_labeling/_07_image_extraction/with_confidence.py
python cookbook/data_labeling/_07_image_extraction/ocr_fields.py
```
Requires `OPENAI_API_KEY`. Swap the URLs for your own images as needed.
@@ -0,0 +1,35 @@
# Test Log - _07_image_extraction
Tested 2026-05-17 against `gpt-5.5` (OpenAIResponses), agno 2.6.6. Inputs are public Wikimedia URLs.
### basic.py
**Status:** PASS
**Description:** Extract a `Scene` (subject, setting, time of day, colors, objects) from a Golden Gate Bridge photo.
**Result:** All fields populated correctly.
---
### with_confidence.py
**Status:** PASS
**Description:** Same task with `ConfidentField` / `ConfidentList` wrapping each output (Eiffel Tower image).
**Result:** All fields populated with `high` confidence; nested confidence wrappers serialize correctly.
---
### ocr_fields.py
**Status:** PASS
**Description:** OCR a sign image into typed fields (primary text, secondary text, color scheme).
**Result:** "Welcome to Fabulous Las Vegas / NEVADA" read correctly with multi-color palette.
**Note:** Original URL (`Stop_sign_London_2020.jpg`) was 404. Replaced with the canonical Las Vegas welcome-sign URL (verified via Wikimedia API); also gives a richer multi-text example than a single-word stop sign.
---
@@ -0,0 +1,58 @@
"""
Image Extraction - Basic
========================
Extract typed scene attributes from an image. The output is a Pydantic
object whose schema you control.
"""
from typing import List, Literal, Optional
from agno.agent import Agent, RunOutput # noqa
from agno.media import Image
from agno.models.openai import OpenAIResponses
from pydantic import BaseModel, Field
from rich.pretty import pprint # noqa
# ---------------------------------------------------------------------------
# Schema
# ---------------------------------------------------------------------------
class Scene(BaseModel):
subject: str = Field(..., description="The main subject of the image")
setting: Literal["indoor", "outdoor", "studio", "unknown"]
time_of_day: Optional[Literal["day", "night", "dawn_or_dusk"]] = None
dominant_colors: List[str] = Field(
default_factory=list, description="Two to four named colors"
)
notable_objects: List[str] = Field(
default_factory=list, description="Up to five named objects in the scene"
)
# ---------------------------------------------------------------------------
# Agent Instructions
# ---------------------------------------------------------------------------
instructions = """\
Describe the image as a structured Scene. Be concrete and observational.
If a field is not determinable from the image, leave it null or empty.
"""
# ---------------------------------------------------------------------------
# Create Agent
# ---------------------------------------------------------------------------
agent = Agent(
model=OpenAIResponses(id="gpt-5.5"),
instructions=instructions,
output_schema=Scene,
)
# ---------------------------------------------------------------------------
# Run Agent
# ---------------------------------------------------------------------------
if __name__ == "__main__":
url = "https://upload.wikimedia.org/wikipedia/commons/0/0c/GoldenGateBridge-001.jpg"
run: RunOutput = agent.run("Extract the scene attributes.", images=[Image(url=url)])
pprint({"url": url, "result": run.content})
@@ -0,0 +1,64 @@
"""
Image Extraction - OCR Fields
=============================
Extract text-heavy fields from an image: signs, menus, receipts, business
cards. The model reads the image and maps the visible text into a typed
schema.
"""
from typing import List, Optional
from agno.agent import Agent, RunOutput # noqa
from agno.media import Image
from agno.models.openai import OpenAIResponses
from pydantic import BaseModel, Field
from rich.pretty import pprint # noqa
# ---------------------------------------------------------------------------
# Schema
# ---------------------------------------------------------------------------
class SignReading(BaseModel):
primary_text: Optional[str] = Field(
None, description="The main word or phrase shown on the sign"
)
secondary_text: List[str] = Field(
default_factory=list,
description="Any additional words shown, in reading order",
)
color_scheme: Optional[str] = Field(
None, description="Dominant colors used on the sign, comma separated"
)
# ---------------------------------------------------------------------------
# Agent Instructions
# ---------------------------------------------------------------------------
instructions = """\
Read the text on the sign in the image. Return what is literally written -
do not translate, expand, or paraphrase. If the sign is partly obscured,
include what is legible and leave the rest null.
"""
# ---------------------------------------------------------------------------
# Create Agent
# ---------------------------------------------------------------------------
agent = Agent(
model=OpenAIResponses(id="gpt-5.5"),
instructions=instructions,
output_schema=SignReading,
)
# ---------------------------------------------------------------------------
# Run Agent
# ---------------------------------------------------------------------------
if __name__ == "__main__":
# The "Welcome to Fabulous Las Vegas" sign, public domain.
url = "https://upload.wikimedia.org/wikipedia/commons/1/10/Welcome_to_Fabulous_Las_Vegas_sign.jpg"
run: RunOutput = agent.run(
"Extract the text fields from this sign.", images=[Image(url=url)]
)
pprint({"url": url, "result": run.content})
@@ -0,0 +1,73 @@
"""
Image Extraction - With Confidence
==================================
Adds per-field confidence to image attribute extraction. Useful when the
input quality varies (low-res, motion blur, partial occlusion) and you
need to flag uncertain fields for review.
"""
from typing import List, Literal, Optional
from agno.agent import Agent, RunOutput # noqa
from agno.media import Image
from agno.models.openai import OpenAIResponses
from pydantic import BaseModel, Field
from rich.pretty import pprint # noqa
Confidence = Literal["high", "medium", "low"]
# ---------------------------------------------------------------------------
# Schema
# ---------------------------------------------------------------------------
class ConfidentStr(BaseModel):
value: Optional[str] = None
confidence: Confidence
class ConfidentList(BaseModel):
values: List[str] = Field(default_factory=list)
confidence: Confidence
class Scene(BaseModel):
subject: ConfidentStr
setting: ConfidentStr
time_of_day: ConfidentStr
dominant_colors: ConfidentList
notable_objects: ConfidentList
# ---------------------------------------------------------------------------
# Agent Instructions
# ---------------------------------------------------------------------------
instructions = """\
Describe the image as a structured Scene. For each field, report
confidence:
- high - clearly determinable from the image
- medium - inferred but well-supported
- low - guessed or partially obscured
Be conservative. If you cannot see a field, mark its value null and
confidence low.
"""
# ---------------------------------------------------------------------------
# Create Agent
# ---------------------------------------------------------------------------
agent = Agent(
model=OpenAIResponses(id="gpt-5.5"),
instructions=instructions,
output_schema=Scene,
)
# ---------------------------------------------------------------------------
# Run Agent
# ---------------------------------------------------------------------------
if __name__ == "__main__":
url = "https://upload.wikimedia.org/wikipedia/commons/a/a8/Tour_Eiffel_Wikimedia_Commons.jpg"
run: RunOutput = agent.run("Extract the scene attributes.", images=[Image(url=url)])
pprint({"url": url, "result": run.content})
@@ -0,0 +1,41 @@
# Image Bounding Boxes
Detect objects in an image and return their bounding boxes. The model
emits normalized coordinates in `[0, 1]` so the result is resolution-
independent.
## Files
- `basic.py` — detect one labeled object with a bounding box.
- `with_confidence.py` — adds per-box confidence.
- `multi_object.py` — detect multiple objects of multiple classes.
## When to use
- Pre-labeling for an object detection training set (human-in-the-loop
refinement on top).
- Crop suggestions for product imagery.
- Coarse spatial routing (counting people, vehicles, defects).
For pixel-accurate masks, this primitive isn't the right tool - a
segmentation model is. For "is X in the image" without coordinates, use
[`_06_image_classification/`](../_06_image_classification/) with multilabel.
## Coordinate convention
Coordinates are normalized to the image dimensions:
- `x`, `y` = top-left corner, in `[0, 1]`
- `width`, `height` = box size, in `[0, 1]`
Multiply by the actual image width/height to get pixel coordinates.
## Run
```bash
python cookbook/data_labeling/_08_image_bounding_boxes/basic.py
python cookbook/data_labeling/_08_image_bounding_boxes/with_confidence.py
python cookbook/data_labeling/_08_image_bounding_boxes/multi_object.py
```
Requires `OPENAI_API_KEY`.
@@ -0,0 +1,35 @@
# Test Log - _08_image_bounding_boxes
Tested 2026-05-17 against `gpt-5.5` (OpenAIResponses), agno 2.6.6.
### basic.py
**Status:** PASS
**Description:** Detect the main subject in a cat photo, return one normalized bounding box.
**Result:** Sensible tight box around the cat (x=0.22, y=0.12, w=0.69, h=0.86).
---
### with_confidence.py
**Status:** PASS
**Description:** Same task with a `confidence` field added to the schema.
**Result:** Returns sensible box (x=0.215, y=0.115, w=0.69, h=0.84) with `high` confidence.
**Note:** Initially produced degenerate boxes (all-zero or all-one). Fix: added per-field descriptions on x/y/w/h and expanded coordinate guidance in the instructions to match `basic.py`. Without that grounding, `gpt-5.5` does not consistently produce sensible coordinates.
---
### multi_object.py
**Status:** PASS
**Description:** Detect multiple objects in an image, return a list of bounding boxes.
**Result:** Multiple boxes returned with reasonable coordinates and labels.
---
@@ -0,0 +1,59 @@
"""
Image Bounding Boxes - Basic
============================
Detect a single object in an image and return its bounding box. The model
emits normalized coordinates so the result is resolution-independent.
"""
from agno.agent import Agent, RunOutput # noqa
from agno.media import Image
from agno.models.openai import OpenAIResponses
from pydantic import BaseModel, Field
from rich.pretty import pprint # noqa
# ---------------------------------------------------------------------------
# Schema
# ---------------------------------------------------------------------------
class BoundingBox(BaseModel):
label: str = Field(..., description="What the box contains")
x: float = Field(..., ge=0.0, le=1.0, description="Top-left x in [0, 1]")
y: float = Field(..., ge=0.0, le=1.0, description="Top-left y in [0, 1]")
width: float = Field(..., ge=0.0, le=1.0, description="Width in [0, 1]")
height: float = Field(..., ge=0.0, le=1.0, description="Height in [0, 1]")
# ---------------------------------------------------------------------------
# Agent Instructions
# ---------------------------------------------------------------------------
instructions = """\
Locate the main subject of the image and return its bounding box in
normalized coordinates. Coordinates are relative to the full image:
- x, y: top-left corner, each in [0, 1]
- width, height: size, each in [0, 1]
The box should be tight: include the subject and exclude as much background
as possible without clipping the subject.
"""
# ---------------------------------------------------------------------------
# Create Agent
# ---------------------------------------------------------------------------
agent = Agent(
model=OpenAIResponses(id="gpt-5.5"),
instructions=instructions,
output_schema=BoundingBox,
)
# ---------------------------------------------------------------------------
# Run Agent
# ---------------------------------------------------------------------------
if __name__ == "__main__":
url = "https://upload.wikimedia.org/wikipedia/commons/4/4d/Cat_November_2010-1a.jpg"
run: RunOutput = agent.run(
"Locate the main subject of this image.", images=[Image(url=url)]
)
pprint({"url": url, "result": run.content})
@@ -0,0 +1,67 @@
"""
Image Bounding Boxes - Multi Object
===================================
Detect multiple objects of multiple classes in one image. The output is a
list of boxes, each with a label and normalized coordinates.
"""
from typing import List
from agno.agent import Agent, RunOutput # noqa
from agno.media import Image
from agno.models.openai import OpenAIResponses
from pydantic import BaseModel, Field
from rich.pretty import pprint # noqa
# ---------------------------------------------------------------------------
# Schema
# ---------------------------------------------------------------------------
class BoundingBox(BaseModel):
label: str = Field(..., description="Object class for this box")
x: float = Field(..., ge=0.0, le=1.0)
y: float = Field(..., ge=0.0, le=1.0)
width: float = Field(..., ge=0.0, le=1.0)
height: float = Field(..., ge=0.0, le=1.0)
class Detection(BaseModel):
boxes: List[BoundingBox] = Field(
default_factory=list, description="All detected objects in the image"
)
# ---------------------------------------------------------------------------
# Agent Instructions
# ---------------------------------------------------------------------------
instructions = """\
Detect every distinct object in the image. For each one, return a label
and a tight bounding box in normalized coordinates (top-left x and y,
width, height - all in [0, 1]).
Skip background elements (sky, road surface) unless they are the subject.
Skip duplicates: if two objects are nearly identical and overlapping,
report a single box.
"""
# ---------------------------------------------------------------------------
# Create Agent
# ---------------------------------------------------------------------------
agent = Agent(
model=OpenAIResponses(id="gpt-5.5"),
instructions=instructions,
output_schema=Detection,
)
# ---------------------------------------------------------------------------
# Run Agent
# ---------------------------------------------------------------------------
if __name__ == "__main__":
url = "https://upload.wikimedia.org/wikipedia/commons/d/d9/Collage_of_Nine_Dogs.jpg"
run: RunOutput = agent.run(
"Detect every object in this image.", images=[Image(url=url)]
)
pprint({"url": url, "result": run.content})
@@ -0,0 +1,69 @@
"""
Image Bounding Boxes - With Confidence
======================================
Adds per-box confidence so downstream consumers can threshold or send
low-confidence boxes to a human reviewer.
"""
from typing import Literal
from agno.agent import Agent, RunOutput # noqa
from agno.media import Image
from agno.models.openai import OpenAIResponses
from pydantic import BaseModel, Field
from rich.pretty import pprint # noqa
# ---------------------------------------------------------------------------
# Schema
# ---------------------------------------------------------------------------
class BoundingBox(BaseModel):
label: str = Field(..., description="What the box contains")
x: float = Field(..., ge=0.0, le=1.0, description="Top-left x in [0, 1]")
y: float = Field(..., ge=0.0, le=1.0, description="Top-left y in [0, 1]")
width: float = Field(..., ge=0.0, le=1.0, description="Width in [0, 1]")
height: float = Field(..., ge=0.0, le=1.0, description="Height in [0, 1]")
confidence: Literal["high", "medium", "low"] = Field(
..., description="Confidence in the box and label"
)
# ---------------------------------------------------------------------------
# Agent Instructions
# ---------------------------------------------------------------------------
instructions = """\
Locate the main subject of the image and return a bounding box plus a
confidence. Coordinates are relative to the full image:
- x, y: top-left corner, each in [0, 1]
- width, height: size, each in [0, 1]
The box should be tight: include the subject and exclude as much background
as possible without clipping the subject.
Confidence reflects both the box and the label:
- high - subject is clearly visible, box is tight and accurate
- medium - subject is identifiable but partly occluded, box is approximate
- low - subject is barely visible or you are guessing at the location
"""
# ---------------------------------------------------------------------------
# Create Agent
# ---------------------------------------------------------------------------
agent = Agent(
model=OpenAIResponses(id="gpt-5.5"),
instructions=instructions,
output_schema=BoundingBox,
)
# ---------------------------------------------------------------------------
# Run Agent
# ---------------------------------------------------------------------------
if __name__ == "__main__":
url = "https://upload.wikimedia.org/wikipedia/commons/4/4d/Cat_November_2010-1a.jpg"
run: RunOutput = agent.run(
"Locate the main subject and report confidence.", images=[Image(url=url)]
)
pprint({"url": url, "result": run.content})
@@ -0,0 +1,36 @@
# Image Extraction to Vector DB
End-to-end pipeline: extract structured descriptions from images with an
agent, then embed and store them in a vector database for similarity
search. The classic "build a searchable media library" labeling workflow.
## Files
- `basic.py` — extract → embed → store → search. All in one file.
No variants. The value here is the full pipeline.
## How it works
1. For each image URL, an agent extracts a structured `ImageDescription`
(subject, setting, mood, key objects).
2. The structured fields are flattened to a single searchable string.
3. The string is embedded with `OpenAIEmbedder` and stored in LanceDb.
4. A text query searches the index and returns the most similar images.
## When to use
- Searching a stock photo library by natural-language query.
- Deduplicating a product catalog by visual + descriptive similarity.
- Building "find more like this" features on user-uploaded media.
## Run
Install the extra dependency for LanceDb first:
```bash
pip install lancedb tantivy
python cookbook/data_labeling/_09_image_extraction_to_vectordb/basic.py
```
Requires `OPENAI_API_KEY`. Writes to `tmp/lancedb/` under the repo root.
@@ -0,0 +1,15 @@
# Test Log - _09_image_extraction_to_vectordb
Tested 2026-05-17 against `gpt-5.5` (OpenAIResponses) + `text-embedding-3-small`, agno 2.6.6.
### basic.py
**Status:** PASS
**Description:** Describe three images, embed the descriptions, store in LanceDb, then run two semantic queries.
**Result:** "a famous landmark" returns Golden Gate and Eiffel; "a sleeping pet" returns the cat as top hit.
**Note:** Requires `lancedb` (not in the `agno[demo]` extras). Install with `uv pip install lancedb` into `.venvs/demo`. Worth either adding to the demo extras or shipping a `requirements.in` for this folder.
---
@@ -0,0 +1,103 @@
"""
Image Extraction to Vector DB - Basic
=====================================
End-to-end labeling pipeline:
extract structured fields from images -> embed -> store -> search.
1. An agent extracts an ImageDescription from each image.
2. The description is flattened to a single searchable string.
3. The string is embedded with OpenAIEmbedder and stored in LanceDb.
4. We query the index with natural language.
Writes to tmp/lancedb/ under the repo root.
"""
from typing import List
from agno.agent import Agent
from agno.knowledge.document import Document
from agno.knowledge.embedder.openai import OpenAIEmbedder
from agno.media import Image
from agno.models.openai import OpenAIResponses
from agno.vectordb.lancedb import LanceDb, SearchType
from pydantic import BaseModel, Field
from rich.pretty import pprint # noqa
# ---------------------------------------------------------------------------
# Schema
# ---------------------------------------------------------------------------
class ImageDescription(BaseModel):
subject: str = Field(..., description="The main subject of the image")
setting: str = Field(..., description="Where the image takes place")
mood: str = Field(..., description="Overall mood or tone")
key_objects: List[str] = Field(
default_factory=list, description="Up to five notable objects"
)
# ---------------------------------------------------------------------------
# Create Extraction Agent
# ---------------------------------------------------------------------------
extractor = Agent(
model=OpenAIResponses(id="gpt-5.5"),
instructions="You describe images as structured, search-friendly metadata.",
output_schema=ImageDescription,
)
# ---------------------------------------------------------------------------
# Vector DB
# ---------------------------------------------------------------------------
vector_db = LanceDb(
uri="tmp/lancedb",
table_name="data_labeling_images",
search_type=SearchType.vector,
embedder=OpenAIEmbedder(id="text-embedding-3-small"),
)
# ---------------------------------------------------------------------------
# Pipeline
# ---------------------------------------------------------------------------
def describe(url: str) -> ImageDescription:
return extractor.run("Describe this image.", images=[Image(url=url)]).content
def to_searchable_text(d: ImageDescription) -> str:
return (
f"Subject: {d.subject}. Setting: {d.setting}. Mood: {d.mood}. "
f"Objects: {', '.join(d.key_objects)}."
)
def index_images(urls: List[str]) -> None:
vector_db.create()
docs: List[Document] = []
for url in urls:
description = describe(url)
docs.append(
Document(
name=url,
content=to_searchable_text(description),
meta_data={"url": url, "subject": description.subject},
)
)
vector_db.insert(content_hash="image_batch_1", documents=docs)
# ---------------------------------------------------------------------------
# Run
# ---------------------------------------------------------------------------
if __name__ == "__main__":
urls = [
"https://upload.wikimedia.org/wikipedia/commons/0/0c/GoldenGateBridge-001.jpg",
"https://upload.wikimedia.org/wikipedia/commons/a/a8/Tour_Eiffel_Wikimedia_Commons.jpg",
"https://upload.wikimedia.org/wikipedia/commons/4/4d/Cat_November_2010-1a.jpg",
]
index_images(urls)
for query in ["a famous landmark", "a sleeping pet"]:
results = vector_db.search(query, limit=2)
pprint({"query": query, "hits": [r.meta_data for r in results]})
@@ -0,0 +1,29 @@
# Audio Classification
Assign a label to an audio clip. Same primitive as text classification
with audio input.
## Files
- `basic.py` — single label per clip (language identification).
- `with_confidence.py` — adds confidence in the label.
## When to use
- Language routing in a multilingual voicemail system.
- Genre tagging for a music library.
- Emotion / tone gating in customer support audio.
## Models
Uses Gemini for native audio input. Replace the `Gemini(...)` line with an
audio-capable model from another provider if needed.
## Run
```bash
python cookbook/data_labeling/_10_audio_classification/basic.py
python cookbook/data_labeling/_10_audio_classification/with_confidence.py
```
Requires `GOOGLE_API_KEY`.
@@ -0,0 +1,23 @@
# Test Log - _10_audio_classification
Tested 2026-05-17 against `gemini-3-flash-preview` (Gemini), agno 2.6.6. Input is `agno-public/demo_data/QA-01.mp3` (English Q&A clip).
### basic.py
**Status:** PASS
**Description:** Classify audio clip language from a closed set.
**Result:** Identified as `english`.
---
### with_confidence.py
**Status:** PASS
**Description:** Same task with a `confidence` field.
**Result:** `english`, confidence `high`.
---
@@ -0,0 +1,49 @@
"""
Audio Classification - Basic
============================
Assign a single label from a closed set to an audio clip. The classic
language-identification primitive shown here applies to any closed-set
audio classification (genre, emotion, speaker, intent).
"""
from typing import Literal
import requests
from agno.agent import Agent, RunOutput # noqa
from agno.media import Audio
from agno.models.google import Gemini
from pydantic import BaseModel, Field
from rich.pretty import pprint # noqa
# ---------------------------------------------------------------------------
# Schema
# ---------------------------------------------------------------------------
class Classification(BaseModel):
language: Literal[
"english", "spanish", "french", "german", "mandarin", "hindi", "other"
] = Field(..., description="Primary language spoken in the clip")
# ---------------------------------------------------------------------------
# Create Agent
# ---------------------------------------------------------------------------
agent = Agent(
model=Gemini(id="gemini-3-flash-preview"),
instructions="You classify audio clips by spoken language.",
output_schema=Classification,
)
# ---------------------------------------------------------------------------
# Run Agent
# ---------------------------------------------------------------------------
if __name__ == "__main__":
url = "https://agno-public.s3.us-east-1.amazonaws.com/demo_data/QA-01.mp3"
audio_bytes = requests.get(url).content
run: RunOutput = agent.run(
"What language is spoken in this audio?",
audio=[Audio(content=audio_bytes)],
)
pprint({"url": url, "result": run.content})
@@ -0,0 +1,62 @@
"""
Audio Classification - With Confidence
======================================
Adds confidence so downstream routing can treat low-confidence labels
differently (escalate to a stronger model, queue for human review).
"""
from typing import Literal
import requests
from agno.agent import Agent, RunOutput # noqa
from agno.media import Audio
from agno.models.google import Gemini
from pydantic import BaseModel, Field
from rich.pretty import pprint # noqa
# ---------------------------------------------------------------------------
# Schema
# ---------------------------------------------------------------------------
class Classification(BaseModel):
language: Literal[
"english", "spanish", "french", "german", "mandarin", "hindi", "other"
] = Field(..., description="Primary language spoken in the clip")
confidence: Literal["high", "medium", "low"] = Field(
..., description="Confidence in the language label"
)
# ---------------------------------------------------------------------------
# Agent Instructions
# ---------------------------------------------------------------------------
instructions = """\
Identify the language and report a confidence:
- high - clear speech, accent is identifiable, no background interference
- medium - speech is audible but accent / dialect is ambiguous
- low - very short, heavily accented, mixed-language, or noisy
"""
# ---------------------------------------------------------------------------
# Create Agent
# ---------------------------------------------------------------------------
agent = Agent(
model=Gemini(id="gemini-3-flash-preview"),
instructions=instructions,
output_schema=Classification,
)
# ---------------------------------------------------------------------------
# Run Agent
# ---------------------------------------------------------------------------
if __name__ == "__main__":
url = "https://agno-public.s3.us-east-1.amazonaws.com/demo_data/QA-01.mp3"
audio_bytes = requests.get(url).content
run: RunOutput = agent.run(
"Identify the language and report confidence.",
audio=[Audio(content=audio_bytes)],
)
pprint({"url": url, "result": run.content})
@@ -0,0 +1,29 @@
# Audio Transcription
Speech-to-text on an audio clip. The simplest labeling primitive in the
audio family: input is raw audio, output is a typed transcript.
## Files
- `basic.py` — flat transcript string.
- `with_diarization.py` — segments labeled with a speaker identifier.
- `with_timestamps.py` — segments with start/end timing in seconds.
## When to use
- Building a training set for ASR fine-tuning.
- Pre-filling a CRM with call transcripts before human cleanup.
- Generating searchable text for an audio archive.
For typed call/meeting fields rather than a raw transcript, see
[`_12_audio_extraction/`](../_12_audio_extraction/).
## Run
```bash
python cookbook/data_labeling/_11_audio_transcription/basic.py
python cookbook/data_labeling/_11_audio_transcription/with_diarization.py
python cookbook/data_labeling/_11_audio_transcription/with_timestamps.py
```
Requires `GOOGLE_API_KEY`.
@@ -0,0 +1,33 @@
# Test Log - _11_audio_transcription
Tested 2026-05-17 against `gemini-3-flash-preview` (Gemini), agno 2.6.6.
### basic.py
**Status:** PASS
**Description:** Verbatim transcript from `QA-01.mp3` (English Q&A).
**Result:** Clean transcript of the full clip.
---
### with_diarization.py
**Status:** PASS
**Description:** Transcript split into speaker turns over `sample_conversation.wav`.
**Result:** Two speakers identified with five turns; speaker labels are consistent across turns.
---
### with_timestamps.py
**Status:** PASS
**Description:** Transcript with `[start, end]` second timestamps per segment.
**Result:** Segments returned with monotonically increasing timestamps.
---
@@ -0,0 +1,53 @@
"""
Audio Transcription - Basic
===========================
Speech-to-text on an audio clip. The output is a flat transcript string.
"""
import requests
from agno.agent import Agent, RunOutput # noqa
from agno.media import Audio
from agno.models.google import Gemini
from pydantic import BaseModel, Field
from rich.pretty import pprint # noqa
# ---------------------------------------------------------------------------
# Schema
# ---------------------------------------------------------------------------
class Transcript(BaseModel):
text: str = Field(..., description="Verbatim transcript of all spoken audio")
# ---------------------------------------------------------------------------
# Agent Instructions
# ---------------------------------------------------------------------------
instructions = """\
Transcribe everything spoken in the audio clip. Preserve disfluencies
(um, uh, like) only if asked - by default produce a clean verbatim
transcript. Do not add commentary.
"""
# ---------------------------------------------------------------------------
# Create Agent
# ---------------------------------------------------------------------------
agent = Agent(
model=Gemini(id="gemini-3-flash-preview"),
instructions=instructions,
output_schema=Transcript,
)
# ---------------------------------------------------------------------------
# Run Agent
# ---------------------------------------------------------------------------
if __name__ == "__main__":
url = "https://agno-public.s3.us-east-1.amazonaws.com/demo_data/QA-01.mp3"
audio_bytes = requests.get(url).content
run: RunOutput = agent.run(
"Transcribe this audio.",
audio=[Audio(content=audio_bytes)],
)
pprint(run.content)
@@ -0,0 +1,65 @@
"""
Audio Transcription - With Diarization
======================================
Speech-to-text with speaker labels. Each segment is attributed to a
speaker identifier (Speaker A, Speaker B, ...). The model assigns labels
consistently across the clip but does not know real names.
"""
from typing import List
import requests
from agno.agent import Agent, RunOutput # noqa
from agno.media import Audio
from agno.models.google import Gemini
from pydantic import BaseModel, Field
from rich.pretty import pprint # noqa
# ---------------------------------------------------------------------------
# Schema
# ---------------------------------------------------------------------------
class DiarizedTurn(BaseModel):
speaker: str = Field(
..., description="Speaker identifier - 'Speaker A', 'Speaker B', etc."
)
text: str = Field(..., description="What this speaker said in this turn")
class DiarizedTranscript(BaseModel):
turns: List[DiarizedTurn]
# ---------------------------------------------------------------------------
# Agent Instructions
# ---------------------------------------------------------------------------
instructions = """\
Transcribe the audio and split it into turns. Each turn is one continuous
stretch of speech by a single speaker. Label speakers consistently across
the whole clip: the first speaker is "Speaker A", the second is
"Speaker B", and so on. Do not invent names.
"""
# ---------------------------------------------------------------------------
# Create Agent
# ---------------------------------------------------------------------------
agent = Agent(
model=Gemini(id="gemini-3-flash-preview"),
instructions=instructions,
output_schema=DiarizedTranscript,
)
# ---------------------------------------------------------------------------
# Run Agent
# ---------------------------------------------------------------------------
if __name__ == "__main__":
url = "https://agno-public.s3.amazonaws.com/demo_data/sample_conversation.wav"
audio_bytes = requests.get(url).content
run: RunOutput = agent.run(
"Transcribe with speaker labels.",
audio=[Audio(content=audio_bytes)],
)
pprint(run.content)
@@ -0,0 +1,63 @@
"""
Audio Transcription - With Timestamps
=====================================
Speech-to-text with start/end timing per segment, in seconds from the
start of the clip. Useful for video captioning, jump-to-quote UI, or
chunked playback.
"""
from typing import List
import requests
from agno.agent import Agent, RunOutput # noqa
from agno.media import Audio
from agno.models.google import Gemini
from pydantic import BaseModel, Field
from rich.pretty import pprint # noqa
# ---------------------------------------------------------------------------
# Schema
# ---------------------------------------------------------------------------
class Segment(BaseModel):
start_seconds: float = Field(..., ge=0.0, description="Start time in seconds")
end_seconds: float = Field(..., ge=0.0, description="End time in seconds")
text: str = Field(..., description="Transcript of this segment")
class TimedTranscript(BaseModel):
segments: List[Segment]
# ---------------------------------------------------------------------------
# Agent Instructions
# ---------------------------------------------------------------------------
instructions = """\
Transcribe the audio and split it into segments of roughly one sentence
each. For each segment, return the start and end time in seconds from the
beginning of the clip. Times should be monotonically non-decreasing.
"""
# ---------------------------------------------------------------------------
# Create Agent
# ---------------------------------------------------------------------------
agent = Agent(
model=Gemini(id="gemini-3-flash-preview"),
instructions=instructions,
output_schema=TimedTranscript,
)
# ---------------------------------------------------------------------------
# Run Agent
# ---------------------------------------------------------------------------
if __name__ == "__main__":
url = "https://agno-public.s3.us-east-1.amazonaws.com/demo_data/QA-01.mp3"
audio_bytes = requests.get(url).content
run: RunOutput = agent.run(
"Transcribe with timestamps.",
audio=[Audio(content=audio_bytes)],
)
pprint(run.content)
@@ -0,0 +1,30 @@
# Audio Extraction
Audio → typed Pydantic object. The model listens to a call, meeting, or
voicemail and lifts structured fields into a domain schema.
## Files
- `basic.py` — generic call summary: caller intent, key topics, action.
- `meeting_notes.py` — meeting-shape schema: attendees, topics, action
items.
- `call_summary.py` — customer support shape: issue, resolution, sentiment.
## When to use
- Populate a CRM from a sales call recording.
- Generate structured meeting minutes from a Zoom audio export.
- Build a training set of `(audio, structured_label)` pairs.
If you only need a transcript, use
[`_11_audio_transcription/`](../_11_audio_transcription/).
## Run
```bash
python cookbook/data_labeling/_12_audio_extraction/basic.py
python cookbook/data_labeling/_12_audio_extraction/meeting_notes.py
python cookbook/data_labeling/_12_audio_extraction/call_summary.py
```
Requires `GOOGLE_API_KEY`.
@@ -0,0 +1,33 @@
# Test Log - _12_audio_extraction
Tested 2026-05-17 against `gemini-3-flash-preview` (Gemini), agno 2.6.6.
### basic.py
**Status:** PASS
**Description:** Extract a generic `CallSummary` (caller intent, key topics, next action) from a conversation clip.
**Result:** Caller intent and next action captured; key topics list is coherent.
---
### call_summary.py
**Status:** PASS
**Description:** Extract a domain-shaped `SupportCall` (issue, resolution status, sentiment, follow-up).
**Result:** All fields populated with sensible values.
---
### meeting_notes.py
**Status:** PASS
**Description:** Extract `MeetingNotes` (attendees, topics, action items) from a conversation clip.
**Result:** Attendees and topics extracted; action_items empty since none were stated in this clip.
---
@@ -0,0 +1,62 @@
"""
Audio Extraction - Basic
========================
Extract typed structured data from an audio clip. The schema here is a
generic call summary; swap in domain-specific shapes for production use.
"""
from typing import List, Optional
import requests
from agno.agent import Agent, RunOutput # noqa
from agno.media import Audio
from agno.models.google import Gemini
from pydantic import BaseModel, Field
from rich.pretty import pprint # noqa
# ---------------------------------------------------------------------------
# Schema
# ---------------------------------------------------------------------------
class CallSummary(BaseModel):
caller_intent: str = Field(..., description="What the caller is trying to do")
key_topics: List[str] = Field(
default_factory=list, description="Main topics discussed"
)
next_action: Optional[str] = Field(
None, description="What should happen next, if mentioned"
)
# ---------------------------------------------------------------------------
# Agent Instructions
# ---------------------------------------------------------------------------
instructions = """\
Listen to the call and extract a structured summary. Use what is actually
said - do not invent topics or actions. If the caller's intent is unclear,
write what you can determine and leave others null.
"""
# ---------------------------------------------------------------------------
# Create Agent
# ---------------------------------------------------------------------------
agent = Agent(
model=Gemini(id="gemini-3-flash-preview"),
instructions=instructions,
output_schema=CallSummary,
)
# ---------------------------------------------------------------------------
# Run Agent
# ---------------------------------------------------------------------------
if __name__ == "__main__":
url = "https://agno-public.s3.amazonaws.com/demo_data/sample_conversation.wav"
audio_bytes = requests.get(url).content
run: RunOutput = agent.run(
"Summarize this call.",
audio=[Audio(content=audio_bytes)],
)
pprint(run.content)
@@ -0,0 +1,67 @@
"""
Audio Extraction - Call Summary
===============================
Customer support call shape: issue, resolution status, customer sentiment.
Common shape for populating ticketing systems from voice channels.
"""
from typing import Literal, Optional
import requests
from agno.agent import Agent, RunOutput # noqa
from agno.media import Audio
from agno.models.google import Gemini
from pydantic import BaseModel, Field
from rich.pretty import pprint # noqa
# ---------------------------------------------------------------------------
# Schema
# ---------------------------------------------------------------------------
class SupportCall(BaseModel):
issue: str = Field(..., description="What the customer is reporting")
resolution_status: Literal["resolved", "pending", "escalated", "unclear"] = Field(
..., description="State of the issue at end of call"
)
customer_sentiment: Literal["positive", "neutral", "negative"]
follow_up_required: bool = Field(
..., description="Whether the agent committed to a follow-up"
)
notes: Optional[str] = Field(
None, description="One sentence of additional context, if useful"
)
# ---------------------------------------------------------------------------
# Agent Instructions
# ---------------------------------------------------------------------------
instructions = """\
You are extracting structured data from a customer support call recording.
Be conservative on resolution status: if you cannot confirm the issue was
resolved on the call, use 'pending' or 'unclear'. Sentiment reflects the
customer's tone, not the support agent's.
"""
# ---------------------------------------------------------------------------
# Create Agent
# ---------------------------------------------------------------------------
agent = Agent(
model=Gemini(id="gemini-3-flash-preview"),
instructions=instructions,
output_schema=SupportCall,
)
# ---------------------------------------------------------------------------
# Run Agent
# ---------------------------------------------------------------------------
if __name__ == "__main__":
url = "https://agno-public.s3.amazonaws.com/demo_data/sample_conversation.wav"
audio_bytes = requests.get(url).content
run: RunOutput = agent.run(
"Extract a support-call summary.",
audio=[Audio(content=audio_bytes)],
)
pprint(run.content)
@@ -0,0 +1,69 @@
"""
Audio Extraction - Meeting Notes
================================
Meeting-shape extraction: attendees, topics, action items. The standard
shape used by note-taking integrations.
"""
from typing import List, Optional
import requests
from agno.agent import Agent, RunOutput # noqa
from agno.media import Audio
from agno.models.google import Gemini
from pydantic import BaseModel, Field
from rich.pretty import pprint # noqa
# ---------------------------------------------------------------------------
# Schema
# ---------------------------------------------------------------------------
class ActionItem(BaseModel):
owner: str = Field(..., description="Person responsible, as named in the meeting")
description: str = Field(..., description="What they committed to do")
due_date: Optional[str] = Field(None, description="ISO yyyy-mm-dd if mentioned")
class MeetingNotes(BaseModel):
attendees: List[str] = Field(
default_factory=list, description="Speakers identifiable in the audio"
)
topics: List[str] = Field(
default_factory=list, description="Main topics discussed, in order"
)
action_items: List[ActionItem] = Field(default_factory=list)
# ---------------------------------------------------------------------------
# Agent Instructions
# ---------------------------------------------------------------------------
instructions = """\
Listen to the meeting audio and produce structured notes. Only include
action items that are explicitly assigned to a named person - skip vague
group asks. Only include attendees you can actually identify from the
audio (named by another speaker or self-introduced).
"""
# ---------------------------------------------------------------------------
# Create Agent
# ---------------------------------------------------------------------------
agent = Agent(
model=Gemini(id="gemini-3-flash-preview"),
instructions=instructions,
output_schema=MeetingNotes,
)
# ---------------------------------------------------------------------------
# Run Agent
# ---------------------------------------------------------------------------
if __name__ == "__main__":
url = "https://agno-public.s3.amazonaws.com/demo_data/sample_conversation.wav"
audio_bytes = requests.get(url).content
run: RunOutput = agent.run(
"Produce meeting notes for this recording.",
audio=[Audio(content=audio_bytes)],
)
pprint(run.content)
@@ -0,0 +1,28 @@
# Video Classification
Assign a clip-level label to a video. Same primitive as image
classification with the temporal dimension folded in - the model watches
the clip and emits one label for the whole thing.
## Files
- `basic.py` — single label per clip.
- `with_confidence.py` — adds confidence in the label.
## When to use
- Content moderation gates on user-uploaded clips.
- Pre-tagging stock footage by scene type.
- Routing security camera clips by event category.
For typed event/scene extraction with timestamps, use
[`_14_video_extraction/`](../_14_video_extraction/).
## Run
```bash
python cookbook/data_labeling/_13_video_classification/basic.py
python cookbook/data_labeling/_13_video_classification/with_confidence.py
```
Requires `GOOGLE_API_KEY`. Uses Gemini for native video input.
@@ -0,0 +1,23 @@
# Test Log - _13_video_classification
Tested 2026-05-17 against `gemini-3-flash-preview` (Gemini), agno 2.6.6. Input is `agno-public/demo/sample_seaview.mp4` (note: actual clip is a lab/microscope scene, not a seaview — sample asset name is misleading).
### basic.py
**Status:** PASS
**Description:** Single-label classification of the clip's scene type.
**Result:** Classified as `indoor` (consistent with the actual lab scene).
---
### with_confidence.py
**Status:** PASS
**Description:** Same task with a `confidence` field.
**Result:** Classified as `people` with `high` confidence (model picks a different valid label than basic.py — both are reasonable for a person in a lab).
---
@@ -0,0 +1,54 @@
"""
Video Classification - Basic
============================
Assign a single label from a closed set to a video clip. The model watches
the clip end-to-end and emits one label for the whole thing.
"""
from typing import Literal
import httpx
from agno.agent import Agent, RunOutput # noqa
from agno.media import Video
from agno.models.google import Gemini
from pydantic import BaseModel, Field
from rich.pretty import pprint # noqa
# ---------------------------------------------------------------------------
# Schema
# ---------------------------------------------------------------------------
class Classification(BaseModel):
scene_type: Literal[
"nature",
"urban",
"indoor",
"people",
"vehicle",
"animal",
"other",
] = Field(..., description="Dominant scene type in the clip")
# ---------------------------------------------------------------------------
# Create Agent
# ---------------------------------------------------------------------------
agent = Agent(
model=Gemini(id="gemini-3-flash-preview"),
instructions="You classify short video clips by dominant scene type.",
output_schema=Classification,
)
# ---------------------------------------------------------------------------
# Run Agent
# ---------------------------------------------------------------------------
if __name__ == "__main__":
url = "https://agno-public.s3.amazonaws.com/demo/sample_seaview.mp4"
video_bytes = httpx.get(url).content
run: RunOutput = agent.run(
"Classify this clip.",
videos=[Video(content=video_bytes, format="mp4")],
)
pprint({"url": url, "result": run.content})
@@ -0,0 +1,69 @@
"""
Video Classification - With Confidence
======================================
Adds confidence so downstream consumers can route low-confidence clips to
human review or a stronger model.
"""
from typing import Literal
import httpx
from agno.agent import Agent, RunOutput # noqa
from agno.media import Video
from agno.models.google import Gemini
from pydantic import BaseModel, Field
from rich.pretty import pprint # noqa
# ---------------------------------------------------------------------------
# Schema
# ---------------------------------------------------------------------------
class Classification(BaseModel):
scene_type: Literal[
"nature",
"urban",
"indoor",
"people",
"vehicle",
"animal",
"other",
] = Field(..., description="Dominant scene type in the clip")
confidence: Literal["high", "medium", "low"] = Field(
..., description="Confidence in the label"
)
# ---------------------------------------------------------------------------
# Agent Instructions
# ---------------------------------------------------------------------------
instructions = """\
Classify the clip and report a confidence:
- high - the dominant scene is unambiguous across the clip
- medium - the dominant scene is identifiable but some shots break the
pattern
- low - the clip mixes several scene types and you had to pick
"""
# ---------------------------------------------------------------------------
# Create Agent
# ---------------------------------------------------------------------------
agent = Agent(
model=Gemini(id="gemini-3-flash-preview"),
instructions=instructions,
output_schema=Classification,
)
# ---------------------------------------------------------------------------
# Run Agent
# ---------------------------------------------------------------------------
if __name__ == "__main__":
url = "https://agno-public.s3.amazonaws.com/demo/sample_seaview.mp4"
video_bytes = httpx.get(url).content
run: RunOutput = agent.run(
"Classify this clip with confidence.",
videos=[Video(content=video_bytes, format="mp4")],
)
pprint({"url": url, "result": run.content})
@@ -0,0 +1,30 @@
# Video Extraction
Video → typed Pydantic object. Pull events, scene descriptions, or
timestamped actions from a clip into a structured schema.
## Files
- `basic.py` — clip-level summary plus a flat list of scenes.
- `scene_descriptions.py` — one structured description per detected scene.
- `action_timestamps.py` — actions or events with start/end times in
seconds.
## When to use
- Indexing a video archive for search.
- Generating chapters / timestamps for a long-form recording.
- Building a `(video, structured_labels)` training set.
For a single clip-level label, use
[`_13_video_classification/`](../_13_video_classification/).
## Run
```bash
python cookbook/data_labeling/_14_video_extraction/basic.py
python cookbook/data_labeling/_14_video_extraction/scene_descriptions.py
python cookbook/data_labeling/_14_video_extraction/action_timestamps.py
```
Requires `GOOGLE_API_KEY`.
@@ -0,0 +1,33 @@
# Test Log - _14_video_extraction
Tested 2026-05-17 against `gemini-3-flash-preview` (Gemini), agno 2.6.6. Input is `agno-public/demo/sample_seaview.mp4` (actually a scientist using a microscope).
### basic.py
**Status:** PASS
**Description:** Extract a `VideoSummary` (overall summary, dominant subject, scenes list).
**Result:** Two-sentence summary plus three scene phrases; matches what's on screen.
---
### scene_descriptions.py
**Status:** PASS
**Description:** Extract a list of `Scene` objects with name, description, and visible objects.
**Result:** One scene returned with detailed description and visible objects list.
---
### action_timestamps.py
**Status:** PASS
**Description:** Extract a list of `Event`s with action descriptions and `[start, end]` second timestamps.
**Result:** Three events returned with monotonically increasing timestamps.
---
@@ -0,0 +1,64 @@
"""
Video Extraction - Action Timestamps
====================================
Detect actions or events in the clip with start/end times in seconds. The
shape used to generate chapter markers, "skip intro" cues, or training
data for temporal action detection.
"""
from typing import List
import httpx
from agno.agent import Agent, RunOutput # noqa
from agno.media import Video
from agno.models.google import Gemini
from pydantic import BaseModel, Field
from rich.pretty import pprint # noqa
# ---------------------------------------------------------------------------
# Schema
# ---------------------------------------------------------------------------
class Event(BaseModel):
action: str = Field(..., description="Short phrase naming the action or event")
start_seconds: float = Field(..., ge=0.0, description="Start time in seconds")
end_seconds: float = Field(..., ge=0.0, description="End time in seconds")
class Events(BaseModel):
events: List[Event]
# ---------------------------------------------------------------------------
# Agent Instructions
# ---------------------------------------------------------------------------
instructions = """\
Detect distinct actions or events in the clip. For each one, return a
short action name and start/end times in seconds from the beginning of
the clip. Times should be monotonically non-decreasing. Skip ambient or
filler content - only include events with a clear beginning and end.
"""
# ---------------------------------------------------------------------------
# Create Agent
# ---------------------------------------------------------------------------
agent = Agent(
model=Gemini(id="gemini-3-flash-preview"),
instructions=instructions,
output_schema=Events,
)
# ---------------------------------------------------------------------------
# Run Agent
# ---------------------------------------------------------------------------
if __name__ == "__main__":
url = "https://agno-public.s3.amazonaws.com/demo/sample_seaview.mp4"
video_bytes = httpx.get(url).content
run: RunOutput = agent.run(
"Detect events with timestamps.",
videos=[Video(content=video_bytes, format="mp4")],
)
pprint(run.content)
@@ -0,0 +1,67 @@
"""
Video Extraction - Basic
========================
Extract a clip-level summary plus a flat list of scenes. The shape most
useful for "give me an overview" tasks - chapters, thumbnails, index
metadata.
"""
from typing import List, Optional
import httpx
from agno.agent import Agent, RunOutput # noqa
from agno.media import Video
from agno.models.google import Gemini
from pydantic import BaseModel, Field
from rich.pretty import pprint # noqa
# ---------------------------------------------------------------------------
# Schema
# ---------------------------------------------------------------------------
class VideoSummary(BaseModel):
overall_summary: str = Field(
..., description="Two-sentence summary of the whole clip"
)
dominant_subject: Optional[str] = Field(
None, description="Main subject of the clip"
)
scenes: List[str] = Field(
default_factory=list,
description="Distinct scenes in order, each as a short phrase",
)
# ---------------------------------------------------------------------------
# Agent Instructions
# ---------------------------------------------------------------------------
instructions = """\
Watch the clip and summarize it. List distinct scenes in chronological
order as short phrases (e.g. "wide shot of harbor", "close-up of seagulls").
Skip transitions and filler. The summary should reflect what is actually
shown, not what the clip is "about" thematically.
"""
# ---------------------------------------------------------------------------
# Create Agent
# ---------------------------------------------------------------------------
agent = Agent(
model=Gemini(id="gemini-3-flash-preview"),
instructions=instructions,
output_schema=VideoSummary,
)
# ---------------------------------------------------------------------------
# Run Agent
# ---------------------------------------------------------------------------
if __name__ == "__main__":
url = "https://agno-public.s3.amazonaws.com/demo/sample_seaview.mp4"
video_bytes = httpx.get(url).content
run: RunOutput = agent.run(
"Summarize this clip and list the scenes.",
videos=[Video(content=video_bytes, format="mp4")],
)
pprint(run.content)
@@ -0,0 +1,66 @@
"""
Video Extraction - Scene Descriptions
=====================================
One structured description per detected scene. Each scene has a name, a
detailed description, and a list of visible objects - the shape used for
video indexing and search.
"""
from typing import List
import httpx
from agno.agent import Agent, RunOutput # noqa
from agno.media import Video
from agno.models.google import Gemini
from pydantic import BaseModel, Field
from rich.pretty import pprint # noqa
# ---------------------------------------------------------------------------
# Schema
# ---------------------------------------------------------------------------
class Scene(BaseModel):
name: str = Field(..., description="Short phrase naming the scene")
description: str = Field(..., description="One to two sentences of detail")
visible_objects: List[str] = Field(
default_factory=list, description="Up to five notable objects in the scene"
)
class ScenesDocument(BaseModel):
scenes: List[Scene]
# ---------------------------------------------------------------------------
# Agent Instructions
# ---------------------------------------------------------------------------
instructions = """\
Watch the clip and split it into distinct scenes. For each scene, return a
short name, a detailed description of what is visually shown, and the
notable objects. A new scene begins when the location, subject, or shot
changes substantially. Do not invent details.
"""
# ---------------------------------------------------------------------------
# Create Agent
# ---------------------------------------------------------------------------
agent = Agent(
model=Gemini(id="gemini-3-flash-preview"),
instructions=instructions,
output_schema=ScenesDocument,
)
# ---------------------------------------------------------------------------
# Run Agent
# ---------------------------------------------------------------------------
if __name__ == "__main__":
url = "https://agno-public.s3.amazonaws.com/demo/sample_seaview.mp4"
video_bytes = httpx.get(url).content
run: RunOutput = agent.run(
"Describe each scene in this clip.",
videos=[Video(content=video_bytes, format="mp4")],
)
pprint(run.content)
@@ -0,0 +1,28 @@
# Document Classification
Assign a document-type label to a (potentially multipage) PDF: invoice,
receipt, contract, spec sheet, report, recipe, etc.
## Files
- `basic.py` — single label per PDF.
- `with_confidence.py` — adds confidence in the label.
## When to use
- Routing inbound documents to the right downstream pipeline (an invoice
goes to AP, a contract to legal review, a recipe to a content team).
- Coarse pre-filtering before a heavier extraction pipeline runs.
If you want fields rather than a label, use
[`_16_document_extraction/`](../_16_document_extraction/).
## Run
```bash
python cookbook/data_labeling/_15_document_classification/basic.py
python cookbook/data_labeling/_15_document_classification/with_confidence.py
```
Requires `OPENAI_API_KEY`. The samples use a public PDF hosted by
agno-public; swap the URL for your own PDF.
@@ -0,0 +1,23 @@
# Test Log - _15_document_classification
Tested 2026-05-17 against `gpt-5.5` (OpenAIResponses), agno 2.6.6. Input is `agno-public/recipes/ThaiRecipes.pdf`.
### basic.py
**Status:** PASS
**Description:** Classify a PDF into one of {invoice, receipt, contract, spec_sheet, report, recipe, other}.
**Result:** Correctly classified as `recipe`.
---
### with_confidence.py
**Status:** PASS
**Description:** Same task with a `confidence` field.
**Result:** Classified as `recipe` with `high` confidence.
---
@@ -0,0 +1,59 @@
"""
Document Classification - Basic
===============================
Assign a single document-type label to a PDF. Use this as a coarse routing
step before a more specific extraction pipeline runs.
"""
from typing import Literal
from agno.agent import Agent, RunOutput # noqa
from agno.media import File
from agno.models.openai import OpenAIResponses
from pydantic import BaseModel, Field
from rich.pretty import pprint # noqa
# ---------------------------------------------------------------------------
# Schema
# ---------------------------------------------------------------------------
class Classification(BaseModel):
label: Literal[
"invoice",
"receipt",
"contract",
"spec_sheet",
"report",
"recipe",
"other",
] = Field(..., description="Document type")
# ---------------------------------------------------------------------------
# Agent Instructions
# ---------------------------------------------------------------------------
instructions = """\
Classify the attached PDF by document type. Use 'other' if it does not
fit any of the listed categories - do not force-fit. Base the decision
on the document's structure and content, not on a few keywords.
"""
# ---------------------------------------------------------------------------
# Create Agent
# ---------------------------------------------------------------------------
agent = Agent(
model=OpenAIResponses(id="gpt-5.5"),
instructions=instructions,
output_schema=Classification,
)
# ---------------------------------------------------------------------------
# Run Agent
# ---------------------------------------------------------------------------
if __name__ == "__main__":
url = "https://agno-public.s3.amazonaws.com/recipes/ThaiRecipes.pdf"
run: RunOutput = agent.run("Classify this document.", files=[File(url=url)])
pprint({"url": url, "result": run.content})
@@ -0,0 +1,66 @@
"""
Document Classification - With Confidence
=========================================
Adds confidence so downstream routing can treat low-confidence documents
differently (escalate, retry, or send to human review).
"""
from typing import Literal
from agno.agent import Agent, RunOutput # noqa
from agno.media import File
from agno.models.openai import OpenAIResponses
from pydantic import BaseModel, Field
from rich.pretty import pprint # noqa
# ---------------------------------------------------------------------------
# Schema
# ---------------------------------------------------------------------------
class Classification(BaseModel):
label: Literal[
"invoice",
"receipt",
"contract",
"spec_sheet",
"report",
"recipe",
"other",
] = Field(..., description="Document type")
confidence: Literal["high", "medium", "low"] = Field(
..., description="Confidence in the label"
)
# ---------------------------------------------------------------------------
# Agent Instructions
# ---------------------------------------------------------------------------
instructions = """\
Classify the attached PDF and report confidence:
- high - document is unambiguously of this type (clear structure, headings)
- medium - mostly clear but with mixed signals (e.g. a contract that
includes an embedded invoice)
- low - could fit several categories; pick the closest and flag low
"""
# ---------------------------------------------------------------------------
# Create Agent
# ---------------------------------------------------------------------------
agent = Agent(
model=OpenAIResponses(id="gpt-5.5"),
instructions=instructions,
output_schema=Classification,
)
# ---------------------------------------------------------------------------
# Run Agent
# ---------------------------------------------------------------------------
if __name__ == "__main__":
url = "https://agno-public.s3.amazonaws.com/recipes/ThaiRecipes.pdf"
run: RunOutput = agent.run(
"Classify this document with confidence.", files=[File(url=url)]
)
pprint({"url": url, "result": run.content})
@@ -0,0 +1,38 @@
# Document Extraction
Multipage PDF → typed Pydantic object. The closest neighbor in production
labeling: invoice fields, contract clauses, statement line items, lab
report fields.
## Files
- `basic.py` — extract top-level document metadata.
- `with_line_items.py` — extract a list of nested sub-objects (the
line-item shape: invoice line items, recipe steps, contract clauses).
- `with_confidence.py` — adds per-field confidence.
These examples use a public recipe book PDF as the demo input so the
cookbook runs out of the box. For the production case, swap the URL or
provide a local `File(filepath="...")` to your own invoice / contract /
report PDF, and adapt the schema.
## When to use
- Lift invoice / receipt / statement fields into a database row.
- Extract contract clauses for review queues.
- Build a structured index over a PDF corpus.
If you only need a document-type label, use
[`_15_document_classification/`](../_15_document_classification/). For multi-agent
quality control on top of this primitive, see
[`_18_quality_review/`](../_18_quality_review/).
## Run
```bash
python cookbook/data_labeling/_16_document_extraction/basic.py
python cookbook/data_labeling/_16_document_extraction/with_line_items.py
python cookbook/data_labeling/_16_document_extraction/with_confidence.py
```
Requires `OPENAI_API_KEY`.
@@ -0,0 +1,35 @@
# Test Log - _16_document_extraction
Tested 2026-05-17 against `gpt-5.5` (OpenAIResponses), agno 2.6.6. Input is `agno-public/recipes/ThaiRecipes.pdf`.
### basic.py
**Status:** PASS
**Description:** Extract document-level metadata (title, cuisine, language, recipe_count) into a `RecipeBook`.
**Result:** Title "Thai SELECT Cookbook", cuisine "Thai cuisine", language "English", recipe_count 10.
---
### with_confidence.py
**Status:** PASS
**Description:** Same task with `ConfidentField` wrapping each value.
**Result:** All four fields populated with sensible confidences.
**Note:** Originally failed with `Invalid schema for response_format 'RecipeBook': $ref cannot have keywords {'description'}` — OpenAI's strict structured-output mode rejects a `description` on a field whose type is itself a referenced model. Fix: removed the `Field(..., description=...)` annotation on `recipe_count` (kept the explanatory text as a code comment).
---
### with_line_items.py
**Status:** PASS
**Description:** Extract a `RecipeBook` with a nested list of `Recipe` line items (name, course, prep time).
**Result:** Multiple recipes extracted with full nested structure.
---
@@ -0,0 +1,56 @@
"""
Document Extraction - Basic
===========================
Extract top-level metadata from a multipage PDF into a typed object.
The schema here is for a recipe book - swap it for an Invoice, Contract,
LabReport, etc. for your domain.
"""
from typing import Optional
from agno.agent import Agent, RunOutput # noqa
from agno.media import File
from agno.models.openai import OpenAIResponses
from pydantic import BaseModel, Field
from rich.pretty import pprint # noqa
# ---------------------------------------------------------------------------
# Schema
# ---------------------------------------------------------------------------
class RecipeBook(BaseModel):
title: Optional[str] = Field(None, description="Book or document title")
cuisine: Optional[str] = Field(None, description="Cuisine or culinary tradition")
language: Optional[str] = Field(None, description="Language of the document")
recipe_count: Optional[int] = Field(
None, description="Number of distinct recipes in the document"
)
# ---------------------------------------------------------------------------
# Agent Instructions
# ---------------------------------------------------------------------------
instructions = """\
Extract document-level metadata from the attached PDF. Use exactly what
the document shows. If a field is not present, leave it null.
"""
# ---------------------------------------------------------------------------
# Create Agent
# ---------------------------------------------------------------------------
agent = Agent(
model=OpenAIResponses(id="gpt-5.5"),
instructions=instructions,
output_schema=RecipeBook,
)
# ---------------------------------------------------------------------------
# Run Agent
# ---------------------------------------------------------------------------
if __name__ == "__main__":
url = "https://agno-public.s3.amazonaws.com/recipes/ThaiRecipes.pdf"
run: RunOutput = agent.run("Extract document metadata.", files=[File(url=url)])
pprint({"url": url, "result": run.content})
@@ -0,0 +1,68 @@
"""
Document Extraction - With Confidence
=====================================
Adds per-field confidence. Useful when input PDFs vary in quality (scans,
faxes, mixed languages) and downstream needs to route uncertain fields to
human review.
"""
from typing import Literal, Optional
from agno.agent import Agent, RunOutput # noqa
from agno.media import File
from agno.models.openai import OpenAIResponses
from pydantic import BaseModel
from rich.pretty import pprint # noqa
Confidence = Literal["high", "medium", "low"]
# ---------------------------------------------------------------------------
# Schema
# ---------------------------------------------------------------------------
class ConfidentField(BaseModel):
value: Optional[str] = None
confidence: Confidence
class RecipeBook(BaseModel):
title: ConfidentField
cuisine: ConfidentField
language: ConfidentField
# Held as a string so per-field confidence applies cleanly to the count.
recipe_count: ConfidentField
# ---------------------------------------------------------------------------
# Agent Instructions
# ---------------------------------------------------------------------------
instructions = """\
Extract document metadata. For each field, report confidence:
- high - explicit in the document
- medium - inferred from structure or context
- low - guessed, partly obscured, or ambiguous
Be conservative. Mark unsure fields low.
"""
# ---------------------------------------------------------------------------
# Create Agent
# ---------------------------------------------------------------------------
agent = Agent(
model=OpenAIResponses(id="gpt-5.5"),
instructions=instructions,
output_schema=RecipeBook,
)
# ---------------------------------------------------------------------------
# Run Agent
# ---------------------------------------------------------------------------
if __name__ == "__main__":
url = "https://agno-public.s3.amazonaws.com/recipes/ThaiRecipes.pdf"
run: RunOutput = agent.run(
"Extract metadata with field-level confidence.", files=[File(url=url)]
)
pprint({"url": url, "result": run.content})
@@ -0,0 +1,62 @@
"""
Document Extraction - With Line Items
=====================================
Extract a nested list of sub-objects from a PDF. Same shape used for
invoice line items, statement transactions, contract clauses - the most
common production extraction pattern.
"""
from typing import List, Optional
from agno.agent import Agent, RunOutput # noqa
from agno.media import File
from agno.models.openai import OpenAIResponses
from pydantic import BaseModel, Field
from rich.pretty import pprint # noqa
# ---------------------------------------------------------------------------
# Schema
# ---------------------------------------------------------------------------
class Recipe(BaseModel):
name: str = Field(..., description="Recipe name as printed")
course: Optional[str] = Field(None, description="Appetizer, main, dessert, etc.")
prep_time_minutes: Optional[int] = None
class RecipeBook(BaseModel):
title: Optional[str] = None
cuisine: Optional[str] = None
recipes: List[Recipe] = Field(default_factory=list)
# ---------------------------------------------------------------------------
# Agent Instructions
# ---------------------------------------------------------------------------
instructions = """\
Extract document metadata and every distinct recipe from the attached PDF.
Each recipe entry should reflect what the document actually shows; leave
fields null if not present. Do not invent recipes or paraphrase names.
"""
# ---------------------------------------------------------------------------
# Create Agent
# ---------------------------------------------------------------------------
agent = Agent(
model=OpenAIResponses(id="gpt-5.5"),
instructions=instructions,
output_schema=RecipeBook,
)
# ---------------------------------------------------------------------------
# Run Agent
# ---------------------------------------------------------------------------
if __name__ == "__main__":
url = "https://agno-public.s3.amazonaws.com/recipes/ThaiRecipes.pdf"
run: RunOutput = agent.run(
"Extract the book metadata and every recipe.", files=[File(url=url)]
)
pprint({"url": url, "result": run.content})
@@ -0,0 +1,29 @@
# LLM as Judge
Score a generated output against criteria. The same machinery as labeling -
input is the (prompt, response) pair, output is a structured score - but
applied to evaluating models rather than producing training labels.
## Files
- `basic.py` — single 1-5 score on overall quality.
- `single_rubric.py` — explicit multi-criterion rubric, per-criterion
scores plus an overall.
- `with_rationale.py` — score plus a one-sentence rationale.
## When to use
- Evaluating model outputs in a test harness.
- Building eval dashboards for production agent workloads.
- Building reward-model training data (combine with
[`_05_text_pairwise_preference/`](../_05_text_pairwise_preference/)).
## Run
```bash
python cookbook/data_labeling/_17_llm_as_judge/basic.py
python cookbook/data_labeling/_17_llm_as_judge/single_rubric.py
python cookbook/data_labeling/_17_llm_as_judge/with_rationale.py
```
Requires `OPENAI_API_KEY`.
@@ -0,0 +1,33 @@
# Test Log - _17_llm_as_judge
Tested 2026-05-17 against `gpt-5.5` (OpenAIResponses), agno 2.6.6.
### basic.py
**Status:** PASS
**Description:** Score response quality on a 1-5 scale against the prompt.
**Result:** Detailed answer scored 5, "It just is." scored 1.
---
### with_rationale.py
**Status:** PASS
**Description:** Same task with a free-text rationale for the score.
**Result:** Score 4 with a rationale that calls out both the positives and a specific weakness.
---
### single_rubric.py
**Status:** PASS
**Description:** Score against an explicit rubric (correctness, completeness, clarity, concision, overall).
**Result:** All rubric dimensions scored independently with a separate overall.
---
@@ -0,0 +1,70 @@
"""
LLM as Judge - Basic
====================
Score a generated response against the prompt on a 1-5 scale. The simplest
eval primitive - and identical machinery to single-label text
classification, just applied to (prompt, response) pairs.
"""
from typing import Literal
from agno.agent import Agent, RunOutput # noqa
from agno.models.openai import OpenAIResponses
from pydantic import BaseModel, Field
from rich.pretty import pprint # noqa
# ---------------------------------------------------------------------------
# Schema
# ---------------------------------------------------------------------------
class Score(BaseModel):
overall: Literal[1, 2, 3, 4, 5] = Field(
..., description="Overall quality on a 1-5 scale where 5 is excellent"
)
# ---------------------------------------------------------------------------
# Agent Instructions
# ---------------------------------------------------------------------------
instructions = """\
Score the response on overall quality:
1 - unusable
2 - poor
3 - acceptable
4 - good
5 - excellent
Use the full scale. Reserve 5 for genuinely excellent responses.
"""
# ---------------------------------------------------------------------------
# Create Agent
# ---------------------------------------------------------------------------
agent = Agent(
model=OpenAIResponses(id="gpt-5.5"),
instructions=instructions,
output_schema=Score,
)
# ---------------------------------------------------------------------------
# Run Agent
# ---------------------------------------------------------------------------
def build_input(prompt: str, response: str) -> str:
return f"Prompt:\n{prompt}\n\nResponse:\n{response}"
if __name__ == "__main__":
prompt = "Explain why the sky is blue, in one sentence."
samples = [
(
"Sunlight scatters off air molecules; shorter (blue) wavelengths "
"scatter more, so blue dominates what we see."
),
"It just is.",
]
for response in samples:
run: RunOutput = agent.run(build_input(prompt, response))
pprint({"response": response, "score": run.content})
@@ -0,0 +1,68 @@
"""
LLM as Judge - Single Rubric
============================
Score against an explicit, named rubric. Each criterion gets its own
score, plus an overall. Use this for eval consistency across runs and
graders.
"""
from typing import Literal
from agno.agent import Agent, RunOutput # noqa
from agno.models.openai import OpenAIResponses
from pydantic import BaseModel, Field
from rich.pretty import pprint # noqa
Score = Literal[1, 2, 3, 4, 5]
# ---------------------------------------------------------------------------
# Schema
# ---------------------------------------------------------------------------
class RubricScore(BaseModel):
correctness: Score = Field(..., description="Is everything stated true")
completeness: Score = Field(..., description="Does it answer the prompt fully")
clarity: Score = Field(..., description="Easy to understand for the target reader")
concision: Score = Field(..., description="Free of wasted words")
overall: Score = Field(..., description="Holistic quality")
# ---------------------------------------------------------------------------
# Agent Instructions
# ---------------------------------------------------------------------------
instructions = """\
Score the response on each rubric criterion using a 1-5 scale:
1 - unusable, 2 - poor, 3 - acceptable, 4 - good, 5 - excellent.
The overall score should reflect the holistic quality, not a simple
average. A response that fails correctness can still score well on
clarity, but the overall should reflect the worst dimension.
"""
# ---------------------------------------------------------------------------
# Create Agent
# ---------------------------------------------------------------------------
agent = Agent(
model=OpenAIResponses(id="gpt-5.5"),
instructions=instructions,
output_schema=RubricScore,
)
# ---------------------------------------------------------------------------
# Run Agent
# ---------------------------------------------------------------------------
def build_input(prompt: str, response: str) -> str:
return f"Prompt:\n{prompt}\n\nResponse:\n{response}"
if __name__ == "__main__":
prompt = "How do I cancel my subscription?"
response = (
"Go to Settings > Subscription and click 'Cancel'. The cancellation "
"takes effect at the end of your current billing period."
)
run: RunOutput = agent.run(build_input(prompt, response))
pprint({"response": response, "scores": run.content})
@@ -0,0 +1,59 @@
"""
LLM as Judge - With Rationale
=============================
Score plus a one-sentence rationale. The rationale lets a human spot-
check the model's reasoning and is itself useful as training data for a
reward model.
"""
from typing import Literal
from agno.agent import Agent, RunOutput # noqa
from agno.models.openai import OpenAIResponses
from pydantic import BaseModel, Field
from rich.pretty import pprint # noqa
# ---------------------------------------------------------------------------
# Schema
# ---------------------------------------------------------------------------
class Score(BaseModel):
overall: Literal[1, 2, 3, 4, 5] = Field(..., description="Overall quality 1-5")
rationale: str = Field(..., description="One sentence explaining the score")
# ---------------------------------------------------------------------------
# Agent Instructions
# ---------------------------------------------------------------------------
instructions = """\
Score the response on a 1-5 scale where 5 is excellent. In the rationale,
name the specific quality (or absence) that drove your score - quote a
phrase from the response when possible.
"""
# ---------------------------------------------------------------------------
# Create Agent
# ---------------------------------------------------------------------------
agent = Agent(
model=OpenAIResponses(id="gpt-5.5"),
instructions=instructions,
output_schema=Score,
)
# ---------------------------------------------------------------------------
# Run Agent
# ---------------------------------------------------------------------------
def build_input(prompt: str, response: str) -> str:
return f"Prompt:\n{prompt}\n\nResponse:\n{response}"
if __name__ == "__main__":
prompt = "Suggest a name for a side project that helps people sleep better."
response = (
"How about 'Drift'? Short, evocative, and the .app domain is probably free."
)
run: RunOutput = agent.run(build_input(prompt, response))
pprint({"response": response, "score": run.content})
@@ -0,0 +1,39 @@
# Quality Review
Multi-agent quality control on top of any extraction primitive: two
labelers (different providers) extract independently, a reviewer
identifies disagreement, an adjudicator resolves it against the original
input. Use this when label quality matters more than throughput.
## Files
- `basic.py` — labeler → reviewer → adjudicator applied to text
extraction (the `_03_text_extraction/basic.py` Contact schema). The same
pattern composes on top of any image / audio / document extraction
cookbook in this directory.
## When to use
- High-value labels where a wrong answer is expensive.
- Building eval / training sets where disagreement is itself signal.
- Regulated workloads where you need an auditable resolution trail.
If you only need single-pass extraction, use the relevant `*_extraction/`
cookbook. If you need provider ensembling for evals rather than labels,
see [`_17_llm_as_judge/`](../_17_llm_as_judge/).
## Composition
`basic.py` is a flat imperative pipeline (three sequential agent calls).
For production traceability and parallelism, wrap the same agents in
`Workflow(Parallel(labeler_a, labeler_b), reviewer, Condition(adjudicator))` -
see `cookbook/04_workflows/04_parallel_execution/parallel_with_condition.py`.
## Run
```bash
python cookbook/data_labeling/_18_quality_review/basic.py
```
Requires `OPENAI_API_KEY` and `ANTHROPIC_API_KEY` (the two labelers run
on different providers for ensemble diversity).
@@ -0,0 +1,13 @@
# Test Log - _18_quality_review
Tested 2026-05-17 against `gpt-5.5` (OpenAIResponses) for labeler A / reviewer / adjudicator and `claude-sonnet-4-5` (Claude) for labeler B, agno 2.6.6.
### basic.py
**Status:** PASS
**Description:** End-to-end labeler / reviewer / adjudicator pipeline on a Contact extraction task. Two labelers (different providers) extract, a reviewer diffs them, an adjudicator runs only if there's disagreement.
**Result:** Labelers agreed; reviewer returned `needs_adjudication=False`; final label returned with `notes='Labelers agreed.'` (adjudicator path not exercised on this input — covered by the original invoice cookbook in git history).
---
@@ -0,0 +1,132 @@
"""
Quality Review - Basic
======================
Multi-agent quality control on a labeling task. Two labelers (different
providers) extract a Contact from the input text. A reviewer compares them
and decides if adjudication is needed. If yes, an adjudicator resolves
disagreements against the original text.
This is what the original cookbook/data_labeling/ invoice pipeline did,
condensed to one file and applied to text input so the pattern stands out.
The same shape composes on top of any extraction primitive in this
directory (text, image, audio, document).
"""
from typing import List, Optional
from agno.agent import Agent
from agno.models.anthropic import Claude
from agno.models.openai import OpenAIResponses
from pydantic import BaseModel, Field
from rich.pretty import pprint
# ---------------------------------------------------------------------------
# Schemas
# ---------------------------------------------------------------------------
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
# ---------------------------------------------------------------------------
# Create Agents
# ---------------------------------------------------------------------------
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=OpenAIResponses(id="gpt-5.5"),
instructions=LABELER_INSTRUCTIONS,
output_schema=Contact,
)
labeler_b = Agent(
name="Labeler B",
model=Claude(id="claude-sonnet-4-5"),
instructions=LABELER_INSTRUCTIONS,
output_schema=Contact,
)
reviewer = Agent(
name="Reviewer",
model=OpenAIResponses(id="gpt-5.5"),
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=OpenAIResponses(id="gpt-5.5"),
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,
)
# ---------------------------------------------------------------------------
# Pipeline
# ---------------------------------------------------------------------------
def label_with_quality_review(text: str) -> FinalLabel:
a = labeler_a.run(text).content
b = labeler_b.run(text).content
review_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(review_prompt).content
if not report.needs_adjudication:
return FinalLabel(contact=a, notes="Labelers agreed.")
adj_prompt = (
f"Original input:\n{text}\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 adjudicator.run(adj_prompt).content
# ---------------------------------------------------------------------------
# Run
# ---------------------------------------------------------------------------
if __name__ == "__main__":
text = (
"Sarah Johnson, VP Marketing at Acme Corp. "
"Reach me at [email protected] or +1-555-0102."
)
pprint(label_with_quality_review(text))