"""
Pattern 73: Pipeline Pattern

Data flows through a series of processing stages with transformations
at each stage. Similar to Unix pipes or data processing pipelines.
"""

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


@dataclass
class PipelineStage:
    """Stage in the processing pipeline."""
    name: str
    processor: Callable[[Any], Any]
    description: str = ""
    input_type: Optional[str] = None
    output_type: Optional[str] = None


@dataclass
class PipelineResult:
    """Result from pipeline execution."""
    success: bool
    input_data: Any
    output_data: Optional[Any]
    stages_executed: List[str]
    stage_outputs: Dict[str, Any]
    execution_time: float
    error: Optional[str] = None


class PipelineAgent:
    """Agent that acts as a pipeline stage."""
    
    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 transform(self, data: Any) -> Any:
        """Transform input data."""
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"You are {self.name}, a {self.role}. "
                      f"Transform the input data according to your role."),
            ("human", f"Input: {data}\n\nProvide transformed output.")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({})
        return str(response.content)


class PipelinePattern:
    """
    Pipeline Pattern - Sequential data transformation chain.
    
    Data flows through stages with each stage transforming the data.
    Stages can be agents, functions, or any callable processor.
    """
    
    def __init__(self):
        self.stages: List[PipelineStage] = []
        self.execution_log: List[PipelineResult] = []
    
    def add_stage(self, stage: PipelineStage):
        """Add stage to pipeline."""
        self.stages.append(stage)
    
    def add_agent_stage(self, agent: PipelineAgent, description: str = ""):
        """Add agent as pipeline stage."""
        stage = PipelineStage(
            name=agent.name,
            processor=agent.transform,
            description=description or f"{agent.role} processing"
        )
        self.stages.append(stage)
    
    def add_function_stage(self, name: str, func: Callable[[Any], Any], description: str = ""):
        """Add function as pipeline stage."""
        stage = PipelineStage(
            name=name,
            processor=func,
            description=description
        )
        self.stages.append(stage)
    
    def execute(self, input_data: Any) -> PipelineResult:
        """Execute data through pipeline."""
        start_time = datetime.now()
        
        stages_executed = []
        stage_outputs = {}
        current_data = input_data
        
        try:
            for stage in self.stages:
                print(f"   Executing stage: {stage.name}...")
                
                # Process data through stage
                current_data = stage.processor(current_data)
                
                # Log execution
                stages_executed.append(stage.name)
                stage_outputs[stage.name] = current_data
            
            execution_time = (datetime.now() - start_time).total_seconds()
            
            result = PipelineResult(
                success=True,
                input_data=input_data,
                output_data=current_data,
                stages_executed=stages_executed,
                stage_outputs=stage_outputs,
                execution_time=execution_time
            )
            
        except Exception as e:
            execution_time = (datetime.now() - start_time).total_seconds()
            
            result = PipelineResult(
                success=False,
                input_data=input_data,
                output_data=None,
                stages_executed=stages_executed,
                stage_outputs=stage_outputs,
                execution_time=execution_time,
                error=str(e)
            )
        
        self.execution_log.append(result)
        return result
    
    def run(self, input_data: Any) -> Dict[str, Any]:
        """Run pipeline and return results."""
        result = self.execute(input_data)
        
        return {
            "pattern": "Pipeline",
            "success": result.success,
            "stages_executed": len(result.stages_executed),
            "total_stages": len(self.stages),
            "execution_time": result.execution_time,
            "output": result.output_data,
            "error": result.error
        }
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get pipeline statistics."""
        if not self.execution_log:
            return {"total_runs": 0}
        
        return {
            "total_runs": len(self.execution_log),
            "successful_runs": sum(1 for r in self.execution_log if r.success),
            "failed_runs": sum(1 for r in self.execution_log if not r.success),
            "avg_execution_time": sum(r.execution_time for r in self.execution_log) / len(self.execution_log),
            "total_stages": len(self.stages)
        }


# Example usage
if __name__ == "__main__":
    print("=== Pipeline Pattern Demo ===\n")
    
    # Create pipeline
    pipeline = PipelinePattern()
    
    # Add agent-based stages
    print("1. Building pipeline with agents...")
    extractor = PipelineAgent("Extractor", "data extractor")
    transformer = PipelineAgent("Transformer", "data transformer")
    validator = PipelineAgent("Validator", "data validator")
    
    pipeline.add_agent_stage(extractor, "Extract key information")
    pipeline.add_agent_stage(transformer, "Transform to target format")
    pipeline.add_agent_stage(validator, "Validate output quality")
    
    # Add function-based stage
    def format_output(data: Any) -> str:
        return f"[FORMATTED] {str(data).upper()}"
    
    pipeline.add_function_stage("Formatter", format_output, "Format final output")
    
    # Execute pipeline
    print("\n2. Executing pipeline...")
    input_text = "Process this customer feedback: The product is great but shipping was slow."
    result = pipeline.run(input_text)
    
    print(f"\n3. Pipeline Results:")
    print(f"   Success: {result['success']}")
    print(f"   Stages Executed: {result['stages_executed']}/{result['total_stages']}")
    print(f"   Execution Time: {result['execution_time']:.2f}s")
    
    if result['output']:
        print(f"\n4. Final Output:")
        print(f"   {str(result['output'])[:150]}...")
    
    # Statistics
    stats = pipeline.get_statistics()
    print(f"\n5. Pipeline Statistics:")
    print(f"   Total Runs: {stats['total_runs']}")
    print(f"   Success Rate: {stats['successful_runs']}/{stats['total_runs']}")
    
    print("\n✓ Pipeline pattern processed data through transformation stages!")

    

Agentic Architecture Pattern: Pipeline

Pattern 73: 73 Pipeline – Line-by-Line Explanation

Overview

Pattern 73: Pipeline Pattern

Data flows through a series of processing stages with transformations at each stage. Similar to Unix pipes or data processing pipelines.

Code Structure Analysis

File Information

• Pattern Number: 73
• Pattern Name: 73 Pipeline
• File Path: patterns\execution_control\73_pipeline.py
• Total Lines: 210

Detailed Code Explanation

Module Docstring

"""
Pattern 73: Pipeline Pattern

Data flows through a series of processing stages with transformations
at each stage. Similar to Unix pipes or data processing pipelines.
"""

Explanation: This module docstring describes the purpose and key features of the 73 Pipeline pattern.

Imports Section

from typing import Dict, Any, List, Optional, Callable
from dataclasses import dataclass, field
from datetime import datetime
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, Callable: 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

PipelineStage (Line 16)

Description: Stage in the processing pipeline.

PipelineResult (Line 26)

Description: Result from pipeline execution.

PipelineAgent (Line 37)

Description: Agent that acts as a pipeline stage.

Methods:

__init__ (Line 40)

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”)

transform (Line 45)

def transform(self, data):
    """Transform input data."""

Purpose: Transform input data.

Parameters: – self: Instance reference – data: Input data to process

PipelinePattern (Line 58)

Description: Pipeline Pattern – Sequential data transformation chain.

Data flows through stages with each stage transforming the data. Stages can be agents, functions, or any callable processor.

Methods:

__init__ (Line 66)

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

Purpose: No method docstring.

Parameters: – self: Instance reference

add_stage (Line 70)

def add_stage(self, stage):
    """Add stage to pipeline."""

Purpose: Add stage to pipeline.

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

add_agent_stage (Line 74)

def add_agent_stage(self, agent, description):
    """Add agent as pipeline stage."""

Purpose: Add agent as pipeline stage.

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

add_function_stage (Line 83)

def add_function_stage(self, name, func, description):
    """Add function as pipeline stage."""

Purpose: Add function as pipeline stage.

Parameters: – self: Instance reference – name: Identifier or label – func: Parameter for func – description: Parameter for description

execute (Line 92)

def execute(self, input_data):
    """Execute data through pipeline."""

Purpose: Execute data through pipeline.

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

run (Line 138)

def run(self, input_data):
    """Run pipeline and return results."""

Purpose: Run pipeline and return results.

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

get_statistics (Line 152)

def get_statistics(self):
    """Get pipeline statistics."""

Purpose: Get pipeline statistics.

Parameters: – self: Instance reference

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