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.
Last updated