Architecture

graph TB
    A["Task Queue<br/>(Redis)"]
    B["Orchestration Engine<br/>(Task Routing & Scheduling)"]
    C["Task Router<br/>(Capability Matching)"]
    D["Shared Memory Layer<br/>(Redis + In-Memory)"]
    E["Message Bus<br/>(Pub/Sub)"]
    F["Metrics Collector<br/>(Prometheus)"]
    
    G["Sales Agent"]
    H["Data Agent"]
    I["Analysis Agent"]
    J["Approval Agent"]
    K["Notification Agent"]
    L["Document Agent"]
    M["Pricing Agent"]
    N["Inventory Agent"]
    
    A -->|submit| B
    B --> C
    C -->|route| G & H & I & J & K & L & M & N
    G & H & I & J & K & L & M & N -->|read/write| D
    G & H & I & J & K & L & M & N -->|publish| E
    E -->|subscribe| G & H & I & J & K & L & M & N
    B & G & H & I & J & K & L & M & N -->|emit| F
    
    style B fill:#4A90E2,stroke:#2E5C8A,color:#fff
    style C fill:#50C878,stroke:#2A7A47,color:#fff
    style D fill:#FFB347,stroke:#CC8A2E,color:#fff
    style E fill:#9B59B6,stroke:#6C3A7F,color:#fff
    style F fill:#E74C3C,stroke:#A93226,color:#fff

Key Features

Performance Benchmarks

Quick Start

Installation

pip install multi-agent-orchestration-framework

Basic Usage

from orchestrator.engine import OrchestrationEngine
from agents.sales_agent import SalesAgent
import asyncio

async def main():
    # Initialize the orchestration engine
    engine = OrchestrationEngine(
        config={
            "max_workers": 8,
            "redis_url": "redis://localhost:6379",
            "log_level": "INFO"
        }
    )
    
    # Register agents
    sales_agent = SalesAgent(name="sales-001")
    await engine.register_agent(sales_agent)
    
    # Submit a task
    task_id = await engine.submit_task(
        task_type="sales_quote",
        payload={
            "customer_id": "ACME-001",
            "product_ids": ["PROD-123", "PROD-456"],
            "quantity": 100
        },
        priority=1
    )
    
    # Monitor task progress
    status = await engine.get_status(task_id)
    print(f"Task {task_id}: {status.state} - {status.progress}%")
    
    # Shutdown gracefully
    await engine.shutdown()

asyncio.run(main())

API Reference

OrchestrationEngine

class OrchestrationEngine:
    """Main orchestration engine for multi-agent coordination."""
    
    async def register_agent(agent: BaseAgent) -> None:
        """Register a new agent with the orchestration engine."""
    
    async def submit_task(
        task_type: str,
        payload: Dict[str, Any],
        priority: int = 0
    ) -> str:
        """Submit a task and return task ID."""
    
    async def get_status(task_id: str) -> TaskStatus:
        """Get current task status and progress."""
    
    async def shutdown() -> None:
        """Gracefully shutdown the engine and all agents."""

BaseAgent

class BaseAgent(ABC):
    """Abstract base class for all agents."""
    
    @abstractmethod
    async def process_task(task: AgentTask) -> TaskResult:
        """Process an assigned task."""
    
    @abstractmethod
    def get_capabilities() -> List[str]:
        """Return list of task types this agent can handle."""
    
    async def health_check() -> bool:
        """Verify agent is operational and responsive."""

Example: Manufacturing Workflow

# See examples/manufacturing_workflow.py for complete working example
# Demonstrates: Order processing → Quality check → Fulfillment → Notification

Contributing

We welcome contributions! Please see CONTRIBUTING.md for guidelines.

License

MIT License - see LICENSE for details.

Built by Sainath Pattipati

Powering intelligent automation at enterprise scale.