"""
Pattern 71: Sequential Agent Pattern

Agents execute tasks in a strict sequential order, with each agent processing
the output of the previous agent in a linear chain.
"""

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


@dataclass
class SequentialTask:
    """Task in sequential execution."""
    task_id: str
    agent_name: str
    input_data: Any
    output_data: Optional[Any] = None
    status: str = "pending"  # pending, running, completed, failed
    start_time: Optional[datetime] = None
    end_time: Optional[datetime] = None
    error: Optional[str] = None


class SequentialState(TypedDict):
    """State for sequential execution workflow."""
    tasks: List[SequentialTask]
    current_index: int
    input_data: Any
    final_output: Optional[Any]
    execution_log: List[str]


class SequentialAgent:
    """Agent that processes data sequentially."""
    
    def __init__(self, name: str, role: str, model_name: str = "gpt-4"):
        self.name = name
        self.role = role
        self.llm = ChatOpenAI(model=model_name, temperature=0.7)
    
    def process(self, input_data: Any) -> Any:
        """Process input and return output."""
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"You are {self.name}, a {self.role}. "
                      f"Process the input and provide your output for the next agent."),
            ("human", f"Input: {input_data}\n\nProvide your processed output.")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({})
        return str(response.content)


class SequentialAgentPattern:
    """
    Sequential Agent Pattern - Linear execution chain.
    
    Uses LangGraph to manage sequential workflow where each agent
    processes the output of the previous agent in strict order.
    """
    
    def __init__(self):
        self.agents: List[SequentialAgent] = []
        self.execution_history: List[SequentialTask] = []
    
    def add_agent(self, agent: SequentialAgent):
        """Add agent to sequential chain."""
        self.agents.append(agent)
    
    def build_graph(self) -> StateGraph:
        """Build sequential execution graph."""
        workflow = StateGraph(SequentialState)
        
        def execute_next(state: SequentialState) -> SequentialState:
            """Execute next agent in sequence."""
            current_idx = state["current_index"]
            tasks = state["tasks"]
            
            if current_idx >= len(tasks):
                # All tasks completed
                if tasks:
                    state["final_output"] = tasks[-1].output_data
                return state
            
            task = tasks[current_idx]
            task.status = "running"
            task.start_time = datetime.now()
            
            try:
                # Get agent
                agent = self.agents[current_idx]
                
                # Get input (previous output or initial input)
                if current_idx == 0:
                    input_data = state["input_data"]
                else:
                    input_data = tasks[current_idx - 1].output_data
                
                # Execute agent
                output = agent.process(input_data)
                
                task.output_data = output
                task.status = "completed"
                task.end_time = datetime.now()
                
                log_msg = f"Agent {agent.name} completed: {str(output)[:50]}..."
                state["execution_log"].append(log_msg)
                
            except Exception as e:
                task.status = "failed"
                task.error = str(e)
                task.end_time = datetime.now()
                state["execution_log"].append(f"Agent {task.agent_name} failed: {str(e)}")
            
            # Move to next
            state["current_index"] = current_idx + 1
            
            return state
        
        def should_continue(state: SequentialState) -> str:
            """Check if should continue to next agent."""
            if state["current_index"] >= len(state["tasks"]):
                return "end"
            
            current_task = state["tasks"][state["current_index"]]
            if current_task.status == "failed":
                return "end"
            
            return "continue"
        
        workflow.add_node("execute", execute_next)
        workflow.set_entry_point("execute")
        
        workflow.add_conditional_edges(
            "execute",
            should_continue,
            {
                "continue": "execute",
                "end": END
            }
        )
        
        return workflow.compile()
    
    def run(self, input_data: Any) -> Dict[str, Any]:
        """Execute sequential agent chain."""
        # Create tasks for each agent
        tasks = [
            SequentialTask(
                task_id=f"task_{i}",
                agent_name=agent.name,
                input_data=input_data if i == 0 else None
            )
            for i, agent in enumerate(self.agents)
        ]
        
        initial_state: SequentialState = {
            "tasks": tasks,
            "current_index": 0,
            "input_data": input_data,
            "final_output": None,
            "execution_log": []
        }
        
        # Build and execute graph
        graph = self.build_graph()
        final_state = graph.invoke(initial_state)
        
        # Store history
        self.execution_history.extend(final_state["tasks"])
        
        return {
            "pattern": "Sequential",
            "agents_executed": len([t for t in final_state["tasks"] if t.status == "completed"]),
            "total_agents": len(self.agents),
            "final_output": final_state["final_output"],
            "execution_log": final_state["execution_log"],
            "success": all(t.status == "completed" for t in final_state["tasks"])
        }


# Example usage
if __name__ == "__main__":
    print("=== Sequential Agent Pattern Demo ===\n")
    
    # Create pattern
    pattern = SequentialAgentPattern()
    
    # Create agent chain
    print("1. Creating sequential agent chain...")
    researcher = SequentialAgent("Researcher", "research analyst")
    writer = SequentialAgent("Writer", "content writer")
    editor = SequentialAgent("Editor", "editor and proofreader")
    formatter = SequentialAgent("Formatter", "document formatter")
    
    # Add agents in order
    pattern.add_agent(researcher)
    pattern.add_agent(writer)
    pattern.add_agent(editor)
    pattern.add_agent(formatter)
    
    # Execute sequential workflow
    print("\n2. Executing sequential workflow...")
    result = pattern.run("Write a blog post about AI agents")
    
    print(f"\n3. Results:")
    print(f"   Success: {result['success']}")
    print(f"   Agents Executed: {result['agents_executed']}/{result['total_agents']}")
    print(f"\n4. Execution Log:")
    for log in result['execution_log']:
        print(f"   - {log}")
    
    print(f"\n5. Final Output:")
    print(f"   {str(result['final_output'])[:200]}...")
    
    print("\n✓ Sequential agent pattern executed tasks in linear order!")

    

Agentic Architecture Pattern: Sequential Agent

Pattern 71: 71 Sequential Agent – Line-by-Line Explanation

Overview

Pattern 71: Sequential Agent Pattern

Agents execute tasks in a strict sequential order, with each agent processing the output of the previous agent in a linear chain.

Code Structure Analysis

File Information

• Pattern Number: 71
• Pattern Name: 71 Sequential Agent
• File Path: patterns\execution_control\71_sequential_agent.py
• Total Lines: 223

Detailed Code Explanation

Module Docstring

"""
Pattern 71: Sequential Agent Pattern

Agents execute tasks in a strict sequential order, with each agent processing
the output of the previous agent in a linear chain.
"""

Explanation: This module docstring describes the purpose and key features of the 71 Sequential Agent pattern.

Imports Section

from typing import Dict, Any, List, Optional
from dataclasses import dataclass, field
from datetime import datetime
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
from typing_extensions import TypedDict  # 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 – from langgraph.graph import StateGraph, END: LangGraph for workflow orchestration – from typing_extensions import TypedDict: Type hints for better code documentation

Classes

SequentialTask (Line 18)

Description: Task in sequential execution.

SequentialState (Line 30)

Description: State for sequential execution workflow.

SequentialAgent (Line 39)

Description: Agent that processes data sequentially.

Methods:

__init__ (Line 42)

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

Purpose: No method docstring.

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

process (Line 47)

def process(self, input_data):
    """Process input and return output."""

Purpose: Process input and return output.

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

SequentialAgentPattern (Line 60)

Description: Sequential Agent Pattern – Linear execution chain.

Uses LangGraph to manage sequential workflow where each agent processes the output of the previous agent in strict order.

Methods:

__init__ (Line 68)

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

Purpose: No method docstring.

Parameters: – self: Instance reference

add_agent (Line 72)

def add_agent(self, agent):
    """Add agent to sequential chain."""

Purpose: Add agent to sequential chain.

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

build_graph (Line 76)

def build_graph(self):
    """Build sequential execution graph."""

Purpose: Build sequential execution graph.

Parameters: – self: Instance reference

run (Line 151)

def run(self, input_data):
    """Execute sequential agent chain."""

Purpose: Execute sequential agent chain.

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

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