aishastr.live

Agentic Architecture Pattern: Plugin

Written by

admin

in


    """
Pattern 95: Plugin Pattern

Dynamically extend agent capabilities by loading plugins at runtime.
Plugins implement a common interface and can be added/removed without
modifying the core agent.
"""

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 Plugin(ABC):
    """Base class for plugins."""
    
    @abstractmethod
    def get_name(self) -> str:
        """Return plugin name."""
        pass
    
    @abstractmethod
    def get_version(self) -> str:
        """Return plugin version."""
        pass
    
    @abstractmethod
    def execute(self, context: Dict[str, Any], input_data: Any) -> Any:
        """Execute plugin functionality."""
        pass
    
    def initialize(self):
        """Optional initialization."""
        pass
    
    def cleanup(self):
        """Optional cleanup."""
        pass


@dataclass
class PluginMetadata:
    """Metadata about an installed plugin."""
    name: str
    version: str
    enabled: bool = True
    load_time: datetime = field(default_factory=datetime.now)
    execution_count: int = 0


class PluginPattern:
    """
    Plugin Pattern - Extensible agent through plugins.
    """
    
    def __init__(self, model_name: str = "gpt-4"):
        self.llm = ChatOpenAI(model=model_name, temperature=0.5)
        self.plugins: Dict[str, Plugin] = {}
        self.metadata: Dict[str, PluginMetadata] = {}
    
    def register_plugin(self, plugin: Plugin):
        """Register a plugin."""
        plugin.initialize()
        name = plugin.get_name()
        self.plugins[name] = plugin
        self.metadata[name] = PluginMetadata(
            name=name,
            version=plugin.get_version()
        )
        print(f"Plugin registered: {name} v{plugin.get_version()}")
    
    def unregister_plugin(self, plugin_name: str):
        """Unregister a plugin."""
        if plugin_name in self.plugins:
            self.plugins[plugin_name].cleanup()
            del self.plugins[plugin_name]
            del self.metadata[plugin_name]
            print(f"Plugin unregistered: {plugin_name}")
    
    def enable_plugin(self, plugin_name: str):
        """Enable a plugin."""
        if plugin_name in self.metadata:
            self.metadata[plugin_name].enabled = True
    
    def disable_plugin(self, plugin_name: str):
        """Disable a plugin."""
        if plugin_name in self.metadata:
            self.metadata[plugin_name].enabled = False
    
    def execute_plugin(self, plugin_name: str, input_data: Any, context: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
        """Execute a specific plugin."""
        if plugin_name not in self.plugins:
            return {"error": f"Plugin {plugin_name} not found"}
        
        if not self.metadata[plugin_name].enabled:
            return {"error": f"Plugin {plugin_name} is disabled"}
        
        context = context or {}
        
        try:
            result = self.plugins[plugin_name].execute(context, input_data)
            self.metadata[plugin_name].execution_count += 1
            return {"success": True, "result": result}
        except Exception as e:
            return {"success": False, "error": str(e)}
    
    def run(self, task: str) -> Dict[str, Any]:
        """Run task using available plugins."""
        # Use LLM to determine which plugins to use
        plugin_list = [f"{name} ({meta.version})" for name, meta in self.metadata.items() if meta.enabled]
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"Available plugins: {', '.join(plugin_list)}. Determine which plugins to use for the task."),
            ("human", "{task}")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({"task": task})
        
        return {
            "task": task,
            "available_plugins": len(plugin_list),
            "recommendation": response.content
        }
    
    def list_plugins(self) -> List[Dict[str, Any]]:
        """List all registered plugins."""
        return [
            {
                "name": meta.name,
                "version": meta.version,
                "enabled": meta.enabled,
                "executions": meta.execution_count
            }
            for meta in self.metadata.values()
        ]


# Example plugins
class TextAnalysisPlugin(Plugin):
    """Plugin for text analysis."""
    
    def get_name(self) -> str:
        return "TextAnalysis"
    
    def get_version(self) -> str:
        return "1.0.0"
    
    def execute(self, context: Dict[str, Any], input_data: Any) -> Any:
        text = str(input_data)
        return {
            "length": len(text),
            "words": len(text.split()),
            "uppercase": text.upper(),
            "lowercase": text.lower()
        }


class MathPlugin(Plugin):
    """Plugin for math operations."""
    
    def get_name(self) -> str:
        return "Math"
    
    def get_version(self) -> str:
        return "1.0.0"
    
    def execute(self, context: Dict[str, Any], input_data: Any) -> Any:
        if isinstance(input_data, (list, tuple)):
            return {
                "sum": sum(input_data),
                "average": sum(input_data) / len(input_data) if input_data else 0,
                "min": min(input_data) if input_data else None,
                "max": max(input_data) if input_data else None
            }
        return {"error": "Input must be a list or tuple"}


class LoggerPlugin(Plugin):
    """Plugin for logging."""
    
    def __init__(self):
        self.logs = []
    
    def get_name(self) -> str:
        return "Logger"
    
    def get_version(self) -> str:
        return "1.0.0"
    
    def execute(self, context: Dict[str, Any], input_data: Any) -> Any:
        log_entry = f"[{datetime.now()}] {input_data}"
        self.logs.append(log_entry)
        return {"logged": True, "entry": log_entry}


# Example usage
if __name__ == "__main__":
    print("=== Plugin Pattern Demo ===\n")
    
    pattern = PluginPattern()
    
    # Register plugins
    pattern.register_plugin(TextAnalysisPlugin())
    pattern.register_plugin(MathPlugin())
    pattern.register_plugin(LoggerPlugin())
    
    print(f"\nRegistered plugins: {len(pattern.list_plugins())}\n")
    
    # Execute plugins
    result1 = pattern.execute_plugin("TextAnalysis", "Hello World!")
    print(f"TextAnalysis result: {result1['result']}\n")
    
    result2 = pattern.execute_plugin("Math", [1, 2, 3, 4, 5])
    print(f"Math result: {result2['result']}\n")
    
    result3 = pattern.execute_plugin("Logger", "Task completed successfully")
    print(f"Logger result: {result3['result']}\n")
    
    # List all plugins
    print("=== Plugin Status ===")
    for plugin in pattern.list_plugins():
        print(f"- {plugin['name']} v{plugin['version']}: "
              f"{'enabled' if plugin['enabled'] else 'disabled'}, "
              f"{plugin['executions']} executions")
    
    # Disable and test
    print("\n=== Testing Plugin Disable ===")
    pattern.disable_plugin("Math")
    result4 = pattern.execute_plugin("Math", [1, 2, 3])
    print(f"Attempt to use disabled plugin: {result4}")

Agentic Architecture Pattern: Plugin

Pattern 95: 95 Plugin – Line-by-Line Explanation

Overview

Pattern 95: Plugin Pattern

Dynamically extend agent capabilities by loading plugins at runtime. Plugins implement a common interface and can be added/removed without modifying the core agent.

Code Structure Analysis

File Information

  • Pattern Number: 95
  • Pattern Name: 95 Plugin
  • File Path: patterns\tool_resource\95_plugin.py
  • Total Lines: 235

Detailed Code Explanation

Module Docstring

"""
Pattern 95: Plugin Pattern

Dynamically extend agent capabilities by loading plugins at runtime.
Plugins implement a common interface and can be added/removed without
modifying the core agent.
"""

Explanation: This module docstring describes the purpose and key features of the 95 Plugin 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

Plugin (Line 17)

Description: Base class for plugins.

Methods:

get_name (Line 21)
def get_name(self):
    """Return plugin name."""

Purpose: Return plugin name.

Parameters: – self: Instance reference

get_version (Line 26)
def get_version(self):
    """Return plugin version."""

Purpose: Return plugin version.

Parameters: – self: Instance reference

execute (Line 31)
def execute(self, context, input_data):
    """Execute plugin functionality."""

Purpose: Execute plugin functionality.

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

initialize (Line 35)
def initialize(self):
    """Optional initialization."""

Purpose: Optional initialization.

Parameters: – self: Instance reference

cleanup (Line 39)
def cleanup(self):
    """Optional cleanup."""

Purpose: Optional cleanup.

Parameters: – self: Instance reference

PluginMetadata (Line 45)

Description: Metadata about an installed plugin.

PluginPattern (Line 54)

Description: Plugin Pattern – Extensible agent through plugins.

Methods:

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

register_plugin (Line 64)
def register_plugin(self, plugin):
    """Register a plugin."""

Purpose: Register a plugin.

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

unregister_plugin (Line 75)
def unregister_plugin(self, plugin_name):
    """Unregister a plugin."""

Purpose: Unregister a plugin.

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

enable_plugin (Line 83)
def enable_plugin(self, plugin_name):
    """Enable a plugin."""

Purpose: Enable a plugin.

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

disable_plugin (Line 88)
def disable_plugin(self, plugin_name):
    """Disable a plugin."""

Purpose: Disable a plugin.

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

execute_plugin (Line 93)
def execute_plugin(self, plugin_name, input_data, context):
    """Execute a specific plugin."""

Purpose: Execute a specific plugin.

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

run (Line 110)
def run(self, task):
    """Run task using available plugins."""

Purpose: Run task using available plugins.

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

list_plugins (Line 129)
def list_plugins(self):
    """List all registered plugins."""

Purpose: List all registered plugins.

Parameters: – self: Instance reference

TextAnalysisPlugin (Line 143)

Description: Plugin for text analysis.

Methods:

get_name (Line 146)
def get_name(self):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference

get_version (Line 149)
def get_version(self):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference

execute (Line 152)
def execute(self, context, input_data):
    """No method docstring."""

Purpose: No method docstring.

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

MathPlugin (Line 162)

Description: Plugin for math operations.

Methods:

get_name (Line 165)
def get_name(self):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference

get_version (Line 168)
def get_version(self):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference

execute (Line 171)
def execute(self, context, input_data):
    """No method docstring."""

Purpose: No method docstring.

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

LoggerPlugin (Line 182)

Description: Plugin for logging.

Methods:

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

Purpose: No method docstring.

Parameters: – self: Instance reference

get_name (Line 188)
def get_name(self):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference

get_version (Line 191)
def get_version(self):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference

execute (Line 194)
def execute(self, context, input_data):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference – context: Parameter for context – 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 = Plugin(name="example", model_name="gpt-4")

# Configure and execute
result = pattern.execute(data={"key": "value"})

# Check results
print(result)

Key Concepts

  • LLM Integration: Uses LangChain/OpenAI for intelligent decision-making
  • Workflow Orchestration: LangGraph manages execution flow through nodes
  • State Management: TypedDict defines workflow state structure
  • Error Handling: Comprehensive error handling and fallback mechanisms
  • Extensibility: Designed for easy customization and extension

Extension Points

This pattern can be extended by: 1. Adding custom workflow nodes 2. Implementing additional validation logic 3. Integrating with external services 4. Customizing LLM prompts and responses 5. Adding monitoring and logging capabilities

Comments

Leave a Reply

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

More posts