Abstract Factory Pattern

The Abstract Factory pattern is perfect for LLM applications that need to create families of related objects, such as different AI providers, model configurations, or complete AI system stacks with consistent interfaces.

Why Abstract Factory for LLM?

LLM applications often require:

• Provider abstraction: Support multiple AI providers (OpenAI, Anthropic, Google, etc.)

• Environment consistency: Different configurations for development, testing, and production

• Family cohesion: Related components that work together (client + embeddings + tools)

• Easy switching: Change entire AI stacks without code modifications

Key LLM Use Cases

1. Multi-Provider AI Client Factory

Creating consistent interfaces for different AI providers:

from abc import ABC, abstractmethod

# Abstract products
class LLMClient(ABC):
    @abstractmethod
    def generate(self, prompt, **kwargs):
        pass
    
    @abstractmethod
    def stream_generate(self, prompt, **kwargs):
        pass

class EmbeddingClient(ABC):
    @abstractmethod
    def embed_text(self, text):
        pass
    
    @abstractmethod
    def embed_documents(self, documents):
        pass

class ImageClient(ABC):
    @abstractmethod
    def generate_image(self, prompt):
        pass
    
    @abstractmethod
    def analyze_image(self, image):
        pass

# Abstract factory
class AIProviderFactory(ABC):
    @abstractmethod
    def create_llm_client(self):
        pass
    
    @abstractmethod
    def create_embedding_client(self):
        pass
    
    @abstractmethod
    def create_image_client(self):
        pass

# Concrete implementations for OpenAI
class OpenAILLMClient(LLMClient):
    def generate(self, prompt, **kwargs):
        return openai.ChatCompletion.create(
            model="gpt-4",
            messages=[{"role": "user", "content": prompt}],
            **kwargs
        )
    
    def stream_generate(self, prompt, **kwargs):
        return openai.ChatCompletion.create(
            model="gpt-4",
            messages=[{"role": "user", "content": prompt}],
            stream=True,
            **kwargs
        )

class OpenAIEmbeddingClient(EmbeddingClient):
    def embed_text(self, text):
        return openai.Embedding.create(
            model="text-embedding-ada-002",
            input=text
        )
    
    def embed_documents(self, documents):
        return [self.embed_text(doc) for doc in documents]

class OpenAIImageClient(ImageClient):
    def generate_image(self, prompt):
        return openai.Image.create(
            prompt=prompt,
            n=1,
            size="1024x1024"
        )
    
    def analyze_image(self, image):
        return openai.ChatCompletion.create(
            model="gpt-4-vision-preview",
            messages=[{
                "role": "user",
                "content": [{"type": "image_url", "image_url": image}]
            }]
        )

class OpenAIFactory(AIProviderFactory):
    def create_llm_client(self):
        return OpenAILLMClient()
    
    def create_embedding_client(self):
        return OpenAIEmbeddingClient()
    
    def create_image_client(self):
        return OpenAIImageClient()

# Concrete implementations for Anthropic
class AnthropicLLMClient(LLMClient):
    def generate(self, prompt, **kwargs):
        return anthropic.Completion.create(
            model="claude-3-opus-20240229",
            prompt=f"Human: {prompt}\n\nAssistant:",
            **kwargs
        )
    
    def stream_generate(self, prompt, **kwargs):
        return anthropic.Completion.create(
            model="claude-3-opus-20240229",
            prompt=f"Human: {prompt}\n\nAssistant:",
            stream=True,
            **kwargs
        )

class AnthropicFactory(AIProviderFactory):
    def create_llm_client(self):
        return AnthropicLLMClient()
    
    def create_embedding_client(self):
        # Anthropic doesn't have embeddings, use OpenAI
        return OpenAIEmbeddingClient()
    
    def create_image_client(self):
        # Anthropic doesn't have image generation, use OpenAI
        return OpenAIImageClient()

# Usage
class AIApplication:
    def __init__(self, factory: AIProviderFactory):
        self.llm = factory.create_llm_client()
        self.embeddings = factory.create_embedding_client()
        self.images = factory.create_image_client()
    
    def process_multimodal_request(self, text_prompt, image_url=None):
        if image_url:
            image_analysis = self.images.analyze_image(image_url)
            combined_prompt = f"{text_prompt}\n\nImage analysis: {image_analysis}"
        else:
            combined_prompt = text_prompt
        
        return self.llm.generate(combined_prompt)

# Easy provider switching
openai_app = AIApplication(OpenAIFactory())
anthropic_app = AIApplication(AnthropicFactory())

Benefits:

• Consistent interface across all AI providers

• Easy switching between providers

• Family of related clients work together

• Centralized provider configuration

2. Environment-Based Configuration Factory

Different configurations for different environments:

class AIConfigFactory(ABC):
    @abstractmethod
    def create_llm_config(self):
        pass
    
    @abstractmethod
    def create_embedding_config(self):
        pass
    
    @abstractmethod
    def create_storage_config(self):
        pass

class DevelopmentConfigFactory(AIConfigFactory):
    def create_llm_config(self):
        return {
            "model": "gpt-3.5-turbo",  # Cheaper for development
            "max_tokens": 100,
            "temperature": 0.7,
            "timeout": 30
        }
    
    def create_embedding_config(self):
        return {
            "model": "text-embedding-ada-002",
            "batch_size": 10,  # Smaller batches for dev
            "cache_enabled": False
        }
    
    def create_storage_config(self):
        return {
            "type": "memory",  # In-memory for quick dev cycles
            "persistence": False
        }

class ProductionConfigFactory(AIConfigFactory):
    def create_llm_config(self):
        return {
            "model": "gpt-4",  # Best model for production
            "max_tokens": 2000,
            "temperature": 0.3,  # More deterministic
            "timeout": 120,
            "retry_attempts": 3
        }
    
    def create_embedding_config(self):
        return {
            "model": "text-embedding-ada-002",
            "batch_size": 100,  # Optimized batching
            "cache_enabled": True,
            "cache_ttl": 3600
        }
    
    def create_storage_config(self):
        return {
            "type": "postgresql",  # Persistent storage
            "connection_pool": 20,
            "backup_enabled": True
        }

class TestConfigFactory(AIConfigFactory):
    def create_llm_config(self):
        return {
            "model": "mock",  # Mock for fast tests
            "deterministic": True,
            "response_delay": 0
        }
    
    def create_embedding_config(self):
        return {
            "model": "mock",
            "fixed_dimensions": 1536
        }
    
    def create_storage_config(self):
        return {
            "type": "memory",
            "clear_on_start": True  # Clean slate for each test
        }

# Environment-based factory selection
def get_ai_factory(environment):
    factories = {
        "development": DevelopmentConfigFactory(),
        "production": ProductionConfigFactory(),
        "test": TestConfigFactory()
    }
    return factories.get(environment, DevelopmentConfigFactory())

Benefits:

• Environment-specific optimizations

• Consistent configuration patterns

• Easy environment switching

• Centralized configuration management

3. RAG System Component Factory

Creating complete RAG systems with consistent component families:

class RAGComponentFactory(ABC):
    @abstractmethod
    def create_vectorstore(self):
        pass
    
    @abstractmethod
    def create_retriever(self):
        pass
    
    @abstractmethod
    def create_reranker(self):
        pass
    
    @abstractmethod
    def create_generator(self):
        pass

class ScientificRAGFactory(RAGComponentFactory):
    def create_vectorstore(self):
        return ChromaVectorStore(
            collection_name="scientific_papers",
            embedding_model="all-mpnet-base-v2",  # Good for scientific text
            metadata_schema={
                "paper_id": str,
                "authors": list,
                "journal": str,
                "citation_count": int
            }
        )
    
    def create_retriever(self):
        return ScientificRetriever(
            similarity_threshold=0.7,
            citation_weight=0.3,  # Favor highly cited papers
            recency_weight=0.2,
            max_results=10
        )
    
    def create_reranker(self):
        return ScientificReranker(
            model="cross-encoder/ms-marco-MiniLM-L-12-v2",
            citation_boost=True,
            peer_review_filter=True
        )
    
    def create_generator(self):
        return ScientificGenerator(
            model="gpt-4",
            system_prompt="""You are a scientific research assistant. 
            Provide accurate, well-cited responses based on peer-reviewed sources.""",
            citation_format="academic"
        )

class LegalRAGFactory(RAGComponentFactory):
    def create_vectorstore(self):
        return PineconeVectorStore(
            index_name="legal_documents",
            embedding_model="law-embedding-model",  # Specialized legal embeddings
            metadata_schema={
                "case_id": str,
                "jurisdiction": str,
                "court_level": str,
                "date": str,
                "legal_area": str
            }
        )
    
    def create_retriever(self):
        return LegalRetriever(
            jurisdiction_filter=True,
            precedent_ranking=True,
            statutory_priority=True,
            max_results=15
        )
    
    def create_reranker(self):
        return LegalReranker(
            model="legal-reranker-v1",
            jurisdiction_boost=True,
            recency_penalty=False  # Older cases can be very relevant
        )
    
    def create_generator(self):
        return LegalGenerator(
            model="gpt-4",
            system_prompt="""You are a legal research assistant. 
            Provide legally accurate information with proper case citations.
            Always include appropriate disclaimers.""",
            citation_format="legal"
        )

class GeneralRAGFactory(RAGComponentFactory):
    def create_vectorstore(self):
        return FAISSVectorStore(
            embedding_model="text-embedding-ada-002",
            index_type="flat",
            normalize_embeddings=True
        )
    
    def create_retriever(self):
        return SemanticRetriever(
            similarity_threshold=0.75,
            max_results=5
        )
    
    def create_reranker(self):
        return CrossEncoderReranker(
            model="cross-encoder/ms-marco-MiniLM-L-6-v2"
        )
    
    def create_generator(self):
        return GeneralGenerator(
            model="gpt-3.5-turbo",
            system_prompt="You are a helpful assistant.",
            max_tokens=1000
        )

# RAG System using factory
class RAGSystem:
    def __init__(self, factory: RAGComponentFactory):
        self.vectorstore = factory.create_vectorstore()
        self.retriever = factory.create_retriever()
        self.reranker = factory.create_reranker()
        self.generator = factory.create_generator()
    
    def query(self, question):
        # Retrieve relevant documents
        docs = self.retriever.retrieve(question, self.vectorstore)
        
        # Rerank for relevance
        ranked_docs = self.reranker.rerank(question, docs)
        
        # Generate response
        return self.generator.generate(question, ranked_docs)

# Domain-specific RAG systems
scientific_rag = RAGSystem(ScientificRAGFactory())
legal_rag = RAGSystem(LegalRAGFactory())
general_rag = RAGSystem(GeneralRAGFactory())

Benefits:

• Domain-optimized RAG components

• Consistent component interfaces

• Easy domain switching

• Specialized configurations per use case

4. Testing and Evaluation Factory

Creating comprehensive testing suites for different scenarios:

class TestSuiteFactory(ABC):
    @abstractmethod
    def create_test_data_generator(self):
        pass
    
    @abstractmethod
    def create_evaluator(self):
        pass
    
    @abstractmethod
    def create_metrics_collector(self):
        pass
    
    @abstractmethod
    def create_reporter(self):
        pass

class PerformanceTestFactory(TestSuiteFactory):
    def create_test_data_generator(self):
        return PerformanceTestDataGenerator(
            query_count=1000,
            concurrent_users=[1, 10, 50, 100],
            query_complexity_distribution={
                "simple": 0.4,
                "medium": 0.4,
                "complex": 0.2
            }
        )
    
    def create_evaluator(self):
        return PerformanceEvaluator(
            timeout_threshold=30,  # seconds
            memory_limit=1024,     # MB
            cpu_threshold=80       # %
        )
    
    def create_metrics_collector(self):
        return PerformanceMetricsCollector(
            metrics=["response_time", "throughput", "error_rate", "resource_usage"],
            sampling_interval=1    # second
        )
    
    def create_reporter(self):
        return PerformanceReporter(
            format="dashboard",
            real_time_updates=True,
            alert_thresholds={
                "response_time": 5.0,
                "error_rate": 0.05
            }
        )

class QualityTestFactory(TestSuiteFactory):
    def create_test_data_generator(self):
        return QualityTestDataGenerator(
            golden_dataset="human_evaluated_qa.json",
            adversarial_examples=True,
            domain_specific_cases=True
        )
    
    def create_evaluator(self):
        return QualityEvaluator(
            metrics=["accuracy", "relevance", "coherence", "factuality"],
            human_evaluation_sample_rate=0.1
        )
    
    def create_metrics_collector(self):
        return QualityMetricsCollector(
            automated_scoring=True,
            llm_as_judge=True,
            judge_model="gpt-4"
        )
    
    def create_reporter(self):
        return QualityReporter(
            format="detailed_analysis",
            include_examples=True,
            failure_analysis=True
        )

# Testing framework using factories
class AITestingFramework:
    def __init__(self, factory: TestSuiteFactory):
        self.data_generator = factory.create_test_data_generator()
        self.evaluator = factory.create_evaluator()
        self.metrics = factory.create_metrics_collector()
        self.reporter = factory.create_reporter()
    
    def run_test_suite(self, model):
        # Generate test data
        test_data = self.data_generator.generate()
        
        # Run evaluation
        results = self.evaluator.evaluate(model, test_data)
        
        # Collect metrics
        metrics = self.metrics.collect(results)
        
        # Generate report
        return self.reporter.generate_report(metrics, results)

# Different testing approaches
performance_testing = AITestingFramework(PerformanceTestFactory())
quality_testing = AITestingFramework(QualityTestFactory())

Benefits:

• Comprehensive testing capabilities

• Consistent testing interfaces

• Easy test type switching

• Specialized testing for different concerns

Implementation Advantages

1. Provider Independence

• Switch between AI providers without code changes

• Consistent interfaces across different services

• Easy migration between providers

• Vendor lock-in prevention

2. Configuration Management

• Environment-specific optimizations

• Centralized configuration control

• Easy deployment across environments

• Consistent component relationships

3. Domain Specialization

• Optimized component families for specific domains

• Domain-specific configurations and behaviors

• Easy domain switching and comparison

• Specialized feature sets per domain

4. Testing and Quality Assurance

• Comprehensive testing frameworks

• Different testing strategies for different concerns

• Consistent evaluation approaches

• Easy comparison of different configurations

Real-World Impact

The Abstract Factory pattern in LLM applications provides:

• Flexibility: Easy switching between providers, environments, and configurations

• Consistency: Guaranteed compatible component families

• Maintainability: Centralized creation logic and configuration management

• Scalability: Easy extension to new providers and environments

This pattern is crucial for production LLM systems where you need to support multiple AI providers, different deployment environments, and domain-specific optimizations while maintaining consistent interfaces and behavior.