⚙️ Session 6: Advanced Orchestration¶

from typing import Dict, List, Any, Optional
import asyncio

class DynamicAtomicOrchestrator:
    """Advanced orchestrator with dynamic component assembly"""

    def __init__(self):
        self.component_registry = {}
        self.workflow_templates = {}
        self.active_workflows = {}

    def register_component_factory(self, component_type: str, factory_func):
        """Register component factory for dynamic instantiation"""
        self.component_registry[component_type] = factory_func

    def create_component_on_demand(self, component_type: str, config: Dict):
        """Create component instance based on runtime requirements"""
        if component_type in self.component_registry:
            factory = self.component_registry[component_type]
            return factory(config)
        else:
            raise ValueError(f"Unknown component type: {component_type}")

    def analyze_data_requirements(self, data: Any) -> List[str]:
        """Analyze data to determine required processing components"""
        requirements = []

        if isinstance(data, str):
            if len(data) > 1000:
                requirements.append("chunking_processor")
            if "json" in data.lower() or "{" in data:
                requirements.append("json_processor")
            if any(char.isdigit() for char in data):
                requirements.append("numeric_processor")

        # Always include validation
        requirements.append("data_validator")

        return requirements

class IntelligentWorkflowRouter:
    """Route data through optimal processing paths"""

    def __init__(self, orchestrator: DynamicAtomicOrchestrator):
        self.orchestrator = orchestrator
        self.routing_rules = {}
        self.performance_metrics = {}

    def add_routing_rule(self, condition: str, workflow_path: List[str]):
        """Add routing rule for specific data conditions"""
        self.routing_rules[condition] = workflow_path

    def determine_optimal_path(self, data: Any) -> List[str]:
        """Determine optimal processing path for given data"""

        # Check explicit routing rules first
        for condition, workflow_path in self.routing_rules.items():
            if self._evaluate_condition(condition, data):
                return workflow_path

        # Fall back to analysis-based routing
        requirements = self.orchestrator.analyze_data_requirements(data)

        # Optimize based on performance metrics
        if "json_processor" in requirements and "chunking_processor" in requirements:
            # JSON processing before chunking is more efficient
            optimized_order = ["json_processor", "chunking_processor", "data_validator"]
            return optimized_order

        return requirements

    def _evaluate_condition(self, condition: str, data: Any) -> bool:
        """Evaluate routing condition against data"""
        if condition == "large_data":
            return len(str(data)) > 5000
        elif condition == "structured_data":
            return isinstance(data, dict) or ("{" in str(data) and "}" in str(data))
        elif condition == "streaming_data":
            return hasattr(data, 'stream') or 'stream' in str(data).lower()

        return False

class FaultTolerantCoordinator:
    """Coordinate processing with fault tolerance and recovery"""

    def __init__(self, max_retries: int = 3):
        self.max_retries = max_retries
        self.failure_log = []
        self.circuit_breakers = {}

    async def execute_with_fault_tolerance(
        self,
        component: Any,
        operation: str,
        data: Any,
        component_id: str
    ) -> Dict:
        """Execute operation with automatic retry and circuit breaking"""

        # Check circuit breaker status
        if self._is_circuit_open(component_id):
            return {
                "status": "circuit_open",
                "error": f"Circuit breaker open for {component_id}",
                "fallback_result": self._execute_fallback(operation, data)
            }

        for attempt in range(self.max_retries):
            try:
                # Execute the operation
                if hasattr(component, operation):
                    method = getattr(component, operation)
                    result = await self._safe_execute(method, data)

                    # Reset circuit breaker on success
                    self._reset_circuit_breaker(component_id)

                    return {
                        "status": "success",
                        "result": result,
                        "attempts": attempt + 1
                    }
                else:
                    # Fallback to generic processing
                    result = component.run(f"Process data: {str(data)[:200]}")
                    return {
                        "status": "success",
                        "result": result,
                        "attempts": attempt + 1,
                        "method": "fallback"
                    }

            except Exception as e:
                self.failure_log.append({
                    "component_id": component_id,
                    "operation": operation,
                    "attempt": attempt + 1,
                    "error": str(e),
                    "timestamp": "current_time"
                })

                # Open circuit breaker if too many failures
                if attempt == self.max_retries - 1:
                    self._open_circuit_breaker(component_id)

                # Wait before retry (exponential backoff)
                await asyncio.sleep(2 ** attempt)

        # All retries failed
        return {
            "status": "failed",
            "attempts": self.max_retries,
            "fallback_result": self._execute_fallback(operation, data)
        }

    def _is_circuit_open(self, component_id: str) -> bool:
        """Check if circuit breaker is open for component"""
        breaker = self.circuit_breakers.get(component_id, {})
        return breaker.get("status", "closed") == "open"

    def _execute_fallback(self, operation: str, data: Any) -> str:
        """Execute fallback processing when component fails"""
        return f"Fallback result for {operation} on data: {str(data)[:100]}"

class AdvancedParallelCoordinator:
    """Coordinate complex parallel processing workflows"""

    def __init__(self):
        self.dependency_graph = {}
        self.completion_callbacks = {}
        self.processing_state = {}

    def add_dependency(self, task_id: str, depends_on: List[str]):
        """Add dependency relationship between tasks"""
        self.dependency_graph[task_id] = depends_on

    async def execute_dependent_workflow(
        self,
        tasks: Dict[str, Dict],
        data: Any
    ) -> Dict:
        """Execute workflow with task dependencies"""

        # Initialize processing state
        for task_id in tasks:
            self.processing_state[task_id] = {
                "status": "pending",
                "result": None,
                "start_time": None,
                "end_time": None
            }

        # Find tasks with no dependencies to start with
        ready_tasks = [
            task_id for task_id in tasks
            if not self.dependency_graph.get(task_id, [])
        ]

        completed_tasks = set()

        while len(completed_tasks) < len(tasks):
            # Execute ready tasks in parallel
            if ready_tasks:
                results = await asyncio.gather(*[
                    self._execute_single_task(task_id, tasks[task_id], data)
                    for task_id in ready_tasks
                ])

                # Update completion status
                for i, task_id in enumerate(ready_tasks):
                    self.processing_state[task_id] = results[i]
                    completed_tasks.add(task_id)

            # Find newly ready tasks
            ready_tasks = []
            for task_id in tasks:
                if task_id not in completed_tasks:
                    dependencies = self.dependency_graph.get(task_id, [])
                    if all(dep in completed_tasks for dep in dependencies):
                        ready_tasks.append(task_id)

            # Prevent infinite loops
            if not ready_tasks and len(completed_tasks) < len(tasks):
                # Handle circular dependencies or other issues
                break

        return {
            "workflow_status": "completed",
            "completed_tasks": len(completed_tasks),
            "total_tasks": len(tasks),
            "processing_state": self.processing_state.copy()
        }

    async def _execute_single_task(
        self,
        task_id: str,
        task_config: Dict,
        data: Any
    ) -> Dict:
        """Execute a single task in the workflow"""

        start_time = "current_timestamp"

        try:
            # Get component for this task
            component_type = task_config.get("component_type")
            component = task_config.get("component")
            operation = task_config.get("operation", "run")

            # Execute the task
            if hasattr(component, operation):
                method = getattr(component, operation)
                result = method(data)
            else:
                result = component.run(f"Execute task {task_id}: {str(data)[:200]}")

            return {
                "status": "completed",
                "result": result,
                "start_time": start_time,
                "end_time": "current_timestamp",
                "task_id": task_id
            }

        except Exception as e:
            return {
                "status": "failed",
                "error": str(e),
                "start_time": start_time,
                "end_time": "current_timestamp",
                "task_id": task_id
            }

class MasterOrchestrationSystem:
    """Complete advanced orchestration system"""

    def __init__(self):
        self.dynamic_orchestrator = DynamicAtomicOrchestrator()
        self.workflow_router = IntelligentWorkflowRouter(self.dynamic_orchestrator)
        self.fault_coordinator = FaultTolerantCoordinator()
        self.parallel_coordinator = AdvancedParallelCoordinator()

        self._setup_component_factories()
        self._setup_routing_rules()

    def _setup_component_factories(self):
        """Register component factories for dynamic creation"""

        def create_json_processor(config):
            return BaseAgent(
                agent_name="json_processor",
                system_prompt="Specialized JSON data processing",
                max_tokens=config.get("max_tokens", 400)
            )

        def create_chunking_processor(config):
            return BaseAgent(
                agent_name="chunking_processor",
                system_prompt="Specialized data chunking and segmentation",
                max_tokens=config.get("max_tokens", 300)
            )

        self.dynamic_orchestrator.register_component_factory("json_processor", create_json_processor)
        self.dynamic_orchestrator.register_component_factory("chunking_processor", create_chunking_processor)

    def _setup_routing_rules(self):
        """Setup intelligent routing rules"""
        self.workflow_router.add_routing_rule(
            "large_data",
            ["chunking_processor", "json_processor", "data_validator"]
        )
        self.workflow_router.add_routing_rule(
            "structured_data",
            ["json_processor", "data_validator"]
        )

    async def process_with_advanced_orchestration(
        self,
        data: Any,
        processing_requirements: Optional[Dict] = None
    ) -> Dict:
        """Execute complete advanced orchestration workflow"""

        orchestration_id = f"orch_{hash(str(data))}"

        # Step 1: Determine optimal processing path
        processing_path = self.workflow_router.determine_optimal_path(data)

        # Step 2: Create required components dynamically
        components = {}
        for component_type in processing_path:
            config = processing_requirements.get(component_type, {}) if processing_requirements else {}
            components[component_type] = self.dynamic_orchestrator.create_component_on_demand(
                component_type, config
            )

        # Step 3: Setup task dependencies
        tasks = {}
        for i, component_type in enumerate(processing_path):
            tasks[f"task_{i}_{component_type}"] = {
                "component_type": component_type,
                "component": components[component_type],
                "operation": "run"
            }

            # Add dependency on previous task
            if i > 0:
                self.parallel_coordinator.add_dependency(
                    f"task_{i}_{component_type}",
                    [f"task_{i-1}_{processing_path[i-1]}"]
                )

        # Step 4: Execute workflow with fault tolerance
        workflow_result = await self.parallel_coordinator.execute_dependent_workflow(
            tasks, data
        )

        return {
            "orchestration_id": orchestration_id,
            "processing_path": processing_path,
            "workflow_result": workflow_result,
            "components_created": list(components.keys()),
            "status": "completed"
        }

class EventDrivenOrchestrator:
    """Orchestrate processing based on data events"""

    def __init__(self):
        self.event_handlers = {}
        self.event_queue = []

    def register_event_handler(self, event_type: str, handler_func):
        """Register handler for specific event types"""
        if event_type not in self.event_handlers:
            self.event_handlers[event_type] = []
        self.event_handlers[event_type].append(handler_func)

    async def process_event(self, event: Dict) -> Dict:
        """Process incoming event through registered handlers"""
        event_type = event.get("type", "unknown")

        if event_type in self.event_handlers:
            results = []
            for handler in self.event_handlers[event_type]:
                result = await handler(event)
                results.append(result)

            return {
                "event_type": event_type,
                "handlers_executed": len(results),
                "results": results
            }

        return {"event_type": event_type, "status": "no_handlers"}

class StreamProcessingOrchestrator:
    """Orchestrate continuous stream processing"""

    def __init__(self):
        self.stream_processors = {}
        self.processing_stats = {}

    async def process_data_stream(
        self,
        stream_id: str,
        data_stream: Any
    ) -> Dict:
        """Process continuous data stream"""

        if stream_id not in self.processing_stats:
            self.processing_stats[stream_id] = {
                "items_processed": 0,
                "errors": 0,
                "start_time": "current_time"
            }

        try:
            # Process stream data
            result = await self._process_stream_batch(data_stream)

            # Update stats
            self.processing_stats[stream_id]["items_processed"] += 1

            return {
                "stream_id": stream_id,
                "status": "processed",
                "result": result,
                "stats": self.processing_stats[stream_id].copy()
            }

        except Exception as e:
            self.processing_stats[stream_id]["errors"] += 1
            return {
                "stream_id": stream_id,
                "status": "error",
                "error": str(e),
                "stats": self.processing_stats[stream_id].copy()
            }

def test_advanced_orchestration():
    """Comprehensive testing of advanced orchestration patterns"""

    # Test dynamic component creation
    orchestrator = DynamicAtomicOrchestrator()

    def test_component_factory(config):
        return BaseAgent(
            agent_name="test_component",
            system_prompt="Test component for validation",
            max_tokens=config.get("max_tokens", 200)
        )

    orchestrator.register_component_factory("test_component", test_component_factory)
    component = orchestrator.create_component_on_demand("test_component", {"max_tokens": 150})

    assert component.agent_name == "test_component"
    assert component.max_tokens == 150

    # Test intelligent routing
    router = IntelligentWorkflowRouter(orchestrator)
    router.add_routing_rule("test_data", ["test_component"])

    path = router.determine_optimal_path("test data")
    assert "test_component" in path or "data_validator" in path

    print("✅ Advanced orchestration tests passed!")