ai_pipeline takes one ticket and returns one dict. Your inbox has 200 performance reviews due for classification by Friday. What changes?
The input. ai_pipeline handled one text — now I need to run the same classifier across a whole list and get a list of labels back.
One agent, one list comprehension, one pass over all 200 items. You create the classifier once outside the loop — so you're not re-instantiating an Agent object on every iteration — then map it over the inputs:
from typing import Literal
def batch_classify(texts: list) -> list:
classifier = Agent(model, result_type=Literal["high", "medium", "low"])
return [classifier.run_sync(t).output for t in texts]
Why define classifier outside the comprehension? Doesn't Agent(model, ...) just configure — it doesn't cost a network call, right?
Correct — the network call is .run_sync(t). But creating the Agent once and reusing it is cleaner and cheaper if there are ever startup costs (tool registration, system-prompt encoding). One object, iterated over every item. The comprehension handles the rest.
I'm chaining these like functions now. ai_pipeline on Day 18 was one text in, one dict out. Now it's a list in, a list of labels out — same pattern, just scaled.
That's the scalability moment. A for loop that took two days of manual triage last quarter is now one line of Python. You could run this nightly on yesterday's tickets — no analyst blocked, no backlog.
I could schedule this. Import the ticket export, call batch_classify, write the labels back to the spreadsheet. That's the entire Thursday morning triage meeting automated.
Exactly. And because the Literal constraint forces each output to be exactly "high", "medium", or "low", the result list is safe to sort, filter, or aggregate without any cleanup step:
labels = batch_classify(tickets)
high_priority = [t for t, l in zip(tickets, labels) if l == "high"]No free-text parsing, no edge cases where the model wrote "urgent" instead of "high". The same comprehension pattern works for any reduction — shortest response, word count, structured dict — once the batch loop is in place, the logic inside is just a swap.
A list comprehension over .run_sync() calls is the simplest batch pattern in PydanticAI.
Create the agent once outside the comprehension. The agent object holds configuration (model, result_type, system_prompt) — instantiating it inside the loop is wasteful and inconsistent.
result_type=Literal["high","medium","low"] constrains every output in the batch. If the model returns anything else, PydanticAI raises before your code sees it — so you get a clean list or a clear error, never silent garbage.
Return value is a plain Python list — ready to zip with your original inputs, write to a spreadsheet, or pass to the next agent.
Any time you have a fixed list and a single classification task: support tickets, review comments, survey responses, email threads. The list comprehension scales linearly — 10 items or 10,000, the structure is identical.
ai_pipeline takes one ticket and returns one dict. Your inbox has 200 performance reviews due for classification by Friday. What changes?
The input. ai_pipeline handled one text — now I need to run the same classifier across a whole list and get a list of labels back.
One agent, one list comprehension, one pass over all 200 items. You create the classifier once outside the loop — so you're not re-instantiating an Agent object on every iteration — then map it over the inputs:
from typing import Literal
def batch_classify(texts: list) -> list:
classifier = Agent(model, result_type=Literal["high", "medium", "low"])
return [classifier.run_sync(t).output for t in texts]Why define classifier outside the comprehension? Doesn't Agent(model, ...) just configure — it doesn't cost a network call, right?
Correct — the network call is .run_sync(t). But creating the Agent once and reusing it is cleaner and cheaper if there are ever startup costs (tool registration, system-prompt encoding). One object, iterated over every item. The comprehension handles the rest.
I'm chaining these like functions now. ai_pipeline on Day 18 was one text in, one dict out. Now it's a list in, a list of labels out — same pattern, just scaled.
That's the scalability moment. A for loop that took two days of manual triage last quarter is now one line of Python. You could run this nightly on yesterday's tickets — no analyst blocked, no backlog.
I could schedule this. Import the ticket export, call batch_classify, write the labels back to the spreadsheet. That's the entire Thursday morning triage meeting automated.
Exactly. And because the Literal constraint forces each output to be exactly "high", "medium", or "low", the result list is safe to sort, filter, or aggregate without any cleanup step:
labels = batch_classify(tickets)
high_priority = [t for t, l in zip(tickets, labels) if l == "high"]No free-text parsing, no edge cases where the model wrote "urgent" instead of "high". The same comprehension pattern works for any reduction — shortest response, word count, structured dict — once the batch loop is in place, the logic inside is just a swap.
A list comprehension over .run_sync() calls is the simplest batch pattern in PydanticAI.
Create the agent once outside the comprehension. The agent object holds configuration (model, result_type, system_prompt) — instantiating it inside the loop is wasteful and inconsistent.
result_type=Literal["high","medium","low"] constrains every output in the batch. If the model returns anything else, PydanticAI raises before your code sees it — so you get a clean list or a clear error, never silent garbage.
Return value is a plain Python list — ready to zip with your original inputs, write to a spreadsheet, or pass to the next agent.
Any time you have a fixed list and a single classification task: support tickets, review comments, survey responses, email threads. The list comprehension scales linearly — 10 items or 10,000, the structure is identical.
Create a free account to get started. Paid plans unlock all tracks.