Tag: llm

Agentic Architecture Pattern: Self Consistency


    """
Pattern 197: Self Consistency Pattern

Purpose: Enhanced implementation with full LangChain/LangGraph integration
"""

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


@dataclass
class SelfConsistencyConfig:
    """Configuration for Self Consistency Pattern."""
    model_name: str = "gpt-4"
    temperature: float = 0.7
    max_iterations: int = 10
    enable_logging: bool = True


@dataclass
class SelfConsistencyState:
    """State information for Self Consistency Pattern."""
    current_step: int = 0
    data: Dict[str, Any] = field(default_factory=dict)
    results: List[Any] = field(default_factory=list)
    metadata: Dict[str, Any] = field(default_factory=dict)
    timestamp: datetime = field(default_factory=datetime.now)


class SelfConsistencyAgent:
    """Agent implementing Self Consistency Pattern."""
    
    def __init__(self, name: str, config: SelfConsistencyConfig = None):
        self.name = name
        self.config = config or SelfConsistencyConfig()
        self.llm = ChatOpenAI(
            model=self.config.model_name,
            temperature=self.config.temperature
        )
        self.state = SelfConsistencyState()
        self.execution_history: List[Dict[str, Any]] = []
    
    def process(self, input_data: Any) -> Dict[str, Any]:
        """Process input according to Self Consistency Pattern."""
        self.state.current_step += 1
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"You are {self.name}, implementing Self Consistency Pattern. "
                      f"Provide clear, structured responses."),
            ("human", "Process this input: {input_data}")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({"input_data": str(input_data)})
        
        result = {
            "agent": self.name,
            "step": self.state.current_step,
            "input": input_data,
            "output": str(response.content),
            "timestamp": datetime.now()
        }
        
        self.execution_history.append(result)
        self.state.results.append(result)
        
        return result
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get agent statistics."""
        return {
            "name": self.name,
            "total_executions": len(self.execution_history),
            "current_step": self.state.current_step,
            "results_count": len(self.state.results)
        }


class SelfConsistencyPattern:
    """
    Self Consistency Pattern - Enhanced implementation
    
    This pattern provides comprehensive functionality for self consistency
    with full LangChain/LangGraph integration.
    
    Features:
    - Multi-agent coordination
    - State management
    - Execution tracking
    - LangGraph workflow integration
    """
    
    def __init__(self, config: SelfConsistencyConfig = None):
        self.config = config or SelfConsistencyConfig()
        self.agents: Dict[str, SelfConsistencyAgent] = {}
        self.workflow = self._build_workflow()
        self.execution_log: List[Dict[str, Any]] = []
    
    def _build_workflow(self) -> StateGraph:
        """Build LangGraph workflow for pattern execution."""
        
        class WorkflowState(TypedDict):
            input: str
            processed: str
            output: str
            metadata: Dict[str, Any]
        
        def process_input(state: WorkflowState) -> WorkflowState:
            """Process input node."""
            return {
                **state,
                "processed": f"Processed: {state['input']}",
                "metadata": {"step": "process_input"}
            }
        
        def generate_output(state: WorkflowState) -> WorkflowState:
            """Generate output node."""
            llm = ChatOpenAI(model=self.config.model_name, temperature=self.config.temperature)
            
            prompt = ChatPromptTemplate.from_messages([
                ("system", "You are implementing Self Consistency Pattern. Generate comprehensive output."),
                ("human", "{input}")
            ])
            
            chain = prompt | llm
            response = chain.invoke({"input": state["processed"]})
            
            return {
                **state,
                "output": str(response.content),
                "metadata": {**state.get("metadata", {}), "step": "generate_output"}
            }
        
        # Build workflow graph
        workflow = StateGraph(WorkflowState)
        workflow.add_node("process", process_input)
        workflow.add_node("generate", generate_output)
        workflow.set_entry_point("process")
        workflow.add_edge("process", "generate")
        workflow.add_edge("generate", END)
        
        return workflow.compile()
    
    def register_agent(self, agent: SelfConsistencyAgent):
        """Register an agent with the pattern."""
        self.agents[agent.name] = agent
        if self.config.enable_logging:
            print(f"Registered agent: {agent.name}")
    
    def execute(self, input_data: Any, agent_name: str = None) -> Dict[str, Any]:
        """Execute the pattern with given input."""
        
        # Use workflow if no specific agent requested
        if agent_name is None:
            result = self.workflow.invoke({
                "input": str(input_data),
                "processed": "",
                "output": "",
                "metadata": {}
            })
            
            execution_record = {
                "pattern": "Self Consistency Pattern",
                "input": input_data,
                "output": result.get("output", ""),
                "metadata": result.get("metadata", {}),
                "timestamp": datetime.now()
            }
            
            self.execution_log.append(execution_record)
            return execution_record
        
        # Execute with specific agent
        agent = self.agents.get(agent_name)
        if not agent:
            raise ValueError(f"Agent {agent_name} not found")
        
        result = agent.process(input_data)
        self.execution_log.append(result)
        return result
    
    def run(self, input_data: Any) -> Dict[str, Any]:
        """Main execution method."""
        return self.execute(input_data)
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get pattern statistics."""
        return {
            "pattern": "Self Consistency Pattern",
            "total_executions": len(self.execution_log),
            "registered_agents": len(self.agents),
            "agents": [agent.get_statistics() for agent in self.agents.values()]
        }


# Example usage
if __name__ == "__main__":
    print("=== Self Consistency Pattern Demo ===\n")
    
    # Create configuration
    config = SelfConsistencyConfig(
        model_name="gpt-4",
        temperature=0.7,
        enable_logging=True
    )
    
    # Create pattern
    pattern = SelfConsistencyPattern(config)
    
    # Create and register agents
    agent1 = SelfConsistencyAgent("Agent1", config)
    agent2 = SelfConsistencyAgent("Agent2", config)
    pattern.register_agent(agent1)
    pattern.register_agent(agent2)
    
    print("\nExecuting pattern workflow...")
    
    # Execute with workflow
    result1 = pattern.run("Demonstrate self consistency functionality")
    print(f"\nWorkflow Result:")
    print(f"  Output: {result1['output'][:150]}...")
    
    # Execute with specific agents
    result2 = pattern.execute("Specific task for Agent1", agent_name="Agent1")
    print(f"\nAgent1 Result:")
    print(f"  Output: {result2['output'][:150]}...")
    
    result3 = pattern.execute("Specific task for Agent2", agent_name="Agent2")
    print(f"\nAgent2 Result:")
    print(f"  Output: {result3['output'][:150]}...")
    
    # Show statistics
    print("\n=== Statistics ===")
    stats = pattern.get_statistics()
    print(f"Pattern: {stats['pattern']}")
    print(f"Total Executions: {stats['total_executions']}")
    print(f"Registered Agents: {stats['registered_agents']}")
    print(f"\nAgent Statistics:")
    for agent_stats in stats['agents']:
        print(f"  {agent_stats['name']}: {agent_stats['total_executions']} executions")

Agentic Architecture Pattern: Self Consistency

Pattern 197: 197 Self Consistency Pattern – Line-by-Line Explanation

Overview

Pattern 197: Self Consistency Pattern

Purpose: Enhanced implementation with full LangChain/LangGraph integration

Code Structure Analysis

File Information

Pattern Number: 197
Pattern Name: 197 Self Consistency Pattern
File Path: patterns\llm_specific\197_self_consistency_pattern.py
Total Lines: 247

Detailed Code Explanation

Module Docstring

"""
Pattern 197: Self Consistency Pattern

Purpose: Enhanced implementation with full LangChain/LangGraph integration
"""

Explanation: This module docstring describes the purpose and key features of the 197 Self Consistency Pattern pattern.

Imports Section

from typing import Dict, Any, List, Optional
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
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 enum import Enum: Enumeration support for fixed sets of values – 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

SelfConsistencyConfig (Line 18)

Description: Configuration for Self Consistency Pattern.

SelfConsistencyState (Line 27)

Description: State information for Self Consistency Pattern.

SelfConsistencyAgent (Line 36)

Description: Agent implementing Self Consistency Pattern.

Methods:

`init` (Line 39)

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

Purpose: No method docstring.

Parameters: – self: Instance reference – name: Identifier or label – config: Configuration settings

`process` (Line 49)

def process(self, input_data):
    """Process input according to Self Consistency Pattern."""

Purpose: Process input according to Self Consistency Pattern.

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

`get_statistics` (Line 75)

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

Purpose: Get agent statistics.

Parameters: – self: Instance reference

SelfConsistencyPattern (Line 85)

Description: Self Consistency Pattern – Enhanced implementation

This pattern provides comprehensive functionality for self consistency with full LangChain/LangGraph integration.

Features: – Multi-agent coordination – State management – Execution tracking – LangGraph workflow integration

Methods:

`init` (Line 99)

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

Purpose: No method docstring.

Parameters: – self: Instance reference – config: Configuration settings

`_build_workflow` (Line 105)

def _build_workflow(self):
    """Build LangGraph workflow for pattern execution."""

Purpose: Build LangGraph workflow for pattern execution.

Parameters: – self: Instance reference

`register_agent` (Line 150)

def register_agent(self, agent):
    """Register an agent with the pattern."""

Purpose: Register an agent with the pattern.

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

`execute` (Line 156)

def execute(self, input_data, agent_name):
    """Execute the pattern with given input."""

Purpose: Execute the pattern with given input.

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

`run` (Line 188)

def run(self, input_data):
    """Main execution method."""

Purpose: Main execution method.

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

`get_statistics` (Line 192)

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

Purpose: Get pattern statistics.

Parameters: – self: Instance reference

WorkflowState (Line 108)

Description: No class docstring.

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

November 4, 2025

Agentic Architecture Pattern: Constitutional Ai


    """
Pattern 196: Constitutional Ai Pattern

Purpose: Enhanced implementation with full LangChain/LangGraph integration
"""

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


@dataclass
class ConstitutionalAiConfig:
    """Configuration for Constitutional Ai Pattern."""
    model_name: str = "gpt-4"
    temperature: float = 0.7
    max_iterations: int = 10
    enable_logging: bool = True


@dataclass
class ConstitutionalAiState:
    """State information for Constitutional Ai Pattern."""
    current_step: int = 0
    data: Dict[str, Any] = field(default_factory=dict)
    results: List[Any] = field(default_factory=list)
    metadata: Dict[str, Any] = field(default_factory=dict)
    timestamp: datetime = field(default_factory=datetime.now)


class ConstitutionalAiAgent:
    """Agent implementing Constitutional Ai Pattern."""
    
    def __init__(self, name: str, config: ConstitutionalAiConfig = None):
        self.name = name
        self.config = config or ConstitutionalAiConfig()
        self.llm = ChatOpenAI(
            model=self.config.model_name,
            temperature=self.config.temperature
        )
        self.state = ConstitutionalAiState()
        self.execution_history: List[Dict[str, Any]] = []
    
    def process(self, input_data: Any) -> Dict[str, Any]:
        """Process input according to Constitutional Ai Pattern."""
        self.state.current_step += 1
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"You are {self.name}, implementing Constitutional Ai Pattern. "
                      f"Provide clear, structured responses."),
            ("human", "Process this input: {input_data}")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({"input_data": str(input_data)})
        
        result = {
            "agent": self.name,
            "step": self.state.current_step,
            "input": input_data,
            "output": str(response.content),
            "timestamp": datetime.now()
        }
        
        self.execution_history.append(result)
        self.state.results.append(result)
        
        return result
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get agent statistics."""
        return {
            "name": self.name,
            "total_executions": len(self.execution_history),
            "current_step": self.state.current_step,
            "results_count": len(self.state.results)
        }


class ConstitutionalAiPattern:
    """
    Constitutional Ai Pattern - Enhanced implementation
    
    This pattern provides comprehensive functionality for constitutional ai
    with full LangChain/LangGraph integration.
    
    Features:
    - Multi-agent coordination
    - State management
    - Execution tracking
    - LangGraph workflow integration
    """
    
    def __init__(self, config: ConstitutionalAiConfig = None):
        self.config = config or ConstitutionalAiConfig()
        self.agents: Dict[str, ConstitutionalAiAgent] = {}
        self.workflow = self._build_workflow()
        self.execution_log: List[Dict[str, Any]] = []
    
    def _build_workflow(self) -> StateGraph:
        """Build LangGraph workflow for pattern execution."""
        
        class WorkflowState(TypedDict):
            input: str
            processed: str
            output: str
            metadata: Dict[str, Any]
        
        def process_input(state: WorkflowState) -> WorkflowState:
            """Process input node."""
            return {
                **state,
                "processed": f"Processed: {state['input']}",
                "metadata": {"step": "process_input"}
            }
        
        def generate_output(state: WorkflowState) -> WorkflowState:
            """Generate output node."""
            llm = ChatOpenAI(model=self.config.model_name, temperature=self.config.temperature)
            
            prompt = ChatPromptTemplate.from_messages([
                ("system", "You are implementing Constitutional Ai Pattern. Generate comprehensive output."),
                ("human", "{input}")
            ])
            
            chain = prompt | llm
            response = chain.invoke({"input": state["processed"]})
            
            return {
                **state,
                "output": str(response.content),
                "metadata": {**state.get("metadata", {}), "step": "generate_output"}
            }
        
        # Build workflow graph
        workflow = StateGraph(WorkflowState)
        workflow.add_node("process", process_input)
        workflow.add_node("generate", generate_output)
        workflow.set_entry_point("process")
        workflow.add_edge("process", "generate")
        workflow.add_edge("generate", END)
        
        return workflow.compile()
    
    def register_agent(self, agent: ConstitutionalAiAgent):
        """Register an agent with the pattern."""
        self.agents[agent.name] = agent
        if self.config.enable_logging:
            print(f"Registered agent: {agent.name}")
    
    def execute(self, input_data: Any, agent_name: str = None) -> Dict[str, Any]:
        """Execute the pattern with given input."""
        
        # Use workflow if no specific agent requested
        if agent_name is None:
            result = self.workflow.invoke({
                "input": str(input_data),
                "processed": "",
                "output": "",
                "metadata": {}
            })
            
            execution_record = {
                "pattern": "Constitutional Ai Pattern",
                "input": input_data,
                "output": result.get("output", ""),
                "metadata": result.get("metadata", {}),
                "timestamp": datetime.now()
            }
            
            self.execution_log.append(execution_record)
            return execution_record
        
        # Execute with specific agent
        agent = self.agents.get(agent_name)
        if not agent:
            raise ValueError(f"Agent {agent_name} not found")
        
        result = agent.process(input_data)
        self.execution_log.append(result)
        return result
    
    def run(self, input_data: Any) -> Dict[str, Any]:
        """Main execution method."""
        return self.execute(input_data)
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get pattern statistics."""
        return {
            "pattern": "Constitutional Ai Pattern",
            "total_executions": len(self.execution_log),
            "registered_agents": len(self.agents),
            "agents": [agent.get_statistics() for agent in self.agents.values()]
        }


# Example usage
if __name__ == "__main__":
    print("=== Constitutional Ai Pattern Demo ===\n")
    
    # Create configuration
    config = ConstitutionalAiConfig(
        model_name="gpt-4",
        temperature=0.7,
        enable_logging=True
    )
    
    # Create pattern
    pattern = ConstitutionalAiPattern(config)
    
    # Create and register agents
    agent1 = ConstitutionalAiAgent("Agent1", config)
    agent2 = ConstitutionalAiAgent("Agent2", config)
    pattern.register_agent(agent1)
    pattern.register_agent(agent2)
    
    print("\nExecuting pattern workflow...")
    
    # Execute with workflow
    result1 = pattern.run("Demonstrate constitutional ai functionality")
    print(f"\nWorkflow Result:")
    print(f"  Output: {result1['output'][:150]}...")
    
    # Execute with specific agents
    result2 = pattern.execute("Specific task for Agent1", agent_name="Agent1")
    print(f"\nAgent1 Result:")
    print(f"  Output: {result2['output'][:150]}...")
    
    result3 = pattern.execute("Specific task for Agent2", agent_name="Agent2")
    print(f"\nAgent2 Result:")
    print(f"  Output: {result3['output'][:150]}...")
    
    # Show statistics
    print("\n=== Statistics ===")
    stats = pattern.get_statistics()
    print(f"Pattern: {stats['pattern']}")
    print(f"Total Executions: {stats['total_executions']}")
    print(f"Registered Agents: {stats['registered_agents']}")
    print(f"\nAgent Statistics:")
    for agent_stats in stats['agents']:
        print(f"  {agent_stats['name']}: {agent_stats['total_executions']} executions")

Agentic Architecture Pattern: Constitutional Ai

Pattern 196: 196 Constitutional Ai Pattern – Line-by-Line Explanation

Overview

Pattern 196: Constitutional Ai Pattern

Purpose: Enhanced implementation with full LangChain/LangGraph integration

Code Structure Analysis

File Information

Pattern Number: 196
Pattern Name: 196 Constitutional Ai Pattern
File Path: patterns\llm_specific\196_constitutional_ai_pattern.py
Total Lines: 247

Detailed Code Explanation

Module Docstring

"""
Pattern 196: Constitutional Ai Pattern

Purpose: Enhanced implementation with full LangChain/LangGraph integration
"""

Explanation: This module docstring describes the purpose and key features of the 196 Constitutional Ai Pattern pattern.

Imports Section

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

Classes

ConstitutionalAiConfig (Line 18)

Description: Configuration for Constitutional Ai Pattern.

ConstitutionalAiState (Line 27)

Description: State information for Constitutional Ai Pattern.

ConstitutionalAiAgent (Line 36)

Description: Agent implementing Constitutional Ai Pattern.

Methods:

`init` (Line 39)

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

Purpose: No method docstring.

Parameters: – self: Instance reference – name: Identifier or label – config: Configuration settings

`process` (Line 49)

def process(self, input_data):
    """Process input according to Constitutional Ai Pattern."""

Purpose: Process input according to Constitutional Ai Pattern.

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

`get_statistics` (Line 75)

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

Purpose: Get agent statistics.

Parameters: – self: Instance reference

ConstitutionalAiPattern (Line 85)

Description: Constitutional Ai Pattern – Enhanced implementation

This pattern provides comprehensive functionality for constitutional ai with full LangChain/LangGraph integration.

Features: – Multi-agent coordination – State management – Execution tracking – LangGraph workflow integration

Methods:

`init` (Line 99)

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

Purpose: No method docstring.

Parameters: – self: Instance reference – config: Configuration settings

`_build_workflow` (Line 105)

def _build_workflow(self):
    """Build LangGraph workflow for pattern execution."""

Purpose: Build LangGraph workflow for pattern execution.

Parameters: – self: Instance reference

`register_agent` (Line 150)

def register_agent(self, agent):
    """Register an agent with the pattern."""

Purpose: Register an agent with the pattern.

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

`execute` (Line 156)

def execute(self, input_data, agent_name):
    """Execute the pattern with given input."""

Purpose: Execute the pattern with given input.

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

`run` (Line 188)

def run(self, input_data):
    """Main execution method."""

Purpose: Main execution method.

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

`get_statistics` (Line 192)

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

Purpose: Get pattern statistics.

Parameters: – self: Instance reference

WorkflowState (Line 108)

Description: No class docstring.

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

November 4, 2025

Agentic Architecture Pattern: Prompt Chaining


    """
Pattern 195: Prompt Chaining Pattern

Purpose: Enhanced implementation with full LangChain/LangGraph integration
"""

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


@dataclass
class PromptChainingConfig:
    """Configuration for Prompt Chaining Pattern."""
    model_name: str = "gpt-4"
    temperature: float = 0.7
    max_iterations: int = 10
    enable_logging: bool = True


@dataclass
class PromptChainingState:
    """State information for Prompt Chaining Pattern."""
    current_step: int = 0
    data: Dict[str, Any] = field(default_factory=dict)
    results: List[Any] = field(default_factory=list)
    metadata: Dict[str, Any] = field(default_factory=dict)
    timestamp: datetime = field(default_factory=datetime.now)


class PromptChainingAgent:
    """Agent implementing Prompt Chaining Pattern."""
    
    def __init__(self, name: str, config: PromptChainingConfig = None):
        self.name = name
        self.config = config or PromptChainingConfig()
        self.llm = ChatOpenAI(
            model=self.config.model_name,
            temperature=self.config.temperature
        )
        self.state = PromptChainingState()
        self.execution_history: List[Dict[str, Any]] = []
    
    def process(self, input_data: Any) -> Dict[str, Any]:
        """Process input according to Prompt Chaining Pattern."""
        self.state.current_step += 1
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"You are {self.name}, implementing Prompt Chaining Pattern. "
                      f"Provide clear, structured responses."),
            ("human", "Process this input: {input_data}")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({"input_data": str(input_data)})
        
        result = {
            "agent": self.name,
            "step": self.state.current_step,
            "input": input_data,
            "output": str(response.content),
            "timestamp": datetime.now()
        }
        
        self.execution_history.append(result)
        self.state.results.append(result)
        
        return result
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get agent statistics."""
        return {
            "name": self.name,
            "total_executions": len(self.execution_history),
            "current_step": self.state.current_step,
            "results_count": len(self.state.results)
        }


class PromptChainingPattern:
    """
    Prompt Chaining Pattern - Enhanced implementation
    
    This pattern provides comprehensive functionality for prompt chaining
    with full LangChain/LangGraph integration.
    
    Features:
    - Multi-agent coordination
    - State management
    - Execution tracking
    - LangGraph workflow integration
    """
    
    def __init__(self, config: PromptChainingConfig = None):
        self.config = config or PromptChainingConfig()
        self.agents: Dict[str, PromptChainingAgent] = {}
        self.workflow = self._build_workflow()
        self.execution_log: List[Dict[str, Any]] = []
    
    def _build_workflow(self) -> StateGraph:
        """Build LangGraph workflow for pattern execution."""
        
        class WorkflowState(TypedDict):
            input: str
            processed: str
            output: str
            metadata: Dict[str, Any]
        
        def process_input(state: WorkflowState) -> WorkflowState:
            """Process input node."""
            return {
                **state,
                "processed": f"Processed: {state['input']}",
                "metadata": {"step": "process_input"}
            }
        
        def generate_output(state: WorkflowState) -> WorkflowState:
            """Generate output node."""
            llm = ChatOpenAI(model=self.config.model_name, temperature=self.config.temperature)
            
            prompt = ChatPromptTemplate.from_messages([
                ("system", "You are implementing Prompt Chaining Pattern. Generate comprehensive output."),
                ("human", "{input}")
            ])
            
            chain = prompt | llm
            response = chain.invoke({"input": state["processed"]})
            
            return {
                **state,
                "output": str(response.content),
                "metadata": {**state.get("metadata", {}), "step": "generate_output"}
            }
        
        # Build workflow graph
        workflow = StateGraph(WorkflowState)
        workflow.add_node("process", process_input)
        workflow.add_node("generate", generate_output)
        workflow.set_entry_point("process")
        workflow.add_edge("process", "generate")
        workflow.add_edge("generate", END)
        
        return workflow.compile()
    
    def register_agent(self, agent: PromptChainingAgent):
        """Register an agent with the pattern."""
        self.agents[agent.name] = agent
        if self.config.enable_logging:
            print(f"Registered agent: {agent.name}")
    
    def execute(self, input_data: Any, agent_name: str = None) -> Dict[str, Any]:
        """Execute the pattern with given input."""
        
        # Use workflow if no specific agent requested
        if agent_name is None:
            result = self.workflow.invoke({
                "input": str(input_data),
                "processed": "",
                "output": "",
                "metadata": {}
            })
            
            execution_record = {
                "pattern": "Prompt Chaining Pattern",
                "input": input_data,
                "output": result.get("output", ""),
                "metadata": result.get("metadata", {}),
                "timestamp": datetime.now()
            }
            
            self.execution_log.append(execution_record)
            return execution_record
        
        # Execute with specific agent
        agent = self.agents.get(agent_name)
        if not agent:
            raise ValueError(f"Agent {agent_name} not found")
        
        result = agent.process(input_data)
        self.execution_log.append(result)
        return result
    
    def run(self, input_data: Any) -> Dict[str, Any]:
        """Main execution method."""
        return self.execute(input_data)
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get pattern statistics."""
        return {
            "pattern": "Prompt Chaining Pattern",
            "total_executions": len(self.execution_log),
            "registered_agents": len(self.agents),
            "agents": [agent.get_statistics() for agent in self.agents.values()]
        }


# Example usage
if __name__ == "__main__":
    print("=== Prompt Chaining Pattern Demo ===\n")
    
    # Create configuration
    config = PromptChainingConfig(
        model_name="gpt-4",
        temperature=0.7,
        enable_logging=True
    )
    
    # Create pattern
    pattern = PromptChainingPattern(config)
    
    # Create and register agents
    agent1 = PromptChainingAgent("Agent1", config)
    agent2 = PromptChainingAgent("Agent2", config)
    pattern.register_agent(agent1)
    pattern.register_agent(agent2)
    
    print("\nExecuting pattern workflow...")
    
    # Execute with workflow
    result1 = pattern.run("Demonstrate prompt chaining functionality")
    print(f"\nWorkflow Result:")
    print(f"  Output: {result1['output'][:150]}...")
    
    # Execute with specific agents
    result2 = pattern.execute("Specific task for Agent1", agent_name="Agent1")
    print(f"\nAgent1 Result:")
    print(f"  Output: {result2['output'][:150]}...")
    
    result3 = pattern.execute("Specific task for Agent2", agent_name="Agent2")
    print(f"\nAgent2 Result:")
    print(f"  Output: {result3['output'][:150]}...")
    
    # Show statistics
    print("\n=== Statistics ===")
    stats = pattern.get_statistics()
    print(f"Pattern: {stats['pattern']}")
    print(f"Total Executions: {stats['total_executions']}")
    print(f"Registered Agents: {stats['registered_agents']}")
    print(f"\nAgent Statistics:")
    for agent_stats in stats['agents']:
        print(f"  {agent_stats['name']}: {agent_stats['total_executions']} executions")

Agentic Architecture Pattern: Prompt Chaining

Pattern 195: 195 Prompt Chaining Pattern – Line-by-Line Explanation

Overview

Pattern 195: Prompt Chaining Pattern

Purpose: Enhanced implementation with full LangChain/LangGraph integration

Code Structure Analysis

File Information

Pattern Number: 195
Pattern Name: 195 Prompt Chaining Pattern
File Path: patterns\llm_specific\195_prompt_chaining_pattern.py
Total Lines: 247

Detailed Code Explanation

Module Docstring

"""
Pattern 195: Prompt Chaining Pattern

Purpose: Enhanced implementation with full LangChain/LangGraph integration
"""

Explanation: This module docstring describes the purpose and key features of the 195 Prompt Chaining Pattern pattern.

Imports Section

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

Classes

PromptChainingConfig (Line 18)

Description: Configuration for Prompt Chaining Pattern.

PromptChainingState (Line 27)

Description: State information for Prompt Chaining Pattern.

PromptChainingAgent (Line 36)

Description: Agent implementing Prompt Chaining Pattern.

Methods:

`init` (Line 39)

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

Purpose: No method docstring.

Parameters: – self: Instance reference – name: Identifier or label – config: Configuration settings

`process` (Line 49)

def process(self, input_data):
    """Process input according to Prompt Chaining Pattern."""

Purpose: Process input according to Prompt Chaining Pattern.

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

`get_statistics` (Line 75)

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

Purpose: Get agent statistics.

Parameters: – self: Instance reference

PromptChainingPattern (Line 85)

Description: Prompt Chaining Pattern – Enhanced implementation

This pattern provides comprehensive functionality for prompt chaining with full LangChain/LangGraph integration.

Features: – Multi-agent coordination – State management – Execution tracking – LangGraph workflow integration

Methods:

`init` (Line 99)

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

Purpose: No method docstring.

Parameters: – self: Instance reference – config: Configuration settings

`_build_workflow` (Line 105)

def _build_workflow(self):
    """Build LangGraph workflow for pattern execution."""

Purpose: Build LangGraph workflow for pattern execution.

Parameters: – self: Instance reference

`register_agent` (Line 150)

def register_agent(self, agent):
    """Register an agent with the pattern."""

Purpose: Register an agent with the pattern.

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

`execute` (Line 156)

def execute(self, input_data, agent_name):
    """Execute the pattern with given input."""

Purpose: Execute the pattern with given input.

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

`run` (Line 188)

def run(self, input_data):
    """Main execution method."""

Purpose: Main execution method.

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

`get_statistics` (Line 192)

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

Purpose: Get pattern statistics.

Parameters: – self: Instance reference

WorkflowState (Line 108)

Description: No class docstring.

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

November 4, 2025

Agentic Architecture Pattern: Chain Prompting


    """
Pattern 194: Chain Prompting Pattern

Breaks complex tasks into a sequence of simpler prompts where each
prompt builds on previous outputs. Creates dependency chain of prompts.
"""

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


@dataclass
class PromptStep:
    """Single step in prompt chain."""
    name: str
    prompt_template: str
    depends_on: List[str] = field(default_factory=list)
    output: Optional[str] = None
    timestamp: Optional[datetime] = None


@dataclass
class ChainExecution:
    """Execution trace of prompt chain."""
    steps: List[PromptStep]
    inputs: Dict[str, Any]
    outputs: Dict[str, str]
    execution_order: List[str]
    total_time: float = 0.0


class ChainPromptingAgent:
    """Agent that executes prompt chains."""
    
    def __init__(self, name: str, model_name: str = "gpt-4"):
        self.name = name
        self.llm = ChatOpenAI(model=model_name, temperature=0.7)
        self.steps: Dict[str, PromptStep] = {}
        self.executions: List[ChainExecution] = []
    
    def add_step(self, step: PromptStep):
        """Add step to chain."""
        self.steps[step.name] = step
        print(f"  Added step: {step.name}")
    
    def _resolve_dependencies(self) -> List[str]:
        """Resolve execution order based on dependencies."""
        executed = set()
        execution_order = []
        
        def can_execute(step_name: str) -> bool:
            step = self.steps[step_name]
            return all(dep in executed for dep in step.depends_on)
        
        while len(executed)  ChainExecution:
        """Execute the prompt chain."""
        print(f"\n  {self.name}: Executing prompt chain...")
        start_time = datetime.now()
        
        # Resolve execution order
        execution_order = self._resolve_dependencies()
        print(f"  Execution order: {' → '.join(execution_order)}")
        
        # Execute steps in order
        outputs = {}
        executed_steps = []
        
        for step_name in execution_order:
            step = self.steps[step_name]
            print(f"\n  Executing: {step_name}")
            
            # Prepare input by gathering dependencies
            step_input = initial_input.copy()
            for dep in step.depends_on:
                if dep in outputs:
                    step_input[dep] = outputs[dep]
            
            # Execute step
            prompt = ChatPromptTemplate.from_messages([
                ("system", "You are part of a multi-step reasoning chain."),
                ("human", step.prompt_template)
            ])
            
            chain = prompt | self.llm
            response = chain.invoke(step_input)
            
            output = response.content
            outputs[step_name] = output
            
            step.output = output
            step.timestamp = datetime.now()
            executed_steps.append(step)
            
            print(f"  Output: {output[:100]}...")
        
        end_time = datetime.now()
        execution = ChainExecution(
            steps=executed_steps,
            inputs=initial_input,
            outputs=outputs,
            execution_order=execution_order,
            total_time=(end_time - start_time).total_seconds()
        )
        
        self.executions.append(execution)
        return execution


class ChainPromptingPattern:
    """
    Chain Prompting Pattern - Sequential prompt dependencies.
    
    Breaks complex tasks into dependent steps where each step
    uses outputs from previous steps.
    """
    
    def __init__(self):
        self.agents: Dict[str, ChainPromptingAgent] = {}
    
    def register_agent(self, agent: ChainPromptingAgent):
        """Register chain prompting agent."""
        self.agents[agent.name] = agent
    
    def run(self, initial_inputs: Dict[str, Any]) -> Dict[str, Any]:
        """Execute prompt chains."""
        print("=== Chain Prompting Pattern ===\n")
        
        results = {}
        for agent in self.agents.values():
            execution = agent.execute_chain(initial_inputs)
            results[agent.name] = execution.outputs
        
        return {
            "pattern": "ChainPrompting",
            "agents": len(self.agents),
            "results": results,
            "total_steps": sum(len(a.steps) for a in self.agents.values())
        }


# Example usage
if __name__ == "__main__":
    print("=== Chain Prompting Pattern Demo ===\n")
    
    # Create pattern and agent
    pattern = ChainPromptingPattern()
    agent = ChainPromptingAgent("ArticleWriter")
    
    # Build prompt chain for article writing
    print("1. Building prompt chain...\n")
    
    # Step 1: Research (no dependencies)
    agent.add_step(PromptStep(
        name="research",
        prompt_template="Research key facts about: {topic}. List 5 important facts.",
        depends_on=[]
    ))
    
    # Step 2: Outline (depends on research)
    agent.add_step(PromptStep(
        name="outline",
        prompt_template="Create an article outline about {topic} using these facts:\n{research}",
        depends_on=["research"]
    ))
    
    # Step 3: Introduction (depends on outline)
    agent.add_step(PromptStep(
        name="introduction",
        prompt_template="Write an engaging introduction for this article:\nTopic: {topic}\nOutline: {outline}",
        depends_on=["outline"]
    ))
    
    # Step 4: Body (depends on outline and research)
    agent.add_step(PromptStep(
        name="body",
        prompt_template="Write the main body using:\nOutline: {outline}\nFacts: {research}",
        depends_on=["outline", "research"]
    ))
    
    # Step 5: Conclusion (depends on all previous)
    agent.add_step(PromptStep(
        name="conclusion",
        prompt_template="Write a conclusion that ties together:\nIntro: {introduction}\nBody: {body}",
        depends_on=["introduction", "body"]
    ))
    
    pattern.register_agent(agent)
    
    # Execute chain
    print("\n2. Executing chain...\n")
    summary = pattern.run({"topic": "Artificial Intelligence in Healthcare"})
    
    print(f"\n3. Summary:")
    print(f"   Total steps: {summary['total_steps']}")
    print(f"   Steps executed: {len(agent.executions[0].execution_order)}")
    
    print(f"\n4. Chain Outputs:")
    for step_name, output in agent.executions[0].outputs.items():
        print(f"\n   {step_name}:")
        print(f"   {output[:150]}...")
    
    print(f"\n   Total execution time: {agent.executions[0].total_time:.2f}s")
    
    print("\n✓ Chain prompting: sequential dependent prompts!")

Agentic Architecture Pattern: Chain Prompting

Pattern 194: 194 Chain Prompting Pattern – Line-by-Line Explanation

Overview

Pattern 194: Chain Prompting Pattern

Breaks complex tasks into a sequence of simpler prompts where each prompt builds on previous outputs. Creates dependency chain of prompts.

Code Structure Analysis

File Information

Pattern Number: 194
Pattern Name: 194 Chain Prompting Pattern
File Path: patterns\llm_specific\194_chain_prompting_pattern.py
Total Lines: 221

Detailed Code Explanation

Module Docstring

"""
Pattern 194: Chain Prompting Pattern

Breaks complex tasks into a sequence of simpler prompts where each
prompt builds on previous outputs. Creates dependency chain of prompts.
"""

Explanation: This module docstring describes the purpose and key features of the 194 Chain Prompting Pattern 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  # First 20 imports shown

Classes

PromptStep (Line 16)

Description: Single step in prompt chain.

ChainExecution (Line 26)

Description: Execution trace of prompt chain.

ChainPromptingAgent (Line 35)

Description: Agent that executes prompt chains.

Methods:

`init` (Line 38)

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

`add_step` (Line 44)

def add_step(self, step):
    """Add step to chain."""

Purpose: Add step to chain.

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

`_resolve_dependencies` (Line 49)

def _resolve_dependencies(self):
    """Resolve execution order based on dependencies."""

Purpose: Resolve execution order based on dependencies.

Parameters: – self: Instance reference

`execute_chain` (Line 71)

def execute_chain(self, initial_input):
    """Execute the prompt chain."""

Purpose: Execute the prompt chain.

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

ChainPromptingPattern (Line 125)

Description: Chain Prompting Pattern – Sequential prompt dependencies.

Breaks complex tasks into dependent steps where each step uses outputs from previous steps.

Methods:

`init` (Line 133)

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

Purpose: No method docstring.

Parameters: – self: Instance reference

`register_agent` (Line 136)

def register_agent(self, agent):
    """Register chain prompting agent."""

Purpose: Register chain prompting agent.

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

`run` (Line 140)

def run(self, initial_inputs):
    """Execute prompt chains."""

Purpose: Execute prompt chains.

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

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

November 4, 2025

Agentic Architecture Pattern: Zero Shot Learning


    """
Pattern 193: Zero-Shot Learning Pattern

LLM performs tasks without any training examples, relying solely on pre-trained knowledge.
Demonstrates ability to generalize to new tasks using instructions alone.
"""

from typing import Dict, Any, List, Optional
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI


class TaskCategory(Enum):
    """Categories of zero-shot tasks."""
    CLASSIFICATION = "classification"
    GENERATION = "generation"
    TRANSLATION = "translation"
    SUMMARIZATION = "summarization"
    QUESTION_ANSWERING = "question_answering"
    REASONING = "reasoning"
    EXTRACTION = "extraction"


@dataclass
class ZeroShotTask:
    """A task to be performed without examples."""
    task_name: str
    category: TaskCategory
    instruction: str
    expected_output_format: Optional[str] = None
    constraints: List[str] = field(default_factory=list)


@dataclass
class ZeroShotResult:
    """Result from zero-shot execution."""
    task_name: str
    category: TaskCategory
    input: str
    output: str
    instruction_used: str
    confidence_estimate: Optional[float] = None
    timestamp: datetime = field(default_factory=datetime.now)


class ZeroShotAgent:
    """Agent that performs tasks without examples."""
    
    def __init__(self, name: str, model_name: str = "gpt-4", temperature: float = 0.7):
        self.name = name
        self.llm = ChatOpenAI(model=model_name, temperature=temperature)
        self.task_registry: Dict[str, ZeroShotTask] = {}
        self.results: List[ZeroShotResult] = []
    
    def register_task(self, task: ZeroShotTask):
        """Register a task the agent can perform."""
        self.task_registry[task.task_name] = task
        print(f"📋 {self.name} registered task: {task.task_name} ({task.category.value})")
    
    def create_zero_shot_prompt(self, task: ZeroShotTask, input_text: str) -> str:
        """Create prompt for zero-shot execution."""
        prompt_parts = []
        
        # Task instruction
        prompt_parts.append(f"Task: {task.instruction}")
        
        # Output format if specified
        if task.expected_output_format:
            prompt_parts.append(f"\nExpected Output Format: {task.expected_output_format}")
        
        # Constraints if any
        if task.constraints:
            prompt_parts.append("\nConstraints:")
            for constraint in task.constraints:
                prompt_parts.append(f"- {constraint}")
        
        # The actual input
        prompt_parts.append(f"\nInput: {input_text}")
        prompt_parts.append("\nOutput:")
        
        return "\n".join(prompt_parts)
    
    def execute_task(self, task_name: str, input_text: str) -> ZeroShotResult:
        """Execute a registered task on input."""
        if task_name not in self.task_registry:
            raise ValueError(f"Task '{task_name}' not registered")
        
        task = self.task_registry[task_name]
        prompt_text = self.create_zero_shot_prompt(task, input_text)
        
        # Execute with LLM
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"You are {self.name}, a capable AI assistant. "
                      f"Follow the instructions precisely and provide high-quality output."),
            ("human", "{prompt}")
        ])
        
        chain = prompt | self.llm | StrOutputParser()
        output = chain.invoke({"prompt": prompt_text})
        
        result = ZeroShotResult(
            task_name=task_name,
            category=task.category,
            input=input_text,
            output=output,
            instruction_used=task.instruction
        )
        
        self.results.append(result)
        return result
    
    def execute_adhoc(self, instruction: str, input_text: str, 
                     category: TaskCategory = TaskCategory.GENERATION) -> ZeroShotResult:
        """Execute an ad-hoc task without prior registration."""
        adhoc_task = ZeroShotTask(
            task_name=f"adhoc_{len(self.results)}",
            category=category,
            instruction=instruction
        )
        
        prompt_text = self.create_zero_shot_prompt(adhoc_task, input_text)
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"You are {self.name}. Follow instructions and provide quality output."),
            ("human", "{prompt}")
        ])
        
        chain = prompt | self.llm | StrOutputParser()
        output = chain.invoke({"prompt": prompt_text})
        
        result = ZeroShotResult(
            task_name=adhoc_task.task_name,
            category=category,
            input=input_text,
            output=output,
            instruction_used=instruction
        )
        
        self.results.append(result)
        return result
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get agent statistics."""
        category_counts = {}
        for result in self.results:
            cat = result.category.value
            category_counts[cat] = category_counts.get(cat, 0) + 1
        
        return {
            "name": self.name,
            "registered_tasks": len(self.task_registry),
            "total_executions": len(self.results),
            "tasks_by_category": category_counts
        }


class ZeroShotLearningPattern:
    """
    Zero-Shot Learning Pattern - Task execution without training examples.
    
    Leverages LLM's pre-trained knowledge to perform tasks based solely
    on natural language instructions, without any demonstration examples.
    """
    
    def __init__(self, model_name: str = "gpt-4"):
        self.model_name = model_name
        self.agents: Dict[str, ZeroShotAgent] = {}
        self.task_library: Dict[str, ZeroShotTask] = {}
    
    def create_agent(self, name: str) -> ZeroShotAgent:
        """Create a zero-shot capable agent."""
        agent = ZeroShotAgent(name, self.model_name)
        self.agents[name] = agent
        return agent
    
    def register_task_template(self, task: ZeroShotTask):
        """Register a task template in the library."""
        self.task_library[task.task_name] = task
        print(f"📚 Task template registered: {task.task_name}")
    
    def create_task(self, task_name: str, category: TaskCategory, 
                   instruction: str, **kwargs) -> ZeroShotTask:
        """Create and register a task template."""
        task = ZeroShotTask(
            task_name=task_name,
            category=category,
            instruction=instruction,
            expected_output_format=kwargs.get('expected_output_format'),
            constraints=kwargs.get('constraints', [])
        )
        self.register_task_template(task)
        return task
    
    def assign_task_to_agent(self, agent_name: str, task_name: str):
        """Assign a task from library to an agent."""
        if agent_name not in self.agents:
            raise ValueError(f"Agent '{agent_name}' not found")
        if task_name not in self.task_library:
            raise ValueError(f"Task '{task_name}' not in library")
        
        agent = self.agents[agent_name]
        task = self.task_library[task_name]
        agent.register_task(task)
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get pattern statistics."""
        return {
            "total_agents": len(self.agents),
            "task_templates": len(self.task_library),
            "agents": {name: agent.get_statistics() for name, agent in self.agents.items()}
        }


# Example usage
if __name__ == "__main__":
    print("=== Zero-Shot Learning Pattern Demo ===\n")
    
    # Create pattern
    print("1. Creating zero-shot learning pattern...")
    pattern = ZeroShotLearningPattern()
    
    # Create agent
    agent = pattern.create_agent("ZeroShotMaster")
    
    # Task 1: Text Classification
    print(f"\n{'='*70}")
    print("📝 Task 1: Text Classification (Zero-Shot)")
    print(f"{'='*70}\n")
    
    classification_task = pattern.create_task(
        task_name="sentiment_classification",
        category=TaskCategory.CLASSIFICATION,
        instruction="Classify the sentiment of the given text as Positive, Negative, or Neutral.",
        expected_output_format="Single word: Positive, Negative, or Neutral",
        constraints=["Provide only the classification", "No explanation needed"]
    )
    
    pattern.assign_task_to_agent("ZeroShotMaster", "sentiment_classification")
    
    test_texts = [
        "This is absolutely fantastic! Best purchase ever!",
        "Completely disappointed. Waste of money.",
        "It's okay, does what it's supposed to do."
    ]
    
    print("Testing zero-shot classification:\n")
    for text in test_texts:
        result = agent.execute_task("sentiment_classification", text)
        print(f"Text: {result.input}")
        print(f"Classification: {result.output.strip()}\n")
    
    # Task 2: Question Answering
    print(f"\n{'='*70}")
    print("❓ Task 2: Question Answering (Zero-Shot)")
    print(f"{'='*70}\n")
    
    qa_task = pattern.create_task(
        task_name="question_answering",
        category=TaskCategory.QUESTION_ANSWERING,
        instruction="Answer the question concisely based on general knowledge.",
        expected_output_format="2-3 sentences maximum",
        constraints=["Be factual and accurate", "Cite key points only"]
    )
    
    pattern.assign_task_to_agent("ZeroShotMaster", "question_answering")
    
    questions = [
        "What is the capital of France?",
        "How does photosynthesis work?",
        "Who wrote Romeo and Juliet?"
    ]
    
    print("Testing zero-shot Q&A:\n")
    for question in questions:
        result = agent.execute_task("question_answering", question)
        print(f"Q: {result.input}")
        print(f"A: {result.output.strip()}\n")
    
    # Task 3: Ad-hoc Execution (no prior registration)
    print(f"\n{'='*70}")
    print("🚀 Task 3: Ad-hoc Execution (Zero-Shot)")
    print(f"{'='*70}\n")
    
    adhoc_result = agent.execute_adhoc(
        instruction="Extract all email addresses from the text",
        input_text="Contact us at support@example.com or sales@company.org for more info.",
        category=TaskCategory.EXTRACTION
    )
    
    print(f"Instruction: {adhoc_result.instruction_used}")
    print(f"Input: {adhoc_result.input}")
    print(f"Extracted: {adhoc_result.output.strip()}\n")
    
    # Task 4: Translation
    print(f"\n{'='*70}")
    print("🌍 Task 4: Translation (Zero-Shot)")
    print(f"{'='*70}\n")
    
    translation_task = pattern.create_task(
        task_name="english_to_spanish",
        category=TaskCategory.TRANSLATION,
        instruction="Translate the English text to Spanish.",
        expected_output_format="Spanish translation only"
    )
    
    pattern.assign_task_to_agent("ZeroShotMaster", "english_to_spanish")
    
    translate_texts = [
        "Hello, how are you?",
        "The weather is beautiful today.",
        "I love learning new things."
    ]
    
    print("Testing zero-shot translation:\n")
    for text in translate_texts:
        result = agent.execute_task("english_to_spanish", text)
        print(f"English: {result.input}")
        print(f"Spanish: {result.output.strip()}\n")
    
    # Statistics
    print(f"\n{'='*70}")
    print("📊 Zero-Shot Learning Statistics")
    print(f"{'='*70}")
    
    stats = pattern.get_statistics()
    
    print(f"   Total agents: {stats['total_agents']}")
    print(f"   Task templates: {stats['task_templates']}")
    
    for agent_name, agent_stats in stats['agents'].items():
        print(f"\n   Agent: {agent_name}")
        print(f"   - Registered tasks: {agent_stats['registered_tasks']}")
        print(f"   - Total executions: {agent_stats['total_executions']}")
        print(f"   - Tasks by category:")
        for category, count in agent_stats['tasks_by_category'].items():
            print(f"     * {category}: {count}")
    
    print("\n✓ Zero-shot learning enabled task execution without examples!")

Agentic Architecture Pattern: Zero Shot Learning

Pattern 193: 193 Zero Shot Learning Pattern – Line-by-Line Explanation

Overview

Pattern 193: Zero-Shot Learning Pattern

LLM performs tasks without any training examples, relying solely on pre-trained knowledge. Demonstrates ability to generalize to new tasks using instructions alone.

Code Structure Analysis

File Information

Pattern Number: 193
Pattern Name: 193 Zero Shot Learning Pattern
File Path: patterns\llm_specific\193_zero_shot_learning_pattern.py
Total Lines: 343

Detailed Code Explanation

Module Docstring

"""
Pattern 193: Zero-Shot Learning Pattern

LLM performs tasks without any training examples, relying solely on pre-trained knowledge.
Demonstrates ability to generalize to new tasks using instructions alone.
"""

Explanation: This module docstring describes the purpose and key features of the 193 Zero Shot Learning Pattern pattern.

Imports Section

from typing import Dict, Any, List, Optional
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
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 enum import Enum: Enumeration support for fixed sets of values – from langchain_core.prompts import ChatPromptTemplate: LangChain framework integration – from langchain_core.output_parsers import StrOutputParser: LangChain framework integration – from langchain_openai import ChatOpenAI: LangChain framework integration

Enumerations

TaskCategory (Line 17)

class TaskCategory(Enum):
    CLASSIFICATION
    GENERATION
    TRANSLATION
    SUMMARIZATION
    QUESTION_ANSWERING
    REASONING
    EXTRACTION

Purpose: Categories of zero-shot tasks.

Values: – CLASSIFICATION: Represents classification – GENERATION: Represents generation – TRANSLATION: Represents translation – SUMMARIZATION: Represents summarization – QUESTION_ANSWERING: Represents question answering – REASONING: Represents reasoning – EXTRACTION: Represents extraction

Classes

TaskCategory (Line 17)

Description: Categories of zero-shot tasks.

ZeroShotTask (Line 29)

Description: A task to be performed without examples.

ZeroShotResult (Line 39)

Description: Result from zero-shot execution.

ZeroShotAgent (Line 50)

Description: Agent that performs tasks without examples.

Methods:

`init` (Line 53)

def __init__(self, name, model_name, temperature):
    """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”) – temperature: Randomness in LLM responses (0.0-1.0)

`register_task` (Line 59)

def register_task(self, task):
    """Register a task the agent can perform."""

Purpose: Register a task the agent can perform.

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

`create_zero_shot_prompt` (Line 64)

def create_zero_shot_prompt(self, task, input_text):
    """Create prompt for zero-shot execution."""

Purpose: Create prompt for zero-shot execution.

Parameters: – self: Instance reference – task: Parameter for task – input_text: Parameter for input_text

`execute_task` (Line 87)

def execute_task(self, task_name, input_text):
    """Execute a registered task on input."""

Purpose: Execute a registered task on input.

Parameters: – self: Instance reference – task_name: Parameter for task_name – input_text: Parameter for input_text

`execute_adhoc` (Line 116)

def execute_adhoc(self, instruction, input_text, category):
    """Execute an ad-hoc task without prior registration."""

Purpose: Execute an ad-hoc task without prior registration.

Parameters: – self: Instance reference – instruction: Parameter for instruction – input_text: Parameter for input_text – category: Parameter for category

`get_statistics` (Line 146)

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

Purpose: Get agent statistics.

Parameters: – self: Instance reference

ZeroShotLearningPattern (Line 161)

Description: Zero-Shot Learning Pattern – Task execution without training examples.

Leverages LLM’s pre-trained knowledge to perform tasks based solely on natural language instructions, without any demonstration examples.

Methods:

`init` (Line 169)

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

`create_agent` (Line 174)

def create_agent(self, name):
    """Create a zero-shot capable agent."""

Purpose: Create a zero-shot capable agent.

Parameters: – self: Instance reference – name: Identifier or label

`register_task_template` (Line 180)

def register_task_template(self, task):
    """Register a task template in the library."""

Purpose: Register a task template in the library.

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

`create_task` (Line 185)

def create_task(self, task_name, category, instruction):
    """Create and register a task template."""

Purpose: Create and register a task template.

Parameters: – self: Instance reference – task_name: Parameter for task_name – category: Parameter for category – instruction: Parameter for instruction

`assign_task_to_agent` (Line 198)

def assign_task_to_agent(self, agent_name, task_name):
    """Assign a task from library to an agent."""

Purpose: Assign a task from library to an agent.

Parameters: – self: Instance reference – agent_name: Parameter for agent_name – task_name: Parameter for task_name

`get_statistics` (Line 209)

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

Purpose: Get pattern 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 = TaskCategory(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

November 4, 2025

Agentic Architecture Pattern: Few Shot Learning


    """
Pattern 192: Few-Shot Learning Pattern

LLM learns to perform tasks from a small number of examples provided in the prompt.
Demonstrates in-context learning where the model adapts based on examples.
"""

from typing import Dict, Any, List, Optional, Callable
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from langchain_core.prompts import ChatPromptTemplate, FewShotPromptTemplate, PromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI


class ExampleFormat(Enum):
    """Format for few-shot examples."""
    QA_PAIRS = "qa_pairs"  # Question-Answer format
    INPUT_OUTPUT = "input_output"  # Input-Output format
    DEMONSTRATION = "demonstration"  # Step-by-step demonstration
    LABELED = "labeled"  # Labeled examples with categories


@dataclass
class Example:
    """A single few-shot example."""
    input: str
    output: str
    explanation: Optional[str] = None
    label: Optional[str] = None
    metadata: Dict[str, Any] = field(default_factory=dict)


@dataclass
class FewShotTask:
    """A task to be performed using few-shot learning."""
    task_name: str
    description: str
    examples: List[Example]
    example_format: ExampleFormat
    instruction: Optional[str] = None


@dataclass
class FewShotResult:
    """Result from few-shot learning."""
    task_name: str
    input: str
    output: str
    examples_used: int
    confidence: Optional[float] = None
    timestamp: datetime = field(default_factory=datetime.now)


class FewShotLearner:
    """Agent that learns from few examples."""
    
    def __init__(self, name: str, model_name: str = "gpt-4", temperature: float = 0.7):
        self.name = name
        self.llm = ChatOpenAI(model=model_name, temperature=temperature)
        self.learned_tasks: Dict[str, FewShotTask] = {}
        self.results: List[FewShotResult] = []
    
    def learn_task(self, task: FewShotTask):
        """Learn a new task from examples."""
        self.learned_tasks[task.task_name] = task
        print(f"📚 {self.name} learned task '{task.task_name}' from {len(task.examples)} examples")
    
    def create_few_shot_prompt(self, task: FewShotTask, input_text: str) -> str:
        """Create prompt with few-shot examples."""
        prompt_parts = []
        
        # Add task description
        if task.description:
            prompt_parts.append(f"Task: {task.description}\n")
        
        if task.instruction:
            prompt_parts.append(f"Instructions: {task.instruction}\n")
        
        # Add examples based on format
        if task.example_format == ExampleFormat.QA_PAIRS:
            prompt_parts.append("Examples:")
            for i, ex in enumerate(task.examples, 1):
                prompt_parts.append(f"\nQ{i}: {ex.input}")
                prompt_parts.append(f"A{i}: {ex.output}")
                if ex.explanation:
                    prompt_parts.append(f"Explanation: {ex.explanation}")
        
        elif task.example_format == ExampleFormat.INPUT_OUTPUT:
            prompt_parts.append("Examples:")
            for i, ex in enumerate(task.examples, 1):
                prompt_parts.append(f"\nInput {i}: {ex.input}")
                prompt_parts.append(f"Output {i}: {ex.output}")
        
        elif task.example_format == ExampleFormat.DEMONSTRATION:
            prompt_parts.append("Here's how to do it:")
            for i, ex in enumerate(task.examples, 1):
                prompt_parts.append(f"\nExample {i}:")
                prompt_parts.append(f"Given: {ex.input}")
                prompt_parts.append(f"Steps: {ex.explanation or 'See output'}")
                prompt_parts.append(f"Result: {ex.output}")
        
        elif task.example_format == ExampleFormat.LABELED:
            prompt_parts.append("Labeled examples:")
            for ex in task.examples:
                label = ex.label or "unlabeled"
                prompt_parts.append(f"\n[{label}] {ex.input} → {ex.output}")
        
        # Add new input
        prompt_parts.append(f"\n\nNow apply the same pattern:")
        if task.example_format == ExampleFormat.QA_PAIRS:
            prompt_parts.append(f"Q: {input_text}")
            prompt_parts.append("A:")
        else:
            prompt_parts.append(f"Input: {input_text}")
            prompt_parts.append("Output:")
        
        return "\n".join(prompt_parts)
    
    def execute_task(self, task_name: str, input_text: str) -> FewShotResult:
        """Execute learned task on new input."""
        if task_name not in self.learned_tasks:
            raise ValueError(f"Task '{task_name}' not learned yet")
        
        task = self.learned_tasks[task_name]
        prompt_text = self.create_few_shot_prompt(task, input_text)
        
        # Execute with LLM
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"You are {self.name}. Follow the pattern shown in the examples."),
            ("human", "{prompt}")
        ])
        
        chain = prompt | self.llm | StrOutputParser()
        output = chain.invoke({"prompt": prompt_text})
        
        result = FewShotResult(
            task_name=task_name,
            input=input_text,
            output=output,
            examples_used=len(task.examples)
        )
        
        self.results.append(result)
        return result
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get learning statistics."""
        return {
            "name": self.name,
            "tasks_learned": len(self.learned_tasks),
            "total_executions": len(self.results),
            "tasks": {name: len(task.examples) for name, task in self.learned_tasks.items()}
        }


class FewShotLearningPattern:
    """
    Few-Shot Learning Pattern - Learn tasks from small number of examples.
    
    Enables LLMs to quickly adapt to new tasks by providing a few
    demonstration examples in the prompt context.
    """
    
    def __init__(self, model_name: str = "gpt-4"):
        self.model_name = model_name
        self.learners: Dict[str, FewShotLearner] = {}
        self.task_library: Dict[str, FewShotTask] = {}
    
    def create_learner(self, name: str) -> FewShotLearner:
        """Create a few-shot learner."""
        learner = FewShotLearner(name, self.model_name)
        self.learners[name] = learner
        return learner
    
    def register_task(self, task: FewShotTask):
        """Register a task in the library."""
        self.task_library[task.task_name] = task
        print(f"📋 Task registered: {task.task_name} ({len(task.examples)} examples)")
    
    def create_task(self, task_name: str, description: str, 
                   examples: List[Example], example_format: ExampleFormat,
                   instruction: Optional[str] = None) -> FewShotTask:
        """Create and register a task."""
        task = FewShotTask(
            task_name=task_name,
            description=description,
            examples=examples,
            example_format=example_format,
            instruction=instruction
        )
        self.register_task(task)
        return task
    
    def teach_learner(self, learner_name: str, task_name: str):
        """Teach a task to a learner."""
        if learner_name not in self.learners:
            raise ValueError(f"Learner '{learner_name}' not found")
        if task_name not in self.task_library:
            raise ValueError(f"Task '{task_name}' not in library")
        
        learner = self.learners[learner_name]
        task = self.task_library[task_name]
        learner.learn_task(task)
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get pattern statistics."""
        return {
            "total_learners": len(self.learners),
            "tasks_in_library": len(self.task_library),
            "learners": {name: learner.get_statistics() for name, learner in self.learners.items()}
        }


# Example usage
if __name__ == "__main__":
    print("=== Few-Shot Learning Pattern Demo ===\n")
    
    # Create pattern
    print("1. Creating few-shot learning pattern...")
    pattern = FewShotLearningPattern()
    
    # Create learner
    learner = pattern.create_learner("QuickLearner")
    
    # Task 1: Sentiment Analysis
    print(f"\n{'='*70}")
    print("📚 Task 1: Sentiment Analysis")
    print(f"{'='*70}\n")
    
    sentiment_examples = [
        Example(
            input="I love this product! It's amazing!",
            output="Positive",
            explanation="Enthusiastic language with positive words"
        ),
        Example(
            input="Terrible experience. Very disappointed.",
            output="Negative",
            explanation="Strong negative words expressing dissatisfaction"
        ),
        Example(
            input="It's okay, nothing special.",
            output="Neutral",
            explanation="Lukewarm response without strong emotion"
        )
    ]
    
    sentiment_task = pattern.create_task(
        task_name="sentiment_analysis",
        description="Classify the sentiment of text as Positive, Negative, or Neutral",
        examples=sentiment_examples,
        example_format=ExampleFormat.QA_PAIRS,
        instruction="Analyze the emotional tone and classify accordingly"
    )
    
    pattern.teach_learner("QuickLearner", "sentiment_analysis")
    
    # Test sentiment analysis
    test_inputs = [
        "This is the best thing ever!",
        "I'm not impressed at all.",
        "It works as expected."
    ]
    
    print("Testing learned task:\n")
    for test_input in test_inputs:
        result = learner.execute_task("sentiment_analysis", test_input)
        print(f"Input: {result.input}")
        print(f"Output: {result.output}")
        print(f"Examples used: {result.examples_used}\n")
    
    # Task 2: Code Documentation
    print(f"\n{'='*70}")
    print("📚 Task 2: Code Documentation")
    print(f"{'='*70}\n")
    
    doc_examples = [
        Example(
            input="def add(a, b): return a + b",
            output="Adds two numbers and returns their sum.",
            label="math_function"
        ),
        Example(
            input="def is_even(n): return n % 2 == 0",
            output="Checks if a number is even, returns True if divisible by 2.",
            label="boolean_function"
        )
    ]
    
    doc_task = pattern.create_task(
        task_name="code_documentation",
        description="Generate concise documentation for Python functions",
        examples=doc_examples,
        example_format=ExampleFormat.LABELED,
        instruction="Describe what the function does in one clear sentence"
    )
    
    pattern.teach_learner("QuickLearner", "code_documentation")
    
    # Test code documentation
    code_inputs = [
        "def factorial(n): return 1 if n == 0 else n * factorial(n-1)",
        "def reverse_string(s): return s[::-1]"
    ]
    
    print("Testing learned task:\n")
    for code_input in code_inputs:
        result = learner.execute_task("code_documentation", code_input)
        print(f"Code: {result.input}")
        print(f"Doc: {result.output}\n")
    
    # Statistics
    print(f"\n{'='*70}")
    print("📊 Learning Statistics")
    print(f"{'='*70}")
    
    stats = pattern.get_statistics()
    
    print(f"   Total learners: {stats['total_learners']}")
    print(f"   Tasks in library: {stats['tasks_in_library']}")
    
    for learner_name, learner_stats in stats['learners'].items():
        print(f"\n   Learner: {learner_name}")
        print(f"   - Tasks learned: {learner_stats['tasks_learned']}")
        print(f"   - Total executions: {learner_stats['total_executions']}")
        print(f"   - Task details:")
        for task_name, num_examples in learner_stats['tasks'].items():
            print(f"     * {task_name}: {num_examples} examples")
    
    print("\n✓ Few-shot learning enabled rapid task adaptation!")

Agentic Architecture Pattern: Few Shot Learning

Pattern 192: 192 Few Shot Learning Pattern – Line-by-Line Explanation

Overview

Pattern 192: Few-Shot Learning Pattern

LLM learns to perform tasks from a small number of examples provided in the prompt. Demonstrates in-context learning where the model adapts based on examples.

Code Structure Analysis

File Information

Pattern Number: 192
Pattern Name: 192 Few Shot Learning Pattern
File Path: patterns\llm_specific\192_few_shot_learning_pattern.py
Total Lines: 333

Detailed Code Explanation

Module Docstring

"""
Pattern 192: Few-Shot Learning Pattern

LLM learns to perform tasks from a small number of examples provided in the prompt.
Demonstrates in-context learning where the model adapts based on examples.
"""

Explanation: This module docstring describes the purpose and key features of the 192 Few Shot Learning Pattern pattern.

Imports Section

from typing import Dict, Any, List, Optional, Callable
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from langchain_core.prompts import ChatPromptTemplate, FewShotPromptTemplate, PromptTemplate
from langchain_core.output_parsers import StrOutputParser
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 enum import Enum: Enumeration support for fixed sets of values – from langchain_core.prompts import ChatPromptTemplate, FewShotPromptTemplate, PromptTemplate: LangChain framework integration – from langchain_core.output_parsers import StrOutputParser: LangChain framework integration – from langchain_openai import ChatOpenAI: LangChain framework integration

Enumerations

ExampleFormat (Line 17)

class ExampleFormat(Enum):
    QA_PAIRS
    INPUT_OUTPUT
    DEMONSTRATION
    LABELED

Purpose: Format for few-shot examples.

Values: – QA_PAIRS: Represents qa pairs – INPUT_OUTPUT: Represents input output – DEMONSTRATION: Represents demonstration – LABELED: Represents labeled

Classes

ExampleFormat (Line 17)

Description: Format for few-shot examples.

Example (Line 26)

Description: A single few-shot example.

FewShotTask (Line 36)

Description: A task to be performed using few-shot learning.

FewShotResult (Line 46)

Description: Result from few-shot learning.

FewShotLearner (Line 56)

Description: Agent that learns from few examples.

Methods:

`init` (Line 59)

def __init__(self, name, model_name, temperature):
    """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”) – temperature: Randomness in LLM responses (0.0-1.0)

`learn_task` (Line 65)

def learn_task(self, task):
    """Learn a new task from examples."""

Purpose: Learn a new task from examples.

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

`create_few_shot_prompt` (Line 70)

def create_few_shot_prompt(self, task, input_text):
    """Create prompt with few-shot examples."""

Purpose: Create prompt with few-shot examples.

Parameters: – self: Instance reference – task: Parameter for task – input_text: Parameter for input_text

`execute_task` (Line 121)

def execute_task(self, task_name, input_text):
    """Execute learned task on new input."""

Purpose: Execute learned task on new input.

Parameters: – self: Instance reference – task_name: Parameter for task_name – input_text: Parameter for input_text

`get_statistics` (Line 148)

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

Purpose: Get learning statistics.

Parameters: – self: Instance reference

FewShotLearningPattern (Line 158)

Description: Few-Shot Learning Pattern – Learn tasks from small number of examples.

Enables LLMs to quickly adapt to new tasks by providing a few demonstration examples in the prompt context.

Methods:

`init` (Line 166)

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

`create_learner` (Line 171)

def create_learner(self, name):
    """Create a few-shot learner."""

Purpose: Create a few-shot learner.

Parameters: – self: Instance reference – name: Identifier or label

`register_task` (Line 177)

def register_task(self, task):
    """Register a task in the library."""

Purpose: Register a task in the library.

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

`create_task` (Line 182)

def create_task(self, task_name, description, examples, example_format, instruction):
    """Create and register a task."""

Purpose: Create and register a task.

Parameters: – self: Instance reference – task_name: Parameter for task_name – description: Parameter for description – examples: Parameter for examples – example_format: Parameter for example_format – instruction: Parameter for instruction

`teach_learner` (Line 196)

def teach_learner(self, learner_name, task_name):
    """Teach a task to a learner."""

Purpose: Teach a task to a learner.

Parameters: – self: Instance reference – learner_name: Parameter for learner_name – task_name: Parameter for task_name

`get_statistics` (Line 207)

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

Purpose: Get pattern 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 = ExampleFormat(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

November 4, 2025

Agentic Architecture Pattern: Prompt Engineering


    """
Pattern 191: Prompt Engineering Pattern

Systematic design and optimization of prompts for LLMs.
Implements multiple prompt engineering techniques for effective LLM interaction.
"""

from typing import Dict, Any, List, Optional, Callable
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from langchain_core.prompts import ChatPromptTemplate, PromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI


class PromptTechnique(Enum):
    """Prompt engineering techniques."""
    ZERO_SHOT = "zero_shot"
    FEW_SHOT = "few_shot"
    CHAIN_OF_THOUGHT = "chain_of_thought"
    ROLE_BASED = "role_based"
    TEMPLATE = "template"
    STRUCTURED = "structured"


@dataclass
class PromptExample:
    """Example for few-shot learning."""
    input: str
    output: str
    explanation: Optional[str] = None


@dataclass
class PromptTemplate:
    """Prompt template with variables."""
    name: str
    technique: PromptTechnique
    system_message: str
    user_template: str
    variables: List[str]
    examples: List[PromptExample] = field(default_factory=list)
    output_format: Optional[str] = None


@dataclass
class PromptResult:
    """Result of prompt execution."""
    prompt_used: str
    technique: PromptTechnique
    response: str
    metadata: Dict[str, Any] = field(default_factory=dict)
    timestamp: datetime = field(default_factory=datetime.now)


class PromptEngineer:
    """Engineer with multiple prompt techniques."""
    
    def __init__(self, model_name: str = "gpt-4", temperature: float = 0.7):
        self.llm = ChatOpenAI(model=model_name, temperature=temperature)
        self.templates: Dict[str, PromptTemplate] = {}
        self.results: List[PromptResult] = []
    
    def create_zero_shot_prompt(self, task: str, context: Optional[str] = None) -> str:
        """Create zero-shot prompt."""
        if context:
            return f"Context: {context}\n\nTask: {task}\n\nProvide a clear and comprehensive answer."
        return f"Task: {task}\n\nProvide a clear and comprehensive answer."
    
    def create_few_shot_prompt(self, task: str, examples: List[PromptExample]) -> str:
        """Create few-shot prompt with examples."""
        prompt_parts = ["Here are some examples:\n"]
        
        for i, ex in enumerate(examples, 1):
            prompt_parts.append(f"\nExample {i}:")
            prompt_parts.append(f"Input: {ex.input}")
            prompt_parts.append(f"Output: {ex.output}")
            if ex.explanation:
                prompt_parts.append(f"Why: {ex.explanation}")
        
        prompt_parts.append(f"\n\nNow, apply the same pattern:")
        prompt_parts.append(f"Input: {task}")
        prompt_parts.append("Output:")
        
        return "\n".join(prompt_parts)
    
    def create_chain_of_thought_prompt(self, task: str) -> str:
        """Create chain-of-thought prompt."""
        return (f"{task}\n\n"
                f"Let's approach this step-by-step:\n"
                f"1. First, identify the key components\n"
                f"2. Then, analyze the relationships\n"
                f"3. Finally, synthesize the answer\n\n"
                f"Think through each step carefully:")
    
    def create_role_based_prompt(self, task: str, role: str, expertise: List[str]) -> str:
        """Create role-based prompt."""
        expertise_text = ", ".join(expertise)
        return (f"You are a {role} with expertise in {expertise_text}. "
                f"Use your specialized knowledge to address this:\n\n{task}")
    
    def execute_prompt(self, prompt: str, technique: PromptTechnique, 
                      system_message: Optional[str] = None) -> PromptResult:
        """Execute prompt and return result."""
        messages = []
        
        if system_message:
            messages.append(("system", system_message))
        
        messages.append(("human", prompt))
        
        chat_prompt = ChatPromptTemplate.from_messages(messages)
        chain = chat_prompt | self.llm | StrOutputParser()
        
        response = chain.invoke({})
        
        result = PromptResult(
            prompt_used=prompt,
            technique=technique,
            response=response,
            metadata={"system_message": system_message}
        )
        
        self.results.append(result)
        return result
    
    def register_template(self, template: PromptTemplate):
        """Register a reusable prompt template."""
        self.templates[template.name] = template
    
    def use_template(self, template_name: str, **kwargs) -> PromptResult:
        """Use a registered template with variables."""
        if template_name not in self.templates:
            raise ValueError(f"Template '{template_name}' not found")
        
        template = self.templates[template_name]
        
        # Substitute variables
        prompt = template.user_template
        for var in template.variables:
            if var in kwargs:
                prompt = prompt.replace(f"{{{var}}}", str(kwargs[var]))
        
        # Add examples if few-shot
        if template.technique == PromptTechnique.FEW_SHOT and template.examples:
            prompt = self.create_few_shot_prompt(prompt, template.examples)
        
        # Add output format if specified
        if template.output_format:
            prompt += f"\n\nOutput format: {template.output_format}"
        
        return self.execute_prompt(prompt, template.technique, template.system_message)


class PromptEngineeringPattern:
    """
    Prompt Engineering Pattern - Systematic prompt design and optimization.
    
    Provides multiple techniques for crafting effective prompts including
    zero-shot, few-shot, chain-of-thought, and role-based prompting.
    """
    
    def __init__(self, model_name: str = "gpt-4"):
        self.engineer = PromptEngineer(model_name)
        self.technique_usage: Dict[PromptTechnique, int] = {t: 0 for t in PromptTechnique}
    
    def zero_shot(self, task: str, context: Optional[str] = None) -> PromptResult:
        """Execute zero-shot prompting."""
        prompt = self.engineer.create_zero_shot_prompt(task, context)
        result = self.engineer.execute_prompt(prompt, PromptTechnique.ZERO_SHOT)
        self.technique_usage[PromptTechnique.ZERO_SHOT] += 1
        return result
    
    def few_shot(self, task: str, examples: List[PromptExample]) -> PromptResult:
        """Execute few-shot prompting."""
        prompt = self.engineer.create_few_shot_prompt(task, examples)
        result = self.engineer.execute_prompt(prompt, PromptTechnique.FEW_SHOT)
        self.technique_usage[PromptTechnique.FEW_SHOT] += 1
        return result
    
    def chain_of_thought(self, task: str) -> PromptResult:
        """Execute chain-of-thought prompting."""
        prompt = self.engineer.create_chain_of_thought_prompt(task)
        result = self.engineer.execute_prompt(
            prompt, 
            PromptTechnique.CHAIN_OF_THOUGHT,
            "You are a careful thinker who explains reasoning step-by-step."
        )
        self.technique_usage[PromptTechnique.CHAIN_OF_THOUGHT] += 1
        return result
    
    def role_based(self, task: str, role: str, expertise: List[str]) -> PromptResult:
        """Execute role-based prompting."""
        prompt = self.engineer.create_role_based_prompt(task, role, expertise)
        result = self.engineer.execute_prompt(prompt, PromptTechnique.ROLE_BASED)
        self.technique_usage[PromptTechnique.ROLE_BASED] += 1
        return result
    
    def create_template(self, name: str, technique: PromptTechnique, 
                       system_message: str, user_template: str, 
                       variables: List[str], **kwargs) -> PromptTemplate:
        """Create and register a prompt template."""
        template = PromptTemplate(
            name=name,
            technique=technique,
            system_message=system_message,
            user_template=user_template,
            variables=variables,
            examples=kwargs.get('examples', []),
            output_format=kwargs.get('output_format')
        )
        self.engineer.register_template(template)
        return template
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get usage statistics."""
        total = sum(self.technique_usage.values())
        
        return {
            "total_prompts": total,
            "techniques_used": {t.value: count for t, count in self.technique_usage.items() if count > 0},
            "unique_templates": len(self.engineer.templates),
            "avg_response_length": sum(len(r.response) for r in self.engineer.results) / max(len(self.engineer.results), 1)
        }


# Example usage
if __name__ == "__main__":
    print("=== Prompt Engineering Pattern Demo ===\n")
    
    # Create pattern
    print("1. Creating prompt engineering pattern...")
    pattern = PromptEngineeringPattern()
    
    # Test zero-shot
    print(f"\n{'='*70}")
    print("📝 Zero-Shot Prompting")
    print(f"{'='*70}\n")
    
    result = pattern.zero_shot(
        "Explain the concept of 'emergence' in complex systems",
        context="Focus on practical examples"
    )
    print(f"Task: Explain emergence")
    print(f"Technique: {result.technique.value}")
    print(f"Response: {result.response[:200]}...\n")
    
    # Test few-shot
    print(f"\n{'='*70}")
    print("📚 Few-Shot Prompting")
    print(f"{'='*70}\n")
    
    examples = [
        PromptExample(
            input="What is photosynthesis?",
            output="Photosynthesis is the process plants use to convert light energy into chemical energy (glucose).",
            explanation="Simple, clear definition with key components"
        ),
        PromptExample(
            input="What is mitosis?",
            output="Mitosis is cell division that produces two identical daughter cells from one parent cell.",
            explanation="Concise explanation of the core process"
        )
    ]
    
    result = pattern.few_shot("What is osmosis?", examples)
    print(f"Task: What is osmosis?")
    print(f"Technique: {result.technique.value}")
    print(f"Examples provided: {len(examples)}")
    print(f"Response: {result.response[:200]}...\n")
    
    # Test chain-of-thought
    print(f"\n{'='*70}")
    print("🔗 Chain-of-Thought Prompting")
    print(f"{'='*70}\n")
    
    result = pattern.chain_of_thought(
        "If a store has a 20% discount and then adds 8% sales tax, "
        "how much would a $50 item cost?"
    )
    print(f"Task: Calculate final price with discount and tax")
    print(f"Technique: {result.technique.value}")
    print(f"Response: {result.response[:300]}...\n")
    
    # Test role-based
    print(f"\n{'='*70}")
    print("🎭 Role-Based Prompting")
    print(f"{'='*70}\n")
    
    result = pattern.role_based(
        "How should I optimize database queries for a high-traffic application?",
        role="Senior Database Administrator",
        expertise=["SQL optimization", "indexing strategies", "query planning"]
    )
    print(f"Task: Database optimization advice")
    print(f"Technique: {result.technique.value}")
    print(f"Role: Senior Database Administrator")
    print(f"Response: {result.response[:200]}...\n")
    
    # Create and use template
    print(f"\n{'='*70}")
    print("📋 Template-Based Prompting")
    print(f"{'='*70}\n")
    
    template = pattern.create_template(
        name="code_review",
        technique=PromptTechnique.STRUCTURED,
        system_message="You are an expert code reviewer focusing on best practices.",
        user_template="Review this {language} code for {aspect}:\n\n{code}",
        variables=["language", "aspect", "code"],
        output_format="1. Issues found\n2. Recommendations\n3. Severity"
    )
    
    result = pattern.engineer.use_template(
        "code_review",
        language="Python",
        aspect="performance and readability",
        code="def calc(x):\n    return [i*2 for i in range(x)]"
    )
    
    print(f"Template: {template.name}")
    print(f"Response: {result.response[:200]}...\n")
    
    # Statistics
    print(f"\n{'='*70}")
    print("📊 Usage Statistics")
    print(f"{'='*70}")
    
    stats = pattern.get_statistics()
    
    print(f"   Total prompts executed: {stats['total_prompts']}")
    print(f"   Unique templates created: {stats['unique_templates']}")
    print(f"   Avg response length: {stats['avg_response_length']:.0f} chars")
    
    print(f"\n   Techniques used:")
    for technique, count in stats['techniques_used'].items():
        print(f"   - {technique}: {count} times")
    
    print("\n✓ Prompt engineering pattern demonstrated multiple techniques!")

Agentic Architecture Pattern: Prompt Engineering

Pattern 191: 191 Prompt Engineering Pattern – Line-by-Line Explanation

Overview

Pattern 191: Prompt Engineering Pattern

Systematic design and optimization of prompts for LLMs. Implements multiple prompt engineering techniques for effective LLM interaction.

Code Structure Analysis

File Information

Pattern Number: 191
Pattern Name: 191 Prompt Engineering Pattern
File Path: patterns\llm_specific\191_prompt_engineering_pattern.py
Total Lines: 341

Detailed Code Explanation

Module Docstring

"""
Pattern 191: Prompt Engineering Pattern

Systematic design and optimization of prompts for LLMs.
Implements multiple prompt engineering techniques for effective LLM interaction.
"""

Explanation: This module docstring describes the purpose and key features of the 191 Prompt Engineering Pattern pattern.

Imports Section

from typing import Dict, Any, List, Optional, Callable
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from langchain_core.prompts import ChatPromptTemplate, PromptTemplate
from langchain_core.output_parsers import StrOutputParser
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 enum import Enum: Enumeration support for fixed sets of values – from langchain_core.prompts import ChatPromptTemplate, PromptTemplate: LangChain framework integration – from langchain_core.output_parsers import StrOutputParser: LangChain framework integration – from langchain_openai import ChatOpenAI: LangChain framework integration

Enumerations

PromptTechnique (Line 17)

class PromptTechnique(Enum):
    ZERO_SHOT
    FEW_SHOT
    CHAIN_OF_THOUGHT
    ROLE_BASED
    TEMPLATE
    STRUCTURED

Purpose: Prompt engineering techniques.

Values: – ZERO_SHOT: Represents zero shot – FEW_SHOT: Represents few shot – CHAIN_OF_THOUGHT: Represents chain of thought – ROLE_BASED: Represents role based – TEMPLATE: Represents template – STRUCTURED: Represents structured

Classes

PromptTechnique (Line 17)

Description: Prompt engineering techniques.

PromptExample (Line 28)

Description: Example for few-shot learning.

PromptTemplate (Line 36)

Description: Prompt template with variables.

PromptResult (Line 48)

Description: Result of prompt execution.

PromptEngineer (Line 57)

Description: Engineer with multiple prompt techniques.

Methods:

`init` (Line 60)

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

Purpose: No method docstring.

Parameters: – self: Instance reference – model_name: LLM model to use (e.g., “gpt-4”) – temperature: Randomness in LLM responses (0.0-1.0)

`create_zero_shot_prompt` (Line 65)

def create_zero_shot_prompt(self, task, context):
    """Create zero-shot prompt."""

Purpose: Create zero-shot prompt.

Parameters: – self: Instance reference – task: Parameter for task – context: Parameter for context

`create_few_shot_prompt` (Line 71)

def create_few_shot_prompt(self, task, examples):
    """Create few-shot prompt with examples."""

Purpose: Create few-shot prompt with examples.

Parameters: – self: Instance reference – task: Parameter for task – examples: Parameter for examples

`create_chain_of_thought_prompt` (Line 88)

def create_chain_of_thought_prompt(self, task):
    """Create chain-of-thought prompt."""

Purpose: Create chain-of-thought prompt.

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

`create_role_based_prompt` (Line 97)

def create_role_based_prompt(self, task, role, expertise):
    """Create role-based prompt."""

Purpose: Create role-based prompt.

Parameters: – self: Instance reference – task: Parameter for task – role: Parameter for role – expertise: Parameter for expertise

`execute_prompt` (Line 103)

def execute_prompt(self, prompt, technique, system_message):
    """Execute prompt and return result."""

Purpose: Execute prompt and return result.

Parameters: – self: Instance reference – prompt: Parameter for prompt – technique: Parameter for technique – system_message: Parameter for system_message

`register_template` (Line 128)

def register_template(self, template):
    """Register a reusable prompt template."""

Purpose: Register a reusable prompt template.

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

`use_template` (Line 132)

def use_template(self, template_name):
    """Use a registered template with variables."""

Purpose: Use a registered template with variables.

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

PromptEngineeringPattern (Line 156)

Description: Prompt Engineering Pattern – Systematic prompt design and optimization.

Provides multiple techniques for crafting effective prompts including zero-shot, few-shot, chain-of-thought, and role-based prompting.

Methods:

`init` (Line 164)

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

`zero_shot` (Line 168)

def zero_shot(self, task, context):
    """Execute zero-shot prompting."""

Purpose: Execute zero-shot prompting.

Parameters: – self: Instance reference – task: Parameter for task – context: Parameter for context

`few_shot` (Line 175)

def few_shot(self, task, examples):
    """Execute few-shot prompting."""

Purpose: Execute few-shot prompting.

Parameters: – self: Instance reference – task: Parameter for task – examples: Parameter for examples

`chain_of_thought` (Line 182)

def chain_of_thought(self, task):
    """Execute chain-of-thought prompting."""

Purpose: Execute chain-of-thought prompting.

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

`role_based` (Line 193)

def role_based(self, task, role, expertise):
    """Execute role-based prompting."""

Purpose: Execute role-based prompting.

Parameters: – self: Instance reference – task: Parameter for task – role: Parameter for role – expertise: Parameter for expertise

`create_template` (Line 200)

def create_template(self, name, technique, system_message, user_template, variables):
    """Create and register a prompt template."""

Purpose: Create and register a prompt template.

Parameters: – self: Instance reference – name: Identifier or label – technique: Parameter for technique – system_message: Parameter for system_message – user_template: Parameter for user_template – variables: Parameter for variables

`get_statistics` (Line 216)

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

Purpose: Get usage 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 = PromptTechnique(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

November 4, 2025

Agentic Architecture Pattern: Compensation


    """
Pattern 190: Compensation Pattern
Purpose: Undo actions when transaction fails
"""

from typing import Dict, Any, List, Optional, Callable
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum


class TransactionStatus(Enum):
    """Transaction status."""
    PENDING = "pending"
    COMMITTED = "committed"
    COMPENSATING = "compensating"
    COMPENSATED = "compensated"
    FAILED = "failed"


@dataclass
class Action:
    """Action with compensation."""
    name: str
    execute: Callable
    compensate: Callable
    executed: bool = False
    compensated: bool = False


@dataclass
class Transaction:
    """Transaction record."""
    id: str
    actions: List[Action]
    status: TransactionStatus = TransactionStatus.PENDING
    timestamp: datetime = field(default_factory=datetime.now)
    error: Optional[str] = None


class CompensationPattern:
    """
    Compensation Pattern - Undo actions when transaction fails.
    
    Features:
    - Transaction management
    - Automatic compensation
    - Rollback on failure
    - Compensation history
    - Saga pattern support
    """
    
    def __init__(self):
        self.transactions: Dict[str, Transaction] = {}
        self.stats = {
            "total_transactions": 0,
            "committed": 0,
            "compensated": 0,
            "failed": 0
        }
    
    def execute_transaction(self, transaction: Transaction) -> bool:
        """Execute transaction with compensation on failure."""
        self.stats["total_transactions"] += 1
        transaction.status = TransactionStatus.PENDING
        executed_actions = []
        
        try:
            # Execute all actions
            for action in transaction.actions:
                print(f"  Executing: {action.name}")
                action.execute()
                action.executed = True
                executed_actions.append(action)
            
            # All succeeded - commit
            transaction.status = TransactionStatus.COMMITTED
            self.stats["committed"] += 1
            self.transactions[transaction.id] = transaction
            print(f"✅ Transaction {transaction.id} committed")
            return True
            
        except Exception as e:
            # Failure - compensate executed actions in reverse
            print(f"❌ Transaction {transaction.id} failed: {e}")
            transaction.error = str(e)
            transaction.status = TransactionStatus.COMPENSATING
            
            print(f"⚠️  Compensating {len(executed_actions)} actions...")
            for action in reversed(executed_actions):
                try:
                    print(f"  Compensating: {action.name}")
                    action.compensate()
                    action.compensated = True
                except Exception as comp_error:
                    print(f"  ⚠️  Compensation failed for {action.name}: {comp_error}")
            
            transaction.status = TransactionStatus.COMPENSATED
            self.stats["compensated"] += 1
            self.transactions[transaction.id] = transaction
            return False
    
    def get_stats(self) -> Dict[str, Any]:
        """Get transaction statistics."""
        success_rate = (
            self.stats["committed"] / self.stats["total_transactions"] * 100
            if self.stats["total_transactions"] > 0 else 0
        )
        
        return {
            **self.stats,
            "success_rate": f"{success_rate:.1f}%"
        }


if __name__ == "__main__":
    print("=== Compensation Pattern Demo ===\n")
    
    pattern = CompensationPattern()
    
    # Define actions with compensation
    db_data = {"balance": 1000}
    
    def debit_account():
        print(f"    Debiting account (balance: {db_data['balance']})")
        db_data['balance'] -= 100
    
    def compensate_debit():
        print(f"    Reversing debit (balance: {db_data['balance']})")
        db_data['balance'] += 100
    
    def send_payment():
        print(f"    Sending payment")
        # Simulate payment
        pass
    
    def compensate_payment():
        print(f"    Reversing payment")
        pass
    
    def send_notification():
        print(f"    Sending notification")
        raise Exception("Notification service unavailable")  # Simulate failure
    
    def compensate_notification():
        print(f"    No notification to reverse")
        pass
    
    # Create transaction
    transaction = Transaction(
        id="TXN-001",
        actions=[
            Action("debit_account", debit_account, compensate_debit),
            Action("send_payment", send_payment, compensate_payment),
            Action("send_notification", send_notification, compensate_notification)
        ]
    )
    
    print("1. Executing transaction with compensation:\n")
    print(f"Initial balance: ${db_data['balance']}")
    success = pattern.execute_transaction(transaction)
    print(f"\nFinal balance: ${db_data['balance']}")
    print(f"Success: {success}\n")
    
    # Successful transaction
    print("\n2. Successful transaction:\n")
    db_data['balance'] = 1000
    
    transaction2 = Transaction(
        id="TXN-002",
        actions=[
            Action("debit_account", debit_account, compensate_debit),
            Action("send_payment", send_payment, compensate_payment)
        ]
    )
    
    print(f"Initial balance: ${db_data['balance']}")
    success = pattern.execute_transaction(transaction2)
    print(f"\nFinal balance: ${db_data['balance']}")
    print(f"Success: {success}\n")
    
    print("\n=== Statistics ===")
    stats = pattern.get_stats()
    print(f"Total transactions: {stats['total_transactions']}")
    print(f"Committed: {stats['committed']}")
    print(f"Compensated: {stats['compensated']}")
    print(f"Success rate: {stats['success_rate']}")

Agentic Architecture Pattern: Compensation

Pattern 190: 190 Compensation – Line-by-Line Explanation

Overview

Pattern 190: Compensation Pattern Purpose: Undo actions when transaction fails

Code Structure Analysis

File Information

Pattern Number: 190
Pattern Name: 190 Compensation
File Path: patterns\error_handling\190_compensation.py
Total Lines: 188

Detailed Code Explanation

Module Docstring

"""
Pattern 190: Compensation Pattern
Purpose: Undo actions when transaction fails
"""

Explanation: This module docstring describes the purpose and key features of the 190 Compensation pattern.

Imports Section

from typing import Dict, Any, List, Optional, Callable
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum  # First 20 imports shown

Enumerations

TransactionStatus (Line 12)

class TransactionStatus(Enum):
    PENDING
    COMMITTED
    COMPENSATING
    COMPENSATED
    FAILED

Purpose: Transaction status.

Values: – PENDING: Initial state, not yet processed – COMMITTED: Represents committed – COMPENSATING: Represents compensating – COMPENSATED: Represents compensated – FAILED: Execution failed

Classes

TransactionStatus (Line 12)

Description: Transaction status.

Action (Line 22)

Description: Action with compensation.

Transaction (Line 32)

Description: Transaction record.

CompensationPattern (Line 41)

Description: Compensation Pattern – Undo actions when transaction fails.

Features: – Transaction management – Automatic compensation – Rollback on failure – Compensation history – Saga pattern support

Methods:

`init` (Line 53)

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

Purpose: No method docstring.

Parameters: – self: Instance reference

`execute_transaction` (Line 62)

def execute_transaction(self, transaction):
    """Execute transaction with compensation on failure."""

Purpose: Execute transaction with compensation on failure.

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

`get_stats` (Line 103)

def get_stats(self):
    """Get transaction statistics."""

Purpose: Get transaction 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 = TransactionStatus(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

November 4, 2025

Agentic Architecture Pattern: Fail Fast


    """
Pattern 189: Fail Fast

Purpose: Enhanced implementation with full LangChain/LangGraph integration
"""

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


@dataclass
class FailFastConfig:
    """Configuration for Fail Fast."""
    model_name: str = "gpt-4"
    temperature: float = 0.7
    max_iterations: int = 10
    enable_logging: bool = True


@dataclass
class FailFastState:
    """State information for Fail Fast."""
    current_step: int = 0
    data: Dict[str, Any] = field(default_factory=dict)
    results: List[Any] = field(default_factory=list)
    metadata: Dict[str, Any] = field(default_factory=dict)
    timestamp: datetime = field(default_factory=datetime.now)


class FailFastAgent:
    """Agent implementing Fail Fast."""
    
    def __init__(self, name: str, config: FailFastConfig = None):
        self.name = name
        self.config = config or FailFastConfig()
        self.llm = ChatOpenAI(
            model=self.config.model_name,
            temperature=self.config.temperature
        )
        self.state = FailFastState()
        self.execution_history: List[Dict[str, Any]] = []
    
    def process(self, input_data: Any) -> Dict[str, Any]:
        """Process input according to Fail Fast."""
        self.state.current_step += 1
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"You are {self.name}, implementing Fail Fast. "
                      f"Provide clear, structured responses."),
            ("human", "Process this input: {input_data}")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({"input_data": str(input_data)})
        
        result = {
            "agent": self.name,
            "step": self.state.current_step,
            "input": input_data,
            "output": str(response.content),
            "timestamp": datetime.now()
        }
        
        self.execution_history.append(result)
        self.state.results.append(result)
        
        return result
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get agent statistics."""
        return {
            "name": self.name,
            "total_executions": len(self.execution_history),
            "current_step": self.state.current_step,
            "results_count": len(self.state.results)
        }


class FailFastPattern:
    """
    Fail Fast - Enhanced implementation
    
    This pattern provides comprehensive functionality for fail fast
    with full LangChain/LangGraph integration.
    
    Features:
    - Multi-agent coordination
    - State management
    - Execution tracking
    - LangGraph workflow integration
    """
    
    def __init__(self, config: FailFastConfig = None):
        self.config = config or FailFastConfig()
        self.agents: Dict[str, FailFastAgent] = {}
        self.workflow = self._build_workflow()
        self.execution_log: List[Dict[str, Any]] = []
    
    def _build_workflow(self) -> StateGraph:
        """Build LangGraph workflow for pattern execution."""
        
        class WorkflowState(TypedDict):
            input: str
            processed: str
            output: str
            metadata: Dict[str, Any]
        
        def process_input(state: WorkflowState) -> WorkflowState:
            """Process input node."""
            return {
                **state,
                "processed": f"Processed: {state['input']}",
                "metadata": {"step": "process_input"}
            }
        
        def generate_output(state: WorkflowState) -> WorkflowState:
            """Generate output node."""
            llm = ChatOpenAI(model=self.config.model_name, temperature=self.config.temperature)
            
            prompt = ChatPromptTemplate.from_messages([
                ("system", "You are implementing Fail Fast. Generate comprehensive output."),
                ("human", "{input}")
            ])
            
            chain = prompt | llm
            response = chain.invoke({"input": state["processed"]})
            
            return {
                **state,
                "output": str(response.content),
                "metadata": {**state.get("metadata", {}), "step": "generate_output"}
            }
        
        # Build workflow graph
        workflow = StateGraph(WorkflowState)
        workflow.add_node("process", process_input)
        workflow.add_node("generate", generate_output)
        workflow.set_entry_point("process")
        workflow.add_edge("process", "generate")
        workflow.add_edge("generate", END)
        
        return workflow.compile()
    
    def register_agent(self, agent: FailFastAgent):
        """Register an agent with the pattern."""
        self.agents[agent.name] = agent
        if self.config.enable_logging:
            print(f"Registered agent: {agent.name}")
    
    def execute(self, input_data: Any, agent_name: str = None) -> Dict[str, Any]:
        """Execute the pattern with given input."""
        
        # Use workflow if no specific agent requested
        if agent_name is None:
            result = self.workflow.invoke({
                "input": str(input_data),
                "processed": "",
                "output": "",
                "metadata": {}
            })
            
            execution_record = {
                "pattern": "Fail Fast",
                "input": input_data,
                "output": result.get("output", ""),
                "metadata": result.get("metadata", {}),
                "timestamp": datetime.now()
            }
            
            self.execution_log.append(execution_record)
            return execution_record
        
        # Execute with specific agent
        agent = self.agents.get(agent_name)
        if not agent:
            raise ValueError(f"Agent {agent_name} not found")
        
        result = agent.process(input_data)
        self.execution_log.append(result)
        return result
    
    def run(self, input_data: Any) -> Dict[str, Any]:
        """Main execution method."""
        return self.execute(input_data)
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get pattern statistics."""
        return {
            "pattern": "Fail Fast",
            "total_executions": len(self.execution_log),
            "registered_agents": len(self.agents),
            "agents": [agent.get_statistics() for agent in self.agents.values()]
        }


# Example usage
if __name__ == "__main__":
    print("=== Fail Fast Demo ===\n")
    
    # Create configuration
    config = FailFastConfig(
        model_name="gpt-4",
        temperature=0.7,
        enable_logging=True
    )
    
    # Create pattern
    pattern = FailFastPattern(config)
    
    # Create and register agents
    agent1 = FailFastAgent("Agent1", config)
    agent2 = FailFastAgent("Agent2", config)
    pattern.register_agent(agent1)
    pattern.register_agent(agent2)
    
    print("\nExecuting pattern workflow...")
    
    # Execute with workflow
    result1 = pattern.run("Demonstrate fail fast functionality")
    print(f"\nWorkflow Result:")
    print(f"  Output: {result1['output'][:150]}...")
    
    # Execute with specific agents
    result2 = pattern.execute("Specific task for Agent1", agent_name="Agent1")
    print(f"\nAgent1 Result:")
    print(f"  Output: {result2['output'][:150]}...")
    
    result3 = pattern.execute("Specific task for Agent2", agent_name="Agent2")
    print(f"\nAgent2 Result:")
    print(f"  Output: {result3['output'][:150]}...")
    
    # Show statistics
    print("\n=== Statistics ===")
    stats = pattern.get_statistics()
    print(f"Pattern: {stats['pattern']}")
    print(f"Total Executions: {stats['total_executions']}")
    print(f"Registered Agents: {stats['registered_agents']}")
    print(f"\nAgent Statistics:")
    for agent_stats in stats['agents']:
        print(f"  {agent_stats['name']}: {agent_stats['total_executions']} executions")

Agentic Architecture Pattern: Fail Fast

Pattern 189: 189 Fail Fast – Line-by-Line Explanation

Overview

Pattern 189: Fail Fast

Purpose: Enhanced implementation with full LangChain/LangGraph integration

Code Structure Analysis

File Information

Pattern Number: 189
Pattern Name: 189 Fail Fast
File Path: patterns\error_handling\189_fail_fast.py
Total Lines: 247

Detailed Code Explanation

Module Docstring

"""
Pattern 189: Fail Fast

Purpose: Enhanced implementation with full LangChain/LangGraph integration
"""

Explanation: This module docstring describes the purpose and key features of the 189 Fail Fast pattern.

Imports Section

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

Classes

FailFastConfig (Line 18)

Description: Configuration for Fail Fast.

FailFastState (Line 27)

Description: State information for Fail Fast.

FailFastAgent (Line 36)

Description: Agent implementing Fail Fast.

Methods:

`init` (Line 39)

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

Purpose: No method docstring.

Parameters: – self: Instance reference – name: Identifier or label – config: Configuration settings

`process` (Line 49)

def process(self, input_data):
    """Process input according to Fail Fast."""

Purpose: Process input according to Fail Fast.

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

`get_statistics` (Line 75)

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

Purpose: Get agent statistics.

Parameters: – self: Instance reference

FailFastPattern (Line 85)

Description: Fail Fast – Enhanced implementation

This pattern provides comprehensive functionality for fail fast with full LangChain/LangGraph integration.

Features: – Multi-agent coordination – State management – Execution tracking – LangGraph workflow integration

Methods:

`init` (Line 99)

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

Purpose: No method docstring.

Parameters: – self: Instance reference – config: Configuration settings

`_build_workflow` (Line 105)

def _build_workflow(self):
    """Build LangGraph workflow for pattern execution."""

Purpose: Build LangGraph workflow for pattern execution.

Parameters: – self: Instance reference

`register_agent` (Line 150)

def register_agent(self, agent):
    """Register an agent with the pattern."""

Purpose: Register an agent with the pattern.

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

`execute` (Line 156)

def execute(self, input_data, agent_name):
    """Execute the pattern with given input."""

Purpose: Execute the pattern with given input.

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

`run` (Line 188)

def run(self, input_data):
    """Main execution method."""

Purpose: Main execution method.

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

`get_statistics` (Line 192)

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

Purpose: Get pattern statistics.

Parameters: – self: Instance reference

WorkflowState (Line 108)

Description: No class docstring.

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

November 4, 2025

Agentic Architecture Pattern: Fail Safe


    """
Pattern 188: Fail Safe

Purpose: Enhanced implementation with full LangChain/LangGraph integration
"""

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


@dataclass
class FailSafeConfig:
    """Configuration for Fail Safe."""
    model_name: str = "gpt-4"
    temperature: float = 0.7
    max_iterations: int = 10
    enable_logging: bool = True


@dataclass
class FailSafeState:
    """State information for Fail Safe."""
    current_step: int = 0
    data: Dict[str, Any] = field(default_factory=dict)
    results: List[Any] = field(default_factory=list)
    metadata: Dict[str, Any] = field(default_factory=dict)
    timestamp: datetime = field(default_factory=datetime.now)


class FailSafeAgent:
    """Agent implementing Fail Safe."""
    
    def __init__(self, name: str, config: FailSafeConfig = None):
        self.name = name
        self.config = config or FailSafeConfig()
        self.llm = ChatOpenAI(
            model=self.config.model_name,
            temperature=self.config.temperature
        )
        self.state = FailSafeState()
        self.execution_history: List[Dict[str, Any]] = []
    
    def process(self, input_data: Any) -> Dict[str, Any]:
        """Process input according to Fail Safe."""
        self.state.current_step += 1
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"You are {self.name}, implementing Fail Safe. "
                      f"Provide clear, structured responses."),
            ("human", "Process this input: {input_data}")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({"input_data": str(input_data)})
        
        result = {
            "agent": self.name,
            "step": self.state.current_step,
            "input": input_data,
            "output": str(response.content),
            "timestamp": datetime.now()
        }
        
        self.execution_history.append(result)
        self.state.results.append(result)
        
        return result
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get agent statistics."""
        return {
            "name": self.name,
            "total_executions": len(self.execution_history),
            "current_step": self.state.current_step,
            "results_count": len(self.state.results)
        }


class FailSafePattern:
    """
    Fail Safe - Enhanced implementation
    
    This pattern provides comprehensive functionality for fail safe
    with full LangChain/LangGraph integration.
    
    Features:
    - Multi-agent coordination
    - State management
    - Execution tracking
    - LangGraph workflow integration
    """
    
    def __init__(self, config: FailSafeConfig = None):
        self.config = config or FailSafeConfig()
        self.agents: Dict[str, FailSafeAgent] = {}
        self.workflow = self._build_workflow()
        self.execution_log: List[Dict[str, Any]] = []
    
    def _build_workflow(self) -> StateGraph:
        """Build LangGraph workflow for pattern execution."""
        
        class WorkflowState(TypedDict):
            input: str
            processed: str
            output: str
            metadata: Dict[str, Any]
        
        def process_input(state: WorkflowState) -> WorkflowState:
            """Process input node."""
            return {
                **state,
                "processed": f"Processed: {state['input']}",
                "metadata": {"step": "process_input"}
            }
        
        def generate_output(state: WorkflowState) -> WorkflowState:
            """Generate output node."""
            llm = ChatOpenAI(model=self.config.model_name, temperature=self.config.temperature)
            
            prompt = ChatPromptTemplate.from_messages([
                ("system", "You are implementing Fail Safe. Generate comprehensive output."),
                ("human", "{input}")
            ])
            
            chain = prompt | llm
            response = chain.invoke({"input": state["processed"]})
            
            return {
                **state,
                "output": str(response.content),
                "metadata": {**state.get("metadata", {}), "step": "generate_output"}
            }
        
        # Build workflow graph
        workflow = StateGraph(WorkflowState)
        workflow.add_node("process", process_input)
        workflow.add_node("generate", generate_output)
        workflow.set_entry_point("process")
        workflow.add_edge("process", "generate")
        workflow.add_edge("generate", END)
        
        return workflow.compile()
    
    def register_agent(self, agent: FailSafeAgent):
        """Register an agent with the pattern."""
        self.agents[agent.name] = agent
        if self.config.enable_logging:
            print(f"Registered agent: {agent.name}")
    
    def execute(self, input_data: Any, agent_name: str = None) -> Dict[str, Any]:
        """Execute the pattern with given input."""
        
        # Use workflow if no specific agent requested
        if agent_name is None:
            result = self.workflow.invoke({
                "input": str(input_data),
                "processed": "",
                "output": "",
                "metadata": {}
            })
            
            execution_record = {
                "pattern": "Fail Safe",
                "input": input_data,
                "output": result.get("output", ""),
                "metadata": result.get("metadata", {}),
                "timestamp": datetime.now()
            }
            
            self.execution_log.append(execution_record)
            return execution_record
        
        # Execute with specific agent
        agent = self.agents.get(agent_name)
        if not agent:
            raise ValueError(f"Agent {agent_name} not found")
        
        result = agent.process(input_data)
        self.execution_log.append(result)
        return result
    
    def run(self, input_data: Any) -> Dict[str, Any]:
        """Main execution method."""
        return self.execute(input_data)
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get pattern statistics."""
        return {
            "pattern": "Fail Safe",
            "total_executions": len(self.execution_log),
            "registered_agents": len(self.agents),
            "agents": [agent.get_statistics() for agent in self.agents.values()]
        }


# Example usage
if __name__ == "__main__":
    print("=== Fail Safe Demo ===\n")
    
    # Create configuration
    config = FailSafeConfig(
        model_name="gpt-4",
        temperature=0.7,
        enable_logging=True
    )
    
    # Create pattern
    pattern = FailSafePattern(config)
    
    # Create and register agents
    agent1 = FailSafeAgent("Agent1", config)
    agent2 = FailSafeAgent("Agent2", config)
    pattern.register_agent(agent1)
    pattern.register_agent(agent2)
    
    print("\nExecuting pattern workflow...")
    
    # Execute with workflow
    result1 = pattern.run("Demonstrate fail safe functionality")
    print(f"\nWorkflow Result:")
    print(f"  Output: {result1['output'][:150]}...")
    
    # Execute with specific agents
    result2 = pattern.execute("Specific task for Agent1", agent_name="Agent1")
    print(f"\nAgent1 Result:")
    print(f"  Output: {result2['output'][:150]}...")
    
    result3 = pattern.execute("Specific task for Agent2", agent_name="Agent2")
    print(f"\nAgent2 Result:")
    print(f"  Output: {result3['output'][:150]}...")
    
    # Show statistics
    print("\n=== Statistics ===")
    stats = pattern.get_statistics()
    print(f"Pattern: {stats['pattern']}")
    print(f"Total Executions: {stats['total_executions']}")
    print(f"Registered Agents: {stats['registered_agents']}")
    print(f"\nAgent Statistics:")
    for agent_stats in stats['agents']:
        print(f"  {agent_stats['name']}: {agent_stats['total_executions']} executions")

Agentic Architecture Pattern: Fail Safe

Pattern 188: 188 Fail Safe – Line-by-Line Explanation

Overview

Pattern 188: Fail Safe

Purpose: Enhanced implementation with full LangChain/LangGraph integration

Code Structure Analysis

File Information

Pattern Number: 188
Pattern Name: 188 Fail Safe
File Path: patterns\error_handling\188_fail_safe.py
Total Lines: 247

Detailed Code Explanation

Module Docstring

"""
Pattern 188: Fail Safe

Purpose: Enhanced implementation with full LangChain/LangGraph integration
"""

Explanation: This module docstring describes the purpose and key features of the 188 Fail Safe pattern.

Imports Section

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

Classes

FailSafeConfig (Line 18)

Description: Configuration for Fail Safe.

FailSafeState (Line 27)

Description: State information for Fail Safe.

FailSafeAgent (Line 36)

Description: Agent implementing Fail Safe.

Methods:

`init` (Line 39)

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

Purpose: No method docstring.

Parameters: – self: Instance reference – name: Identifier or label – config: Configuration settings

`process` (Line 49)

def process(self, input_data):
    """Process input according to Fail Safe."""

Purpose: Process input according to Fail Safe.

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

`get_statistics` (Line 75)

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

Purpose: Get agent statistics.

Parameters: – self: Instance reference

FailSafePattern (Line 85)

Description: Fail Safe – Enhanced implementation

This pattern provides comprehensive functionality for fail safe with full LangChain/LangGraph integration.

Features: – Multi-agent coordination – State management – Execution tracking – LangGraph workflow integration

Methods:

`init` (Line 99)

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

Purpose: No method docstring.

Parameters: – self: Instance reference – config: Configuration settings

`_build_workflow` (Line 105)

def _build_workflow(self):
    """Build LangGraph workflow for pattern execution."""

Purpose: Build LangGraph workflow for pattern execution.

Parameters: – self: Instance reference

`register_agent` (Line 150)

def register_agent(self, agent):
    """Register an agent with the pattern."""

Purpose: Register an agent with the pattern.

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

`execute` (Line 156)

def execute(self, input_data, agent_name):
    """Execute the pattern with given input."""

Purpose: Execute the pattern with given input.

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

`run` (Line 188)

def run(self, input_data):
    """Main execution method."""

Purpose: Main execution method.

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

`get_statistics` (Line 192)

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

Purpose: Get pattern statistics.

Parameters: – self: Instance reference

WorkflowState (Line 108)

Description: No class docstring.

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

November 4, 2025

Tag: llm

Agentic Architecture Pattern: Self Consistency

Agentic Architecture Pattern: Self Consistency

Pattern 197: 197 Self Consistency Pattern – Line-by-Line Explanation

Overview

Code Structure Analysis

File Information

Detailed Code Explanation

Module Docstring

Imports Section

Classes

SelfConsistencyConfig (Line 18)

SelfConsistencyState (Line 27)

SelfConsistencyAgent (Line 36)

__init__ (Line 39)

process (Line 49)

get_statistics (Line 75)

SelfConsistencyPattern (Line 85)

__init__ (Line 99)

_build_workflow (Line 105)

register_agent (Line 150)

execute (Line 156)

run (Line 188)

get_statistics (Line 192)

WorkflowState (Line 108)

Design Patterns Used

Usage Example

Key Concepts

Extension Points

Agentic Architecture Pattern: Constitutional Ai

Agentic Architecture Pattern: Constitutional Ai

Pattern 196: 196 Constitutional Ai Pattern – Line-by-Line Explanation

Overview

Code Structure Analysis

File Information

Detailed Code Explanation

Module Docstring

Imports Section

Classes

ConstitutionalAiConfig (Line 18)

ConstitutionalAiState (Line 27)

ConstitutionalAiAgent (Line 36)

__init__ (Line 39)

process (Line 49)

get_statistics (Line 75)

ConstitutionalAiPattern (Line 85)

__init__ (Line 99)

_build_workflow (Line 105)

register_agent (Line 150)

execute (Line 156)

run (Line 188)

get_statistics (Line 192)

WorkflowState (Line 108)

Design Patterns Used

Usage Example

Key Concepts

Extension Points

Agentic Architecture Pattern: Prompt Chaining

Agentic Architecture Pattern: Prompt Chaining

Pattern 195: 195 Prompt Chaining Pattern – Line-by-Line Explanation

Overview

Code Structure Analysis

File Information

Detailed Code Explanation

Module Docstring

Imports Section

Classes

PromptChainingConfig (Line 18)

PromptChainingState (Line 27)

PromptChainingAgent (Line 36)

__init__ (Line 39)

process (Line 49)

get_statistics (Line 75)

PromptChainingPattern (Line 85)

__init__ (Line 99)

_build_workflow (Line 105)

register_agent (Line 150)

execute (Line 156)

run (Line 188)

get_statistics (Line 192)

`init` (Line 39)

`process` (Line 49)

`get_statistics` (Line 75)

`init` (Line 99)

`_build_workflow` (Line 105)

`register_agent` (Line 150)

`execute` (Line 156)

`run` (Line 188)

`get_statistics` (Line 192)

`init` (Line 39)

`process` (Line 49)

`get_statistics` (Line 75)

`init` (Line 99)

`_build_workflow` (Line 105)

`register_agent` (Line 150)

`execute` (Line 156)

`run` (Line 188)

`get_statistics` (Line 192)

`init` (Line 39)

`process` (Line 49)

`get_statistics` (Line 75)

`init` (Line 99)

`_build_workflow` (Line 105)

`register_agent` (Line 150)

`execute` (Line 156)

`run` (Line 188)

`get_statistics` (Line 192)

`init` (Line 38)

`add_step` (Line 44)

`_resolve_dependencies` (Line 49)

`execute_chain` (Line 71)

`init` (Line 133)

`register_agent` (Line 136)

`run` (Line 140)

`init` (Line 53)

`register_task` (Line 59)

`create_zero_shot_prompt` (Line 64)

`execute_task` (Line 87)

`execute_adhoc` (Line 116)

`get_statistics` (Line 146)

`init` (Line 169)

`create_agent` (Line 174)

`register_task_template` (Line 180)

`create_task` (Line 185)

`assign_task_to_agent` (Line 198)

`get_statistics` (Line 209)