LangChain with OpenAI vs Claude vs Gemini vs Local LLMs: Performance Analysis

from langchain.llms import OpenAI
from langchain.chat_models import ChatOpenAI
from langchain.callbacks import get_openai_callback
import time

# Initialize OpenAI model
chat_model = ChatOpenAI(
    model_name="gpt-4",
    temperature=0.1,
    max_tokens=1000,
    request_timeout=30
)

# Performance tracking
def benchmark_openai(prompt, iterations=100):
    latencies = []
    costs = []
    
    for i in range(iterations):
        start_time = time.time()
        
        with get_openai_callback() as cb:
            response = chat_model.predict(prompt)
            
        latency = (time.time() - start_time) * 1000  # ms
        latencies.append(latency)
        costs.append(cb.total_cost)
    
    return {
        'avg_latency': sum(latencies) / len(latencies),
        'avg_cost': sum(costs) / len(costs),
        'throughput': len(prompt.split()) / (sum(latencies) / 1000 / len(latencies))
    }

# Benchmark results
results = benchmark_openai("Explain quantum computing in simple terms")
print("Average Latency: {:.2f}ms".format(results['avg_latency']))
print("Average Cost: " + "{:.4f}".format(results['avg_cost']))
print("Throughput: {:.2f} tokens/sec".format(results['throughput']))

from langchain.chat_models import ChatAnthropic
from langchain.callbacks import get_anthropic_callback
import time

# Initialize Claude model
claude_model = ChatAnthropic(
    model="claude-3-opus-20240229",
    temperature=0.1,
    max_tokens=1000,
    timeout=30
)

# Performance tracking for Claude
def benchmark_claude(prompt, iterations=100):
    latencies = []
    costs = []
    
    for i in range(iterations):
        start_time = time.time()
        
        with get_anthropic_callback() as cb:
            response = claude_model.predict(prompt)
            
        latency = (time.time() - start_time) * 1000
        latencies.append(latency)
        costs.append(cb.total_cost)
    
    return {
        'avg_latency': sum(latencies) / len(latencies),
        'avg_cost': sum(costs) / len(costs),
        'quality_score': evaluate_response_quality(response)
    }

# Advanced use case: Code analysis
def analyze_code_with_claude(code_snippet):
    prompt = f"""
    Analyze this code for potential issues, performance optimizations, 
    and security vulnerabilities:
    
    {code_snippet}
    
    Provide detailed feedback with specific recommendations.
    """
    
    start_time = time.time()
    response = claude_model.predict(prompt)
    processing_time = time.time() - start_time
    
    return {
        'analysis': response,
        'processing_time': processing_time,
        'tokens_per_second': len(response.split()) / processing_time
    }

# Benchmark complex reasoning task
reasoning_prompt = """
Solve this multi-step problem:
1. Calculate the ROI of implementing AI in a company with 1000 employees
2. Consider implementation costs, training, and productivity gains
3. Provide a 3-year projection with monthly breakdown
"""

claude_results = benchmark_claude(reasoning_prompt)
print("Claude Latency: {:.2f}ms".format(claude_results['avg_latency']))
print("Claude Cost: " + "{:.4f}".format(claude_results['avg_cost']))

from langchain.llms import GooglePalm
from langchain.chat_models import ChatGooglePalm
import google.generativeai as genai
import time

# Initialize Gemini model
genai.configure(api_key="your-api-key")
gemini_model = ChatGooglePalm(
    model_name="gemini-pro",
    temperature=0.1,
    max_output_tokens=1000
)

# Performance tracking for Gemini
def benchmark_gemini(prompt, iterations=100):
    latencies = []
    token_usage = []
    
    for i in range(iterations):
        start_time = time.time()
        
        response = gemini_model.predict(prompt)
        
        latency = (time.time() - start_time) * 1000
        latencies.append(latency)
        token_usage.append(len(response.split()) + len(prompt.split()))
    
    return {
        'avg_latency': sum(latencies) / len(latencies),
        'avg_tokens': sum(token_usage) / len(token_usage),
        'cost_per_request': (sum(token_usage) / len(token_usage)) * 0.001  # Gemini pricing
    }

# High-volume processing test
def process_batch_with_gemini(prompts_batch):
    results = []
    start_time = time.time()
    
    for prompt in prompts_batch:
        response = gemini_model.predict(prompt)
        results.append({
            'prompt': prompt,
            'response': response,
            'timestamp': time.time()
        })
    
    total_time = time.time() - start_time
    
    return {
        'results': results,
        'total_time': total_time,
        'requests_per_second': len(prompts_batch) / total_time,
        'total_cost': len(prompts_batch) * 0.001  # Estimated cost
    }

# Test batch processing
test_prompts = [
    "Summarize the key points of quantum computing",
    "Explain machine learning in simple terms",
    "What are the benefits of renewable energy?",
    "Describe the process of photosynthesis",
    "How does blockchain technology work?"
]

batch_results = process_batch_with_gemini(test_prompts)
print("Processed {} requests in {:.2f}s".format(len(test_prompts), batch_results['total_time']))
print("Cost: " + "{:.4f}".format(batch_results['total_cost']))

from langchain.llms import LlamaCpp
from langchain.callbacks.manager import CallbackManager
from langchain.callbacks.streaming_stdout import StreamingStdOutCallbackHandler
import psutil
import time

# Initialize local model
callback_manager = CallbackManager([StreamingStdOutCallbackHandler()])

local_llm = LlamaCpp(
    model_path="./models/llama-2-7b-chat.ggmlv3.q4_0.bin",
    temperature=0.1,
    max_tokens=1000,
    top_p=1,
    callback_manager=callback_manager,
    verbose=True,
    n_ctx=2048,
    n_gpu_layers=35  # Use GPU acceleration
)

# Performance tracking for local models
def benchmark_local_llm(prompt, iterations=50):
    latencies = []
    cpu_usage = []
    memory_usage = []
    
    for i in range(iterations):
        # Monitor system resources
        cpu_before = psutil.cpu_percent()
        memory_before = psutil.virtual_memory().percent
        
        start_time = time.time()
        response = local_llm.predict(prompt)
        latency = (time.time() - start_time) * 1000
        
        cpu_after = psutil.cpu_percent()
        memory_after = psutil.virtual_memory().percent
        
        latencies.append(latency)
        cpu_usage.append(cpu_after - cpu_before)
        memory_usage.append(memory_after - memory_before)
    
    return {
        'avg_latency': sum(latencies) / len(latencies),
        'avg_cpu_usage': sum(cpu_usage) / len(cpu_usage),
        'avg_memory_usage': sum(memory_usage) / len(memory_usage),
        'infrastructure_cost': calculate_infrastructure_cost()
    }

# Infrastructure cost calculation
def calculate_infrastructure_cost():
    # Assuming AWS p3.2xlarge instance
    hourly_rate = 3.06  # USD per hour
    utilization = 0.7  # 70% utilization
    cost_per_hour = hourly_rate * utilization
    
    # Cost per 1K tokens (estimated)
    tokens_per_hour = 50000  # Conservative estimate
    cost_per_1k_tokens = (cost_per_hour / tokens_per_hour) * 1000
    
    return {
        'hourly_cost': cost_per_hour,
        'cost_per_1k_tokens': cost_per_1k_tokens,
        'monthly_cost': cost_per_hour * 24 * 30
    }

# Benchmark local model
local_results = benchmark_local_llm("Explain the benefits of edge computing")
print("Local LLM Latency: {:.2f}ms".format(local_results['avg_latency']))
print("CPU Usage: {:.1f}%".format(local_results['avg_cpu_usage']))
print("Memory Usage: {:.1f}%".format(local_results['avg_memory_usage']))

from langchain.llms import OpenAI
from langchain.chat_models import ChatOpenAI, ChatAnthropic
from langchain.callbacks import get_openai_callback, get_anthropic_callback
import time
import random
from typing import List, Dict, Any
from dataclasses import dataclass
from enum import Enum

class ProviderStatus(Enum):
    HEALTHY = "healthy"
    DEGRADED = "degraded"
    UNAVAILABLE = "unavailable"

@dataclass
class ProviderConfig:
    name: str
    model: Any
    priority: int
    max_retries: int
    timeout: int
    cost_per_token: float
    quality_score: float
    status: ProviderStatus = ProviderStatus.HEALTHY

class MultiProviderLLMManager:
    def __init__(self):
        self.providers = self._initialize_providers()
        self.usage_stats = {provider.name: {'calls': 0, 'failures': 0, 'total_cost': 0} 
                           for provider in self.providers}
        self.circuit_breaker = {}
    
    def _initialize_providers(self) -> List[ProviderConfig]:
        return [
            ProviderConfig(
                name="openai_gpt4",
                model=ChatOpenAI(model_name="gpt-4", temperature=0.1),
                priority=1,
                max_retries=3,
                timeout=30,
                cost_per_token=0.03,
                quality_score=9.2
            ),
            ProviderConfig(
                name="claude_opus",
                model=ChatAnthropic(model="claude-3-opus-20240229", temperature=0.1),
                priority=2,
                max_retries=3,
                timeout=25,
                cost_per_token=0.015,
                quality_score=9.4
            ),
            ProviderConfig(
                name="gemini_pro",
                model=ChatGooglePalm(model_name="gemini-pro", temperature=0.1),
                priority=3,
                max_retries=2,
                timeout=20,
                cost_per_token=0.0005,
                quality_score=8.8
            )
        ]
    
    def generate_response(self, prompt: str, 
                         requirements: Dict[str, Any] = None) -> Dict[str, Any]:
        """Generate response with fallback strategy"""
        if requirements is None:
            requirements = {}
        
        # Track all attempts
        attempts = []
        
        # Try providers in order of preference
        for attempt in range(3):  # Maximum 3 attempts
            try:
                provider = self._select_provider(requirements)
                result = self._execute_with_provider(provider, prompt)
                attempts.append(result)
                
                if result['success']:
                    return {
                        'response': result['response'],
                        'provider_used': result['provider'],
                        'cost': result['cost'],
                        'execution_time': result['execution_time'],
                        'attempts': attempts
                    }
                
                # If failed, try next provider
                continue
                
            except Exception as e:
                attempts.append({
                    'provider': 'unknown',
                    'success': False,
                    'error': str(e)
                })
        
        # All providers failed
        raise Exception("All providers failed. Attempts: {}".format(attempts))

# Usage example
manager = MultiProviderLLMManager()

# Cost-optimized request
cost_optimized_result = manager.generate_response(
    "Summarize the key benefits of renewable energy",
    requirements={'prioritize_cost': True}
)

# Quality-optimized request
quality_optimized_result = manager.generate_response(
    "Analyze the economic implications of artificial intelligence adoption",
    requirements={'prioritize_quality': True}
)