"""
Pattern 96: Extension Pattern

Extend agent behavior by adding new capabilities without modifying existing code.
Extensions inherit from base agent and add specific features.
"""

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


class BaseAgent(ABC):
    """Base agent that can be extended."""
    
    def __init__(self, name: str, model_name: str = "gpt-4"):
        self.name = name
        self.llm = ChatOpenAI(model=model_name, temperature=0.7)
    
    @abstractmethod
    def core_capability(self, input_data: str) -> str:
        """Core capability of the agent."""
        pass


@dataclass
class Extension:
    """Extension that adds capabilities."""
    name: str
    capability: Callable
    description: str
    priority: int = 0
    enabled: bool = True
    call_count: int = 0


class ExtensibleAgent(BaseAgent):
    """Agent that supports extensions."""
    
    def __init__(self, name: str, model_name: str = "gpt-4"):
        super().__init__(name, model_name)
        self.extensions: Dict[str, Extension] = {}
    
    def core_capability(self, input_data: str) -> str:
        """Core capability - basic text processing."""
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"You are {self.name}, a base AI agent."),
            ("human", "{input}")
        ])
        chain = prompt | self.llm
        response = chain.invoke({"input": input_data})
        return str(response.content)
    
    def add_extension(self, name: str, capability: Callable, 
                     description: str, priority: int = 0):
        """Add an extension."""
        extension = Extension(
            name=name,
            capability=capability,
            description=description,
            priority=priority
        )
        self.extensions[name] = extension
        print(f"Extension added: {name} (priority: {priority})")
    
    def remove_extension(self, name: str):
        """Remove an extension."""
        if name in self.extensions:
            del self.extensions[name]
            print(f"Extension removed: {name}")
    
    def enable_extension(self, name: str):
        """Enable an extension."""
        if name in self.extensions:
            self.extensions[name].enabled = True
    
    def disable_extension(self, name: str):
        """Disable an extension."""
        if name in self.extensions:
            self.extensions[name].enabled = False
    
    def execute_with_extensions(self, input_data: str) -> Dict[str, Any]:
        """Execute with all enabled extensions."""
        results = {
            "core_result": self.core_capability(input_data),
            "extension_results": {}
        }
        
        # Sort extensions by priority (higher priority first)
        sorted_extensions = sorted(
            self.extensions.values(),
            key=lambda e: e.priority,
            reverse=True
        )
        
        # Execute enabled extensions
        for ext in sorted_extensions:
            if ext.enabled:
                try:
                    result = ext.capability(input_data)
                    results["extension_results"][ext.name] = result
                    ext.call_count += 1
                except Exception as e:
                    results["extension_results"][ext.name] = f"Error: {str(e)}"
        
        return results
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get extension statistics."""
        return {
            "total_extensions": len(self.extensions),
            "enabled_extensions": sum(1 for e in self.extensions.values() if e.enabled),
            "extension_calls": {
                name: ext.call_count
                for name, ext in self.extensions.items()
            }
        }


class ExtensionPattern:
    """
    Extension Pattern - Manage extensible agents.
    """
    
    def __init__(self, model_name: str = "gpt-4"):
        self.model_name = model_name
        self.agents: Dict[str, ExtensibleAgent] = {}
    
    def create_agent(self, name: str) -> ExtensibleAgent:
        """Create a new extensible agent."""
        agent = ExtensibleAgent(name, self.model_name)
        self.agents[name] = agent
        return agent
    
    def run(self, agent_name: str, input_data: str) -> Dict[str, Any]:
        """Run agent with extensions."""
        if agent_name not in self.agents:
            return {"error": f"Agent {agent_name} not found"}
        
        agent = self.agents[agent_name]
        result = agent.execute_with_extensions(input_data)
        
        return {
            "agent": agent_name,
            "input": input_data,
            "results": result,
            "statistics": agent.get_statistics()
        }


# Example usage
if __name__ == "__main__":
    # Create pattern
    pattern = ExtensionPattern()
    
    # Create agent
    agent = pattern.create_agent("SmartAssistant")
    
    # Add extensions
    agent.add_extension(
        name="sentiment_analyzer",
        capability=lambda text: "positive" if "good" in text.lower() or "great" in text.lower() else "neutral",
        description="Analyzes sentiment of text",
        priority=10
    )
    
    agent.add_extension(
        name="word_counter",
        capability=lambda text: f"{len(text.split())} words",
        description="Counts words in text",
        priority=5
    )
    
    agent.add_extension(
        name="text_summarizer",
        capability=lambda text: text[:50] + "..." if len(text) > 50 else text,
        description="Summarizes long text",
        priority=3
    )
    
    # Test
    print("=== Extension Pattern Demo ===\n")
    
    # Run with extensions
    result = pattern.run("SmartAssistant", "This is a great day to learn about AI extensions!")
    
    print(f"Agent: {result['agent']}")
    print(f"\nInput: {result['input']}")
    print(f"\nCore Result: {result['results']['core_result'][:100]}...")
    print(f"\nExtension Results:")
    for ext_name, ext_result in result['results']['extension_results'].items():
        print(f"  - {ext_name}: {ext_result}")
    
    print(f"\nStatistics:")
    stats = result['statistics']
    print(f"  Total extensions: {stats['total_extensions']}")
    print(f"  Enabled extensions: {stats['enabled_extensions']}")
    print(f"  Extension calls: {stats['extension_calls']}")
    
    # Test with disabled extension
    print("\n\n=== Disabling sentiment_analyzer ===")
    agent.disable_extension("sentiment_analyzer")
    
    result2 = pattern.run("SmartAssistant", "Another test message")
    print(f"\nExtension Results:")
    for ext_name, ext_result in result2['results']['extension_results'].items():
        print(f"  - {ext_name}: {ext_result}")

    

Agentic Architecture Pattern: Extension

Pattern 96: 96 Extension – Line-by-Line Explanation

Overview

Pattern 96: Extension Pattern

Extend agent behavior by adding new capabilities without modifying existing code. Extensions inherit from base agent and add specific features.

Code Structure Analysis

File Information

• Pattern Number: 96
• Pattern Name: 96 Extension
• File Path: patterns\tool_resource\96_extension.py
• Total Lines: 211

Detailed Code Explanation

Module Docstring

"""
Pattern 96: Extension Pattern

Extend agent behavior by adding new capabilities without modifying existing code.
Extensions inherit from base agent and add specific features.
"""

Explanation: This module docstring describes the purpose and key features of the 96 Extension pattern.

Imports Section

from typing import Dict, Any, List, Optional, Callable
from dataclasses import dataclass, field
from datetime import datetime
from abc import ABC, abstractmethod
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI  # First 20 imports shown

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

Classes

BaseAgent (Line 16)

Description: Base agent that can be extended.

Methods:

__init__ (Line 19)

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

core_capability (Line 24)

def core_capability(self, input_data):
    """Core capability of the agent."""

Purpose: Core capability of the agent.

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

Extension (Line 30)

Description: Extension that adds capabilities.

ExtensibleAgent (Line 40)

Description: Agent that supports extensions.

Methods:

__init__ (Line 43)

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

core_capability (Line 47)

def core_capability(self, input_data):
    """Core capability - basic text processing."""

Purpose: Core capability – basic text processing.

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

add_extension (Line 57)

def add_extension(self, name, capability, description, priority):
    """Add an extension."""

Purpose: Add an extension.

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

remove_extension (Line 69)

def remove_extension(self, name):
    """Remove an extension."""

Purpose: Remove an extension.

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

enable_extension (Line 75)

def enable_extension(self, name):
    """Enable an extension."""

Purpose: Enable an extension.

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

disable_extension (Line 80)

def disable_extension(self, name):
    """Disable an extension."""

Purpose: Disable an extension.

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

execute_with_extensions (Line 85)

def execute_with_extensions(self, input_data):
    """Execute with all enabled extensions."""

Purpose: Execute with all enabled extensions.

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

get_statistics (Line 111)

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

Purpose: Get extension statistics.

Parameters: – self: Instance reference

ExtensionPattern (Line 123)

Description: Extension Pattern – Manage extensible agents.

Methods:

__init__ (Line 128)

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

def create_agent(self, name):
    """Create a new extensible agent."""

Purpose: Create a new extensible agent.

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

run (Line 138)

def run(self, agent_name, input_data):
    """Run agent with extensions."""

Purpose: Run agent with extensions.

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

Design Patterns Used

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

Usage Example

# Initialize the pattern
pattern = BaseAgent(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