Open Source Llama 3.2 vs Proprietary GPT-5 for Enterprise Code Assistants: Data Privacy Analysis

import os
import json
import time
import logging
import hashlib
from typing import List, Dict, Optional
from vllm import LLM, SamplingParams
from vllm.inputs import TokensPrompt
import sqlalchemy
from sqlalchemy import create_engine, Column, String, Integer, DateTime, Text
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from datetime import datetime

# Configure audit logging for data privacy compliance
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s",
    handlers=[logging.FileHandler("llama_audit.log"), logging.StreamHandler()]
)
logger = logging.getLogger(__name__)

Base = declarative_base()

class InferenceAuditLog(Base):
    """Audit log table for all Llama 3.2 inference requests to meet SOC 2 requirements"""
    __tablename__ = "inference_audit_logs"
    id = Column(Integer, primary_key=True)
    request_id = Column(String(64), nullable=False)
    prompt_hash = Column(String(64), nullable=False)  # Hash only, no raw prompt stored
    completion_hash = Column(String(64), nullable=False)
    latency_ms = Column(Integer, nullable=False)
    token_count = Column(Integer, nullable=False)
    timestamp = Column(DateTime, default=datetime.utcnow)

def init_db(db_path: str = "sqlite:///audit.db"):
    """Initialize audit database"""
    engine = create_engine(db_path)
    Base.metadata.create_all(engine)
    return sessionmaker(bind=engine)()

def deploy_llama_3_2_70b(
    model_path: str = "meta-llama/Llama-3.2-70B-Instruct",
    gpu_count: int = 4,
    max_model_len: int = 128_000
) -> LLM:
    """Deploy self-hosted Llama 3.2 70B with vLLM for code completion"""
    try:
        llm = LLM(
            model=model_path,
            tensor_parallel_size=gpu_count,
            max_model_len=max_model_len,
            trust_remote_code=True,
            gpu_memory_utilization=0.95
        )
        logger.info(f"Deployed Llama 3.2 70B on {gpu_count} GPUs with {max_model_len} token context")
        return llm
    except Exception as e:
        logger.error(f"Failed to deploy Llama 3.2: {str(e)}")
        raise

def generate_code_completion(
    llm: LLM,
    prompt: str,
    session,
    max_tokens: int = 512,
    temperature: float = 0.2
) -> str:
    """Generate code completion with audit logging and error handling"""
    request_id = hashlib.sha256(f"{prompt}{time.time()}".encode()).hexdigest()
    start_time = time.time()
    try:
        # Sampling params optimized for enterprise code completion
        sampling_params = SamplingParams(
            temperature=temperature,
            top_p=0.9,
            max_tokens=max_tokens,
            stop=["\n\n", ""]  # Stop on double newline or code fence
        )
        # Hash prompt for audit (no raw prompt stored to minimize data retention)
        prompt_hash = hashlib.sha256(prompt.encode()).hexdigest()
        outputs = llm.generate([prompt], sampling_params)
        completion = outputs[0].outputs[0].text.strip()
        completion_hash = hashlib.sha256(completion.encode()).hexdigest()
        latency_ms = int((time.time() - start_time) * 1000)
        token_count = len(outputs[0].outputs[0].token_ids)

        # Write audit log entry
        audit_entry = InferenceAuditLog(
            request_id=request_id,
            prompt_hash=prompt_hash,
            completion_hash=completion_hash,
            latency_ms=latency_ms,
            token_count=token_count
        )
        session.add(audit_entry)
        session.commit()
        logger.info(f"Request {request_id}: Generated {token_count} tokens in {latency_ms}ms")
        return completion
    except Exception as e:
        logger.error(f"Request {request_id} failed: {str(e)}")
        session.rollback()
        raise

if __name__ == "__main__":
    # Initialize components
    db_session = init_db()
    llm = deploy_llama_3_2_70b(gpu_count=4)

    # Example enterprise code prompt (Python FastAPI endpoint)
    code_prompt = """def create_user(db: Session, user: UserCreate):
    # Hash password with bcrypt
    hashed_password = bcrypt.hashpw(user.password.encode(), bcrypt.gensalt())
    # TODO: Implement user creation logic with audit logging
    """

    try:
        completion = generate_code_completion(llm, code_prompt, db_session)
        print(f"Generated Completion:\n{completion}")
    except Exception as e:
        print(f"Failed to generate completion: {str(e)}")
        db_session.rollback()
    finally:
        db_session.close()

import os
import re
import json
import time
import hashlib
import logging
from typing import List, Dict, Optional
from openai import OpenAI
from presidio_analyzer import AnalyzerEngine
from presidio_anonymizer import AnonymizerEngine
from presidio_anonymizer.entities import OperatorConfig
import sqlalchemy
from sqlalchemy import create_engine, Column, String, Integer, DateTime, Text
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from datetime import datetime

# Configure logging for API request tracking
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s",
    handlers=[logging.FileHandler("gpt5_audit.log"), logging.StreamHandler()]
)
logger = logging.getLogger(__name__)

Base = declarative_base()

class GPT5AuditLog(Base):
    """Audit log for GPT-5 API requests (stores only redacted prompt hashes per OpenAI terms)"""
    __tablename__ = "gpt5_audit_logs"
    id = Column(Integer, primary_key=True)
    request_id = Column(String(64), nullable=False)
    redacted_prompt_hash = Column(String(64), nullable=False)
    completion_hash = Column(String(64), nullable=False)
    latency_ms = Column(Integer, nullable=False)
    token_count = Column(Integer, nullable=False)
    pii_detected = Column(Integer, nullable=False)  # 1 if PII found, 0 otherwise
    timestamp = Column(DateTime, default=datetime.utcnow)

def init_db(db_path: str = "sqlite:///gpt5_audit.db"):
    engine = create_engine(db_path)
    Base.metadata.create_all(engine)
    return sessionmaker(bind=engine)()

def init_pii_redaction():
    """Initialize Presidio PII detection/anonymization for GPT-5 prompt sanitization"""
    analyzer = AnalyzerEngine()
    anonymizer = AnonymizerEngine()
    return analyzer, anonymizer

def redact_pii(
    text: str,
    analyzer: AnalyzerEngine,
    anonymizer: AnonymizerEngine
) -> tuple[str, int]:
    """Redact PII from prompts before sending to GPT-5 API to reduce data risk"""
    try:
        # Detect PII entities (email, phone, SSN, IP address, etc.)
        analyzer_results = analyzer.analyze(text=text, language="en")
        pii_count = len(analyzer_results)
        if pii_count == 0:
            return text, 0
        # Replace PII with placeholders (e.g., , )
        anonymized_result = anonymizer.anonymize(
            text=text,
            analyzer_results=analyzer_results,
            operators={"DEFAULT": OperatorConfig("replace", {"new_value": ""})}
        )
        logger.warning(f"Redacted {pii_count} PII entities from prompt")
        return anonymized_result.text, pii_count
    except Exception as e:
        logger.error(f"PII redaction failed: {str(e)}")
        raise

def generate_gpt5_completion(
    client: OpenAI,
    prompt: str,
    session,
    analyzer: AnalyzerEngine,
    anonymizer: AnonymizerEngine,
    max_tokens: int = 512,
    temperature: float = 0.2
) -> str:
    """Generate code completion via GPT-5 API with PII redaction and audit logging"""
    request_id = hashlib.sha256(f"{prompt}{time.time()}".encode()).hexdigest()
    start_time = time.time()
    try:
        # Redact PII before sending to OpenAI
        redacted_prompt, pii_count = redact_pii(prompt, analyzer, anonymizer)
        redacted_prompt_hash = hashlib.sha256(redacted_prompt.encode()).hexdigest()

        # Call GPT-5 API (v2024-10)
        response = client.chat.completions.create(
            model="gpt-5",
            messages=[{"role": "user", "content": redacted_prompt}],
            max_tokens=max_tokens,
            temperature=temperature,
            top_p=0.9
        )
        completion = response.choices[0].message.content.strip()
        completion_hash = hashlib.sha256(completion.encode()).hexdigest()
        latency_ms = int((time.time() - start_time) * 1000)
        token_count = response.usage.total_tokens

        # Log audit entry (no raw prompt/completion stored)
        audit_entry = GPT5AuditLog(
            request_id=request_id,
            redacted_prompt_hash=redacted_prompt_hash,
            completion_hash=completion_hash,
            latency_ms=latency_ms,
            token_count=token_count,
            pii_detected=pii_count
        )
        session.add(audit_entry)
        session.commit()
        logger.info(f"Request {request_id}: Generated {token_count} tokens in {latency_ms}ms, PII detected: {pii_count}")
        return completion
    except Exception as e:
        logger.error(f"Request {request_id} failed: {str(e)}")
        session.rollback()
        raise

if __name__ == "__main__":
    # Initialize components (requires OPENAI_API_KEY environment variable)
    if not os.getenv("OPENAI_API_KEY"):
        raise ValueError("OPENAI_API_KEY environment variable not set")
    client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
    db_session = init_db()
    analyzer, anonymizer = init_pii_redaction()

    # Example prompt with PII (simulating enterprise codebase leak risk)
    code_prompt = """def send_welcome_email(user_email: str, user_phone: str):
    # TODO: Implement welcome email sending for user jdoe@example.com, phone 555-123-4567
    """

    try:
        completion = generate_gpt5_completion(client, code_prompt, db_session, analyzer, anonymizer)
        print(f"Generated Completion (redacted prompt sent to GPT-5):\n{completion}")
    except Exception as e:
        print(f"Failed to generate completion: {str(e)}")
        db_session.rollback()
    finally:
        db_session.close()

import os
import time
import json
import hashlib
import logging
from typing import List, Dict, Tuple
from vllm import LLM, SamplingParams
from openai import OpenAI
import sqlalchemy
from sqlalchemy import create_engine, Column, String, Integer, DateTime, Float, Text
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from datetime import datetime

# Configure benchmark logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s",
    handlers=[logging.FileHandler("benchmark_results.log"), logging.StreamHandler()]
)
logger = logging.getLogger(__name__)

Base = declarative_base()

class BenchmarkResult(Base):
    """Store benchmark results for Llama 3.2 vs GPT-5 comparison"""
    __tablename__ = "benchmark_results"
    id = Column(Integer, primary_key=True)
    model_name = Column(String(50), nullable=False)
    prompt_size_tokens = Column(Integer, nullable=False)
    p99_latency_ms = Column(Float, nullable=False)
    throughput_tokens_per_sec = Column(Float, nullable=False)
    data_egress_bytes = Column(Integer, nullable=False)  # 0 for Llama, >0 for GPT-5
    timestamp = Column(DateTime, default=datetime.utcnow)

def init_benchmark_db(db_path: str = "sqlite:///benchmark.db"):
    engine = create_engine(db_path)
    Base.metadata.create_all(engine)
    return sessionmaker(bind=engine)()

def run_llama_benchmark(
    llm: LLM,
    session,
    num_iterations: int = 100,
    prompt_lines: int = 100
) -> Dict:
    """Run latency/throughput benchmark for self-hosted Llama 3.2 70B"""
    logger.info(f"Starting Llama 3.2 benchmark: {num_iterations} iterations, {prompt_lines}-line prompts")
    # Generate 100-line Python code prompt for consistency
    base_prompt = "def process_data(records: List[Dict]) -> List[Dict]:\n"
    base_prompt += "\n".join([f"    # Process record {i}" for i in range(prompt_lines - 1)])
    sampling_params = SamplingParams(temperature=0.2, max_tokens=512, top_p=0.9)
    latencies = []
    total_tokens = 0
    data_egress = 0  # No egress for self-hosted

    for i in range(num_iterations):
        start_time = time.time()
        try:
            outputs = llm.generate([base_prompt], sampling_params)
            latency_ms = (time.time() - start_time) * 1000
            latencies.append(latency_ms)
            total_tokens += len(outputs[0].outputs[0].token_ids)
            if (i + 1) % 10 == 0:
                logger.info(f"Llama iteration {i+1}/{num_iterations}")
        except Exception as e:
            logger.error(f"Llama benchmark iteration {i} failed: {str(e)}")
            continue

    if not latencies:
        raise RuntimeError("No successful Llama benchmark iterations")

    p99_latency = sorted(latencies)[int(0.99 * len(latencies))]
    avg_throughput = total_tokens / (sum(latencies) / 1000)  # tokens per second
    result = {
        "model_name": "llama-3.2-70b",
        "p99_latency_ms": p99_latency,
        "throughput_tokens_per_sec": avg_throughput,
        "data_egress_bytes": data_egress
    }
    # Save to DB
    benchmark_entry = BenchmarkResult(
        model_name=result["model_name"],
        prompt_size_tokens=len(base_prompt.split()),
        p99_latency_ms=p99_latency,
        throughput_tokens_per_sec=avg_throughput,
        data_egress_bytes=data_egress
    )
    session.add(benchmark_entry)
    session.commit()
    logger.info(f"Llama benchmark complete: p99 {p99_latency}ms, throughput {avg_throughput:.2f} t/s")
    return result

def run_gpt5_benchmark(
    client: OpenAI,
    session,
    num_iterations: int = 100,
    prompt_lines: int = 100
) -> Dict:
    """Run latency/throughput benchmark for GPT-5 API, measure data egress"""
    logger.info(f"Starting GPT-5 benchmark: {num_iterations} iterations, {prompt_lines}-line prompts")
    base_prompt = "def process_data(records: List[Dict]) -> List[Dict]:\n"
    base_prompt += "\n".join([f"    # Process record {i}" for i in range(prompt_lines - 1)])
    latencies = []
    total_tokens = 0
    data_egress = 0  # Track egress as prompt + response size

    for i in range(num_iterations):
        start_time = time.time()
        try:
            response = client.chat.completions.create(
                model="gpt-5",
                messages=[{"role": "user", "content": base_prompt}],
                max_tokens=512,
                temperature=0.2
            )
            latency_ms = (time.time() - start_time) * 1000
            latencies.append(latency_ms)
            total_tokens += response.usage.total_tokens
            # Calculate egress: prompt + response size in bytes
            data_egress += len(base_prompt.encode()) + len(response.choices[0].message.content.encode())
            if (i + 1) % 10 == 0:
                logger.info(f"GPT-5 iteration {i+1}/{num_iterations}")
        except Exception as e:
            logger.error(f"GPT-5 benchmark iteration {i} failed: {str(e)}")
            continue

    if not latencies:
        raise RuntimeError("No successful GPT-5 benchmark iterations")

    p99_latency = sorted(latencies)[int(0.99 * len(latencies))]
    avg_throughput = total_tokens / (sum(latencies) / 1000)
    result = {
        "model_name": "gpt-5-api",
        "p99_latency_ms": p99_latency,
        "throughput_tokens_per_sec": avg_throughput,
        "data_egress_bytes": data_egress
    }
    # Save to DB
    benchmark_entry = BenchmarkResult(
        model_name=result["model_name"],
        prompt_size_tokens=len(base_prompt.split()),
        p99_latency_ms=p99_latency,
        throughput_tokens_per_sec=avg_throughput,
        data_egress_bytes=data_egress
    )
    session.add(benchmark_entry)
    session.commit()
    logger.info(f"GPT-5 benchmark complete: p99 {p99_latency}ms, throughput {avg_throughput:.2f} t/s, egress {data_egress} bytes")
    return result

if __name__ == "__main__":
    # Initialize components
    benchmark_session = init_benchmark_db()

    # Run Llama benchmark (requires 4x A100 80GB)
    llama_llm = LLM(
        model="meta-llama/Llama-3.2-70B-Instruct",
        tensor_parallel_size=4,
        max_model_len=128_000
    )
    llama_results = run_llama_benchmark(llama_llm, benchmark_session)

    # Run GPT-5 benchmark (requires OPENAI_API_KEY)
    gpt_client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
    gpt_results = run_gpt5_benchmark(gpt_client, benchmark_session)

    # Print comparison
    print("\n=== Benchmark Comparison ===")
    print(f"Llama 3.2 70B p99 Latency: {llama_results['p99_latency_ms']:.2f}ms")
    print(f"GPT-5 API p99 Latency: {gpt_results['p99_latency_ms']:.2f}ms")
    print(f"Llama Throughput: {llama_results['throughput_tokens_per_sec']:.2f} tokens/sec")
    print(f"GPT-5 Throughput: {gpt_results['throughput_tokens_per_sec']:.2f} tokens/sec")
    print(f"Llama Data Egress: {llama_results['data_egress_bytes']} bytes")
    print(f"GPT-5 Data Egress: {gpt_results['data_egress_bytes']} bytes")
    benchmark_session.close()

# Minimal vLLM deployment for Llama 3.2 8B
from vllm import LLM, SamplingParams
llm = LLM(model="meta-llama/Llama-3.2-8B-Instruct", gpu_memory_utilization=0.9)
params = SamplingParams(temperature=0.2, max_tokens=512)
print(llm.generate(["def add(a,b):"], params)[0].outputs[0].text)

# Quick Presidio PII redaction example
from presidio_analyzer import AnalyzerEngine
from presidio_anonymizer import AnonymizerEngine
analyzer = AnalyzerEngine()
anonymizer = AnonymizerEngine()
text = "Call me at 555-123-4567 or email jdoe@example.com"
results = analyzer.analyze(text=text, language="en")
print(anonymizer.anonymize(text=text, analyzer_results=results).text)

# Calculate 3-year TCO for 150M tokens/month
llama_tco = 42000  # Hardware amortized over 3y
gpt5_monthly = 150_000_000 / 1000 * 0.12  # $18k/month
gpt5_tco = gpt5_monthly * 36  # $648k over 3y
print(f"Llama TCO: ${llama_tco}, GPT-5 TCO: ${gpt5_tco}")