"""
Pattern 72: Parallel Agent Pattern

Multiple agents execute tasks concurrently for faster processing.
Features parallel execution, result aggregation, and synchronization.
"""

from typing import Dict, Any, List, Optional
from dataclasses import dataclass, field
from datetime import datetime
import asyncio
import time
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI


@dataclass
class ParallelTask:
    """Task to be executed in parallel."""
    task_id: str
    description: str
    agent_name: Optional[str] = None
    input_data: Any = None
    result: Optional[str] = None
    status: str = "pending"  # pending, running, completed, failed
    start_time: Optional[datetime] = None
    end_time: Optional[datetime] = None
    duration: float = 0


@dataclass
class ParallelExecution:
    """Record of parallel execution."""
    execution_id: str
    tasks: List[ParallelTask]
    total_time: float
    speedup: float  # vs sequential execution
    efficiency: float  # speedup / num_agents
    timestamp: datetime = field(default_factory=datetime.now)


class ParallelAgent:
    """Agent that can execute tasks concurrently."""
    
    def __init__(self, name: str, model_name: str = "gpt-4"):
        self.name = name
        self.llm = ChatOpenAI(model=model_name, temperature=0.7)
        self.tasks_completed = 0
    
    async def execute_task_async(self, task: ParallelTask) -> str:
        """Execute task asynchronously."""
        task.status = "running"
        task.start_time = datetime.now()
        task.agent_name = self.name
        
        # Simulate work with LLM
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"You are {self.name}, executing a task in parallel with other agents."),
            ("human", f"Task: {task.description}\n"
                     f"Input: {task.input_data}\n\n"
                     f"Execute this task (provide brief result).")
        ])
        
        chain = prompt | self.llm
        
        # Simulate some processing time
        await asyncio.sleep(0.1)  # Small delay to simulate work
        
        response = chain.invoke({})
        result = str(response.content)[:200]
        
        task.end_time = datetime.now()
        task.duration = (task.end_time - task.start_time).total_seconds()
        task.result = result
        task.status = "completed"
        self.tasks_completed += 1
        
        return result


class ParallelAgentPattern:
    """
    Parallel Agent Pattern - Concurrent task execution.
    
    Multiple agents work simultaneously on different tasks,
    reducing total execution time through parallelism.
    """
    
    def __init__(self):
        self.agents: List[ParallelAgent] = []
        self.execution_history: List[ParallelExecution] = []
    
    def register_agent(self, agent: ParallelAgent):
        """Register agent for parallel execution."""
        self.agents.append(agent)
    
    async def execute_parallel(self, tasks: List[ParallelTask]) -> ParallelExecution:
        """Execute tasks in parallel across available agents."""
        start_time = time.time()
        
        print(f"\n   Executing {len(tasks)} tasks in parallel with {len(self.agents)} agents...")
        
        # Assign tasks to agents (round-robin)
        task_assignments = []
        for i, task in enumerate(tasks):
            agent = self.agents[i % len(self.agents)]
            task_assignments.append(agent.execute_task_async(task))
        
        # Execute all tasks concurrently
        await asyncio.gather(*task_assignments)
        
        total_time = time.time() - start_time
        
        # Calculate metrics
        avg_task_duration = sum(t.duration for t in tasks) / len(tasks)
        sequential_time = sum(t.duration for t in tasks)
        speedup = sequential_time / total_time if total_time > 0 else 1
        efficiency = speedup / len(self.agents) if self.agents else 0
        
        print(f"   ✓ Completed in {total_time:.2f}s")
        print(f"   Speedup: {speedup:.2f}x (vs sequential: {sequential_time:.2f}s)")
        print(f"   Efficiency: {efficiency:.1%}")
        
        # Record execution
        execution = ParallelExecution(
            execution_id=f"exec_{len(self.execution_history)}",
            tasks=tasks,
            total_time=total_time,
            speedup=speedup,
            efficiency=efficiency
        )
        self.execution_history.append(execution)
        
        return execution
    
    def execute_parallel_sync(self, tasks: List[ParallelTask]) -> ParallelExecution:
        """Synchronous wrapper for parallel execution."""
        return asyncio.run(self.execute_parallel(tasks))
    
    def aggregate_results(self, tasks: List[ParallelTask]) -> Dict[str, Any]:
        """Aggregate results from parallel tasks."""
        completed = [t for t in tasks if t.status == "completed"]
        failed = [t for t in tasks if t.status == "failed"]
        
        results = {
            "total_tasks": len(tasks),
            "completed": len(completed),
            "failed": len(failed),
            "results": {t.task_id: t.result for t in completed},
            "agent_contributions": {}
        }
        
        # Count tasks per agent
        for task in completed:
            if task.agent_name:
                results["agent_contributions"][task.agent_name] = \
                    results["agent_contributions"].get(task.agent_name, 0) + 1
        
        return results
    
    def run(self, scenario: str) -> Dict[str, Any]:
        """Execute parallel agent pattern."""
        total_executions = len(self.execution_history)
        avg_speedup = (sum(e.speedup for e in self.execution_history) / total_executions 
                      if total_executions > 0 else 0)
        avg_efficiency = (sum(e.efficiency for e in self.execution_history) / total_executions 
                         if total_executions > 0 else 0)
        
        return {
            "pattern": "ParallelAgent",
            "total_agents": len(self.agents),
            "total_executions": total_executions,
            "average_speedup": avg_speedup,
            "average_efficiency": avg_efficiency,
            "scenario": scenario
        }


# Example usage
if __name__ == "__main__":
    print("=== Parallel Agent Pattern Demo ===\n")
    
    # Create pattern
    pattern = ParallelAgentPattern()
    
    # Register parallel agents
    print("1. Registering parallel agents...")
    agents = [
        ParallelAgent("Worker1"),
        ParallelAgent("Worker2"),
        ParallelAgent("Worker3"),
        ParallelAgent("Worker4"),
    ]
    
    for agent in agents:
        pattern.register_agent(agent)
    
    print(f"   Registered {len(agents)} agents")
    
    # Create tasks for parallel execution
    print("\n2. Creating tasks...")
    tasks = [
        ParallelTask("t1", "Analyze customer feedback", input_data="feedback_batch_1"),
        ParallelTask("t2", "Process invoice data", input_data="invoices_q4"),
        ParallelTask("t3", "Generate sales report", input_data="sales_data_nov"),
        ParallelTask("t4", "Classify support tickets", input_data="tickets_batch_5"),
        ParallelTask("t5", "Extract document insights", input_data="documents_set_3"),
        ParallelTask("t6", "Summarize meeting notes", input_data="meetings_week_45"),
        ParallelTask("t7", "Validate data quality", input_data="dataset_v2"),
        ParallelTask("t8", "Transform data format", input_data="raw_exports"),
    ]
    
    print(f"   Created {len(tasks)} tasks")
    
    # Execute in parallel
    print("\n3. Executing tasks in parallel...")
    execution = pattern.execute_parallel_sync(tasks)
    
    # Show task assignments
    print("\n4. Task Distribution:")
    for agent in agents:
        agent_tasks = [t for t in tasks if t.agent_name == agent.name]
        print(f"   {agent.name}: {len(agent_tasks)} tasks")
        for task in agent_tasks:
            print(f"      - {task.task_id}: {task.description} ({task.duration:.2f}s)")
    
    # Aggregate results
    print("\n5. Results Aggregation:")
    aggregated = pattern.aggregate_results(tasks)
    print(f"   Total Tasks: {aggregated['total_tasks']}")
    print(f"   Completed: {aggregated['completed']}")
    print(f"   Failed: {aggregated['failed']}")
    
    # Performance metrics
    print("\n6. Performance Metrics:")
    summary = pattern.run("Parallel data processing")
    print(f"   Agents: {summary['total_agents']}")
    print(f"   Average Speedup: {summary['average_speedup']:.2f}x")
    print(f"   Average Efficiency: {summary['average_efficiency']:.1%}")
    
    print("\n✓ Parallel agents executed tasks concurrently for faster processing!")

    

Agentic Architecture Pattern: Parallel Agent

Pattern 72: 72 Parallel Agent – Line-by-Line Explanation

Overview

Pattern 72: Parallel Agent Pattern

Multiple agents execute tasks concurrently for faster processing. Features parallel execution, result aggregation, and synchronization.

Code Structure Analysis

File Information

• Pattern Number: 72
• Pattern Name: 72 Parallel Agent
• File Path: patterns\execution_control\72_parallel_agent.py
• Total Lines: 242

Detailed Code Explanation

Module Docstring

"""
Pattern 72: Parallel Agent Pattern

Multiple agents execute tasks concurrently for faster processing.
Features parallel execution, result aggregation, and synchronization.
"""

Explanation: This module docstring describes the purpose and key features of the 72 Parallel Agent pattern.

Imports Section

from typing import Dict, Any, List, Optional
from dataclasses import dataclass, field
from datetime import datetime
import asyncio
import time
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI  # First 20 imports shown

Explanation: The module imports necessary dependencies: – from typing import Dict, Any, List, Optional: Type hints for better code documentation – from dataclasses import dataclass, field: Data class decorators for clean data structures – from datetime import datetime: Date and time utilities – from langchain_core.prompts import ChatPromptTemplate: LangChain framework integration – from langchain_openai import ChatOpenAI: LangChain framework integration

Classes

ParallelTask (Line 18)

Description: Task to be executed in parallel.

ParallelExecution (Line 32)

Description: Record of parallel execution.

ParallelAgent (Line 42)

Description: Agent that can execute tasks concurrently.

Methods:

__init__ (Line 45)

def __init__(self, name, model_name):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference – name: Identifier or label – model_name: LLM model to use (e.g., “gpt-4”)

ParallelAgentPattern (Line 81)

Description: Parallel Agent Pattern – Concurrent task execution.

Multiple agents work simultaneously on different tasks, reducing total execution time through parallelism.

Methods:

__init__ (Line 89)

def __init__(self):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference

register_agent (Line 93)

def register_agent(self, agent):
    """Register agent for parallel execution."""

Purpose: Register agent for parallel execution.

Parameters: – self: Instance reference – agent: Parameter for agent

execute_parallel_sync (Line 136)

def execute_parallel_sync(self, tasks):
    """Synchronous wrapper for parallel execution."""

Purpose: Synchronous wrapper for parallel execution.

Parameters: – self: Instance reference – tasks: Parameter for tasks

aggregate_results (Line 140)

def aggregate_results(self, tasks):
    """Aggregate results from parallel tasks."""

Purpose: Aggregate results from parallel tasks.

Parameters: – self: Instance reference – tasks: Parameter for tasks

run (Line 161)

def run(self, scenario):
    """Execute parallel agent pattern."""

Purpose: Execute parallel agent pattern.

Parameters: – self: Instance reference – scenario: Parameter for scenario

Design Patterns Used

• Strategy Pattern: Different execution strategies for various scenarios
• State Pattern: Managing different states in the workflow
• Chain of Responsibility: Sequential processing through workflow nodes
• Observer Pattern: Tracking and monitoring pattern execution
• Factory Pattern: Creating instances with proper configuration

Usage Example

# Initialize the pattern
pattern = ParallelTask(name="example", model_name="gpt-4")

# Configure and execute
result = pattern.execute(data={"key": "value"})

# Check results
print(result)

Key Concepts

• LLM Integration: Uses LangChain/OpenAI for intelligent decision-making
• Workflow Orchestration: LangGraph manages execution flow through nodes
• State Management: TypedDict defines workflow state structure
• Error Handling: Comprehensive error handling and fallback mechanisms
• Extensibility: Designed for easy customization and extension

Extension Points

This pattern can be extended by: 1. Adding custom workflow nodes 2. Implementing additional validation logic 3. Integrating with external services 4. Customizing LLM prompts and responses 5. Adding monitoring and logging capabilities