Choosing Between OpenAI, Claude, and Open Source

import json
import time
from openai import OpenAI
from anthropic import Anthropic

def evaluate_model(provider, model, test_cases, judge_fn):
    """Run a set of test cases against a model and score results."""
    results = []

    for case in test_cases:
        start = time.time()

        if provider == "openai":
            client = OpenAI()
            response = client.chat.completions.create(
                model=model,
                messages=[
                    {"role": "system", "content": case["system"]},
                    {"role": "user", "content": case["input"]},
                ],
                temperature=0.0,
                max_tokens=case.get("max_tokens", 1024),
            )
            output = response.choices[0].message.content
            tokens_in = response.usage.prompt_tokens
            tokens_out = response.usage.completion_tokens

        elif provider == "anthropic":
            client = Anthropic()
            response = client.messages.create(
                model=model,
                system=case["system"],
                messages=[{"role": "user", "content": case["input"]}],
                temperature=0.0,
                max_tokens=case.get("max_tokens", 1024),
            )
            output = response.content[0].text
            tokens_in = response.usage.input_tokens
            tokens_out = response.usage.output_tokens

        latency = time.time() - start
        score = judge_fn(case["expected"], output)

        results.append({
            "case_id": case["id"],
            "score": score,
            "latency": latency,
            "tokens_in": tokens_in,
            "tokens_out": tokens_out,
        })

    return {
        "model": model,
        "avg_score": sum(r["score"] for r in results) / len(results),
        "avg_latency": sum(r["latency"] for r in results) / len(results),
        "total_tokens": sum(r["tokens_in"] + r["tokens_out"] for r in results),
        "details": results,
    }

# Define your requirements as a checklist
requirements = {
    "max_latency_ms": 3000,       # User-facing? Need < 3s
    "max_cost_per_request": 0.01, # Budget constraint
    "min_quality_score": 0.85,    # From your own eval (0-1)
    "context_needed": 8000,       # Typical input size in tokens
    "data_privacy": "standard",   # "standard", "strict", "air-gapped"
    "output_format": "json",      # "text", "json", "code"
    "requests_per_month": 500000, # Scale requirement
    "availability": 0.999,        # Uptime requirement
}

1. Does your data need to stay on your infrastructure?
   YES -> Self-hosted: Llama 3.1 70B or Mistral Large
   NO  -> Continue to step 2

2. Do you need >200K token context?
   YES -> Gemini 1.5 Pro (2M context)
   NO  -> Continue to step 3

3. Is this a complex reasoning / math / logic task?
   YES -> o3 (if budget allows) or Claude Opus
   NO  -> Continue to step 4

4. Is latency critical (< 2s response)?
   YES -> GPT-4o mini, Gemini 2.0 Flash, or Claude Haiku
   NO  -> Continue to step 5

5. Is cost the primary constraint (< $0.005/request)?
   YES -> GPT-4o mini, Mistral Small, or self-hosted Llama 8B
   NO  -> Continue to step 6

6. Is this a code-heavy or instruction-following task?
   YES -> Claude Sonnet
   NO  -> GPT-4o (safe default)

# Minimum viable model evaluation
test_cases = [
    {
        "id": "extract_1",
        "system": "Extract the customer name, issue, and sentiment from the support ticket. Return JSON.",
        "input": "Hi, I'm John Smith. My order #4521 arrived damaged. The box was completely crushed. Very disappointed with the packaging quality.",
        "expected": {"name": "John Smith", "issue": "damaged order", "sentiment": "negative"},
        "max_tokens": 200,
    },
    # Add 20-50 representative test cases from your actual workload
]

def judge_extraction(expected: dict, output: str) -> float:
    """Score extraction quality. Returns 0.0-1.0."""
    try:
        parsed = json.loads(output)
        score = 0.0
        for key in expected:
            if key in parsed and str(parsed[key]).lower() == str(expected[key]).lower():
                score += 1.0 / len(expected)
        return score
    except json.JSONDecodeError:
        return 0.0  # Failed to produce valid JSON

# Test multiple models
candidates = [
    ("openai", "gpt-4o"),
    ("openai", "gpt-4o-mini"),
    ("anthropic", "claude-sonnet-4-20250514"),
    ("anthropic", "claude-3-5-haiku-20241022"),
]

for provider, model in candidates:
    result = evaluate_model(provider, model, test_cases, judge_extraction)
    monthly_cost = (
        result["total_tokens"] / len(test_cases) *
        requirements["requests_per_month"] / 1_000_000 * 5.0  # Rough avg price
    )
    print(f"{model:>30}: score={result['avg_score']:.2f}  "
          f"latency={result['avg_latency']:.1f}s  "
          f"est_monthly=${monthly_cost:,.0f}")

def self_host_vs_api_breakeven(
    monthly_requests: int,
    avg_tokens_per_request: int,  # Input + output combined
    api_price_per_m_tokens: float,
    gpu_monthly_cost: float,      # e.g., A100 80GB at ~$2/hr = $1,460/mo
    gpu_throughput_tokens_per_sec: int,
):
    """Compare API vs self-hosted costs."""
    # API cost
    total_tokens = monthly_requests * avg_tokens_per_request
    api_cost = (total_tokens / 1_000_000) * api_price_per_m_tokens

    # Self-hosted cost: how many GPUs do you need?
    tokens_per_month = monthly_requests * avg_tokens_per_request
    seconds_needed = tokens_per_month / gpu_throughput_tokens_per_sec
    hours_needed = seconds_needed / 3600
    gpus_needed = max(1, int(hours_needed / (30 * 24)) + 1)  # Hours per month
    self_hosted_cost = gpus_needed * gpu_monthly_cost

    return {
        "api_monthly_cost": round(api_cost, 2),
        "self_hosted_monthly_cost": round(self_hosted_cost, 2),
        "gpus_needed": gpus_needed,
        "savings": round(api_cost - self_hosted_cost, 2),
        "cheaper": "self-hosted" if self_hosted_cost < api_cost else "api",
    }

# Scenario: 1M requests/month, 2K tokens each, using GPT-4o equivalent
result = self_host_vs_api_breakeven(
    monthly_requests=1_000_000,
    avg_tokens_per_request=2000,
    api_price_per_m_tokens=6.25,     # Blended GPT-4o rate
    gpu_monthly_cost=1460,           # A100 80GB on-demand
    gpu_throughput_tokens_per_sec=500, # Llama 70B on A100, batched
)
print(json.dumps(result, indent=2))

{
  "api_monthly_cost": 12500.0,
  "self_hosted_monthly_cost": 4380.0,
  "gpus_needed": 3,
  "savings": 8120.0,
  "cheaper": "self-hosted"
}

# Install ollama
curl -fsSL https://ollama.com/install.sh | sh

# Pull and run a model
ollama pull llama3.1:70b
ollama run llama3.1:70b "Explain transformers in one paragraph"

from openai import OpenAI

# Point the OpenAI client at Ollama's local server
client = OpenAI(
    base_url="http://localhost:11434/v1",
    api_key="ollama",  # Required but unused
)

response = client.chat.completions.create(
    model="llama3.1:70b",
    messages=[{"role": "user", "content": "Hello"}],
    temperature=0.7,
    max_tokens=1024,
)

print(response.choices[0].message.content)

# Install vLLM
pip install vllm

# Start the server (requires GPU)
python -m vllm.entrypoints.openai.api_server \
    --model meta-llama/Meta-Llama-3.1-70B-Instruct \
    --tensor-parallel-size 2 \
    --max-model-len 8192 \
    --port 8000

from openai import OpenAI

# vLLM also exposes an OpenAI-compatible API
client = OpenAI(
    base_url="http://your-server:8000/v1",
    api_key="unused",
)

response = client.chat.completions.create(
    model="meta-llama/Meta-Llama-3.1-70B-Instruct",
    messages=[
        {"role": "system", "content": "You are a helpful assistant."},
        {"role": "user", "content": "Explain vLLM in 3 sentences."},
    ],
    temperature=0.0,
    max_tokens=512,
)

from openai import OpenAI

# Together.ai — hosted open-source models
client = OpenAI(
    base_url="https://api.together.xyz/v1",
    api_key="your-together-api-key",
)

response = client.chat.completions.create(
    model="meta-llama/Meta-Llama-3.1-70B-Instruct-Turbo",
    messages=[{"role": "user", "content": "Hello"}],
    temperature=0.7,
    max_tokens=1024,
)

# Groq — optimized for speed
client_groq = OpenAI(
    base_url="https://api.groq.com/openai/v1",
    api_key="your-groq-api-key",
)

response = client_groq.chat.completions.create(
    model="llama-3.1-70b-versatile",
    messages=[{"role": "user", "content": "Hello"}],
)

from openai import OpenAI
from anthropic import Anthropic

openai_client = OpenAI()
anthropic_client = Anthropic()

def classify_complexity(query: str) -> str:
    """Use a cheap model to classify query complexity."""
    response = openai_client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": (
                "Classify the complexity of this query as 'simple' or 'complex'. "
                "Simple: factual questions, basic tasks, classification. "
                "Complex: multi-step reasoning, code generation, analysis. "
                "Return ONLY the word 'simple' or 'complex'."
            )},
            {"role": "user", "content": query},
        ],
        temperature=0.0,
        max_tokens=10,
    )
    return response.choices[0].message.content.strip().lower()

def route_and_respond(query: str, system_prompt: str) -> dict:
    """Route query to appropriate model based on complexity."""
    complexity = classify_complexity(query)

    if complexity == "simple":
        # Cheap and fast model for simple queries
        response = openai_client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[
                {"role": "system", "content": system_prompt},
                {"role": "user", "content": query},
            ],
            temperature=0.0,
            max_tokens=1024,
        )
        return {
            "model": "gpt-4o-mini",
            "response": response.choices[0].message.content,
            "cost_tier": "low",
        }
    else:
        # Powerful model for complex queries
        response = anthropic_client.messages.create(
            model="claude-sonnet-4-20250514",
            system=system_prompt,
            messages=[{"role": "user", "content": query}],
            temperature=0.0,
            max_tokens=4096,
        )
        return {
            "model": "claude-sonnet-4-20250514",
            "response": response.content[0].text,
            "cost_tier": "high",
        }

# Usage
result = route_and_respond(
    "What is the capital of France?",
    "You are a helpful assistant. Be concise."
)
print(f"Routed to: {result['model']} (cost tier: {result['cost_tier']})")
print(f"Response: {result['response']}")

MODEL_CONFIG = {
    "classification": {
        "provider": "openai",
        "model": "gpt-4o-mini",
        "temperature": 0.0,
        "max_tokens": 50,
    },
    "extraction": {
        "provider": "openai",
        "model": "gpt-4o-mini",
        "temperature": 0.0,
        "max_tokens": 500,
    },
    "code_generation": {
        "provider": "anthropic",
        "model": "claude-sonnet-4-20250514",
        "temperature": 0.0,
        "max_tokens": 4096,
    },
    "summarization": {
        "provider": "openai",
        "model": "gpt-4o",
        "temperature": 0.3,
        "max_tokens": 1024,
    },
    "reasoning": {
        "provider": "openai",
        "model": "o3",
        "temperature": 1.0,  # o3 requires temperature=1
        "max_tokens": 8192,
    },
}

def call_model(task_type: str, messages: list[dict]) -> str:
    """Call the appropriate model based on task type."""
    config = MODEL_CONFIG[task_type]

    if config["provider"] == "openai":
        client = OpenAI()
        response = client.chat.completions.create(
            model=config["model"],
            messages=messages,
            temperature=config["temperature"],
            max_tokens=config["max_tokens"],
        )
        return response.choices[0].message.content

    elif config["provider"] == "anthropic":
        client = Anthropic()
        system = next((m["content"] for m in messages if m["role"] == "system"), "")
        user_msgs = [m for m in messages if m["role"] != "system"]
        response = client.messages.create(
            model=config["model"],
            system=system,
            messages=user_msgs,
            temperature=config["temperature"],
            max_tokens=config["max_tokens"],
        )
        return response.content[0].text

def monthly_cost(
    input_tokens: int,
    output_tokens: int,
    requests: int,
    input_price: float,  # per 1M tokens
    output_price: float, # per 1M tokens
) -> float:
    total_input = input_tokens * requests
    total_output = output_tokens * requests
    return (
        (total_input / 1_000_000) * input_price +
        (total_output / 1_000_000) * output_price
    )

scenarios = [
    ("GPT-4o",           2.50,  10.00),
    ("GPT-4o mini",      0.15,   0.60),
    ("Claude Sonnet",    3.00,  15.00),
    ("Claude Haiku",     0.80,   4.00),
    ("Gemini 2.0 Flash", 0.10,   0.40),
    ("Mistral Small",    0.10,   0.30),
    ("Llama 70B (Together)", 0.88, 0.88),
    ("Llama 8B (Together)",  0.18, 0.18),
]

print(f"{'Model':<25} {'Monthly cost':>14} {'Per request':>12}")
print("-" * 53)
for name, inp, out in scenarios:
    cost = monthly_cost(1500, 500, 1_000_000, inp, out)
    per_req = cost / 1_000_000
    print(f"{name:<25} ${cost:>12,.2f} ${per_req:>10.6f}")

Model                      Monthly cost   Per request
-----------------------------------------------------
GPT-4o                    $   8,750.00 $  0.008750
GPT-4o mini               $     525.00 $  0.000525
Claude Sonnet             $  11,250.00 $  0.011250
Claude Haiku              $   3,200.00 $  0.003200
Gemini 2.0 Flash          $     350.00 $  0.000350
Mistral Small             $     300.00 $  0.000300
Llama 70B (Together)      $   1,760.00 $  0.001760
Llama 8B (Together)       $     360.00 $  0.000360

from abc import ABC, abstractmethod
from dataclasses import dataclass

@dataclass
class LLMResponse:
    content: str
    input_tokens: int
    output_tokens: int
    model: str
    latency_ms: float

class LLMProvider(ABC):
    @abstractmethod
    def complete(
        self,
        messages: list[dict],
        temperature: float = 0.0,
        max_tokens: int = 1024,
    ) -> LLMResponse:
        pass

class OpenAIProvider(LLMProvider):
    def __init__(self, model: str = "gpt-4o"):
        from openai import OpenAI
        self.client = OpenAI()
        self.model = model

    def complete(self, messages, temperature=0.0, max_tokens=1024):
        import time
        start = time.time()
        response = self.client.chat.completions.create(
            model=self.model,
            messages=messages,
            temperature=temperature,
            max_tokens=max_tokens,
        )
        latency = (time.time() - start) * 1000
        return LLMResponse(
            content=response.choices[0].message.content,
            input_tokens=response.usage.prompt_tokens,
            output_tokens=response.usage.completion_tokens,
            model=self.model,
            latency_ms=latency,
        )

class AnthropicProvider(LLMProvider):
    def __init__(self, model: str = "claude-sonnet-4-20250514"):
        from anthropic import Anthropic
        self.client = Anthropic()
        self.model = model

    def complete(self, messages, temperature=0.0, max_tokens=1024):
        import time
        start = time.time()

        system = ""
        user_messages = []
        for m in messages:
            if m["role"] == "system":
                system = m["content"]
            else:
                user_messages.append(m)

        response = self.client.messages.create(
            model=self.model,
            system=system,
            messages=user_messages,
            temperature=temperature,
            max_tokens=max_tokens,
        )
        latency = (time.time() - start) * 1000
        return LLMResponse(
            content=response.content[0].text,
            input_tokens=response.usage.input_tokens,
            output_tokens=response.usage.output_tokens,
            model=self.model,
            latency_ms=latency,
        )

class OllamaProvider(LLMProvider):
    def __init__(self, model: str = "llama3.1:70b"):
        from openai import OpenAI
        self.client = OpenAI(base_url="http://localhost:11434/v1", api_key="ollama")
        self.model = model

    def complete(self, messages, temperature=0.0, max_tokens=1024):
        import time
        start = time.time()
        response = self.client.chat.completions.create(
            model=self.model,
            messages=messages,
            temperature=temperature,
            max_tokens=max_tokens,
        )
        latency = (time.time() - start) * 1000
        return LLMResponse(
            content=response.choices[0].message.content,
            input_tokens=response.usage.prompt_tokens if response.usage else 0,
            output_tokens=response.usage.completion_tokens if response.usage else 0,
            model=self.model,
            latency_ms=latency,
        )

# Usage — swap providers by changing one line
provider = OpenAIProvider("gpt-4o-mini")
# provider = AnthropicProvider("claude-sonnet-4-20250514")
# provider = OllamaProvider("llama3.1:70b")

result = provider.complete(
    messages=[
        {"role": "system", "content": "You are a helpful assistant."},
        {"role": "user", "content": "Explain DNS in one sentence."},
    ],
    temperature=0.0,
)

print(f"Model: {result.model}")
print(f"Response: {result.content}")
print(f"Tokens: {result.input_tokens} in, {result.output_tokens} out")
print(f"Latency: {result.latency_ms:.0f}ms")

## Model Selection Checklist

### Requirements
- [ ] What is the task? (classification, generation, extraction, reasoning, code)
- [ ] What is the acceptable latency? (< 1s, < 5s, < 30s, doesn't matter)
- [ ] What is the budget per request? (< $0.001, < $0.01, < $0.10)
- [ ] What is the expected monthly volume? (< 10K, < 100K, < 1M, > 1M)
- [ ] What context length is needed? (< 4K, < 32K, < 128K, > 128K)

### Constraints
- [ ] Does data need to stay on-premise? (eliminates API providers)
- [ ] Are there compliance requirements? (SOC2, HIPAA, GDPR)
- [ ] Is the team able to operate self-hosted infrastructure?
- [ ] Is there an existing provider relationship or commitment?

### Evaluation
- [ ] Created 20+ test cases from real production data
- [ ] Tested at least 3 candidate models
- [ ] Measured: quality score, latency, cost per request
- [ ] Tested edge cases: long inputs, adversarial inputs, multilingual
- [ ] Confirmed output format reliability (JSON parsing success rate)

### Production Readiness
- [ ] Provider abstraction layer implemented (can swap models)
- [ ] Fallback model configured for primary provider outages
- [ ] Cost monitoring and alerting in place
- [ ] Rate limit handling with exponential backoff
- [ ] Model version pinned (not using "latest" alias in production)