aishastr.live

Agentic Design Pattern: Adapter Wrapper

Written by

admin

in


    """
Pattern 157: Adapter/Wrapper Pattern

Description:
    Provides unified interfaces to different LLM providers and models, abstracting
    away provider-specific details. Enables easy switching between models without
    changing application code.

Components:
    - Base interface definition
    - Provider-specific adapters
    - Response normalization
    - Error handling translation
    - Configuration management

Use Cases:
    - Multi-provider support
    - Model switching
    - Vendor independence
    - Testing with different models
    - Cost optimization

Benefits:
    - Provider independence
    - Simplified integration
    - Easy model switching
    - Consistent interface
    - Testability

Trade-offs:
    - Abstraction overhead
    - May limit provider-specific features
    - Mapping complexity
    - Performance overhead

LangChain Implementation:
    Uses abstract base classes, provider-specific implementations,
    and normalization layers for consistent responses
"""

import os
from typing import Dict, Any, List, Optional, Protocol
from dataclasses import dataclass
from abc import ABC, abstractmethod
from enum import Enum
from dotenv import load_dotenv

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage, SystemMessage

load_dotenv()


class ModelProvider(Enum):
    """Supported model providers"""
    OPENAI = "openai"
    ANTHROPIC = "anthropic"
    GOOGLE = "google"
    LOCAL = "local"


@dataclass
class ModelConfig:
    """Model configuration"""
    provider: ModelProvider
    model_name: str
    temperature: float = 0.7
    max_tokens: Optional[int] = None
    api_key: Optional[str] = None


@dataclass
class Message:
    """Standardized message format"""
    role: str  # system, user, assistant
    content: str
    metadata: Dict[str, Any] = None
    
    def __post_init__(self):
        if self.metadata is None:
            self.metadata = {}


@dataclass
class CompletionResponse:
    """Standardized completion response"""
    content: str
    model: str
    provider: ModelProvider
    tokens_used: Optional[int] = None
    finish_reason: Optional[str] = None
    metadata: Dict[str, Any] = None
    
    def __post_init__(self):
        if self.metadata is None:
            self.metadata = {}


class LLMAdapter(ABC):
    """Abstract base class for LLM adapters"""
    
    @abstractmethod
    def complete(self, messages: List[Message], **kwargs) -> CompletionResponse:
        """Generate completion from messages"""
        pass
    
    @abstractmethod
    def stream(self, messages: List[Message], **kwargs):
        """Stream completion"""
        pass
    
    @abstractmethod
    def get_model_info(self) -> Dict[str, Any]:
        """Get model information"""
        pass


class OpenAIAdapter(LLMAdapter):
    """Adapter for OpenAI models"""
    
    def __init__(self, config: ModelConfig):
        """Initialize OpenAI adapter"""
        self.config = config
        self.llm = ChatOpenAI(
            model=config.model_name,
            temperature=config.temperature,
            max_tokens=config.max_tokens,
            api_key=config.api_key or os.getenv("OPENAI_API_KEY")
        )
    
    def complete(self, messages: List[Message], **kwargs) -> CompletionResponse:
        """Generate completion"""
        # Convert messages to LangChain format
        lc_messages = self._convert_messages(messages)
        
        try:
            # Generate response
            response = self.llm.invoke(lc_messages)
            
            return CompletionResponse(
                content=response.content,
                model=self.config.model_name,
                provider=ModelProvider.OPENAI,
                finish_reason="stop",
                metadata={
                    "response_type": type(response).__name__
                }
            )
        except Exception as e:
            raise RuntimeError(f"OpenAI completion failed: {str(e)}")
    
    def stream(self, messages: List[Message], **kwargs):
        """Stream completion"""
        lc_messages = self._convert_messages(messages)
        
        try:
            for chunk in self.llm.stream(lc_messages):
                yield chunk.content
        except Exception as e:
            raise RuntimeError(f"OpenAI streaming failed: {str(e)}")
    
    def get_model_info(self) -> Dict[str, Any]:
        """Get model information"""
        return {
            "provider": ModelProvider.OPENAI.value,
            "model": self.config.model_name,
            "temperature": self.config.temperature,
            "max_tokens": self.config.max_tokens
        }
    
    def _convert_messages(self, messages: List[Message]) -> List[BaseMessage]:
        """Convert standard messages to LangChain messages"""
        lc_messages = []
        for msg in messages:
            if msg.role == "system":
                lc_messages.append(SystemMessage(content=msg.content))
            elif msg.role == "user":
                lc_messages.append(HumanMessage(content=msg.content))
            elif msg.role == "assistant":
                lc_messages.append(AIMessage(content=msg.content))
        return lc_messages


class AnthropicAdapter(LLMAdapter):
    """Adapter for Anthropic models (simulated)"""
    
    def __init__(self, config: ModelConfig):
        """Initialize Anthropic adapter"""
        self.config = config
        # In real implementation, use actual Anthropic client
        # For demo, we'll use OpenAI as fallback
        self.llm = ChatOpenAI(
            model="gpt-3.5-turbo",
            temperature=config.temperature
        )
    
    def complete(self, messages: List[Message], **kwargs) -> CompletionResponse:
        """Generate completion"""
        # Convert to format expected by Anthropic (simulated)
        prompt = self._format_anthropic_prompt(messages)
        
        try:
            response = self.llm.invoke([HumanMessage(content=prompt)])
            
            return CompletionResponse(
                content=response.content,
                model=self.config.model_name,
                provider=ModelProvider.ANTHROPIC,
                finish_reason="end_turn",
                metadata={
                    "simulated": True
                }
            )
        except Exception as e:
            raise RuntimeError(f"Anthropic completion failed: {str(e)}")
    
    def stream(self, messages: List[Message], **kwargs):
        """Stream completion"""
        prompt = self._format_anthropic_prompt(messages)
        
        try:
            for chunk in self.llm.stream([HumanMessage(content=prompt)]):
                yield chunk.content
        except Exception as e:
            raise RuntimeError(f"Anthropic streaming failed: {str(e)}")
    
    def get_model_info(self) -> Dict[str, Any]:
        """Get model information"""
        return {
            "provider": ModelProvider.ANTHROPIC.value,
            "model": self.config.model_name,
            "temperature": self.config.temperature,
            "simulated": True
        }
    
    def _format_anthropic_prompt(self, messages: List[Message]) -> str:
        """Format messages for Anthropic API"""
        # Anthropic uses a different format
        formatted = []
        for msg in messages:
            if msg.role == "system":
                formatted.append(f"System: {msg.content}")
            elif msg.role == "user":
                formatted.append(f"Human: {msg.content}")
            elif msg.role == "assistant":
                formatted.append(f"Assistant: {msg.content}")
        formatted.append("Assistant:")
        return "\n\n".join(formatted)


class GoogleAdapter(LLMAdapter):
    """Adapter for Google models (simulated)"""
    
    def __init__(self, config: ModelConfig):
        """Initialize Google adapter"""
        self.config = config
        # In real implementation, use actual Google client
        self.llm = ChatOpenAI(
            model="gpt-3.5-turbo",
            temperature=config.temperature
        )
    
    def complete(self, messages: List[Message], **kwargs) -> CompletionResponse:
        """Generate completion"""
        # Convert to Google format (simulated)
        lc_messages = []
        for msg in messages:
            if msg.role == "system":
                lc_messages.append(SystemMessage(content=msg.content))
            elif msg.role == "user":
                lc_messages.append(HumanMessage(content=msg.content))
        
        try:
            response = self.llm.invoke(lc_messages)
            
            return CompletionResponse(
                content=response.content,
                model=self.config.model_name,
                provider=ModelProvider.GOOGLE,
                finish_reason="STOP",
                metadata={
                    "simulated": True
                }
            )
        except Exception as e:
            raise RuntimeError(f"Google completion failed: {str(e)}")
    
    def stream(self, messages: List[Message], **kwargs):
        """Stream completion"""
        lc_messages = []
        for msg in messages:
            if msg.role == "user":
                lc_messages.append(HumanMessage(content=msg.content))
        
        try:
            for chunk in self.llm.stream(lc_messages):
                yield chunk.content
        except Exception as e:
            raise RuntimeError(f"Google streaming failed: {str(e)}")
    
    def get_model_info(self) -> Dict[str, Any]:
        """Get model information"""
        return {
            "provider": ModelProvider.GOOGLE.value,
            "model": self.config.model_name,
            "temperature": self.config.temperature,
            "simulated": True
        }


class UnifiedLLMInterface:
    """Unified interface for multiple LLM providers"""
    
    def __init__(self):
        """Initialize unified interface"""
        self.adapters: Dict[ModelProvider, LLMAdapter] = {}
        self.default_provider: Optional[ModelProvider] = None
    
    def register_adapter(self, provider: ModelProvider, adapter: LLMAdapter):
        """Register a provider adapter"""
        self.adapters[provider] = adapter
        if self.default_provider is None:
            self.default_provider = provider
        print(f"✓ Registered adapter for {provider.value}")
    
    def set_default_provider(self, provider: ModelProvider):
        """Set default provider"""
        if provider not in self.adapters:
            raise ValueError(f"Provider {provider.value} not registered")
        self.default_provider = provider
        print(f"✓ Set default provider to {provider.value}")
    
    def complete(self, messages: List[Message], 
                provider: Optional[ModelProvider] = None, 
                **kwargs) -> CompletionResponse:
        """Generate completion using specified or default provider"""
        provider = provider or self.default_provider
        
        if provider not in self.adapters:
            raise ValueError(f"Provider {provider.value} not registered")
        
        adapter = self.adapters[provider]
        return adapter.complete(messages, **kwargs)
    
    def stream(self, messages: List[Message], 
              provider: Optional[ModelProvider] = None, 
              **kwargs):
        """Stream completion using specified or default provider"""
        provider = provider or self.default_provider
        
        if provider not in self.adapters:
            raise ValueError(f"Provider {provider.value} not registered")
        
        adapter = self.adapters[provider]
        return adapter.stream(messages, **kwargs)
    
    def complete_with_fallback(self, messages: List[Message], 
                              providers: Optional[List[ModelProvider]] = None,
                              **kwargs) -> CompletionResponse:
        """Try completion with fallback providers"""
        if providers is None:
            providers = [self.default_provider] + [
                p for p in self.adapters.keys() if p != self.default_provider
            ]
        
        last_error = None
        for provider in providers:
            try:
                print(f"Trying provider: {provider.value}")
                return self.complete(messages, provider=provider, **kwargs)
            except Exception as e:
                print(f"  Failed: {str(e)}")
                last_error = e
                continue
        
        raise RuntimeError(f"All providers failed. Last error: {last_error}")
    
    def get_all_model_info(self) -> Dict[str, Dict[str, Any]]:
        """Get information about all registered models"""
        return {
            provider.value: adapter.get_model_info()
            for provider, adapter in self.adapters.items()
        }


def demonstrate_adapter_pattern():
    """Demonstrate adapter/wrapper pattern"""
    print("=" * 80)
    print("ADAPTER/WRAPPER PATTERN DEMONSTRATION")
    print("=" * 80)
    
    # Initialize unified interface
    interface = UnifiedLLMInterface()
    
    # Register adapters for different providers
    print("\n" + "="*80)
    print("SETUP: Registering Provider Adapters")
    print("="*80)
    
    openai_config = ModelConfig(
        provider=ModelProvider.OPENAI,
        model_name="gpt-3.5-turbo",
        temperature=0.7
    )
    interface.register_adapter(
        ModelProvider.OPENAI,
        OpenAIAdapter(openai_config)
    )
    
    anthropic_config = ModelConfig(
        provider=ModelProvider.ANTHROPIC,
        model_name="claude-3-sonnet",
        temperature=0.7
    )
    interface.register_adapter(
        ModelProvider.ANTHROPIC,
        AnthropicAdapter(anthropic_config)
    )
    
    google_config = ModelConfig(
        provider=ModelProvider.GOOGLE,
        model_name="gemini-pro",
        temperature=0.7
    )
    interface.register_adapter(
        ModelProvider.GOOGLE,
        GoogleAdapter(google_config)
    )
    
    # Example 1: Complete with default provider
    print("\n" + "="*80)
    print("EXAMPLE 1: Completion with Default Provider")
    print("="*80)
    messages = [
        Message(role="system", content="You are a helpful assistant."),
        Message(role="user", content="What is the capital of France?")
    ]
    response = interface.complete(messages)
    print(f"Provider: {response.provider.value}")
    print(f"Model: {response.model}")
    print(f"Response: {response.content[:200]}...")
    
    # Example 2: Complete with specific provider
    print("\n" + "="*80)
    print("EXAMPLE 2: Completion with Specific Provider (Anthropic)")
    print("="*80)
    response = interface.complete(messages, provider=ModelProvider.ANTHROPIC)
    print(f"Provider: {response.provider.value}")
    print(f"Model: {response.model}")
    print(f"Response: {response.content[:200]}...")
    
    # Example 3: Switch between providers
    print("\n" + "="*80)
    print("EXAMPLE 3: Switching Between Providers")
    print("="*80)
    question = Message(role="user", content="Explain quantum computing briefly.")
    
    for provider in [ModelProvider.OPENAI, ModelProvider.GOOGLE, ModelProvider.ANTHROPIC]:
        response = interface.complete([question], provider=provider)
        print(f"\n{provider.value}:")
        print(f"  {response.content[:150]}...")
    
    # Example 4: Streaming response
    print("\n" + "="*80)
    print("EXAMPLE 4: Streaming Response")
    print("="*80)
    stream_messages = [
        Message(role="user", content="Count from 1 to 5.")
    ]
    print("Streaming: ", end="", flush=True)
    for chunk in interface.stream(stream_messages):
        print(chunk, end="", flush=True)
    print()
    
    # Example 5: Fallback mechanism
    print("\n" + "="*80)
    print("EXAMPLE 5: Fallback Mechanism")
    print("="*80)
    messages = [
        Message(role="user", content="What is machine learning?")
    ]
    response = interface.complete_with_fallback(messages)
    print(f"Succeeded with provider: {response.provider.value}")
    print(f"Response: {response.content[:200]}...")
    
    # Example 6: Model information
    print("\n" + "="*80)
    print("EXAMPLE 6: Model Information")
    print("="*80)
    all_info = interface.get_all_model_info()
    for provider, info in all_info.items():
        print(f"\n{provider}:")
        for key, value in info.items():
            print(f"  {key}: {value}")
    
    # Example 7: Conversation across providers
    print("\n" + "="*80)
    print("EXAMPLE 7: Multi-Turn Conversation")
    print("="*80)
    conversation = [
        Message(role="user", content="What is Python?")
    ]
    
    # First turn with OpenAI
    response1 = interface.complete(conversation, provider=ModelProvider.OPENAI)
    print(f"Turn 1 ({response1.provider.value}): {response1.content[:100]}...")
    
    # Add response and continue
    conversation.append(Message(role="assistant", content=response1.content))
    conversation.append(Message(role="user", content="What are its main features?"))
    
    # Second turn with different provider
    response2 = interface.complete(conversation, provider=ModelProvider.GOOGLE)
    print(f"Turn 2 ({response2.provider.value}): {response2.content[:100]}...")
    
    # Summary
    print("\n" + "="*80)
    print("SUMMARY: Adapter/Wrapper Pattern Best Practices")
    print("="*80)
    print("""
1. INTERFACE DESIGN:
   - Define clear, provider-agnostic interface
   - Standardize message formats
   - Normalize response structures
   - Handle provider-specific features gracefully

2. ADAPTER IMPLEMENTATION:
   - Encapsulate provider-specific logic
   - Handle format conversions
   - Map error codes consistently
   - Maintain provider-specific optimizations

3. RESPONSE NORMALIZATION:
   - Standardize response format
   - Map finish reasons consistently
   - Include provider metadata
   - Handle streaming uniformly

4. ERROR HANDLING:
   - Translate provider-specific errors
   - Provide meaningful error messages
   - Implement fallback mechanisms
   - Log provider-specific details

5. CONFIGURATION:
   - Centralize configuration management
   - Support provider-specific parameters
   - Enable easy provider switching
   - Validate configurations

6. TESTING:
   - Test each adapter independently
   - Verify format conversions
   - Test fallback mechanisms
   - Mock external API calls

Benefits:
✓ Provider independence and flexibility
✓ Easy model switching
✓ Simplified application code
✓ Consistent interface across providers
✓ Facilitates testing and mocking
✓ Vendor lock-in prevention

Challenges:
- Abstraction may limit provider features
- Mapping complexity for edge cases
- Performance overhead
- Maintaining feature parity
- Handling provider-specific capabilities
    """)


if __name__ == "__main__":
    demonstrate_adapter_pattern()

Agentic Design Pattern: Adapter Wrapper

Pattern 157: Adapter Wrapper – Line-by-Line Explanation

Overview

This document provides a detailed line-by-line explanation of the Adapter Wrapper pattern implementation.

Pattern Description

Pattern 157: Adapter/Wrapper Pattern

Description: Provides unified interfaces to different LLM providers and models, abstracting away provider-specific details. Enables easy switching between models without changing application code.

Components: – Base interface definition – Provider-specific adapters – Response normalization – Error handling translation – Configuration management

Use Cases: – Multi-provider support – Model switching – Vendor independence – Testing with different models – Cost optimization

Benefits: – Provider independence – Simplified integration – Easy model switching – Consistent interface – Testability

Trade-offs: – Abstraction overhead – May limit provider-specific features – Mapping complexity – Performance overhead

LangChain Implementation: Uses abstract base classes, provider-specific implementations, and normalization layers for consistent responses

File Statistics

  • Total Lines: 578
  • Classes: 9
  • Functions: 1
  • Constants: 0

Module Documentation

"""
Pattern 157: Adapter/Wrapper Pattern

Description:
    Provides unified interfaces to different LLM providers and models, abstracting
    away provider-specific details. Enables easy switching between models without
    changing application code.

Components:
    - Base interface definition
    - Provider-specific adapters
    - Response normalization
    - Error handling translation
    - Configuration management

Use Cases:
    - Multi-provider support
    - Model switching
    - Vendor independence
    - Testing with different models
    - Cost optimization

Benefits:
    - Provider independence
    - Simplified integration
    - Easy model switching
    - Consistent interface
    - Testability

Trade-offs:
    - Abstraction overhead
    - May limit provider-specific features
    - Mapping complexity
    - Performance overhead

LangChain Implementation:
    Uses abstract base classes, provider-specific implementations,
    and normalization layers for consistent responses
"""

Purpose: This module implements the Adapter Wrapper pattern, providing a structured approach to agent design and implementation.

Import Statements

Standard Library Imports

import dataclasses.dataclass
import enum.Enum
import langchain_core.messages.AIMessage
import langchain_core.messages.BaseMessage
import langchain_core.messages.HumanMessage
import langchain_core.messages.SystemMessage
import langchain_core.output_parsers.StrOutputParser
import langchain_core.prompts.ChatPromptTemplate
import os
import typing.Any
import typing.Dict
import typing.List
import typing.Optional
import typing.Protocol

Purpose: Core Python functionality: – dataclasses.dataclass: Simplified class definitions with @dataclass – enum.Enum: Enumeration support – langchain_core.messages.AIMessage: Regular expression operations – langchain_core.messages.BaseMessage: Regular expression operations – langchain_core.messages.HumanMessage: Regular expression operations – langchain_core.messages.SystemMessage: Regular expression operations – langchain_core.output_parsers.StrOutputParser: Regular expression operations – langchain_core.prompts.ChatPromptTemplate: Regular expression operations – os: Environment variable access and system operations – typing.Any: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Dict: Type hints for better code documentation (List, Dict, Any, Optional) – typing.List: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Optional: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Protocol: Type hints for better code documentation (List, Dict, Any, Optional)

LangChain Imports

from langchain_openai.ChatOpenAI

Purpose: LangChain framework components: – LLM Interface: Wrapper for OpenAI models – Chains: Composable building blocks for complex workflows

Other Imports

from abc.ABC
from abc.abstractmethod
from dotenv.load_dotenv

Purpose: Additional dependencies for specialized functionality.

Classes

Class: ModelProvider

Lines 56-61

Inherits from: Enum

Purpose:

Supported model providers

Class: ModelConfig

Lines 65-71

Purpose:

Model configuration

Class: Message

Lines 75-83

Purpose:

Standardized message format

Methods:

__post_init__(self)

Lines 81-83

Purpose: Post Init operation.

Parameters:

Class: CompletionResponse

Lines 87-98

Purpose:

Standardized completion response

Methods:

__post_init__(self)

Lines 96-98

Purpose: Post Init operation.

Parameters:

Class: LLMAdapter

Lines 101-117

Inherits from: ABC

Purpose:

Abstract base class for LLM adapters

Methods:

complete(self, messages)

Lines 105-107

Decorators: abstractmethod

Purpose:

Generate completion from messages

Parameters: – messages: Messages

Returns: CompletionResponse

stream(self, messages)

Lines 110-112

Decorators: abstractmethod

Purpose:

Stream completion

Parameters: – messages: Messages

get_model_info(self)

Lines 115-117

Decorators: abstractmethod

Purpose:

Get model information

Parameters:

Returns: <ast.Subscript object at 0x000001A06D75BC50>

Class: OpenAIAdapter

Lines 120-183

Inherits from: LLMAdapter

Purpose:

Adapter for OpenAI models

Methods:

__init__(self, config)

Lines 123-131

Purpose:

Initialize OpenAI adapter

Parameters: – config: Config

complete(self, messages)

Lines 133-152

Purpose:

Generate completion

Parameters: – messages: Messages

Returns: CompletionResponse

stream(self, messages)

Lines 154-162

Purpose:

Stream completion

Parameters: – messages: Messages

get_model_info(self)

Lines 164-171

Purpose:

Get model information

Parameters:

Returns: <ast.Subscript object at 0x000001A06D753690>

_convert_messages(self, messages)

Lines 173-183

Purpose:

Convert standard messages to LangChain messages

Parameters: – messages: Messages

Returns: <ast.Subscript object at 0x000001A06D7CB290>

Class: AnthropicAdapter

Lines 186-250

Inherits from: LLMAdapter

Purpose:

Adapter for Anthropic models (simulated)

Methods:

__init__(self, config)

Lines 189-197

Purpose:

Initialize Anthropic adapter

Parameters: – config: Config

complete(self, messages)

Lines 199-217

Purpose:

Generate completion

Parameters: – messages: Messages

Returns: CompletionResponse

stream(self, messages)

Lines 219-227

Purpose:

Stream completion

Parameters: – messages: Messages

get_model_info(self)

Lines 229-236

Purpose:

Get model information

Parameters:

Returns: <ast.Subscript object at 0x000001A06D7AB310>

_format_anthropic_prompt(self, messages)

Lines 238-250

Purpose:

Format messages for Anthropic API

Parameters: – messages: Messages

Returns: str

Class: GoogleAdapter

Lines 253-310

Inherits from: LLMAdapter

Purpose:

Adapter for Google models (simulated)

Methods:

__init__(self, config)

Lines 256-263

Purpose:

Initialize Google adapter

Parameters: – config: Config

complete(self, messages)

Lines 265-288

Purpose:

Generate completion

Parameters: – messages: Messages

Returns: CompletionResponse

stream(self, messages)

Lines 290-301

Purpose:

Stream completion

Parameters: – messages: Messages

get_model_info(self)

Lines 303-310

Purpose:

Get model information

Parameters:

Returns: <ast.Subscript object at 0x000001A06D80FD50>

Class: UnifiedLLMInterface

Lines 313-385

Purpose:

Unified interface for multiple LLM providers

Methods:

__init__(self)

Lines 316-319

Purpose:

Initialize unified interface

Parameters:

register_adapter(self, provider, adapter)

Lines 321-326

Purpose:

Register a provider adapter

Parameters: – provider: Provider – adapter: Adapter

set_default_provider(self, provider)

Lines 328-333

Purpose:

Set default provider

Parameters: – provider: Provider

complete(self, messages, provider)

Lines 335-345

Purpose:

Generate completion using specified or default provider

Parameters: – messages: Messages – provider: Provider

Returns: CompletionResponse

stream(self, messages, provider)

Lines 347-357

Purpose:

Stream completion using specified or default provider

Parameters: – messages: Messages – provider: Provider

complete_with_fallback(self, messages, providers)

Lines 359-378

Purpose:

Try completion with fallback providers

Parameters: – messages: Messages – providers: Providers

Returns: CompletionResponse

get_all_model_info(self)

Lines 380-385

Purpose:

Get information about all registered models

Parameters:

Returns: <ast.Subscript object at 0x000001A06D622590>

Functions

demonstrate_adapter_pattern()

Lines 388-573

Purpose:

Demonstrate adapter/wrapper pattern

Key Concepts Summary

Pattern Overview

The Adapter Wrapper pattern provides a structured approach to building AI agents with specific capabilities and behaviors. This implementation leverages LangChain’s composable components to create a robust and maintainable solution.

Core Components

  1. Agent Class: Main implementation of the pattern logic
  2. Prompt Engineering: Carefully crafted prompts to guide LLM behavior
  3. Tool Integration: Optional tools for extended functionality
  4. Memory Management: State handling across interactions
  5. Output Parsing: Structured response processing

When to Use

✅ Good for: – Scenarios requiring this specific pattern’s strengths – Applications needing structured agent behavior – Systems where explainability is important – Production environments requiring reliability

❌ Not ideal for: – Simple single-prompt tasks – Scenarios requiring different pattern characteristics – Resource-constrained environments (if computationally intensive)

Best Practices Demonstrated

1. Environment Configuration

  • ✅ Use .env files for API keys
  • ✅ Never hardcode credentials
  • ✅ Validate environment variables on startup

2. Error Handling

  • ✅ Graceful degradation on failures
  • ✅ Informative error messages
  • ✅ Logging for debugging

3. Prompt Engineering

  • ✅ Clear, specific instructions
  • ✅ Structured output formats
  • ✅ Few-shot examples when beneficial

4. Code Organization

  • ✅ Separation of concerns (class-based design)
  • ✅ Type hints for clarity
  • ✅ Docstrings for documentation

5. Performance Optimization

  • ✅ Appropriate temperature settings
  • ✅ Token budget management
  • ✅ Caching where applicable

LangChain Components Used

  • ChatOpenAI: OpenAI model interface
  • Chains: Sequential component composition

Why LangChain? – Composable abstractions for LLM applications – Rich ecosystem of integrations – Production-ready patterns – Active community and development

Execution Flow

1. Initialize agent with configuration
   ↓
2. Load environment and dependencies
   ↓
3. Prepare prompt with input data
   ↓
4. Invoke LLM through LangChain
   ↓
5. Parse and validate response
   ↓
6. Return formatted result

Key Decision Points: – Model selection (temperature, model type) – Prompt structure (few-shot vs zero-shot) – Error handling strategy – Output format validation

Performance Considerations

Computational Cost

  • LLM Calls: Primary cost factor (based on tokens)
  • Latency: Depends on model and input size
  • Memory: Minimal for most patterns

Optimization Strategies

  1. Prompt Optimization: Minimize tokens while maintaining clarity
  2. Caching: Cache repeated queries/responses
  3. Model Selection: Use smaller models when appropriate
  4. Batch Processing: Group similar requests
  5. Streaming: For long responses, stream tokens

Cost Estimates (Approximate)

  • GPT-3.5-turbo: ~$0.002 per request
  • GPT-4: ~$0.03 per request
  • GPT-4-turbo: ~$0.01 per request

Actual costs vary based on input/output token counts

Common Pitfalls to Avoid

1. Prompt Design Issues

❌ Vague or ambiguous instructions
✅ Clear, specific, testable prompts

2. Error Handling

❌ Ignoring API errors or rate limits
✅ Implement retry logic and graceful degradation

3. Token Management

❌ Exceeding model token limits
✅ Truncate inputs, chunk processing

4. Security

❌ Exposing API keys in code
✅ Use environment variables

5. Testing

❌ Only testing happy paths
✅ Test edge cases, failures, malformed inputs

6. Cost Management

❌ Unmonitored API usage
✅ Implement budgets, alerts, and monitoring

Further Improvements

Production Enhancements

  1. Monitoring: Add logging, metrics, tracing
  2. Validation: Input/output schema validation
  3. Rate Limiting: Implement request throttling
  4. Caching: Redis or in-memory cache for responses
  5. Async Support: Non-blocking operations for scalability

Feature Extensions

  1. Multi-Model Support: Switch between providers
  2. Fallback Models: Graceful degradation strategy
  3. Custom Tools: Domain-specific function calling
  4. Advanced Memory: Vector stores for long-term memory
  5. Evaluation: Automated testing and benchmarking

Code Quality

  1. Testing: Unit tests, integration tests
  2. Documentation: API docs, usage examples
  3. Type Safety: Stricter type hints
  4. Configuration: Externalized config files

Related Patterns

  • Pattern 156
  • Pattern 158

Pattern Families: – Reasoning Patterns (001-011) – Multi-Agent Patterns (012-019) – Tool Use Patterns (020-025) – Memory Patterns (026-031) – And many more…

Summary

The Adapter Wrapper pattern provides a powerful approach to building AI agents with specific capabilities. This implementation demonstrates:

✅ Key Strengths: – Clean, maintainable code structure – LangChain integration for composability – Type hints and documentation – Production-ready patterns

🎯 Best Use Cases: – Applications requiring this pattern’s characteristics – Systems needing structured agent behavior – Production environments requiring reliability

📚 Learning Resources: – LangChain Documentation: https://python.langchain.com/ – Pattern Documentation: See related patterns – OpenAI API Reference: https://platform.openai.com/docs/

File: 157_adapter_wrapper.py
Total Lines: 578
Pattern Category: AI Agent Design Pattern #157

Generated from actual implementation code
For educational and reference purposes

Comments

Leave a Reply

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

More posts