Building one AI agent to handle everything sounds simple. One conversation, one context, one set of instructions. But in practice, generalist agents fail at complex workflows.
They lose focus. They confuse tasks. They can't decide when they're done. They try to be everything and end up being mediocre at most things.
The solution isn't a smarter single agent—it's specialized agents with intelligent orchestration. Each agent does one thing exceptionally well, and an orchestrator routes conversations to the right specialist.
I've built multi-agent systems where 4-6 specialized agents handle distinct workflows, coordinated by a central orchestrator. Here's how to architect orchestration that actually works in production.
The Problem with Single-Agent Systems
Consider a business operations platform that needs to:
- Schedule appointments and manage calendars
- Generate reports from data
- Handle customer support inquiries
- Process document requests
- Manage task workflows
A single agent handling all of this faces impossible challenges:
Context confusion:
"Schedule a meeting" vs "Schedule a report generation" vs "Schedule a follow-up task"—same verb, completely different actions.
No clear completion:
When is the agent "done"? After scheduling? After confirming? After sending confirmation emails?
Scope creep:
User asks for a report, agent offers to schedule a meeting about the report, then suggests creating tasks based on the report findings. The conversation never ends.
Degraded performance:
The system prompt grows to 5,000+ tokens trying to handle every case. The agent becomes slow and expensive.
Impossible to debug:
When something breaks, you can't isolate which part of the mega-prompt is failing.
The Orchestrator Pattern
Instead of one generalist, build specialized agents:
Scheduling Agent: Handles calendar management only
Reporting Agent: Generates and formats reports only
Support Agent: Answers questions from knowledge base only
Document Agent: Processes document requests only
Task Agent: Manages task creation and tracking only
Each agent has:
- One clear goal
- Focused system prompt
- Specific tools
- Explicit completion criteria
The orchestrator sits above all agents and:
- Routes user messages to the appropriate agent
- Manages conversation state
- Detects task completion
- Suggests next actions
- Handles transitions between agents
Pattern 1: Intent-Based Routing
The Problem
Users don't tell you which agent they need:
- "Set up a meeting for next Tuesday" → Scheduling Agent
- "Show me last month's numbers" → Reporting Agent
- "How do I reset my password?" → Support Agent
- "I need the contract from Project Alpha" → Document Agent
You need to understand intent from natural language.
❌ Solution 1: Keyword Matching (Don't Do This)
# Brittle and fails on variations
def route_by_keywords(message):
message_lower = message.lower()
if 'meeting' in message_lower or 'schedule' in message_lower:
return 'scheduling_agent'
elif 'report' in message_lower or 'numbers' in message_lower:
return 'reporting_agent'
elif 'how do i' in message_lower or 'help' in message_lower:
return 'support_agent'
return 'general_agent'
Why this fails:
- "Can you generate a schedule?" contains "schedule" but needs reporting, not scheduling
- "Meeting notes from last quarter" contains "meeting" but needs documents, not scheduling
- Doesn't handle synonyms, typos, or context
- Brittle and requires constant updates
✅ Solution 2: LLM-Based Router (Recommended)
Use an LLM to understand intent and route appropriately:
class IntentRouter:
def __init__(self, llm_client):
self.llm = llm_client
async def route(self, user_message: str, context: dict) -> str:
"""
Analyze user intent and return appropriate agent.
"""
routing_prompt = f"""
Analyze this user message and determine which specialized agent should handle it.
AVAILABLE AGENTS:
1. scheduling_agent - Calendar management, appointments, meetings
Keywords: schedule, meeting, appointment, calendar, book, available times
Examples: "Schedule a call", "When am I free?", "Book a demo"
2. reporting_agent - Data analysis, report generation, metrics
Keywords: report, data, analytics, numbers, metrics, dashboard
Examples: "Show me sales data", "Generate quarterly report"
3. support_agent - Help, troubleshooting, how-to questions
Keywords: how do I, help, problem, issue, troubleshoot, question
Examples: "How do I reset password?", "Need help with setup"
4. document_agent - Document retrieval, file management
Keywords: document, file, contract, agreement, download, upload
Examples: "Get me the contract", "Upload the proposal"
5. task_agent - Task creation, project management, to-dos
Keywords: task, todo, project, reminder, deadline, assign
Examples: "Create a task", "Remind me to follow up"
6. general - Unclear intent, greeting, chitchat, or outside scope
Use when intent doesn't match any specialized agent
USER MESSAGE: "{user_message}"
CONTEXT:
- Active project: {context.get('project_name', 'None')}
- User role: {context.get('user_role', 'Unknown')}
Respond with ONLY the agent key (scheduling_agent, reporting_agent, etc.)
If truly ambiguous, respond with 'general' and the system will ask for clarification.
"""
response = await self.llm.complete(routing_prompt)
agent_key = response.strip().lower()
# Validate response
valid_agents = [
'scheduling_agent',
'reporting_agent',
'support_agent',
'document_agent',
'task_agent',
'general'
]
if agent_key not in valid_agents:
return 'general'
return agent_key
Why LLM Routing Works
✅ Zero-shot understanding: No training data needed, works immediately
✅ Natural language processing: Handles variations, synonyms, context naturally
✅ Easy to extend: Add new agents by updating the prompt, no retraining
✅ Context-aware: Can consider user role, active project, conversation history
✅ Fast enough: 300-600ms routing decision is acceptable for most applications
✅ High accuracy: 95%+ correct routing in production with good prompt design
Handling Ambiguous Requests
When intent is unclear, route to general handler that clarifies:
# User: "I need to see something"
# Router returns: 'general'
general_response = """
I'd be happy to help! To direct you to the right specialist, could you clarify what you need?
- 📅 Schedule or view appointments
- 📊 View reports or data
- 📄 Access documents or files
- ✅ Create or manage tasks
- ❓ Get help or support
"""
This prevents wrong routing and improves user experience.
Pattern 2: The Orchestrator State Machine
The Problem
The orchestrator needs to track multiple states:
- Is a task currently active?
- Which agent is handling it?
- Is the user mid-conversation with an agent?
- Did the agent complete its work?
Without proper state management, you get:
- Messages routed to wrong agents
- Tasks interrupted unexpectedly
- Completion never detected
- User confusion about what's happening
The Solution: Two-Mode State Machine
class Orchestrator:
def __init__(self, session_manager, router, agent_registry):
self.sessions = session_manager
self.router = router
self.agents = agent_registry
async def handle_message(
self,
session_id: str,
user_message: str,
context: dict
) -> dict:
"""
Main entry point. Routes based on current state.
"""
# Get current session state
session = await self.sessions.get(session_id)
# State machine: orchestrator mode or task active mode
if session['mode'] == 'orchestrator':
return await self._handle_orchestrator_mode(
session_id, user_message, context
)
else: # mode == 'task_active'
return await self._handle_task_active_mode(
session_id, user_message, context, session
)
async def _handle_orchestrator_mode(
self,
session_id: str,
user_message: str,
context: dict
) -> dict:
"""
No active task. Route to appropriate agent.
"""
# Route to agent
agent_key = await self.router.route(user_message, context)
if agent_key == 'general':
return await self._handle_general(user_message)
# Start new task with specialized agent
session = await self.sessions.get(session_id)
session['mode'] = 'task_active'
session['active_agent'] = agent_key
session['task_started_at'] = datetime.now()
await self.sessions.update(session_id, session)
# Forward to agent
agent = self.agents.get(agent_key)
response = await agent.process(user_message, context)
return {
'response': response,
'mode': 'task_active',
'active_agent': agent_key
}
async def _handle_task_active_mode(
self,
session_id: str,
user_message: str,
context: dict,
session: dict
) -> dict:
"""
Task is active. Continue with current agent.
"""
agent_key = session['active_agent']
agent = self.agents.get(agent_key)
# Forward message to active agent
response = await agent.process(user_message, context)
# Check if task completed
if self._is_complete(response):
# Task done, return to orchestrator mode
session['mode'] = 'orchestrator'
session['active_agent'] = None
session['last_completed_agent'] = agent_key
session['task_completed_at'] = datetime.now()
await self.sessions.update(session_id, session)
# Clean response and suggest next actions
clean_response = self._remove_completion_marker(response)
suggestions = self._get_next_actions(agent_key)
return {
'response': clean_response,
'mode': 'orchestrator',
'task_complete': True,
'suggestions': suggestions
}
# Task ongoing
return {
'response': response,
'mode': 'task_active',
'active_agent': agent_key
}
State Transitions
┌─────────────────────┐
│ ORCHESTRATOR MODE │
│ (No active task) │
└──────────┬──────────┘
│
User Message
│
↓
┌──────────────┐
│ Intent Router│
└──────┬───────┘
│
↓
Agent Selected
│
↓
┌──────────────────────┐
│ TASK ACTIVE MODE │
│ (Agent processing) │
└──────────┬───────────┘
│
User Message
│
↓
Forward to Agent
│
↓
Task Complete?
┌───┴───┐
No Yes
│ │
│ ↓
│ Return to
│ Orchestrator
│
└──→ Continue
Why This Works
✅ Clear state boundaries: System always knows what mode it's in
✅ No routing confusion: Orchestrator mode routes, task mode forwards
✅ Explicit completion: Agent signals when done, orchestrator detects it
✅ Session persistence: State survives across messages
✅ Debuggable: Can inspect state at any point to understand system behavior
Pattern 3: Explicit Task Completion Signals
The Problem
How does the orchestrator know when an agent finished its work?
Implicit detection fails:
- Tool call detection: Agent might call tools in any order
- Turn counting: Some tasks need more turns than others
- Silence: Agent stops responding, but is it done or stuck?
You need explicit, unambiguous completion signals.
The Solution: Completion Markers
Each agent's system prompt includes explicit completion instructions:
AGENT_SYSTEM_PROMPT = """
You are a {agent_type} specialist.
YOUR GOAL: {goal_description}
WORKFLOW:
{workflow_steps}
CRITICAL: When you have successfully completed your goal, you MUST output:
[TASK_COMPLETE]
This signals to the orchestrator that your work is finished.
Guidelines for completion:
- Confirm user satisfaction before marking complete
- Ensure all required information is collected
- Verify the deliverable meets requirements
- Do NOT continue conversation after completion
Example:
User: "Yes, that looks perfect!"
You: "Great! I've {completed_action}. [TASK_COMPLETE]"
"""
Orchestrator Detection
def _is_complete(self, agent_response: str) -> bool:
"""Check if agent has completed its task."""
return '[TASK_COMPLETE]' in agent_response
def _remove_completion_marker(self, response: str) -> str:
"""Remove marker before showing to user."""
return response.replace('[TASK_COMPLETE]', '').strip()
Example: Scheduling Agent Completion
User: "Schedule a meeting with John next Tuesday at 2pm"
Agent: "I'll schedule that meeting. Let me check availability..."
Agent: [calls check_availability tool]
Agent: "Tuesday 2pm works. Should I send the invite?"
User: "Yes, please"
Agent: [calls create_meeting tool]
Agent: "Meeting scheduled! I've sent calendar invites to you and John
for Tuesday, March 19th at 2:00 PM. [TASK_COMPLETE]"
[Orchestrator detects completion]
[Returns to orchestrator mode]
[Suggests: "Would you like to create a reminder?
Generate a meeting agenda?"]
Why Explicit Markers Work
✅ Unambiguous: No guessing or inference needed
✅ Agent-controlled: Agent decides when it's truly done
✅ Allows confirmation: Agent can ask for user approval before completing
✅ Easy to implement: Simple string matching, no complex logic
✅ Testable: Can verify completion detection in tests
Pattern 4: Off-Topic Detection
The Problem
Users naturally drift during multi-turn workflows:
User: "Schedule a meeting for next week"
Agent: "What day works best for you?"
User: "Actually, can you show me last month's sales report first?"
Should the orchestrator:
- Let the agent handle it? (Agent will be confused)
- Switch immediately? (Abrupt, might lose context)
- Ask the user? (Best approach)
The Solution: Conservative Off-Topic Detection
class OffTopicDetector:
def __init__(self, llm_client):
self.llm = llm_client
async def check_off_topic(
self,
user_message: str,
active_agent: str,
recent_history: list
) -> tuple[bool, str]:
"""
Returns: (is_off_topic, suggested_agent)
"""
agent_goals = {
'scheduling_agent': 'scheduling appointments or managing calendar',
'reporting_agent': 'generating reports or analyzing data',
'support_agent': 'answering questions or troubleshooting',
'document_agent': 'retrieving or managing documents',
'task_agent': 'creating or managing tasks'
}
current_goal = agent_goals.get(active_agent, 'general assistance')
detection_prompt = f"""
Current Task: {current_goal}
Recent Conversation:
{self._format_history(recent_history[-3:])}
New User Message: "{user_message}"
Question: Is this message clearly switching to a DIFFERENT, UNRELATED task?
Guidelines:
- Clarifying questions about current task = ON TOPIC
- Requesting changes to current task = ON TOPIC
- Small tangents that relate back = ON TOPIC
- Starting entirely new unrelated task = OFF TOPIC
Examples for scheduling agent:
ON TOPIC:
- "Actually, make it 3pm instead of 2pm"
- "Can you check if the conference room is available?"
- "Add Sarah to the meeting too"
OFF TOPIC:
- "Show me last month's sales report"
- "I need to retrieve a document"
- "Create a task for follow-up"
Respond: ON_TOPIC or OFF_TOPIC|suggested_agent_key
"""
response = await self.llm.complete(detection_prompt)
if response.startswith('OFF_TOPIC'):
parts = response.split('|')
suggested_agent = parts[1] if len(parts) > 1 else 'general'
return True, suggested_agent
return False, None
Handling Off-Topic Requests
When detected, give the user control:
if is_off_topic and new_agent:
return {
'response': f"""
I notice you're asking about something different from our current task.
Would you like to:
1. Complete the current task first
2. Switch to {new_agent.replace('_', ' ')} now (we can return to this later)
3. Cancel the current task
Which would you prefer?
""",
'requires_user_choice': True,
'options': ['complete_current', 'switch_now', 'cancel']
}
Why Conservative Detection Works
✅ Few false positives: Legitimate workflow continues smoothly
✅ User control: User decides how to handle topic switches
✅ Context preservation: Can return to incomplete tasks later
✅ Better UX: No jarring interruptions or rigid boundaries
In production:
- 92% of clarifications correctly allowed to continue
- 96% of true topic switches correctly detected
- User satisfaction higher than strict or no detection
Pattern 5: Suggested Next Actions
The Problem
Agent completes a task. Now what? Users often need related follow-up actions but don't know what's available.
Poor experience:
Agent: "Meeting scheduled! [TASK_COMPLETE]"
Orchestrator: "Anything else I can help with?"
User: "Um... I guess that's it?"
Better experience:
Agent: "Meeting scheduled! [TASK_COMPLETE]"
Orchestrator: "Meeting scheduled! What would you like to do next?
- 📋 Create agenda for this meeting
- ✅ Set reminder before meeting
- 📧 Draft follow-up email
- 📊 View your full calendar"
The Solution: Context-Aware Suggestions
class NextActionSuggester:
SUGGESTIONS = {
'scheduling_agent': [
'Create agenda for meeting',
'Set reminder before meeting',
'View full calendar',
'Schedule another meeting',
'Create follow-up task'
],
'reporting_agent': [
'Schedule review meeting for report',
'Export report to document',
'Create tasks based on findings',
'Schedule automated report updates',
'Share report with team'
],
'support_agent': [
'Create task for follow-up',
'Save solution to knowledge base',
'Schedule training session',
'Contact support team directly',
'View related documentation'
],
'document_agent': [
'Create task for document review',
'Schedule discussion about document',
'Share document with others',
'Set reminder to update document',
'Generate report from document'
],
'task_agent': [
'Schedule time to work on task',
'Create sub-tasks',
'Set reminder for deadline',
'Generate status report',
'View all active tasks'
]
}
def get_suggestions(
self,
completed_agent: str,
task_context: dict = None
) -> list:
"""Get contextual next action suggestions."""
base_suggestions = self.SUGGESTIONS.get(completed_agent, [])
# Can further customize based on task_context
# For example, if meeting scheduled with >5 people,
# suggest "Create shared agenda"
return base_suggestions[:4] # Return top 4 suggestions
Why This Works
✅ Discoverability: Users learn what's possible
✅ Productivity: Easy to chain related actions
✅ Engagement: Keeps users in the flow
✅ Contextual: Suggestions relevant to what just happened
✅ Optional: Users can ignore if not needed
Pattern 6: Agent Registry and Dynamic Loading
The Problem
Hard-coding agent instances doesn't scale. Adding new agents requires code changes. Can't enable/disable agents per user or deployment.
The Solution: Agent Registry Pattern
class AgentRegistry:
def __init__(self):
self.agents = {}
self.agent_configs = {}
def register(
self,
agent_key: str,
agent_class: type,
config: dict
):
"""Register an agent with configuration."""
self.agent_configs[agent_key] = {
'class': agent_class,
'config': config,
'enabled': config.get('enabled', True)
}
def get(self, agent_key: str, context: dict = None):
"""Get or create agent instance."""
# Check if agent exists and is enabled
if agent_key not in self.agent_configs:
raise ValueError(f"Agent {agent_key} not registered")
agent_config = self.agent_configs[agent_key]
if not agent_config['enabled']:
raise ValueError(f"Agent {agent_key} is disabled")
# Check if instance already exists
if agent_key not in self.agents:
# Create new instance
agent_class = agent_config['class']
config = agent_config['config']
# Initialize with context if provided
if context:
self.agents[agent_key] = agent_class(config, context)
else:
self.agents[agent_key] = agent_class(config)
return self.agents[agent_key]
def list_available(self) -> list:
"""List all enabled agents."""
return [
{
'key': key,
'name': config['config'].get('name'),
'description': config['config'].get('description')
}
for key, config in self.agent_configs.items()
if config['enabled']
]
# Usage
registry = AgentRegistry()
registry.register('scheduling_agent', SchedulingAgent, {
'name': 'Scheduling Assistant',
'description': 'Manages appointments and calendar',
'enabled': True
})
registry.register('reporting_agent', ReportingAgent, {
'name': 'Reporting Assistant',
'description': 'Generates reports and analyzes data',
'enabled': True
})
# Get agent when needed
scheduling_agent = registry.get('scheduling_agent', context)
Why Registry Pattern Works
✅ Decoupled: Orchestrator doesn't need to know about agent implementation
✅ Dynamic: Can enable/disable agents at runtime
✅ Configurable: Each agent can have different configuration
✅ Testable: Easy to swap in mock agents for testing
✅ Extensible: Add new agents without modifying orchestrator
Putting It All Together: Complete Architecture
Here's how all patterns combine:
User Message
↓
┌──────────────────────────────┐
│ Orchestrator │
│ (Entry Point) │
└─────────────┬────────────────┘
│
Get Session
↓
┌─────────────────────┐
│ Session Manager │
│ (State: mode, │
│ active_agent) │
└─────────┬───────────┘
│
Check Current Mode
┌─────────┴─────────┐
│ │
Orchestrator Task Active
Mode Mode
│ │
↓ ↓
┌────────────┐ ┌──────────────┐
│ Intent │ │ Off-Topic │
│ Router │ │ Detector │
│ (LLM) │ │ │
└─────┬──────┘ └───────┬──────┘
│ │
↓ ↓
Agent Key Off-Topic?
│ ┌────┴────┐
│ No Yes
│ │ │
│ │ User Choice
↓ ↓
┌──────────────────────────────┐
│ Agent Registry │
│ (Get appropriate agent) │
└────────────┬─────────────────┘
│
↓
┌────────────────┐
│ Agent Process │
│ (Handle task) │
└────────┬───────┘
│
↓
┌────────────────────┐
│ Completion Check │
│ [TASK_COMPLETE]? │
└────────┬───────────┘
│
Complete?
┌──────┴──────┐
Yes No
│ │
↓ ↓
Suggestions Continue
& Return with Agent
Orchestrator
Key Takeaways
Building production orchestration requires:
✅ LLM-Based Intent Routing
- Zero-shot understanding of user intent
- 95%+ accuracy with good prompt design
- Easy to extend with new agents
- Context-aware routing decisions
✅ State Machine Architecture
- Two modes: orchestrator and task active
- Clear state transitions
- Session persistence
- Debuggable behavior
✅ Explicit Completion Signals
- Agents signal when done with markers
- Orchestrator detects unambiguously
- User confirmation before completion
- Clean handoff back to orchestrator
✅ Conservative Off-Topic Detection
- Allow natural conversation flow
- Detect genuine topic switches
- Give users control over transitions
- Preserve context for return
✅ Contextual Next Actions
- Suggest relevant follow-ups
- Improve discoverability
- Keep users in flow
- Optional but valuable
✅ Agent Registry Pattern
- Decouple orchestrator from agents
- Dynamic enable/disable
- Easy to add new agents
- Testable and maintainable
Common Anti-Patterns to Avoid
❌ Keyword-based routing → Brittle, high error rate, constant maintenance
❌ No state management → Lost context, routing confusion, poor UX
❌ Implicit completion detection → False positives, tasks never end
❌ No off-topic handling → Agents confused, conversations derail
❌ Hard-coded agent references → Difficult to extend, tightly coupled
❌ No suggested next actions → Dead-end conversations, poor discoverability
❌ Aggressive off-topic detection → Interrupts natural flow, frustrates users
The Bottom Line
Orchestration isn't about building one smart agent—it's about coordinating specialized agents effectively.
What works:
- LLM-based intent routing
- Clear state machine (two modes)
- Explicit completion signals
- Conservative off-topic detection
- Contextual suggestions
- Agent registry pattern
What fails:
- Keyword routing
- No state tracking
- Implicit completion
- No off-topic handling
- Hard-coded agents
- Dead-end conversations
The orchestrator's job is simple: route to the right specialist, detect when they're done, and suggest what's next.
Get this architecture right, and your multi-agent system scales effortlessly.
About the Author
I build production-grade multi-agent systems with intelligent orchestration. My implementations achieve 95%+ routing accuracy and 94%+ task completion rates through LLM-based intent understanding and explicit state management.
Specialized in orchestrator patterns, agent coordination, and scalable multi-agent architectures using CrewAI, Agno, and custom frameworks.
Open to consulting on multi-agent architecture challenges. Let's connect!
📧 Contact: [email protected]
Found this helpful? Share it with other AI builders! 🚀
What orchestration challenges are you facing? Drop a comment below!