Adapter Pattern
The Adapter pattern bridges incompatible interfaces, making different systems work together seamlessly. In LLM applications, adapters excel at creating unified interfaces across multiple AI providers, handling different API formats, and integrating with legacy systems.
Why Adapter Pattern for LLM?
LLM applications often need:
Multi-provider support: Integrate OpenAI, Anthropic, Google, Azure, and local models
API standardization: Create consistent interfaces despite different provider APIs
Legacy integration: Connect modern LLM capabilities with existing systems
Data format adaptation: Handle different input/output formats across providers
Protocol bridging: Support REST, GraphQL, gRPC, and WebSocket protocols
Key LLM Use Cases
1. Multi-Provider LLM Adapter
Create a unified interface for different LLM providers:
from abc import ABC, abstractmethod
from typing import Dict, Any, Optional, List
import json
class LLMResponse:
"""Standardized response format"""
def __init__(self, content: str, model: str, usage: Dict[str, int], metadata: Dict[str, Any] = None):
self.content = content
self.model = model
self.usage = usage # {'input_tokens': int, 'output_tokens': int, 'total_tokens': int}
self.metadata = metadata or {}
class LLMProvider(ABC):
"""Abstract base class for LLM providers"""
@abstractmethod
def generate(self, prompt: str, **kwargs) -> LLMResponse:
pass
@abstractmethod
def get_models(self) -> List[str]:
pass
class OpenAIAdapter(LLMProvider):
"""Adapter for OpenAI API"""
def __init__(self, api_key: str, base_url: str = "https://api.openai.com/v1"):
self.api_key = api_key
self.base_url = base_url
self.models = ["gpt-4", "gpt-3.5-turbo", "gpt-4-turbo"]
def generate(self, prompt: str, model: str = "gpt-3.5-turbo", **kwargs) -> LLMResponse:
"""Generate response using OpenAI API format"""
# Simulate OpenAI API call
openai_request = {
"model": model,
"messages": [{"role": "user", "content": prompt}],
"temperature": kwargs.get("temperature", 0.7),
"max_tokens": kwargs.get("max_tokens", 1000)
}
# Mock OpenAI response format
openai_response = {
"choices": [{"message": {"content": f"OpenAI {model} response to: {prompt[:50]}..."}}],
"usage": {"prompt_tokens": 20, "completion_tokens": 50, "total_tokens": 70},
"model": model
}
# Adapt to standard format
return LLMResponse(
content=openai_response["choices"][0]["message"]["content"],
model=openai_response["model"],
usage={
"input_tokens": openai_response["usage"]["prompt_tokens"],
"output_tokens": openai_response["usage"]["completion_tokens"],
"total_tokens": openai_response["usage"]["total_tokens"]
},
metadata={"provider": "openai", "request": openai_request}
)
def get_models(self) -> List[str]:
return self.models
class AnthropicAdapter(LLMProvider):
"""Adapter for Anthropic Claude API"""
def __init__(self, api_key: str):
self.api_key = api_key
self.models = ["claude-3-opus", "claude-3-sonnet", "claude-3-haiku"]
def generate(self, prompt: str, model: str = "claude-3-sonnet", **kwargs) -> LLMResponse:
"""Generate response using Anthropic API format"""
# Simulate Anthropic API call
anthropic_request = {
"model": model,
"messages": [{"role": "user", "content": prompt}],
"max_tokens": kwargs.get("max_tokens", 1000),
"temperature": kwargs.get("temperature", 0.7)
}
# Mock Anthropic response format
anthropic_response = {
"content": [{"text": f"Claude {model} response to: {prompt[:50]}..."}],
"usage": {"input_tokens": 18, "output_tokens": 55},
"model": model
}
# Adapt to standard format
return LLMResponse(
content=anthropic_response["content"][0]["text"],
model=anthropic_response["model"],
usage={
"input_tokens": anthropic_response["usage"]["input_tokens"],
"output_tokens": anthropic_response["usage"]["output_tokens"],
"total_tokens": anthropic_response["usage"]["input_tokens"] + anthropic_response["usage"]["output_tokens"]
},
metadata={"provider": "anthropic", "request": anthropic_request}
)
def get_models(self) -> List[str]:
return self.models
class GoogleGeminiAdapter(LLMProvider):
"""Adapter for Google Gemini API"""
def __init__(self, api_key: str):
self.api_key = api_key
self.models = ["gemini-pro", "gemini-pro-vision", "gemini-ultra"]
def generate(self, prompt: str, model: str = "gemini-pro", **kwargs) -> LLMResponse:
"""Generate response using Google Gemini API format"""
# Simulate Gemini API call
gemini_request = {
"model": f"models/{model}",
"contents": [{"parts": [{"text": prompt}]}],
"generationConfig": {
"temperature": kwargs.get("temperature", 0.7),
"maxOutputTokens": kwargs.get("max_tokens", 1000)
}
}
# Mock Gemini response format
gemini_response = {
"candidates": [{"content": {"parts": [{"text": f"Gemini {model} response to: {prompt[:50]}..."}]}}],
"usageMetadata": {"promptTokenCount": 15, "candidatesTokenCount": 60, "totalTokenCount": 75},
"modelVersion": model
}
# Adapt to standard format
return LLMResponse(
content=gemini_response["candidates"][0]["content"]["parts"][0]["text"],
model=gemini_response["modelVersion"],
usage={
"input_tokens": gemini_response["usageMetadata"]["promptTokenCount"],
"output_tokens": gemini_response["usageMetadata"]["candidatesTokenCount"],
"total_tokens": gemini_response["usageMetadata"]["totalTokenCount"]
},
metadata={"provider": "google", "request": gemini_request}
)
def get_models(self) -> List[str]:
return self.models
class UnifiedLLMClient:
"""Unified client that uses adapter pattern to support multiple providers"""
def __init__(self):
self.providers: Dict[str, LLMProvider] = {}
self.default_provider = None
def add_provider(self, name: str, provider: LLMProvider, set_as_default: bool = False):
"""Add a new provider adapter"""
self.providers[name] = provider
if set_as_default or self.default_provider is None:
self.default_provider = name
def generate(self, prompt: str, provider: str = None, **kwargs) -> LLMResponse:
"""Generate response using specified or default provider"""
provider_name = provider or self.default_provider
if provider_name not in self.providers:
raise ValueError(f"Provider {provider_name} not found. Available: {list(self.providers.keys())}")
return self.providers[provider_name].generate(prompt, **kwargs)
def get_all_models(self) -> Dict[str, List[str]]:
"""Get all available models from all providers"""
return {name: provider.get_models() for name, provider in self.providers.items()}
# Usage example
def demonstrate_multi_provider_adapter():
"""Demonstrate unified LLM client with multiple providers"""
# Create unified client
llm_client = UnifiedLLMClient()
# Add different provider adapters
llm_client.add_provider("openai", OpenAIAdapter("sk-dummy-key"))
llm_client.add_provider("anthropic", AnthropicAdapter("sk-ant-dummy-key"))
llm_client.add_provider("google", GoogleGeminiAdapter("google-dummy-key"))
# Use same interface for different providers
prompt = "Explain quantum computing in simple terms"
for provider_name in llm_client.providers.keys():
print(f"\n🔄 Using {provider_name.upper()} provider:")
response = llm_client.generate(prompt, provider=provider_name)
print(f"📝 Response: {response.content[:100]}...")
print(f"🎯 Model: {response.model}")
print(f"📊 Usage: {response.usage}")
print(f"🏷️ Provider: {response.metadata['provider']}")
# Show all available models
print(f"\n📋 Available models: {llm_client.get_all_models()}")Benefits:
Seamless provider switching without code changes
Consistent interface across all LLM providers
Easy to add new providers
Centralized configuration and management
2. Legacy System Integration Adapter
Connect modern LLM capabilities with existing enterprise systems:
from datetime import datetime
from typing import Dict, Any, Union
import xml.etree.ElementTree as ET
class LegacySystemResponse:
"""Legacy system response format"""
def __init__(self, status: str, data: Dict[str, Any], message: str = ""):
self.status = status
self.data = data
self.message = message
self.timestamp = datetime.now()
class XMLLegacySystemAdapter:
"""Adapter for XML-based legacy systems"""
def __init__(self, system_name: str, llm_client: UnifiedLLMClient):
self.system_name = system_name
self.llm_client = llm_client
def process_xml_request(self, xml_string: str) -> str:
"""Process XML request from legacy system and return XML response"""
try:
# Parse incoming XML
root = ET.fromstring(xml_string)
request_type = root.find('type').text
content = root.find('content').text
# Convert to modern LLM request
if request_type == "SUMMARIZE":
prompt = f"Provide a concise summary of the following text:\n\n{content}"
elif request_type == "ANALYZE":
prompt = f"Analyze the following content and provide key insights:\n\n{content}"
elif request_type == "TRANSLATE":
target_lang = root.find('target_language').text
prompt = f"Translate the following text to {target_lang}:\n\n{content}"
else:
prompt = content
# Get LLM response
llm_response = self.llm_client.generate(prompt, provider="openai")
# Convert back to XML format
response_xml = self._create_xml_response(
"SUCCESS",
llm_response.content,
{"model": llm_response.model, "tokens": llm_response.usage["total_tokens"]}
)
return response_xml
except Exception as e:
return self._create_xml_response("ERROR", str(e), {})
def _create_xml_response(self, status: str, content: str, metadata: Dict[str, Any]) -> str:
"""Create XML response in legacy system format"""
root = ET.Element("response")
# Status
status_elem = ET.SubElement(root, "status")
status_elem.text = status
# Timestamp
timestamp_elem = ET.SubElement(root, "timestamp")
timestamp_elem.text = datetime.now().isoformat()
# Content
content_elem = ET.SubElement(root, "content")
content_elem.text = content
# Metadata
metadata_elem = ET.SubElement(root, "metadata")
for key, value in metadata.items():
meta_item = ET.SubElement(metadata_elem, key)
meta_item.text = str(value)
return ET.tostring(root, encoding='unicode')
class SOAPServiceAdapter:
"""Adapter for SOAP-based enterprise services"""
def __init__(self, service_url: str, llm_client: UnifiedLLMClient):
self.service_url = service_url
self.llm_client = llm_client
def process_soap_request(self, soap_envelope: str) -> str:
"""Process SOAP request and return SOAP response"""
# Simplified SOAP processing - in reality would use proper SOAP library
try:
# Extract content from SOAP envelope (simplified)
content_start = soap_envelope.find("<content>") + len("<content>")
content_end = soap_envelope.find("</content>")
content = soap_envelope[content_start:content_end]
# Process with LLM
prompt = f"Process this enterprise request: {content}"
llm_response = self.llm_client.generate(prompt, provider="anthropic")
# Return SOAP response
return f"""<?xml version="1.0" encoding="UTF-8"?>
<soap:Envelope xmlns:soap="http://schemas.xmlsoap.org/soap/envelope/">
<soap:Body>
<ProcessResponse>
<Status>SUCCESS</Status>
<Result>{llm_response.content}</Result>
<Metadata>
<Model>{llm_response.model}</Model>
<Tokens>{llm_response.usage['total_tokens']}</Tokens>
<Provider>anthropic</Provider>
</Metadata>
</ProcessResponse>
</soap:Body>
</soap:Envelope>"""
except Exception as e:
return f"""<?xml version="1.0" encoding="UTF-8"?>
<soap:Envelope xmlns:soap="http://schemas.xmlsoap.org/soap/envelope/">
<soap:Body>
<soap:Fault>
<faultcode>PROCESSING_ERROR</faultcode>
<faultstring>{str(e)}</faultstring>
</soap:Fault>
</soap:Body>
</soap:Envelope>"""
class DatabaseAdapterLLM:
"""Adapter to connect LLM with database query systems"""
def __init__(self, llm_client: UnifiedLLMClient):
self.llm_client = llm_client
self.schema_info = {}
def natural_language_to_sql(self, question: str, schema: str) -> Dict[str, Any]:
"""Convert natural language question to SQL query"""
prompt = f"""
Given the following database schema:
{schema}
Convert this natural language question to a SQL query:
"{question}"
Respond with only the SQL query, no explanation.
"""
llm_response = self.llm_client.generate(prompt, provider="google")
return {
"sql_query": llm_response.content.strip(),
"original_question": question,
"model_used": llm_response.model,
"confidence": "high" if "SELECT" in llm_response.content.upper() else "low"
}
def explain_query_results(self, sql_query: str, results: List[Dict[str, Any]]) -> str:
"""Explain database query results in natural language"""
prompt = f"""
SQL Query: {sql_query}
Results: {json.dumps(results, indent=2)}
Explain these query results in natural language, focusing on key insights and patterns.
"""
llm_response = self.llm_client.generate(prompt, provider="openai")
return llm_response.content
# Usage example
def demonstrate_legacy_integration():
"""Demonstrate legacy system integration with LLM"""
# Setup
llm_client = UnifiedLLMClient()
llm_client.add_provider("openai", OpenAIAdapter("sk-dummy-key"))
llm_client.add_provider("anthropic", AnthropicAdapter("sk-ant-dummy-key"))
llm_client.add_provider("google", GoogleGeminiAdapter("google-dummy-key"))
# XML Legacy System
xml_adapter = XMLLegacySystemAdapter("MainframeSystem", llm_client)
xml_request = """<?xml version="1.0"?>
<request>
<type>SUMMARIZE</type>
<content>
Our company's quarterly report shows significant growth in the AI sector.
Revenue increased by 45% compared to last quarter, primarily driven by
our new machine learning products. Customer satisfaction improved by 23%,
and we expanded our team by 150 new employees across engineering and sales.
</content>
</request>"""
print("🏢 Legacy XML System Request:")
xml_response = xml_adapter.process_xml_request(xml_request)
print(f"📤 Response: {xml_response[:200]}...")
# Database Integration
db_adapter = DatabaseAdapterLLM(llm_client)
schema = """
CREATE TABLE employees (
id INT PRIMARY KEY,
name VARCHAR(100),
department VARCHAR(50),
salary DECIMAL(10,2),
hire_date DATE
);
"""
question = "What is the average salary of employees in the engineering department?"
sql_result = db_adapter.natural_language_to_sql(question, schema)
print(f"\n💾 Database Query Generation:")
print(f"❓ Question: {question}")
print(f"🔍 Generated SQL: {sql_result['sql_query']}")
print(f"🎯 Confidence: {sql_result['confidence']}")Benefits:
Modernizes legacy systems without replacing them
Provides AI capabilities to existing enterprise infrastructure
Maintains compatibility with established workflows
Gradual migration path to modern architectures
3. Data Format Adaptation
Handle different data formats and protocols:
from typing import Union, Dict, Any, List
import csv
import io
import json
import yaml
class DataFormatAdapter:
"""Adapter for different data formats in LLM processing"""
def __init__(self, llm_client: UnifiedLLMClient):
self.llm_client = llm_client
def process_data(self, data: Union[str, Dict, List],
input_format: str, output_format: str,
task: str = "analyze") -> Union[str, Dict, List]:
"""Process data with format adaptation"""
# Convert input to standard format
standardized_data = self._standardize_input(data, input_format)
# Create appropriate prompt based on task
prompt = self._create_task_prompt(standardized_data, task)
# Get LLM response
llm_response = self.llm_client.generate(prompt, provider="openai")
# Convert output to requested format
return self._format_output(llm_response.content, output_format)
def _standardize_input(self, data: Union[str, Dict, List], input_format: str) -> str:
"""Convert input data to standardized string format"""
if input_format.lower() == "json":
if isinstance(data, str):
return data
return json.dumps(data, indent=2)
elif input_format.lower() == "yaml":
if isinstance(data, str):
return data
return yaml.dump(data, default_flow_style=False)
elif input_format.lower() == "csv":
if isinstance(data, str):
return data
# Convert list of dicts to CSV string
if isinstance(data, list) and data and isinstance(data[0], dict):
output = io.StringIO()
fieldnames = data[0].keys()
writer = csv.DictWriter(output, fieldnames=fieldnames)
writer.writeheader()
writer.writerows(data)
return output.getvalue()
elif input_format.lower() == "xml":
return str(data) # Assume already XML string
else:
return str(data)
def _create_task_prompt(self, data: str, task: str) -> str:
"""Create appropriate prompt based on task"""
task_prompts = {
"analyze": f"Analyze the following data and provide key insights:\n\n{data}",
"summarize": f"Provide a concise summary of this data:\n\n{data}",
"transform": f"Transform this data into a more readable format:\n\n{data}",
"extract": f"Extract the most important information from this data:\n\n{data}",
"validate": f"Validate this data and identify any issues or inconsistencies:\n\n{data}"
}
return task_prompts.get(task.lower(), f"Process the following data: {data}")
def _format_output(self, content: str, output_format: str) -> Union[str, Dict, List]:
"""Format LLM output to requested format"""
if output_format.lower() == "json":
try:
# Try to extract JSON from response
if "{" in content and "}" in content:
json_start = content.find("{")
json_end = content.rfind("}") + 1
return json.loads(content[json_start:json_end])
else:
return {"analysis": content}
except:
return {"analysis": content, "format_note": "Could not parse as JSON"}
elif output_format.lower() == "yaml":
return yaml.dump({"analysis": content}, default_flow_style=False)
elif output_format.lower() == "markdown":
return f"# Analysis Report\n\n{content}\n\n---\n*Generated by LLM*"
else:
return content
class ProtocolAdapter:
"""Adapter for different communication protocols"""
def __init__(self, llm_client: UnifiedLLMClient):
self.llm_client = llm_client
def handle_rest_request(self, method: str, endpoint: str, data: Dict[str, Any]) -> Dict[str, Any]:
"""Handle REST API requests"""
if method.upper() == "POST" and endpoint == "/analyze":
prompt = data.get("text", "")
response = self.llm_client.generate(f"Analyze: {prompt}")
return {
"status": "success",
"data": {
"analysis": response.content,
"model": response.model,
"usage": response.usage
},
"timestamp": datetime.now().isoformat()
}
def handle_graphql_request(self, query: str, variables: Dict[str, Any] = None) -> Dict[str, Any]:
"""Handle GraphQL requests"""
# Simplified GraphQL handling
if "analyze" in query.lower():
text = variables.get("text", "") if variables else ""
response = self.llm_client.generate(f"Analyze: {text}")
return {
"data": {
"analyze": {
"result": response.content,
"metadata": {
"model": response.model,
"tokens": response.usage["total_tokens"]
}
}
}
}
def handle_websocket_message(self, message: Dict[str, Any]) -> Dict[str, Any]:
"""Handle WebSocket messages"""
message_type = message.get("type", "")
if message_type == "generate":
prompt = message.get("prompt", "")
response = self.llm_client.generate(prompt)
return {
"type": "response",
"id": message.get("id"),
"data": {
"content": response.content,
"model": response.model,
"usage": response.usage
}
}
# Usage example
def demonstrate_data_format_adaptation():
"""Demonstrate data format adaptation"""
# Setup
llm_client = UnifiedLLMClient()
llm_client.add_provider("openai", OpenAIAdapter("sk-dummy-key"))
format_adapter = DataFormatAdapter(llm_client)
# JSON input to Markdown output
json_data = {
"sales": [
{"product": "AI Assistant", "revenue": 50000, "growth": 25},
{"product": "ML Platform", "revenue": 75000, "growth": 40},
{"product": "Data Analytics", "revenue": 30000, "growth": 15}
]
}
print("📊 JSON to Markdown Analysis:")
result = format_adapter.process_data(
json_data,
input_format="json",
output_format="markdown",
task="analyze"
)
print(result[:300] + "...")
# CSV input to JSON output
csv_data = """product,revenue,growth
AI Assistant,50000,25
ML Platform,75000,40
Data Analytics,30000,15"""
print("\n📈 CSV to JSON Analysis:")
result = format_adapter.process_data(
csv_data,
input_format="csv",
output_format="json",
task="summarize"
)
print(json.dumps(result, indent=2))Benefits:
Seamless data format conversion
Protocol-agnostic LLM integration
Flexible input/output handling
Easy integration with existing systems
Implementation Advantages
1. Interface Unification
Single interface for multiple providers
Consistent behavior across different systems
Reduced coupling between client code and providers
Easy provider switching and A/B testing
2. Legacy System Modernization
Gradual modernization without complete replacement
Maintains existing workflows while adding AI capabilities
Minimal disruption to established processes
Cost-effective AI integration
3. Extensibility
Easy to add new providers without changing client code
Support for future LLM providers and protocols
Pluggable architecture for different data formats
Backwards compatibility with existing integrations
4. Error Handling and Resilience
Centralized error handling across providers
Graceful degradation when providers fail
Standardized error responses
Failover mechanisms between providers
Real-World Impact
The Adapter pattern in LLM applications provides:
Vendor Independence: Avoid lock-in to specific LLM providers
Integration Flexibility: Connect AI with any existing system
Development Efficiency: Write once, work with any provider
Future-Proofing: Easy adoption of new AI technologies
This pattern is essential for enterprise LLM systems where integration with existing infrastructure and flexibility across providers are critical requirements.
Last updated