Skip to main content

Lesson 4: Multi-Agent Orchestration

Welcome to the pinnacle of AgentMap development! In this advanced lesson, you'll learn to orchestrate multiple specialized agents, create intelligent routing systems, and build sophisticated multi-agent workflows that can handle complex business processes.

Learning Objectives​

By the end of this lesson, you will:

  • ✅ Design multi-agent workflows with intelligent routing
  • ✅ Use OrchestratorAgent for dynamic agent selection
  • ✅ Implement keyword-based and AI-based routing strategies
  • ✅ Coordinate multiple custom agents effectively
  • ✅ Build fault-tolerant, scalable agent systems
  • ✅ Apply advanced workflow patterns and best practices

Overview: What We're Building​

We'll create an Intelligent Document Processing System that:

  1. Analyzes incoming documents to determine type and priority
  2. Routes documents to specialized processing agents
  3. Coordinates multiple AI specialists for different document types
  4. Consolidates results into comprehensive reports
  5. Handles errors and exceptions gracefully

Step 1: Download the Complete System​

Let's get all the files for our multi-agent orchestration system:

Main Workflow File​

Sample Document for Testing​

Custom Agent 1: Contract Analyzer​

Custom Agent 2: Results Consolidator​

Step 2: Understanding Multi-Agent Architecture​

Orchestration Patterns​

Our system demonstrates several advanced orchestration patterns:

1. Classification-Based Routing​

2. Parallel Processing​

Multiple agents can process the same document simultaneously for different aspects.

3. Results Consolidation​

A specialized agent combines outputs from multiple processors.

4. Error Handling​

Each agent has independent error handling with graceful degradation.

Agent Coordination Strategies​

  1. Sequential Processing: One agent passes results to the next
  2. Parallel Processing: Multiple agents work simultaneously
  3. Conditional Routing: Route based on content or metadata
  4. Result Merging: Combine outputs intelligently

Step 3: The OrchestratorAgent Deep Dive​

How OrchestratorAgent Works​

RouteToProcessor,Route to specialized processor based on document type,orchestrator,ContractProcessor,ErrorHandler,"document_content,classification_result",processor_selection,,"{""nodes"": ""ContractProcessor|InvoiceProcessor|ReportProcessor|EmailProcessor|GeneralProcessor""}"

Key Configuration:

  • nodes: Pipe-separated list of possible destinations
  • Routing Logic: Based on classification_result category
  • Fallback: Default to first node if no match

Routing Decision Logic​

The orchestrator uses this logic:

  1. Extract Category: From classification_result.category
  2. Map to Agent:
    • CONTRACT → ContractProcessor
    • INVOICE → InvoiceProcessor
    • REPORT → ReportProcessor
    • EMAIL → EmailProcessor
    • Default → GeneralProcessor

Advanced Routing Strategies​

Keyword-Based Routing​

{
  "routing_strategy": "keywords",
  "routes": {
    "ContractProcessor": ["contract", "agreement", "terms"],
    "InvoiceProcessor": ["invoice", "bill", "$", "payment"],
    "ReportProcessor": ["report", "analysis", "data", "metrics"]
  }
}

Confidence-Based Routing​

{
  "routing_strategy": "confidence",
  "confidence_threshold": 0.8,
  "fallback_node": "GeneralProcessor"
}

Step 4: Running the Complete System​

Setup Steps​

  1. Create Directory Structure:
mkdir -p agentmap-learning/data
cd agentmap-learning

  1. Download All Files:

    • lesson4.csv (main workflow)
    • sample_document.txt (test document)
    • contract_analyzer.py (specialized agent)
    • results_consolidator.py (consolidation agent)

  2. Set API Keys:

export ANTHROPIC_API_KEY="your-api-key"
  1. Run the Workflow:
agentmap run lesson4.csv

Expected Execution Flow​

📄 Loading document...
🧠 Classifying document type...
🎯 Routing to ContractProcessor...
⚖️ Analyzing contract terms...
📊 Consolidating results...
📋 Generating final report...
💾 Saving results...
✅ Processing complete!

Step 5: Examining Multi-Agent Results​

Check the Final Report​

cat data/processing_report.md

You'll see a comprehensive report with:

# 📋 DOCUMENT PROCESSING REPORT

## Executive Summary
The document has been classified as a SOFTWARE DEVELOPMENT SERVICES AGREEMENT 
with high confidence (0.95). Specialized contract analysis has been performed, 
revealing key terms, financial obligations, and risk factors.

## 📊 Processing Details
- **Document Type**: CONTRACT (95% confidence)
- **Processing Time**: ~45 seconds
- **Urgency Level**: MEDIUM

## 🔍 Detailed Analysis
### Contract Analysis
- **Contract Type**: Software Development Services Agreement
- **Total Value**: $125,000 USD
- **Timeline**: 16 weeks with 4 milestone payments
- **Risk Assessment**: Medium risk with standard terms
- **Compliance**: 85% standard clauses present

## 📈 Key Insights & Findings
- Well-structured milestone-based payment schedule
- Comprehensive IP and confidentiality clauses
- Standard termination and warranty provisions
- Clear scope of work and deliverables

## ⚠️ Risk Assessment
- **Low Risk**: Standard commercial terms
- **Considerations**: Ensure milestone acceptance criteria are clear
- **Recommendations**: Legal review recommended before signing

## 📋 Recommended Actions
1. Review milestone acceptance criteria with technical team
2. Confirm intellectual property ownership terms
3. Validate scope against internal requirements
4. Legal review before execution

## 🎯 Conclusion
This appears to be a well-structured software development agreement with 
standard commercial terms. Recommend proceeding with legal review.

Step 6: Advanced Multi-Agent Patterns​

Pattern 1: Hierarchical Processing​

Pattern 2: Consensus Building​

Multiple agents analyze the same content and vote on the result:

class ConsensusAgent(BaseAgent):
    def process(self, inputs):
        results = inputs.get('agent_results', [])
        
        # Collect votes from different agents
        votes = {}
        for result in results:
            category = result.get('category')
            confidence = result.get('confidence', 0.0)
            
            if category not in votes:
                votes[category] = []
            votes[category].append(confidence)
        
        # Calculate weighted consensus
        consensus = {}
        for category, confidences in votes.items():
            consensus[category] = sum(confidences) / len(confidences)
        
        # Return highest confidence category
        winner = max(consensus.items(), key=lambda x: x[1])
        return {
            'consensus_category': winner[0],
            'consensus_confidence': winner[1],
            'all_votes': votes
        }

Pattern 3: Pipeline Processing​

Sequential processing where each agent enriches the data:

Pipeline,Step1,Normalize data,data_normalizer,Step2,Error,raw_data,normalized_data,,
Pipeline,Step2,Extract features,feature_extractor,Step3,Error,normalized_data,features,,
Pipeline,Step3,Classify content,classifier,Step4,Error,features,classification,,
Pipeline,Step4,Generate insights,insight_generator,End,Error,classification,insights,,

Step 7: Building Your Own Multi-Agent System​

Exercise 1: Add New Specialist Agents​

Create additional specialist agents:

Invoice Analyzer Agent​

class InvoiceAnalyzerAgent(BaseAgent):
    def process(self, inputs):
        # Extract invoice-specific information
        invoice_analysis = {
            'invoice_number': self._extract_invoice_number(content),
            'amounts': self._extract_amounts(content),
            'tax_analysis': self._analyze_taxes(content),
            'vendor_info': self._extract_vendor_info(content),
            'payment_terms': self._extract_payment_terms(content)
        }
        return invoice_analysis

Email Analyzer Agent​

class EmailAnalyzerAgent(BaseAgent):
    def process(self, inputs):
        # Analyze email-specific features
        email_analysis = {
            'sentiment': self._analyze_sentiment(content),
            'action_items': self._extract_action_items(content),
            'urgency': self._assess_urgency(content),
            'recipients': self._extract_recipients(content),
            'thread_analysis': self._analyze_thread(content)
        }
        return email_analysis

Exercise 2: Implement Advanced Routing​

Create a smart router that uses multiple criteria:

class SmartRouterAgent(BaseAgent):
    def process(self, inputs):
        classification = inputs.get('classification_result', {})
        content = inputs.get('document_content', '')
        
        # Multi-criteria routing
        routing_score = {}
        
        # Factor 1: Classification confidence
        category = classification.get('category', '')
        confidence = classification.get('confidence', 0.0)
        
        # Factor 2: Content complexity
        complexity = self._assess_complexity(content)
        
        # Factor 3: Urgency level
        urgency = classification.get('urgency', 'MEDIUM')
        
        # Route based on combined score
        if category == 'CONTRACT' and confidence > 0.8:
            return {'route': 'ContractProcessor', 'priority': 'HIGH'}
        elif complexity > 0.7:
            return {'route': 'AdvancedProcessor', 'priority': 'HIGH'}
        else:
            return {'route': 'GeneralProcessor', 'priority': 'MEDIUM'}

Exercise 3: Add Quality Assurance​

Implement a QA agent that validates other agents' outputs:

class QualityAssuranceAgent(BaseAgent):
    def process(self, inputs):
        analyses = inputs.get('all_analyses', {})
        
        qa_report = {
            'consistency_check': self._check_consistency(analyses),
            'completeness_check': self._check_completeness(analyses),
            'accuracy_assessment': self._assess_accuracy(analyses),
            'quality_score': 0.0,
            'recommendations': []
        }
        
        # Calculate overall quality score
        qa_report['quality_score'] = self._calculate_quality_score(qa_report)
        
        return qa_report

Step 8: Performance and Scalability​

Optimizing Multi-Agent Workflows​

1. Parallel Processing​

Run independent agents simultaneously:

import asyncio

async def run_parallel_agents(document, agents):
    tasks = []
    for agent in agents:
        task = asyncio.create_task(agent.process_async(document))
        tasks.append(task)
    
    results = await asyncio.gather(*tasks)
    return results

2. Caching Results​

Cache expensive operations:

from functools import lru_cache

class CachingAgent(BaseAgent):
    @lru_cache(maxsize=128)
    def _cached_analysis(self, content_hash):
        # Expensive analysis here
        return analysis_result

3. Load Balancing​

Distribute work across multiple instances:

class LoadBalancedRouter(BaseAgent):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.agent_pool = self._initialize_agent_pool()
        self.current_index = 0
    
    def process(self, inputs):
        # Round-robin agent selection
        selected_agent = self.agent_pool[self.current_index]
        self.current_index = (self.current_index + 1) % len(self.agent_pool)
        
        return selected_agent.process(inputs)

Key Concepts Mastered​

1. Multi-Agent Architecture​

  • Orchestration patterns for complex workflows
  • Agent specialization and coordination
  • Intelligent routing and decision making
  • Results consolidation and synthesis

2. Advanced Orchestration​

  • OrchestratorAgent configuration and usage
  • Multiple routing strategies (classification, keywords, confidence)
  • Error handling in distributed systems
  • Workflow fault tolerance

3. Agent Coordination​

  • Sequential vs. parallel processing patterns
  • Data flow between specialized agents
  • Conflict resolution and consensus building
  • Quality assurance and validation

4. Scalability Patterns​

  • Performance optimization techniques
  • Caching and load balancing strategies
  • Async processing capabilities
  • Resource management

Best Practices for Multi-Agent Systems​

1. Design Principles​

  • Single responsibility for each agent
  • Clear interfaces between agents
  • Fault tolerance at every level
  • Comprehensive logging and monitoring

2. Error Handling​

  • Graceful degradation when agents fail
  • Circuit breaker patterns for unstable agents
  • Retry logic with exponential backoff
  • Comprehensive error reporting

3. Performance​

  • Profile and optimize bottleneck agents
  • Use async processing where appropriate
  • Implement intelligent caching strategies
  • Monitor resource usage

4. Maintainability​

  • Modular agent design for easy updates
  • Comprehensive testing strategies
  • Clear documentation and examples
  • Version management for agent updates

Troubleshooting Multi-Agent Systems​

Common Issues and Solutions​

Issue: Agent routing failures​

Solution:

  • Check orchestrator node names match actual agents
  • Verify classification output format
  • Add fallback routing logic

Issue: Results consolidation errors​

Solution:

  • Ensure all agents return structured data
  • Add input validation in consolidator
  • Handle missing or malformed results

Issue: Performance bottlenecks​

Solution:

  • Profile individual agent performance
  • Implement parallel processing where possible
  • Add caching for expensive operations

Issue: Inconsistent results​

Solution:

  • Add quality assurance agents
  • Implement consensus mechanisms
  • Validate agent outputs

Congratulations!​

You've completed the AgentMap Learning Guide! You now have mastery over:

  • ✅ Basic Workflow Creation - Building fundamental agent chains
  • ✅ Advanced Prompt Engineering - Creating intelligent, adaptive AI
  • ✅ Custom Agent Development - Building specialized business logic
  • ✅ Multi-Agent Orchestration - Coordinating complex agent systems

What You Can Build Now​

With these skills, you can create:

  • Intelligent Document Processing Systems
  • Multi-Stage Data Pipelines
  • AI-Powered Business Process Automation
  • Complex Decision Support Systems
  • Adaptive Workflow Orchestration

Next Steps in Your Journey​

Advanced Topics to Explore​

Advanced Patterns and Examples​

Additional Resources​

🎉 Congratulations on completing the AgentMap Learning Guide! You've journeyed from basic agent usage to sophisticated multi-agent orchestration. The patterns and techniques you've learned will serve as the foundation for building powerful, intelligent automation systems. Welcome to the AgentMap community of builders!

Join the Community:

  • Share your projects and get feedback
  • Contribute to the open-source project
  • Help other learners on their journey
  • Stay updated with the latest features and patterns

The future of intelligent automation is in your hands - now go build something amazing! 🚀