Factory Pattern

The Factory pattern is essential for LLM applications that need to create different types of objects based on runtime conditions, such as selecting AI models, creating specialized tools, or instantiating different processing pipelines.

Why Factory Pattern for LLM?

LLM applications often require:

• Dynamic object creation: Choose specific implementations based on runtime conditions

• Decoupled instantiation: Separate object creation from object usage

• Configuration-driven creation: Create objects based on user preferences or system settings

• Easy extensibility: Add new implementations without modifying existing code

Key LLM Use Cases

1. AI Model Factory

Creating different AI models based on requirements:

from abc import ABC, abstractmethod
from enum import Enum

class ModelType(Enum):
    FAST = "fast"
    BALANCED = "balanced" 
    ADVANCED = "advanced"
    SPECIALIZED = "specialized"

class AIModel(ABC):
    @abstractmethod
    def generate(self, prompt, **kwargs):
        pass
    
    @abstractmethod
    def get_model_info(self):
        pass

class FastModel(AIModel):
    def __init__(self):
        self.model_name = "gpt-3.5-turbo"
        self.max_tokens = 1000
        self.cost_per_token = 0.002
    
    def generate(self, prompt, **kwargs):
        # Fast, lightweight model for simple queries
        return f"Fast response to: {prompt}"
    
    def get_model_info(self):
        return {
            "name": self.model_name,
            "speed": "high",
            "cost": "low",
            "quality": "good"
        }

class BalancedModel(AIModel):
    def __init__(self):
        self.model_name = "gpt-4"
        self.max_tokens = 2000
        self.cost_per_token = 0.03
    
    def generate(self, prompt, **kwargs):
        # Balanced model for general use
        return f"Balanced response to: {prompt}"
    
    def get_model_info(self):
        return {
            "name": self.model_name,
            "speed": "medium",
            "cost": "medium", 
            "quality": "high"
        }

class AdvancedModel(AIModel):
    def __init__(self):
        self.model_name = "gpt-4-turbo"
        self.max_tokens = 4000
        self.cost_per_token = 0.06
    
    def generate(self, prompt, **kwargs):
        # Most capable model for complex tasks
        return f"Advanced response to: {prompt}"
    
    def get_model_info(self):
        return {
            "name": self.model_name,
            "speed": "low",
            "cost": "high",
            "quality": "excellent"
        }

class SpecializedModel(AIModel):
    def __init__(self, domain):
        self.domain = domain
        self.model_name = f"specialized-{domain}-model"
        self.max_tokens = 2000
    
    def generate(self, prompt, **kwargs):
        return f"Specialized {self.domain} response to: {prompt}"
    
    def get_model_info(self):
        return {
            "name": self.model_name,
            "domain": self.domain,
            "quality": "domain-optimized"
        }

class AIModelFactory:
    @staticmethod
    def create_model(model_type: ModelType, **kwargs):
        """Factory method to create AI models based on type"""
        if model_type == ModelType.FAST:
            return FastModel()
        elif model_type == ModelType.BALANCED:
            return BalancedModel()
        elif model_type == ModelType.ADVANCED:
            return AdvancedModel()
        elif model_type == ModelType.SPECIALIZED:
            domain = kwargs.get('domain', 'general')
            return SpecializedModel(domain)
        else:
            raise ValueError(f"Unknown model type: {model_type}")
    
    @staticmethod
    def create_model_by_requirements(requirements):
        """Create model based on specific requirements"""
        if requirements.get('speed_priority', False):
            return AIModelFactory.create_model(ModelType.FAST)
        elif requirements.get('cost_sensitive', False):
            return AIModelFactory.create_model(ModelType.FAST)
        elif requirements.get('quality_priority', False):
            return AIModelFactory.create_model(ModelType.ADVANCED)
        elif requirements.get('domain'):
            return AIModelFactory.create_model(
                ModelType.SPECIALIZED, 
                domain=requirements['domain']
            )
        else:
            return AIModelFactory.create_model(ModelType.BALANCED)

# Usage examples
def handle_user_query(query, user_preferences):
    # Create appropriate model based on user preferences
    model = AIModelFactory.create_model_by_requirements(user_preferences)
    response = model.generate(query)
    return response, model.get_model_info()

# Different usage scenarios
urgent_response = handle_user_query(
    "Quick math question", 
    {"speed_priority": True}
)

research_response = handle_user_query(
    "Complex analysis needed",
    {"quality_priority": True}
)

medical_response = handle_user_query(
    "Medical consultation",
    {"domain": "medical"}
)

Benefits:

• Dynamic model selection based on requirements

• Easy addition of new models

• Centralized model creation logic

• Requirement-based intelligent selection

2. Tool Factory for AI Agents

Creating different tools for AI agents based on needs:

class AITool(ABC):
    @abstractmethod
    def execute(self, input_data):
        pass
    
    @abstractmethod
    def get_tool_info(self):
        pass

class CalculatorTool(AITool):
    def execute(self, expression):
        try:
            # Safe evaluation of mathematical expressions
            result = eval(expression)  # In production, use ast.literal_eval or safer methods
            return {"result": result, "success": True}
        except Exception as e:
            return {"error": str(e), "success": False}
    
    def get_tool_info(self):
        return {
            "name": "calculator",
            "description": "Performs mathematical calculations",
            "input_type": "mathematical_expression"
        }

class WebSearchTool(AITool):
    def execute(self, query):
        # Simulate web search
        return {
            "results": [
                f"Web result 1 for: {query}",
                f"Web result 2 for: {query}",
                f"Web result 3 for: {query}"
            ],
            "success": True
        }
    
    def get_tool_info(self):
        return {
            "name": "web_search",
            "description": "Searches the internet for information",
            "input_type": "search_query"
        }

class CodeExecutorTool(AITool):
    def execute(self, code):
        # Simulate code execution (in production, use sandboxed environment)
        return {
            "output": f"Executed: {code[:50]}...",
            "success": True
        }
    
    def get_tool_info(self):
        return {
            "name": "code_executor",
            "description": "Executes code in a sandboxed environment",
            "input_type": "code"
        }

class FileReaderTool(AITool):
    def execute(self, file_path):
        try:
            with open(file_path, 'r') as f:
                content = f.read()
            return {"content": content, "success": True}
        except Exception as e:
            return {"error": str(e), "success": False}
    
    def get_tool_info(self):
        return {
            "name": "file_reader",
            "description": "Reads content from files",
            "input_type": "file_path"
        }

class ToolFactory:
    _tools = {
        "calculator": CalculatorTool,
        "web_search": WebSearchTool,
        "code_executor": CodeExecutorTool,
        "file_reader": FileReaderTool
    }
    
    @classmethod
    def create_tool(cls, tool_name):
        """Create a tool by name"""
        if tool_name in cls._tools:
            return cls._tools[tool_name]()
        else:
            raise ValueError(f"Unknown tool: {tool_name}")
    
    @classmethod
    def create_tools_for_agent_type(cls, agent_type):
        """Create appropriate tools based on agent type"""
        tool_mappings = {
            "math_tutor": ["calculator"],
            "research_assistant": ["web_search", "file_reader"],
            "coding_assistant": ["code_executor", "file_reader"],
            "general_assistant": ["calculator", "web_search", "file_reader"]
        }
        
        tool_names = tool_mappings.get(agent_type, ["calculator"])
        return [cls.create_tool(name) for name in tool_names]
    
    @classmethod
    def register_tool(cls, name, tool_class):
        """Register a new tool type"""
        cls._tools[name] = tool_class
    
    @classmethod
    def get_available_tools(cls):
        """Get list of available tool names"""
        return list(cls._tools.keys())

# Usage in AI Agent
class AIAgent:
    def __init__(self, agent_type):
        self.agent_type = agent_type
        self.tools = ToolFactory.create_tools_for_agent_type(agent_type)
        self.tool_map = {tool.get_tool_info()["name"]: tool for tool in self.tools}
    
    def use_tool(self, tool_name, input_data):
        if tool_name in self.tool_map:
            return self.tool_map[tool_name].execute(input_data)
        else:
            return {"error": f"Tool {tool_name} not available", "success": False}
    
    def get_available_tools(self):
        return [tool.get_tool_info() for tool in self.tools]

# Different agent types with appropriate tools
math_agent = AIAgent("math_tutor")
research_agent = AIAgent("research_assistant")
coding_agent = AIAgent("coding_assistant")

Benefits:

• Dynamic tool creation based on agent requirements

• Easy registration of new tools

• Type-specific tool sets

• Centralized tool management

3. Prompt Template Factory

Creating different prompt templates based on task types:

class PromptTemplate(ABC):
    @abstractmethod
    def format(self, **kwargs):
        pass
    
    @abstractmethod
    def get_required_params(self):
        pass

class QuestionAnsweringTemplate(PromptTemplate):
    def format(self, context, question, **kwargs):
        return f"""
Based on the following context:
{context}

Please answer this question: {question}

Provide a clear and accurate answer based only on the given context.
"""
    
    def get_required_params(self):
        return ["context", "question"]

class CodeGenerationTemplate(PromptTemplate):
    def format(self, task_description, language="Python", **kwargs):
        return f"""
Write {language} code to accomplish the following task:
{task_description}

Requirements:
- Write clean, well-commented code
- Include error handling where appropriate
- Follow {language} best practices

Code:
"""
    
    def get_required_params(self):
        return ["task_description"]

class SummarizationTemplate(PromptTemplate):
    def format(self, text, max_length=200, **kwargs):
        return f"""
Please summarize the following text in approximately {max_length} words:

{text}

Summary:
"""
    
    def get_required_params(self):
        return ["text"]

class TranslationTemplate(PromptTemplate):
    def format(self, text, source_language, target_language, **kwargs):
        return f"""
Translate the following text from {source_language} to {target_language}:

{text}

Translation:
"""
    
    def get_required_params(self):
        return ["text", "source_language", "target_language"]

class ChainOfThoughtTemplate(PromptTemplate):
    def format(self, problem, **kwargs):
        return f"""
Let's solve this step by step:

Problem: {problem}

Step 1: Understand the problem
Step 2: Break it down into smaller parts
Step 3: Solve each part
Step 4: Combine the solutions

Let's begin:
"""
    
    def get_required_params(self):
        return ["problem"]

class PromptTemplateFactory:
    _templates = {
        "qa": QuestionAnsweringTemplate,
        "code_generation": CodeGenerationTemplate,
        "summarization": SummarizationTemplate,
        "translation": TranslationTemplate,
        "chain_of_thought": ChainOfThoughtTemplate
    }
    
    @classmethod
    def create_template(cls, template_type):
        """Create a template by type"""
        if template_type in cls._templates:
            return cls._templates[template_type]()
        else:
            raise ValueError(f"Unknown template type: {template_type}")
    
    @classmethod
    def create_template_by_task(cls, task_description):
        """Auto-select template based on task description"""
        task_lower = task_description.lower()
        
        if "translate" in task_lower:
            return cls.create_template("translation")
        elif "summarize" in task_lower or "summary" in task_lower:
            return cls.create_template("summarization")
        elif "code" in task_lower or "program" in task_lower:
            return cls.create_template("code_generation")
        elif "step" in task_lower or "explain" in task_lower:
            return cls.create_template("chain_of_thought")
        else:
            return cls.create_template("qa")
    
    @classmethod
    def register_template(cls, name, template_class):
        """Register a new template type"""
        cls._templates[name] = template_class

# Usage
class PromptProcessor:
    def process_request(self, task_description, **params):
        # Auto-select appropriate template
        template = PromptTemplateFactory.create_template_by_task(task_description)
        
        # Format the prompt
        formatted_prompt = template.format(**params)
        
        return formatted_prompt, template.get_required_params()

# Examples
processor = PromptProcessor()

# Auto-selects summarization template
summary_prompt, _ = processor.process_request(
    "Please summarize this document",
    text="Long document content here...",
    max_length=150
)

# Auto-selects code generation template  
code_prompt, _ = processor.process_request(
    "Write a Python function to sort a list",
    task_description="Create a function that sorts a list of numbers",
    language="Python"
)

Benefits:

• Automatic template selection based on task

• Consistent prompt formatting

• Easy template registration and extension

• Parameter validation and requirements

4. Evaluation Metric Factory

Creating different evaluation metrics based on task type:

class EvaluationMetric(ABC):
    @abstractmethod
    def calculate(self, predictions, references, **kwargs):
        pass
    
    @abstractmethod
    def get_metric_info(self):
        pass

class AccuracyMetric(EvaluationMetric):
    def calculate(self, predictions, references, **kwargs):
        correct = sum(p == r for p, r in zip(predictions, references))
        total = len(predictions)
        return correct / total if total > 0 else 0
    
    def get_metric_info(self):
        return {
            "name": "accuracy",
            "range": [0, 1],
            "higher_is_better": True
        }

class BleuMetric(EvaluationMetric):
    def calculate(self, predictions, references, **kwargs):
        # Simplified BLEU calculation
        # In practice, use nltk.translate.bleu_score or similar
        scores = []
        for pred, ref in zip(predictions, references):
            # Simple word overlap calculation
            pred_words = set(pred.split())
            ref_words = set(ref.split())
            if len(pred_words) == 0:
                scores.append(0)
            else:
                overlap = len(pred_words.intersection(ref_words))
                scores.append(overlap / len(pred_words))
        
        return sum(scores) / len(scores) if scores else 0
    
    def get_metric_info(self):
        return {
            "name": "bleu",
            "range": [0, 1],
            "higher_is_better": True
        }

class RougeMetric(EvaluationMetric):
    def calculate(self, predictions, references, **kwargs):
        # Simplified ROUGE calculation
        scores = []
        for pred, ref in zip(predictions, references):
            pred_words = pred.split()
            ref_words = ref.split()
            
            if len(ref_words) == 0:
                scores.append(0)
            else:
                overlap = len(set(pred_words).intersection(set(ref_words)))
                scores.append(overlap / len(ref_words))
        
        return sum(scores) / len(scores) if scores else 0
    
    def get_metric_info(self):
        return {
            "name": "rouge",
            "range": [0, 1],
            "higher_is_better": True
        }

class SemanticSimilarityMetric(EvaluationMetric):
    def calculate(self, predictions, references, **kwargs):
        # Simplified semantic similarity using sentence transformers
        # In practice, use actual embedding models
        similarities = []
        for pred, ref in zip(predictions, references):
            # Placeholder for actual semantic similarity calculation
            # This would use embedding models in practice
            word_overlap = len(set(pred.split()).intersection(set(ref.split())))
            max_words = max(len(pred.split()), len(ref.split()))
            similarity = word_overlap / max_words if max_words > 0 else 0
            similarities.append(similarity)
        
        return sum(similarities) / len(similarities) if similarities else 0
    
    def get_metric_info(self):
        return {
            "name": "semantic_similarity",
            "range": [0, 1],
            "higher_is_better": True
        }

class MetricFactory:
    _metrics = {
        "accuracy": AccuracyMetric,
        "bleu": BleuMetric,
        "rouge": RougeMetric,
        "semantic_similarity": SemanticSimilarityMetric
    }
    
    @classmethod
    def create_metric(cls, metric_name):
        """Create a metric by name"""
        if metric_name in cls._metrics:
            return cls._metrics[metric_name]()
        else:
            raise ValueError(f"Unknown metric: {metric_name}")
    
    @classmethod
    def create_metrics_for_task(cls, task_type):
        """Create appropriate metrics based on task type"""
        task_metrics = {
            "classification": ["accuracy"],
            "translation": ["bleu", "semantic_similarity"],
            "summarization": ["rouge", "semantic_similarity"],
            "question_answering": ["accuracy", "semantic_similarity"],
            "text_generation": ["bleu", "rouge", "semantic_similarity"]
        }
        
        metric_names = task_metrics.get(task_type, ["accuracy"])
        return [cls.create_metric(name) for name in metric_names]

# Usage in evaluation system
class EvaluationSystem:
    def __init__(self, task_type):
        self.metrics = MetricFactory.create_metrics_for_task(task_type)
    
    def evaluate(self, predictions, references):
        results = {}
        for metric in self.metrics:
            metric_info = metric.get_metric_info()
            score = metric.calculate(predictions, references)
            results[metric_info["name"]] = {
                "score": score,
                "info": metric_info
            }
        return results

# Task-specific evaluation
translation_eval = EvaluationSystem("translation")
summarization_eval = EvaluationSystem("summarization")
qa_eval = EvaluationSystem("question_answering")

Benefits:

• Task-appropriate metric selection

• Consistent evaluation interfaces

• Easy metric registration and extension

• Comprehensive evaluation capabilities

Implementation Advantages

1. Flexibility

• Runtime object creation based on conditions

• Easy switching between different implementations

• Configuration-driven object instantiation

• Dynamic behavior modification

2. Extensibility

• Easy addition of new types without modifying existing code

• Registration mechanism for new implementations

• Plugin-like architecture

• Modular component addition

3. Maintainability

• Centralized creation logic

• Clear separation of concerns

• Consistent object interfaces

• Simplified testing and debugging

4. Scalability

• Support for numerous object types

• Efficient object creation and management

• Resource optimization through appropriate selection

• Performance tuning through type-specific implementations

Real-World Impact

The Factory pattern in LLM applications provides:

• Dynamic Adaptation: Create appropriate objects based on runtime requirements

• System Flexibility: Easy switching between different implementations and configurations

• Development Efficiency: Reduced boilerplate code and simplified object management

• Quality Assurance: Consistent object creation patterns and validation

---

🔗 Interactive Implementation

📓 Factory Pattern Notebook - Multi-provider AI client system with intelligent selection, failover, and cost optimization.

This pattern is fundamental for production LLM systems where you need to create different types of objects dynamically based on user requirements, system conditions, or configuration settings while maintaining clean, maintainable code.