6. Multiple agents¶
Questions
What are multi-agent workflows?
When does it make sense to use multi-agent workflows and when does it not?
How do we connect multiple agents together, and are there different design patterns for doing so?
Objectives
Understand how to design a multi-agent workflow.
Analyze the pros and cons for using them.
Why multi-agent workflows¶
So far we have understood and played with single agent workflows. Often a single agent, with:
a broad role,
several capabilities
unstructured outputs
data validation only within tools, and none before and after the agent responds.
When the number of capabilities grows (various tools, skills, MCP servers, Memory etc.) in a single agent, the potential permutations in which these will execute increases exponentially. This can make it difficult to reason about the trajectory of the agent’s execution and the quality of the output. By breaking down the agent into multiple agents, each with a limited set of tools and capabilities, we can reduce the complexity of the agent’s execution and improve the quality of the output.
Also, with multi-agents, each with a specific role – allows to generate structured output, with output validators. In other words, we could perform type validation and logic validation, thereby catching some set of known failures early.
There are more benefits and drawbacks when it comes to multi-agent workflows. It will become evident as we explore them. However here is a teaser:
Potential benefits of multi-agent workflows
Per-agent model selection: different models for different tasks, since some tasks might be simple.
Separation of concerns: limited set of tools per agent, similar to encapsulation in Object-oriented programming.
Saving context size: ability to (somewhat) stateless, and separate message history per agent.
Adding output type validation and data validators after each agent.
Ability for the developer to split tasks into logical chunks and assign role.
The exact benefits can also vary depending on the architecture or design pattern one chooses to adopt.
Uncertainities of multi-agent workflows
Depending on the constuction of the multi-agent workflow, there are some uncertainties with the following:
Cost: can result in fewer or more model calls, more handoffs, more places to fail.
Better, similar or worse complexity when it comes to possible trajectories of tool-calls compared to single-agent workflows.
Design patterns of multi-agent workflows¶
Since this is an emerging field the names and understanding of these “patterns” vary. In 2026, according to both LangChain and Pydantic AI, these 4 patterns are mentioned:
Agent delegation — agents using another agent via tools, either wired up by hand or with the Harness’s
SubAgentscapabilityProgrammatic agent hand-off — one agent runs, then application code calls another agent (or a deterministic function) and so on
Graph based control flow — for the most complex cases, a graph-based state machine can be used to control the execution of multiple agents
Deep Agents — autonomous agents with planning, file operations, task delegation, and sandboxed code execution
And before we get too excited, we also find this gem of an advice:
If Pydantic AI agents are a hammer, and multi-agent workflows are a sledgehammer, then graphs are a nail gun… graphs are a powerful tool, but they’re not the right tool for every job. Please consider other multi-agent approaches before proceeding.
And therefore we will limit our attention to two patterns, namely agent delegation and programmatic agent hand-off.
Programmatic agent hand-off¶
Programmatic hand-off is a design pattern where the output of one agent is used as the input to another agent. With the handoffs pattern, when you need to enforce sequential constraints (unlock capabilities only after preconditions are met), and allows interaction with the user at specific points of the workflow.
Here is a typical hand-off multi-agent workflow.
graph TB
U[User] -->|Prompt| H[Function 0]
H --> A[Agent A]
A --> |Parse and validate input| B[Agent B]
B --> |Failed| A
B --> |Prepare question| F[Function 1]
M --> |Read| B
U <-.->|Ask for details| F
B --> |Validated input| C
F --> C[Agent C]
C <-.->|Read/Write| M[Capabilities: Database, MCP, Memory]
C <-.->|Confirm| U
C --> G[Function 2]
G -->|Continue or exit| U
G --> Z[Exit]
classDef func fill:#F6FFDB,stroke:#6E8900,stroke-width:2px,color:#2E3900
classDef agent fill:#E5F4FF,stroke:#006DDD,stroke-width:2px,color:#030710
class U,F,G,H func
class A,B,C,,E agent
Notice that in this pattern, the hand-off can occur from agent to agent, or from agent to function and back to another agent. This allows the developer to implement the workflow in a way that is most natural for the problem domain and to add deterministic steps where needed.
Agent delegation a.k.a Subagents¶
graph TB
A[User] --> B[Orchestrator agent]
B --> C[Sub-agent A]
B --> D[Sub-agent B]
B --> E[Sub-agent C]
C --> B
D --> B
E --> B
B --> F[User response]
classDef trigger fill:#F6FFDB,stroke:#6E8900,stroke-width:2px,color:#2E3900
classDef process fill:#E5F4FF,stroke:#006DDD,stroke-width:2px,color:#030710
class A,F trigger
class B,C,D,E process
Agent delegation is a design pattern where one “orchestrator” agent delegates a task to child agents or “sub-agents”. The sub-agent runs with its own message history, so task must be self-contained.
Breaking down the problem into different agents¶
The flight booking problem naturally splits into four steps: understand where
and when the user wants to fly, search for flights, pick a seat, and book.
Each step has a clear input and output, which makes it a good fit for the
programmatic hand-off pattern. In flight_booking/agents.py each step becomes
one agent with a narrow role, its own output type, and its own validation –
together they play the role of agents A, B and C in the diagram above.
The data models define the “contract” between the agents:
@dataclass
class Deps:
req_origin: str = ''
req_destination: str = ''
class TripRequest(BaseModel):
"""Details of the trip request by the user."""
req_origin: str
req_destination: str
req_date: str = Field(
description=(
'Travel date exactly as the user phrased it, e.g. "tomorrow", '
'"next Friday", or "2025-01-15". Use "today" if unspecified.'
)
)
class FlightDetails(BaseModel):
"""Details of the most suitable flight."""
flight_number: str
price: int
origin: str = Field(description='Departure city')
destination: str = Field(description='Arrival city')
date: datetime.date
departure_time: datetime.time
arrival_time: datetime.time
class NoFlightFound(BaseModel):
"""When no valid flight is found."""
class SeatPreference(BaseModel):
row: int = Field(ge=1, le=30)
seat: Literal['A', 'B', 'C', 'D', 'E', 'F']
class Failed(BaseModel):
"""Unable to extract a seat selection."""
Deps carries the requested origin and destination to tools and validators
via the context object ctx.deps.
TripRequest, FlightDetails and SeatPreference are the
structured outputs of the first three agents. Note how the models themselves
already encode data types and instructions: the req_date field description
tells the model to keep the date exactly as the user phrased it, and
SeatPreference restricts rows to 1-30 and seats to a valid letter, so
nonsense values fail validation before they can reach the next agent.
Finally, NoFlightFound and Failed are empty marker data models: “no flights”
and “could not extract a seat” are valid outcomes, not errors, so downstream
code can branch on the output type instead of parsing error strings.
The trip extractor agent¶
trip_extractor_agent = Agent[TripRequest](
model,
deps_type=Deps,
output_type=TripRequest,
instructions=(
"You are a helpful assistant that helps the user find a flight. "
"Extract the origin, destination, and travel date of the trip the user "
"is looking for. Keep the date exactly as the user phrased it (e.g. "
"'tomorrow' or 'next Friday') and do not resolve it yourself; use 'today' "
"if the user did not specify a date."
),
retries=3,
)
The first agent has a single job: turn the user’s free-text prompt into a
TripRequest, sometimes with small corrections if needed. For example, it
transforms a prompt:
Find me a flight from stockhom to helsinki next week
into
TripRequest(req_origin='Stockholm', req_destination='Helsinki', req_date='next week')
It has no tools at all (for now) – anything it cannot extract from the prompt
is a job for a later agent. The retries=3 setting gives the model three
attempts to produce output that passes validation.
The search agent¶
# Its tools (get_current_datetime and search_flights) come from the lesson 04
# MCP server (03_mcp_server.py), which must be running at MCP_URL (see config.py).
search_agent = Agent[Deps, list[FlightDetails] | NoFlightFound](
model,
output_type=list[FlightDetails] | NoFlightFound,
deps_type=Deps,
retries=4,
capabilities=[MCP(url=MCP_URL)], # , allowed_tools=["get_current_time", "search_flights"])],
instructions=(
'Your job is to find flights for the user on the requested date. '
'First resolve the travel date to a concrete YYYY-MM-DD date, using the '
'get_current_datetime tool for relative dates like "tomorrow" or "next Friday". '
'Then use search_flights to look up matching flights and return all of them '
'sorted by price (cheapest first), each with its date set to the resolved '
'travel date. Add the arrival and departure time as well. Return NoFlightFound only if there are no matching flights.'
),
)
@search_agent.output_validator
async def validate_flights(
ctx: RunContext[Deps], output: list[FlightDetails] | NoFlightFound
) -> list[FlightDetails] | NoFlightFound:
"""Procedural validation that every returned flight meets the constraints."""
if isinstance(output, NoFlightFound):
return output
errors: list[str] = []
for flight in output:
if flight.origin.lower() != ctx.deps.req_origin.lower():
errors.append(
f'Flight {flight.flight_number} should have origin '
f'{ctx.deps.req_origin}, not {flight.origin}'
)
if flight.destination.lower() != ctx.deps.req_destination.lower():
errors.append(
f'Flight {flight.flight_number} should have destination '
f'{ctx.deps.req_destination}, not {flight.destination}'
)
if errors:
raise ModelRetry('\n'.join(errors))
else:
return output
The search agent resolves the relative date (“tomorrow”, “next Friday”) to a
concrete date using the get_current_datetime tool, then calls
search_flights and returns all matches sorted by price. Both tools come
from the MCP server we built in episode 4, attached here as a single
capability. The output type is a union: list[FlightDetails] | NoFlightFound
– an empty search result is a valid answer, not an error.
The output validator performs procedural validation: every returned flight must
match the origin and destination in ctx.deps, and any mismatch raises a
ModelRetry, which sends the errors back to the model so it can correct itself
within its retries budget.
The seat preference agent¶
seat_preference_agent = Agent[object, SeatPreference | Failed](
model,
output_type=SeatPreference | Failed,
instructions=(
"Extract the user's seat preference. "
'Seats A and F are window seats. '
'Seats B and E are middle seats and C and D are aisle seats'
'Row 1 is the front row and has extra leg room. '
'Rows 14, and 20 also have extra leg room. '
),
)
The simplest of the four: no tools, no deps, no validator – just a mapping
from the user’s words onto the constrained SeatPreference model. The domain
knowledge (which seats are window or aisle, which rows have extra legroom)
lives in the instructions, so the model can for example infer a specific seat
when the user asks for “a window seat with extra legroom”. If it cannot
extract a selection, it returns Failed rather than guessing.
The book agent¶
# Memory files are stored in hands-on/agent-memory/main/
MEMORY_DIR = "agent-memory"
book_agent = Agent[tuple[FlightDetails, SeatPreference], str](
model,
deps_type=tuple[FlightDetails, SeatPreference],
output_type=str,
capabilities=[
MCP(url=MCP_URL), #, allowed_tools=["book_flight"]),
# Resolve paths relative to the "hands-on" directory
LocalWorkspace(Path(__file__).parent.parent),
Skills("./agent-skills", include=["travel-policy"]),
# The agent writes notes to MEMORY.md, which is injected into every request
Memory(FileStore(MEMORY_DIR)),
],
instructions=(
'You are a flight booking agent. '
'Do not request the user for further input and use deps and memory to fill in missing information. '
'Before booking ensure with the skill that it complies with travel policy. '
'If all checks pass, book the flight for the user and return the booking confirmation in plain text. '
)
)
The final agent performs the side-effectful step: booking the flight. It
receives the chosen flight and seat as a tuple via deps, and its
instructions tell it to not ask the user for further input. This is where the
capabilities of the previous episodes come together in a single agent: the
MCP server provides the book_flight tool, the travel-policy skill checks
that the booking complies with the travel policy before the agent commits.
Memory persists data such as passenger name from MEMORY.md, injecting them
into every request.
Wiring the agents together¶
Using hand-off¶
The agents to use the programmatically hand-off pattern, are composed together in
a single main function in flight_booking/hand_off.py. The function is
structured as a loop that starts with the user prompt, and ends when the user
types quit. The loop is structured as a sequence of agents that are called
in turn, and each agent can either return a value to the next agent, or raise
an exception to signal that the user needs to change their trip request.
graph TB
prompt(("user prompt")) -->|"str"| extract["infer_trip_request_via_trip_extractor_agent"]
extract -->|"TripRequest"| search["infer_flights_via_search_agent"]
search -->|"NoFlightFound: user changes trip"| prompt
search -->|"list[FlightDetails]"| select["select_flight"]
select -->|"None: change trip"| prompt
select --> seat["infer_seat_via_seat_preference_agent"]
select -->|"FlightDetails"| book
seat -->|"SeatPreference"| book["infer_purchase_via_book_agent"]
book -->|"next request"| prompt
prompt -->|"quit"| END
The above flowchart is implemented as follows, where all the infer_* functions
are helper functions that call the respective agents.
async def main(test_prompt: str | None = None):
user_prompt = test_prompt or input(
'🤖 Hi! I am a flight search assistant. '
'You can tell me the locations and dates of your trip and I will find the best flight for you\n'
'You: '
)
conversation_history: list[ModelMessage] | None = None
deps = Deps()
usage: RunUsage = RunUsage()
# TODO: Add trip request tool and output validator
# Start agent loop
while user_prompt.strip().lower() not in QUIT_WORDS:
trip, conversation_history = await infer_trip_request_via_trip_extractor_agent(
user_prompt, deps, conversation_history, usage
)
flights = await infer_flights_via_search_agent(trip, deps, usage)
if isinstance(flights, NoFlightFound):
print(
f'No flights from {trip.req_origin} to {trip.req_destination} '
f'on {trip.req_date}.'
)
# nothing bought yet: let the user change the date, destination, etc.
user_prompt = input(
'🤖 Tell me what to change (e.g. a different date or destination), '
f'or type {QUIT_WORDS} to stop\n'
'You: '
)
else:
display_flights(flights)
flight = select_flight(flights)
if flight is not None:
seat = await infer_seat_via_seat_preference_agent(usage)
booking_history = await infer_purchase_via_book_agent(usage, flight, seat)
conversation_history.extend(booking_history)
user_prompt = input(
'🤖 Tell me what to do next (e.g. change dates or book another flight), '
f'or type {QUIT_WORDS} to stop\n'
'You: '
)
Demo
Run
uv run 06_multi_agent_flight_booking.pyand try one of the following prompts.
Book me a flight from Frankfurt to Helsinki
I want to fly from Stockholm to Netherlands in 3 months
Flight to Amsterdam from Warsaw on Friday
Pick a flight
For seat preference answer in one of the following:
the best seat
22, aisle seat
7A
Exercise 6.1: Adding output validator to the hand-off workflow¶
Exercise
Run uv run 06_multi_agent_flight_booking.py.
Try to prompt as follows, and notice how long it takes. In what step did the multi-agent workflow realize that this is invalid?
Flight from Singapore to Helsinki
Exit the workflow.
Add a tool and output validator to the first agent inside
main_handoff.py, allowing the model to recover from a wrong input, early!
@trip_extractor_agent.tool_plain
def ask_user_for_clarification(question: str) -> str:
"""Ask the user a question and return their answer."""
return input(f'🤖 {question}\nYou: ')
@trip_extractor_agent.output_validator
async def validate_trip_request(ctx: RunContext[Deps], output: TripRequest) -> TripRequest:
"""Validate origin and destination of the trip."""
VALID_CITIES = {'Amsterdam', 'Barcelona', 'Copenhagen', 'Dublin', 'Frankfurt', 'Helsinki', 'Istanbul', 'Lisbon', 'London', 'Madrid', 'Munich', 'Oslo', 'Paris', 'Rome', 'Stockholm', 'Vienna', 'Warsaw', 'Zurich' }
if (
output.req_origin not in VALID_CITIES
or output.req_destination not in VALID_CITIES
):
raise ModelRetry(
f'Invalid origin or destination city. Valid cities are: {VALID_CITIES}'
)
return output
Tip
You can inject the above code in flight_booking/main_handoff.py at the highlighted section
# TODO: Add trip request tool and output validator
# Start agent loop
Re-prompt the workflow, starting with an incorrect origin and/or destination.
Exercise 6.2: Using a different model for seat preference¶
Exercise
Seat preference being the simplest of the agents, can be set to a smaller model.
Replace the default
modelused for the seat preference agent with a different one inflight_booking/agents.py
seat_preference_agent = Agent[object, SeatPreference | Failed](
model,
output_type=SeatPreference | Failed,
instructions=(
"Extract the user's seat preference. "
'Seats A and F are window seats. '
'Seats B and E are middle seats and C and D are aisle seats'
'Row 1 is the front row and has extra leg room. '
'Rows 14, and 20 also have extra leg room. '
),
)
Preferably use a smaller model, but anything running / available will do too. Run
uv run 00_check_inference_server.py
to find out what is running.
Tip
In order to do so, copy the lines to define model and make a new model_small, in flight_booking/config.py with a different model name.
# default model
model = OpenAIChatModel(
os.getenv("MODEL_NAME", "google/gemma-4-31b-it"),
provider=OpenAIProvider(openai_client=client),
settings=model_settings,
)
# TODO: make a small model
# model_small =
Import model_small into flight_booking/agents.py module and use it
Quickly test the change with one of the test prompts:
uv run 06_multi_agent_flight_booking.py --test
Using agent-delegation¶
In Pydantic AI an “orchestrator” agent is a regular agent whose SubAgents
capability exposes the specialist agents to it as a single
delegate_task(agent_name, task) tool. Where the hand-off pattern hardcodes
the pipeline in main, here the orchestrator’s instructions describe the
booking workflow, and the model decides for itself which sub-agent to call at
each step.
graph LR
prompt(("user")) -->|"prompt"| orch["orchestrator"]
orch -->|"delegate_task(agent_name, task)"| subs
subgraph subs["SubAgents capability"]
direction LR
extract["trip_extractor_agent"]
search["flight_searcher"]
seat["seat_extractor"]
book["flight_booker"]
end
subs -->|"task result"| orch
orch -->|"question"| ask_user["ask_user"]
ask_user -->|"answer"| orch
orch -->|"ask_user"| prompt
prompt -->|"quit"| END
The four specialists from agents.py are registered under a name and a
description. The description is all the orchestrator sees when routing, so it
doubles as the routing prompt. The ask_user tool lets the orchestrator talk
to the user mid-run, for example to present the found flights or ask for a
seat.
orchestrator = Agent(
model,
deps_type=Deps,
capabilities=[
SubAgents(
agents=[
SubAgent(
trip_extractor_agent,
name='trip_extractor',
description=(
'Extracts the origin, destination and travel date '
'from a user request'
),
),
SubAgent(
search_agent,
name='flight_searcher',
description=(
'Resolves the current date and time. '
'Then based on the travel date, origin and destination '
'of the trip requested, search flights.'
),
),
SubAgent(
seat_preference_agent,
name='seat_extractor',
description=(
"Extracts a seat preference (row number and seat "
"letter) from the user's answer"
),
),
SubAgent(
book_agent,
name='flight_booker',
description=(
'Books a chosen flight and seat, checking the travel '
'policy first, and finally purchasing a ticket.'
),
),
]
)
],
instructions=(
'You are the orchestrator of a flight booking system. The user talks '
'to you, and you delegate the specialist work to sub-agents with the '
'delegate_task tool. Follow this workflow for a booking request:\n'
'1. Delegate the user request verbatim to `trip_extractor` to get the '
'origin, destination and travel date.\n'
"2. Delegate to `flight_searcher` with a task like 'Find me a flight "
'from <origin> to <destination> on <date>.\'\n'
'3. If it reports NoFlightFound, tell the user and ask what to change.\n'
'4. Otherwise present the flights to the user and use ask_user to ask '
'which one to buy. If the user wants to change the trip instead, start '
'again from step 1 with their new request.\n'
'5. Use ask_user to ask for their seat preference, then delegate their '
'answer to `seat_extractor`. If it cannot extract a valid seat, ask '
'the user to phrase it differently and delegate the new answer.\n'
'6. Delegate the purchase to `flight_booker` with a self-contained task '
'that includes every field of the chosen flight (flight number, '
'origin, destination, date, departure and arrival times, price) and '
'the chosen seat (row and seat letter), then confirm the purchase to '
'the user.'
),
)
@orchestrator.tool_plain
def ask_user(question: str) -> str:
"""Ask the user a question and return their answer."""
return input(f'🤖 {question} ')
Each delegation runs the sub-agent with its own message history, so a task must be self-contained – this is why the instructions tell the orchestrator to spell out every field of the chosen flight and seat when delegating the purchase.
With the routing moved into the agent, main shrinks to a single
orchestrator.run call per turn; the conversation history and the shared
usage limits are all the state the loop keeps, with the history only visible
to the orchestrator agent.
async def main(test_prompt: str | None = None):
user_prompt = test_prompt or input(
'🤖 Hi! I am a flight search assistant. '
'You can tell me the locations and dates of your trip and I will find the best flight for you\n'
'You: '
)
conversation_history: list[ModelMessage] | None = None
deps = Deps()
while user_prompt.strip().lower() not in QUIT_WORDS:
# the orchestrator keeps the conversation history and decides itself
# which sub-agent to delegate to at each step; the shared usage limits
# apply across the whole agent tree
with logfire.span('Orchestrator: delegate to sub-agents'):
result = await orchestrator.run(
user_prompt,
deps=deps,
message_history=conversation_history,
usage_limits=usage_limits,
)
conversation_history = result.all_messages()
print(f'🤖 {result.output}')
user_prompt = input(
'🤖 Tell me what to do next (e.g. book another trip), '
'or type "quit" to stop\nYou: '
)
Demo
Run
uv run 06_multi_agent_flight_booking.py --pattern delegation
Try one of the prompt as in the hand-off demo, and watch the orchestrator present the flights and ask for your seat before delegating the purchase.
Tracing and visualizing agent trajectories¶
TODO: Describe tracing with Jaeger / MLflow
Exercise 6.3: Discuss the differences¶
Discussion
Which multi-agent pattern was more flexible and less sticter to use?
Which multi-agent pattern had a more predictable behaviour?
Solution
The agent-delegation pattern is more flexible and less strict to use.
The programmatic hand-off pattern has a more predictable behaviour.
Summary¶
Keypoints
Multi-agent workflows split a complex task into agents with narrow roles, giving per-agent model selection, separation of concerns, smaller context per agent, and output validation after each step – at the cost of more model calls and more places to fail.
The two patterns covered are programmatic hand-off, where application code chains agents in a fixed sequence, and agent delegation, where an orchestrator agent routes tasks to sub-agents via a
delegate_tasktool.Hand-off is more predictable since the pipeline is hardcoded in
main, while delegation is more flexible since the orchestrator decides which sub-agent to call,
Here are some of the pages for your reference and further reading.