Benchmark: Llama 3 vs. Claude 3.5 for 2026 Chatbots

import os
import time
import json
import argparse
from vllm import LLM, SamplingParams
from vllm.inputs import TokensPrompt
import torch
from typing import List, Dict, Optional

# Configuration constants for Llama 3 70B benchmarking
MODEL_PATH = "/models/meta-llama/Meta-Llama3-70B-Instruct"
MAX_MODEL_LEN = 128000  # Llama 3 70B max context
TENSOR_PARALLEL_SIZE = 8  # Match 8xA100 hardware
DTYPE = torch.bfloat16  # Optimized for A100 80GB

def init_llama3_llm() -> LLM:
    """Initialize vLLM-served Llama 3 70B instance with error handling."""
    try:
        llm = LLM(
            model=MODEL_PATH,
            tensor_parallel_size=TENSOR_PARALLEL_SIZE,
            max_model_len=MAX_MODEL_LEN,
            dtype=DTYPE,
            trust_remote_code=True,  # Required for Llama 3 instruct template
            gpu_memory_utilization=0.95  # Maximize A100 80GB usage
        )
        print(f"Initialized Llama 3 70B LLM on {TENSOR_PARALLEL_SIZE} GPUs")
        return llm
    except Exception as e:
        raise RuntimeError(f"Failed to initialize Llama 3 LLM: {str(e)}") from e

def run_benchmark_prompt(
    llm: LLM,
    prompts: List[str],
    max_tokens: int = 1024,
    temperature: float = 0.7
) -> List[Dict]:
    """Run benchmark prompts and collect latency/accuracy metrics."""
    sampling_params = SamplingParams(
        temperature=temperature,
        max_tokens=max_tokens,
        top_p=0.9,
        repetition_penalty=1.1
    )
    results = []
    for idx, prompt in enumerate(prompts):
        start_time = time.perf_counter()
        try:
            outputs = llm.generate(
                prompts=[TokensPrompt(prompt_token_ids=llm.get_tokenizer().encode(prompt))],
                sampling_params=sampling_params
            )
            latency_ms = (time.perf_counter() - start_time) * 1000
            generated_text = outputs[0].outputs[0].text
            token_count = len(outputs[0].outputs[0].token_ids)
            results.append({
                "prompt_id": idx,
                "latency_ms": latency_ms,
                "token_count": token_count,
                "generated_text": generated_text,
                "error": None
            })
        except Exception as e:
            results.append({
                "prompt_id": idx,
                "latency_ms": None,
                "token_count": None,
                "generated_text": None,
                "error": str(e)
            })
            print(f"Prompt {idx} failed: {str(e)}")
    return results

def save_benchmark_results(results: List[Dict], output_path: str) -> None:
    """Persist benchmark results to JSON with error handling."""
    try:
        with open(output_path, "w", encoding="utf-8") as f:
            json.dump(results, f, indent=2, ensure_ascii=False)
        print(f"Saved {len(results)} benchmark results to {output_path}")
    except IOError as e:
        raise RuntimeError(f"Failed to write results to {output_path}: {str(e)}") from e

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Benchmark Llama 3 70B Instruct with vLLM")
    parser.add_argument("--prompt-file", type=str, required=True, help="Path to JSON file with benchmark prompts")
    parser.add_argument("--output-file", type=str, default="llama3_bench_results.json", help="Output path for results")
    parser.add_argument("--max-tokens", type=int, default=1024, help="Max tokens per generation")
    args = parser.parse_args()

    # Load prompts from file
    try:
        with open(args.prompt_file, "r", encoding="utf-8") as f:
            prompts = json.load(f)
        print(f"Loaded {len(prompts)} prompts from {args.prompt_file}")
    except Exception as e:
        raise RuntimeError(f"Failed to load prompts: {str(e)}") from e

    # Initialize model and run benchmark
    llm = init_llama3_llm()
    benchmark_results = run_benchmark_prompt(llm, prompts, max_tokens=args.max_tokens)

    # Save results
    save_benchmark_results(benchmark_results, args.output_file)

    # Calculate aggregate metrics
    valid_results = [r for r in benchmark_results if r["error"] is None]
    if valid_results:
        avg_latency = sum(r["latency_ms"] for r in valid_results) / len(valid_results)
        print(f"Benchmark complete: {len(valid_results)}/{len(prompts)} prompts successful, avg latency {avg_latency:.2f}ms")
    else:
        print("No valid benchmark results generated.")

import os
import time
import json
import argparse
from anthropic import Anthropic, AnthropicError
from typing import List, Dict, Optional

# Configuration constants for Claude 3.5 Sonnet benchmarking
API_KEY = os.environ.get("ANTHROPIC_API_KEY")
if not API_KEY:
    raise ValueError("ANTHROPIC_API_KEY environment variable must be set")
MODEL_NAME = "claude-3-5-sonnet-20260315"  # Pinned version for reproducibility
MAX_TOKENS = 1024
TEMPERATURE = 0.7
SYSTEM_PROMPT = "You are a helpful production chatbot assistant. Respond concisely and accurately."

def init_claude_client() -> Anthropic:
    """Initialize Anthropic client with error handling."""
    try:
        client = Anthropic(api_key=API_KEY)
        # Test connection with a minimal prompt
        test_response = client.messages.create(
            model=MODEL_NAME,
            max_tokens=10,
            temperature=0,
            system=SYSTEM_PROMPT,
            messages=[{"role": "user", "content": "Test connection"}]
        )
        print(f"Initialized Claude 3.5 Sonnet client (model: {MODEL_NAME})")
        return client
    except AnthropicError as e:
        raise RuntimeError(f"Anthropic API error: {str(e)}") from e
    except Exception as e:
        raise RuntimeError(f"Failed to initialize Claude client: {str(e)}") from e

def run_claude_benchmark(
    client: Anthropic,
    prompts: List[str]
) -> List[Dict]:
    """Run benchmark prompts against Claude 3.5 Sonnet and collect metrics."""
    results = []
    for idx, prompt in enumerate(prompts):
        start_time = time.perf_counter()
        try:
            response = client.messages.create(
                model=MODEL_NAME,
                max_tokens=MAX_TOKENS,
                temperature=TEMPERATURE,
                system=SYSTEM_PROMPT,
                messages=[{"role": "user", "content": prompt}]
            )
            latency_ms = (time.perf_counter() - start_time) * 1000
            generated_text = response.content[0].text
            input_tokens = response.usage.input_tokens
            output_tokens = response.usage.output_tokens
            results.append({
                "prompt_id": idx,
                "latency_ms": latency_ms,
                "input_tokens": input_tokens,
                "output_tokens": output_tokens,
                "generated_text": generated_text,
                "error": None
            })
        except AnthropicError as e:
            results.append({
                "prompt_id": idx,
                "latency_ms": None,
                "input_tokens": None,
                "output_tokens": None,
                "generated_text": None,
                "error": f"Anthropic API error: {str(e)}"
            })
            print(f"Prompt {idx} failed: {str(e)}")
        except Exception as e:
            results.append({
                "prompt_id": idx,
                "latency_ms": None,
                "input_tokens": None,
                "output_tokens": None,
                "generated_text": None,
                "error": f"Unexpected error: {str(e)}"
            })
            print(f"Prompt {idx} failed unexpectedly: {str(e)}")
    return results

def save_claude_results(results: List[Dict], output_path: str) -> None:
    """Persist Claude benchmark results to JSON with error handling."""
    try:
        # Redact any sensitive data from prompts/responses
        sanitized_results = []
        for r in results:
            sanitized = r.copy()
            if sanitized.get("generated_text") and len(sanitized["generated_text"]) > 200:
                sanitized["generated_text"] = sanitized["generated_text"][:200] + "... [truncated]"
            sanitized_results.append(sanitized)
        with open(output_path, "w", encoding="utf-8") as f:
            json.dump(sanitized_results, f, indent=2, ensure_ascii=False)
        print(f"Saved {len(results)} Claude benchmark results to {output_path}")
    except IOError as e:
        raise RuntimeError(f"Failed to write Claude results to {output_path}: {str(e)}") from e

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Benchmark Claude 3.5 Sonnet via Anthropic API")
    parser.add_argument("--prompt-file", type=str, required=True, help="Path to JSON file with benchmark prompts")
    parser.add_argument("--output-file", type=str, default="claude35_bench_results.json", help="Output path for results")
    args = parser.parse_args()

    # Load prompts
    try:
        with open(args.prompt_file, "r", encoding="utf-8") as f:
            prompts = json.load(f)
        print(f"Loaded {len(prompts)} prompts from {args.prompt_file}")
    except Exception as e:
        raise RuntimeError(f"Failed to load prompts: {str(e)}") from e

    # Initialize client and run benchmark
    client = init_claude_client()
    benchmark_results = run_claude_benchmark(client, prompts)

    # Save results
    save_claude_results(benchmark_results, args.output_file)

    # Calculate aggregate metrics
    valid_results = [r for r in benchmark_results if r["error"] is None]
    if valid_results:
        avg_latency = sum(r["latency_ms"] for r in valid_results) / len(valid_results)
        total_output_tokens = sum(r["output_tokens"] for r in valid_results)
        print(f"Benchmark complete: {len(valid_results)}/{len(prompts)} prompts successful, avg latency {avg_latency:.2f}ms, total output tokens {total_output_tokens}")
    else:
        print("No valid Claude benchmark results generated.")

import json
import argparse
from typing import Dict, List
from dataclasses import dataclass

@dataclass
class ModelCostConfig:
    """Configuration for model cost calculations."""
    name: str
    self_hosted_cost_per_1k: float  # For Llama 3: infrastructure + GPU amortization
    api_cost_per_1k: float  # For Claude 3.5: Anthropic API pricing
    max_context: int
    is_self_hostable: bool

@dataclass
class WorkloadConfig:
    """Configuration for a chatbot workload."""
    name: str
    monthly_input_tokens: int
    monthly_output_tokens: int
    requires_on_prem: bool
    max_context_needed: int

def load_configs(config_path: str) -> Dict:
    """Load model and workload configs from JSON file."""
    try:
        with open(config_path, "r", encoding="utf-8") as f:
            config = json.load(f)
        # Validate required fields
        required_model_fields = ["name", "self_hosted_cost_per_1k", "api_cost_per_1k", "max_context", "is_self_hostable"]
        for model in config.get("models", []):
            for field in required_model_fields:
                if field not in model:
                    raise ValueError(f"Model config missing required field: {field}")
        required_workload_fields = ["name", "monthly_input_tokens", "monthly_output_tokens", "requires_on_prem", "max_context_needed"]
        for workload in config.get("workloads", []):
            for field in required_workload_fields:
                if field not in workload:
                    raise ValueError(f"Workload config missing required field: {field}")
        return config
    except json.JSONDecodeError as e:
        raise ValueError(f"Invalid JSON in config file: {str(e)}") from e
    except Exception as e:
        raise RuntimeError(f"Failed to load config from {config_path}: {str(e)}") from e

def calculate_monthly_cost(
    model: ModelCostConfig,
    workload: WorkloadConfig
) -> Dict:
    """Calculate monthly cost for a model-workload pair."""
    total_monthly_tokens = workload.monthly_input_tokens + workload.monthly_output_tokens
    cost_per_1k = model.self_hosted_cost_per_1k if model.is_self_hostable else model.api_cost_per_1k
    monthly_cost = (total_monthly_tokens / 1000) * cost_per_1k

    # Check compliance constraints
    compliance_issue = None
    if workload.requires_on_prem and not model.is_self_hostable:
        compliance_issue = f"{model.name} is not self-hostable, but workload requires on-prem"
    if workload.max_context_needed > model.max_context:
        compliance_issue = f"{model.name} max context {model.max_context} < workload needed {workload.max_context_needed}"

    return {
        "model_name": model.name,
        "workload_name": workload.name,
        "total_monthly_tokens": total_monthly_tokens,
        "cost_per_1k_tokens": cost_per_1k,
        "monthly_cost_usd": round(monthly_cost, 2),
        "compliance_issue": compliance_issue,
        "is_viable": compliance_issue is None
    }

def generate_cost_report(models: List[ModelCostConfig], workloads: List[WorkloadConfig]) -> List[Dict]:
    """Generate cost comparison report for all model-workload pairs."""
    report = []
    for workload in workloads:
        for model in models:
            cost_result = calculate_monthly_cost(model, workload)
            report.append(cost_result)
    return report

def save_cost_report(report: List[Dict], output_path: str) -> None:
    """Save cost report to JSON with error handling."""
    try:
        with open(output_path, "w", encoding="utf-8") as f:
            json.dump(report, f, indent=2, ensure_ascii=False)
        print(f"Saved cost report with {len(report)} entries to {output_path}")
    except IOError as e:
        raise RuntimeError(f"Failed to write cost report to {output_path}: {str(e)}") from e

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Compare Llama 3 vs Claude 3.5 monthly costs for chatbot workloads")
    parser.add_argument("--config-file", type=str, required=True, help="Path to cost config JSON file")
    parser.add_argument("--output-file", type=str, default="cost_comparison_report.json", help="Output path for cost report")
    args = parser.parse_args()

    # Load config
    config = load_configs(args.config_file)

    # Parse models
    models = [
        ModelCostConfig(
            name=model["name"],
            self_hosted_cost_per_1k=model["self_hosted_cost_per_1k"],
            api_cost_per_1k=model["api_cost_per_1k"],
            max_context=model["max_context"],
            is_self_hostable=model["is_self_hostable"]
        ) for model in config["models"]
    ]

    # Parse workloads
    workloads = [
        WorkloadConfig(
            name=workload["name"],
            monthly_input_tokens=workload["monthly_input_tokens"],
            monthly_output_tokens=workload["monthly_output_tokens"],
            requires_on_prem=workload["requires_on_prem"],
            max_context_needed=workload["max_context_needed"]
        ) for workload in config["workloads"]
    ]

    # Generate and save report
    cost_report = generate_cost_report(models, workloads)
    save_cost_report(cost_report, args.output_file)

    # Print summary to console
    print("\nCost Comparison Summary:")
    for entry in cost_report:
        status = "✅ VIABLE" if entry["is_viable"] else f"❌ NOT VIABLE: {entry['compliance_issue']}"
        print(f"{entry['workload_name']} + {entry['model_name']}: ${entry['monthly_cost_usd']}/month {status}")

from vllm import LLM

llm = LLM(
    model="/models/llama3-70b-instruct",
    tensor_parallel_size=8,
    gpu_memory_utilization=0.95,
    max_num_seqs=256,  # Continuous batching max concurrent sequences
    enable_prefix_caching=True  # Cache repeated system prompts
)

import os
from anthropic import Anthropic

client = Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"])

def truncate_prompt(prompt: str, max_input_tokens: int = 1024) -> str:
    """Truncate prompt to max_input_tokens using Anthropic tokenizer."""
    token_count = client.messages.count_tokens(
        model="claude-3-5-sonnet-20260315",
        messages=[{"role": "user", "content": prompt}]
    ).input_tokens
    if token_count > max_input_tokens:
        # Truncate to last max_input_tokens tokens (simplified)
        return prompt[-max_input_tokens:] + "... [truncated]"
    return prompt

from transformers import AutoModelForCausalLM, BitsAndBytesConfig

bnb_config = BitsAndBytesConfig(
    load_in_4bit=True,
    bnb_4bit_use_double_quant=True,
    bnb_4bit_quant_type="nf4",
    bnb_4bit_compute_dtype="bfloat16"
)

model = AutoModelForCausalLM.from_pretrained(
    "meta-llama/Meta-Llama3-70B-Instruct",
    quantization_config=bnb_config,
    trust_remote_code=True
)
# Add LoRA adapters here per QLoRA guidelines