aishastr.live

Agentic Architecture Pattern: Orchestrator

Written by

admin

in


    """
Pattern 143: Orchestrator Pattern

Central orchestrator that coordinates complex workflows involving multiple agents.
Implements workflow orchestration with LangGraph state machines.
"""

from dataclasses import dataclass, field
from datetime import datetime
from typing import Dict, Any, List, Optional, Callable
from enum import Enum
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langgraph.graph import StateGraph, END
from typing_extensions import TypedDict


class StepStatus(Enum):
    """Status of workflow step."""
    PENDING = "pending"
    RUNNING = "running"
    COMPLETED = "completed"
    FAILED = "failed"
    SKIPPED = "skipped"


@dataclass
class WorkflowStep:
    """Represents a step in a workflow."""
    step_id: str
    name: str
    agent_id: str
    depends_on: List[str] = field(default_factory=list)
    status: StepStatus = StepStatus.PENDING
    input_data: Optional[Any] = None
    output_data: Optional[Any] = None
    started_at: Optional[datetime] = None
    completed_at: Optional[datetime] = None
    error: Optional[str] = None


@dataclass
class Workflow:
    """Represents a complete workflow."""
    workflow_id: str
    name: str
    description: str
    steps: List[WorkflowStep] = field(default_factory=list)
    status: str = "pending"
    created_at: datetime = field(default_factory=datetime.now)
    completed_at: Optional[datetime] = None


class OrchestratedAgent:
    """Agent that can be orchestrated."""
    
    def __init__(self, agent_id: str, name: str, capabilities: List[str], model_name: str = "gpt-4"):
        self.agent_id = agent_id
        self.name = name
        self.capabilities = capabilities
        self.llm = ChatOpenAI(model=model_name, temperature=0.7)
        
    def execute(self, step: WorkflowStep) -> Any:
        """Execute a workflow step."""
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"You are {self.name}, an agent with capabilities: {', '.join(self.capabilities)}"),
            ("human", """Execute this workflow step:

Step: {step_name}
Input: {input_data}

Provide the output for this step:""")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({
            "step_name": step.name,
            "input_data": step.input_data or "No input provided"
        })
        
        return str(response.content)


class OrchestratorAgent:
    """
    Orchestrator that coordinates workflow execution across multiple agents.
    """
    
    def __init__(self, name: str = "Orchestrator", model_name: str = "gpt-4"):
        self.name = name
        self.llm = ChatOpenAI(model=model_name, temperature=0.3)
        self.agents: Dict[str, OrchestratedAgent] = {}
        self.workflows: Dict[str, Workflow] = {}
        
    def register_agent(self, agent: OrchestratedAgent):
        """Register an agent for orchestration."""
        self.agents[agent.agent_id] = agent
        
    def create_workflow(self, workflow: Workflow):
        """Register a workflow."""
        self.workflows[workflow.workflow_id] = workflow
        
    def execute_workflow(self, workflow_id: str) -> Dict[str, Any]:
        """Execute a workflow by orchestrating its steps."""
        if workflow_id not in self.workflows:
            return {"success": False, "error": "Workflow not found"}
        
        workflow = self.workflows[workflow_id]
        workflow.status = "running"
        
        execution_results = []
        completed_steps = set()
        
        # Build dependency graph
        steps_by_id = {step.step_id: step for step in workflow.steps}
        
        # Execute steps in dependency order
        while len(completed_steps)  Dict[str, Any]:
        """Execute a single workflow step."""
        step.status = StepStatus.RUNNING
        step.started_at = datetime.now()
        
        # Get agent for this step
        agent = self.agents.get(step.agent_id)
        if not agent:
            step.status = StepStatus.FAILED
            step.error = f"Agent {step.agent_id} not found"
            return {
                "success": False,
                "step_id": step.step_id,
                "error": step.error
            }
        
        # Collect outputs from dependency steps
        if step.depends_on:
            dependency_outputs = {}
            for dep_id in step.depends_on:
                if dep_id in steps_by_id:
                    dep_step = steps_by_id[dep_id]
                    dependency_outputs[dep_id] = dep_step.output_data
            step.input_data = dependency_outputs
        
        try:
            # Execute step
            output = agent.execute(step)
            
            step.output_data = output
            step.status = StepStatus.COMPLETED
            step.completed_at = datetime.now()
            
            return {
                "success": True,
                "step_id": step.step_id,
                "step_name": step.name,
                "agent_id": step.agent_id,
                "output": output
            }
            
        except Exception as e:
            step.status = StepStatus.FAILED
            step.error = str(e)
            return {
                "success": False,
                "step_id": step.step_id,
                "error": str(e)
            }
    
    def get_workflow_status(self, workflow_id: str) -> Dict[str, Any]:
        """Get status of a workflow."""
        if workflow_id not in self.workflows:
            return {"error": "Workflow not found"}
        
        workflow = self.workflows[workflow_id]
        
        steps_status = [
            {
                "step_id": step.step_id,
                "name": step.name,
                "status": step.status.value,
                "agent": step.agent_id
            }
            for step in workflow.steps
        ]
        
        return {
            "workflow_id": workflow_id,
            "name": workflow.name,
            "status": workflow.status,
            "total_steps": len(workflow.steps),
            "completed_steps": len([s for s in workflow.steps if s.status == StepStatus.COMPLETED]),
            "steps": steps_status
        }


class OrchestratorPattern:
    """
    LangGraph-based orchestrator pattern for complex workflows.
    """
    
    def __init__(self, model_name: str = "gpt-4"):
        self.orchestrator = OrchestratorAgent(model_name=model_name)
        
    def run(self, workflow_id: str) -> Dict[str, Any]:
        """Run an orchestrated workflow."""
        return self.orchestrator.execute_workflow(workflow_id)


def main():
    """Example usage of Orchestrator Pattern."""
    print("=" * 80)
    print("Orchestrator Pattern - Complex Workflow Demo")
    print("=" * 80)
    
    # Create orchestrator pattern
    pattern = OrchestratorPattern()
    orchestrator = pattern.orchestrator
    
    # Create agents with different capabilities
    agents = [
        OrchestratedAgent(
            agent_id="data-collector",
            name="Data Collector",
            capabilities=["data_collection", "api_calls", "web_scraping"]
        ),
        OrchestratedAgent(
            agent_id="data-processor",
            name="Data Processor",
            capabilities=["data_cleaning", "transformation", "validation"]
        ),
        OrchestratedAgent(
            agent_id="analyzer",
            name="Data Analyzer",
            capabilities=["analysis", "statistics", "insights"]
        ),
        OrchestratedAgent(
            agent_id="visualizer",
            name="Visualizer",
            capabilities=["charts", "graphs", "dashboards"]
        ),
        OrchestratedAgent(
            agent_id="reporter",
            name="Report Generator",
            capabilities=["report_writing", "summarization", "formatting"]
        )
    ]
    
    # Register agents
    for agent in agents:
        orchestrator.register_agent(agent)
    
    # Create a complex workflow: Data Analysis Pipeline
    workflow = Workflow(
        workflow_id="wf-analysis-001",
        name="Data Analysis Pipeline",
        description="Collect, process, analyze, visualize, and report on data",
        steps=[
            WorkflowStep(
                step_id="step-1",
                name="Collect Data",
                agent_id="data-collector",
                depends_on=[],
                input_data="Collect customer transaction data"
            ),
            WorkflowStep(
                step_id="step-2",
                name="Process Data",
                agent_id="data-processor",
                depends_on=["step-1"],
            ),
            WorkflowStep(
                step_id="step-3",
                name="Analyze Data",
                agent_id="analyzer",
                depends_on=["step-2"],
            ),
            WorkflowStep(
                step_id="step-4",
                name="Create Visualizations",
                agent_id="visualizer",
                depends_on=["step-3"],
            ),
            WorkflowStep(
                step_id="step-5",
                name="Generate Report",
                agent_id="reporter",
                depends_on=["step-3", "step-4"],
            )
        ]
    )
    
    # Register workflow
    orchestrator.create_workflow(workflow)
    
    # Display workflow structure
    print("\nšŸ“‹ Workflow Structure:")
    print(f"ID: {workflow.workflow_id}")
    print(f"Name: {workflow.name}")
    print(f"Description: {workflow.description}")
    print(f"\nSteps:")
    for step in workflow.steps:
        deps = f" (depends on: {', '.join(step.depends_on)})" if step.depends_on else ""
        print(f"  {step.step_id}: {step.name} ā†’ {step.agent_id}{deps}")
    
    # Execute workflow
    print("\n" + "=" * 80)
    print("šŸ”„ Executing Workflow...")
    print("=" * 80)
    
    result = pattern.run(workflow.workflow_id)
    
    # Display results
    print("\nāœ… Execution Results:")
    print(f"Success: {result['success']}")
    
    if result["success"]:
        print(f"\nWorkflow: {result['workflow_id']}")
        print(f"Steps Executed: {len(result['results'])}")
        
        print("\n" + "=" * 80)
        print("Step Results:")
        print("=" * 80)
        for step_result in result["results"]:
            print(f"\n{step_result['step_name']} ({step_result['step_id']})")
            print(f"  Agent: {step_result['agent_id']}")
            print(f"  Output: {str(step_result['output'])[:150]}...")
    else:
        print(f"Error: {result.get('error', 'Unknown error')}")
    
    # Display workflow status
    print("\n" + "=" * 80)
    print("šŸ“Š Workflow Status")
    print("=" * 80)
    status = orchestrator.get_workflow_status(workflow.workflow_id)
    print(f"Status: {status['status']}")
    print(f"Progress: {status['completed_steps']}/{status['total_steps']} steps")
    
    print("\nStep Status:")
    for step_status in status["steps"]:
        print(f"  {step_status['name']}: {step_status['status']}")


if __name__ == "__main__":
    main()

Agentic Architecture Pattern: Orchestrator

Pattern 143: 143 Orchestrator – Line-by-Line Explanation

Overview

Pattern 143: Orchestrator Pattern

Central orchestrator that coordinates complex workflows involving multiple agents. Implements workflow orchestration with LangGraph state machines.

Code Structure Analysis

File Information

  • Pattern Number: 143
  • Pattern Name: 143 Orchestrator
  • File Path: patterns\workflow\143_orchestrator.py
  • Total Lines: 393

Detailed Code Explanation

Module Docstring

"""
Pattern 143: Orchestrator Pattern

Central orchestrator that coordinates complex workflows involving multiple agents.
Implements workflow orchestration with LangGraph state machines.
"""

Explanation: This module docstring describes the purpose and key features of the 143 Orchestrator pattern.

Imports Section

from dataclasses import dataclass, field
from datetime import datetime
from typing import Dict, Any, List, Optional, Callable
from enum import Enum
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langgraph.graph import StateGraph, END
from typing_extensions import TypedDict  # First 20 imports shown

Explanation: The module imports necessary dependencies: – from dataclasses import dataclass, field: Data class decorators for clean data structures – from datetime import datetime: Date and time utilities – from typing import Dict, Any, List, Optional, Callable: Type hints for better code documentation – from enum import Enum: Enumeration support for fixed sets of values – from langchain_openai import ChatOpenAI: LangChain framework integration – from langchain_core.prompts import ChatPromptTemplate: LangChain framework integration – from langgraph.graph import StateGraph, END: LangGraph for workflow orchestration – from typing_extensions import TypedDict: Type hints for better code documentation

Enumerations

StepStatus (Line 18)

class StepStatus(Enum):
    PENDING
    RUNNING
    COMPLETED
    FAILED
    SKIPPED

Purpose: Status of workflow step.

Values: – PENDING: Initial state, not yet processed – RUNNING: Currently executing – COMPLETED: Successfully finished – FAILED: Execution failed – SKIPPED: Represents skipped

Classes

StepStatus (Line 18)

Description: Status of workflow step.

WorkflowStep (Line 28)

Description: Represents a step in a workflow.

Workflow (Line 43)

Description: Represents a complete workflow.

OrchestratedAgent (Line 54)

Description: Agent that can be orchestrated.

Methods:

__init__ (Line 57)
def __init__(self, agent_id, name, capabilities, model_name):
    """No method docstring."""

Purpose: No method docstring.

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

execute (Line 63)
def execute(self, step):
    """Execute a workflow step."""

Purpose: Execute a workflow step.

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

OrchestratorAgent (Line 84)

Description: Orchestrator that coordinates workflow execution across multiple agents.

Methods:

__init__ (Line 89)
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”)

register_agent (Line 95)
def register_agent(self, agent):
    """Register an agent for orchestration."""

Purpose: Register an agent for orchestration.

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

create_workflow (Line 99)
def create_workflow(self, workflow):
    """Register a workflow."""

Purpose: Register a workflow.

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

execute_workflow (Line 103)
def execute_workflow(self, workflow_id):
    """Execute a workflow by orchestrating its steps."""

Purpose: Execute a workflow by orchestrating its steps.

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

_execute_step (Line 162)
def _execute_step(self, step, steps_by_id, completed_steps):
    """Execute a single workflow step."""

Purpose: Execute a single workflow step.

Parameters: – self: Instance reference – step: Parameter for step – steps_by_id: Parameter for steps_by_id – completed_steps: Parameter for completed_steps

get_workflow_status (Line 217)
def get_workflow_status(self, workflow_id):
    """Get status of a workflow."""

Purpose: Get status of a workflow.

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

OrchestratorPattern (Line 244)

Description: LangGraph-based orchestrator pattern for complex workflows.

Methods:

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

Purpose: No method docstring.

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

run (Line 252)
def run(self, workflow_id):
    """Run an orchestrated workflow."""

Purpose: Run an orchestrated workflow.

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

Functions

main (Line 257)

def main():
    """Example usage of Orchestrator Pattern."""

Purpose: Example usage of Orchestrator Pattern.

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 = StepStatus(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

Comments

Leave a Reply

Your email address will not be published. Required fields are marked *

More posts