Single-agent prototypes fail when tasked with multi-domain enterprise workflows. An agent asked to simultaneously browse the web, write SQL queries, verify mathematical models, and write marketing copy eventually hallucinates or loses track of its primary objective.
The solution is the Supervisor Pattern: a central orchestrator that breaks problems down and delegates to specialized worker agents with dedicated toolsets.
1. Architecture: The Hierarchical Command Structure
┌───────────────────────┐
│ Supervisor Agent │
│ (State Orchestrator)│
└───────────┬───────────┘
│
┌────────────────────────────┼────────────────────────────┐
▼ ▼ ▼
┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐
│ Search Worker │ │ SQL Analyst │ │ Copywriter │
│ (Tavily Tools) │ │ (Postgres Tool)│ │ (Brand Guide) │
└─────────────────┘ └─────────────────┘ └─────────────────┘
2. Implementation with PydanticAI & LangGraph
from typing import Literal
from pydantic import BaseModel, Field
from pydantic_ai import Agent
class RoutingDecision(BaseModel):
next_worker: Literal["sql_analyst", "web_researcher", "FINISH"]
instructions_for_worker: str
supervisor = Agent(
"google-gla:gemini-2.5-flash",
result_type=RoutingDecision,
system_prompt="You are an enterprise research supervisor. Route user requests to specialists or declare FINISH."
)
sql_worker = Agent("google-gla:gemini-2.5-flash", system_prompt="Execute SQL queries against enterprise warehouses.")
web_worker = Agent("google-gla:gemini-2.5-flash", system_prompt="Search live documentation and technical feeds.")
async def execute_task(user_query: str):
context = user_query
for iteration in range(5):
decision = await supervisor.run(f"Current state:
{context}
Decide next step.")
route = decision.data
if route.next_worker == "FINISH":
print("🏁 Workflow completed successfully!")
break
if route.next_worker == "sql_analyst":
worker_result = await sql_worker.run(route.instructions_for_worker)
context += f"
[SQL Worker Output]: {worker_result.data}"
elif route.next_worker == "web_researcher":
worker_result = await web_worker.run(route.instructions_for_worker)
context += f"
[Web Worker Output]: {worker_result.data}"
return context
3. Enterprise Benefits
- Isolated Prompts: Workers only carry specialized instructions, avoiding prompt pollution and token bloat.
- Modular Toolsets: Security credentials (e.g. database write access) remain isolated exclusively to the SQL worker.
- Resilient Error Recovery: If a worker fails, the supervisor observes the error and re-prompts or routes to a fallback agent.
FREE CODE TEMPLATE
Download the Complete PydanticAI Document Parser Blueprint
Get the complete, type-safe invoice and ID card parsing codebase in Python + a ready-to-run Docker environment. 100% free.
