Why We Stopped Using Proprietary LLMs for Internal Tooling in 2026

import os
import time
import logging
import json
from dataclasses import dataclass
from typing import Optional, Dict, Any, List
from prometheus_client import Counter, Histogram, start_http_server
import requests
from requests.exceptions import RequestException, Timeout, ConnectionError

# Configure logging for production audit trails
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger("internal_llm_client")

# Prometheus metrics for observability (required for internal tooling compliance)
INFERENCE_REQUESTS = Counter(
    "internal_llm_inference_total",
    "Total LLM inference requests",
    ["model", "status"]
)
INFERENCE_LATENCY = Histogram(
    "internal_llm_inference_latency_seconds",
    "LLM inference latency in seconds",
    ["model"]
)
INFERENCE_ERRORS = Counter(
    "internal_llm_inference_errors_total",
    "Total LLM inference errors",
    ["model", "error_type"]
)

@dataclass
class InferenceConfig:
    """Configuration for self-hosted LLM inference endpoint"""
    base_url: str
    model_name: str
    max_retries: int = 3
    timeout: int = 30
    rate_limit_per_min: int = 100
    api_key: Optional[str] = None  # Optional for self-hosted with auth

class InternalLLMClient:
    """Production-grade client for self-hosted LLM inference, replacing proprietary APIs"""

    def __init__(self, config: InferenceConfig):
        self.config = config
        self.session = requests.Session()
        # Set default headers for vLLM-compatible endpoints
        self.session.headers.update({
            "Content-Type": "application/json",
            "User-Agent": "InternalToolingLLMClient/1.0"
        })
        if self.config.api_key:
            self.session.headers["Authorization"] = f"Bearer {self.config.api_key}"
        self._request_timestamps: List[float] = []
        logger.info(f"Initialized LLM client for model {config.model_name} at {config.base_url}")

    def _enforce_rate_limit(self) -> None:
        """Enforce per-minute rate limiting to prevent self-hosted cluster overload"""
        now = time.time()
        # Prune timestamps older than 60 seconds
        self._request_timestamps = [t for t in self._request_timestamps if now - t < 60]
        if len(self._request_timestamps) >= self.config.rate_limit_per_min:
            sleep_time = 60 - (now - self._request_timestamps[0])
            logger.warning(f"Rate limit hit, sleeping {sleep_time:.2f}s")
            time.sleep(max(sleep_time, 0))
        self._request_timestamps.append(now)

    def generate(self, prompt: str, max_tokens: int = 512, temperature: float = 0.1) -> Optional[str]:
        """
        Generate text from self-hosted LLM with retries, error handling, and metrics.

        Args:
            prompt: Input prompt for the LLM
            max_tokens: Maximum tokens to generate
            temperature: Sampling temperature (0.0 = deterministic)

        Returns:
            Generated text, or None if all retries fail
        """
        payload = {
            "model": self.config.model_name,
            "prompt": prompt,
            "max_tokens": max_tokens,
            "temperature": temperature,
            "stream": False
        }
        last_error = None

        for attempt in range(self.config.max_retries + 1):
            self._enforce_rate_limit()
            start_time = time.time()

            try:
                with INFERENCE_LATENCY.labels(model=self.config.model_name).time():
                    response = self.session.post(
                        f"{self.config.base_url}/v1/completions",
                        json=payload,
                        timeout=self.config.timeout
                    )
                response.raise_for_status()
                result = response.json()
                generated_text = result["choices"][0]["text"].strip()

                INFERENCE_REQUESTS.labels(model=self.config.model_name, status="success").inc()
                logger.debug(f"Generated {len(generated_text)} chars in {time.time() - start_time:.2f}s")
                return generated_text

            except Timeout:
                last_error = "timeout"
                INFERENCE_ERRORS.labels(model=self.config.model_name, error_type="timeout").inc()
                logger.warning(f"Attempt {attempt + 1} timed out after {self.config.timeout}s")
            except ConnectionError:
                last_error = "connection_error"
                INFERENCE_ERRORS.labels(model=self.config.model_name, error_type="connection_error").inc()
                logger.warning(f"Attempt {attempt + 1} failed: could not connect to {self.config.base_url}")
            except RequestException as e:
                last_error = f"request_error: {str(e)}"
                INFERENCE_ERRORS.labels(model=self.config.model_name, error_type="request_error").inc()
                logger.error(f"Attempt {attempt + 1} failed: {str(e)}")
            except KeyError as e:
                last_error = f"response_parse_error: missing key {str(e)}"
                INFERENCE_ERRORS.labels(model=self.config.model_name, error_type="parse_error").inc()
                logger.error(f"Attempt {attempt + 1} failed to parse response: {str(e)}")

            # Exponential backoff for retries
            if attempt < self.config.max_retries:
                backoff = 2 ** attempt
                logger.info(f"Retrying in {backoff}s...")
                time.sleep(backoff)

        INFERENCE_REQUESTS.labels(model=self.config.model_name, status="failure").inc()
        logger.error(f"All {self.config.max_retries + 1} attempts failed. Last error: {last_error}")
        return None

if __name__ == "__main__":
    # Start Prometheus metrics server for internal monitoring
    start_http_server(8000)

    # Load config from environment variables (12-factor app compliant)
    config = InferenceConfig(
        base_url=os.getenv("LLM_ENDPOINT_URL", "http://llm-inference.internal:8000"),
        model_name=os.getenv("LLM_MODEL_NAME", "mistralai/Mistral-7B-Instruct-v0.3"),
        max_retries=int(os.getenv("LLM_MAX_RETRIES", "3")),
        timeout=int(os.getenv("LLM_TIMEOUT", "30")),
        rate_limit_per_min=int(os.getenv("LLM_RATE_LIMIT", "100")),
        api_key=os.getenv("LLM_API_KEY")  # None if not set
    )

    client = InternalLLMClient(config)

    # Example internal tooling use case: generate SQL from natural language
    test_prompt = """### Instruction:
Convert the following natural language query to PostgreSQL-compatible SQL:
"Find all users who signed up in the last 30 days and have made at least 2 purchases"

### Response:"""

    result = client.generate(test_prompt, max_tokens=256)
    if result:
        print(f"Generated SQL:\n{result}")
    else:
        print("Failed to generate SQL")

import os
import json
import logging
import argparse
from typing import List, Dict
from datasets import load_dataset, Dataset
from transformers import (
    AutoTokenizer,
    AutoModelForCausalLM,
    TrainingArguments,
    Trainer,
    DataCollatorForLanguageModeling
)
from peft import LoraConfig, get_peft_model, prepare_model_for_kbit_training
from accelerate import Accelerator
import torch

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger("lora_fine_tuner")

def load_internal_docs(dataset_path: str) -> Dataset:
    """
    Load internal documentation from JSONL files for fine-tuning.
    Expected format per line: {"prompt": "...", "completion": "..."}
    """
    data = []
    try:
        with open(dataset_path, "r") as f:
            for line_num, line in enumerate(f, 1):
                line = line.strip()
                if not line:
                    continue
                try:
                    item = json.loads(line)
                    if "prompt" not in item or "completion" not in item:
                        logger.warning(f"Line {line_num}: missing prompt/completion, skipping")
                        continue
                    data.append(item)
                except json.JSONDecodeError as e:
                    logger.error(f"Line {line_num}: invalid JSON: {str(e)}")
    except FileNotFoundError:
        logger.error(f"Dataset file {dataset_path} not found")
        raise
    except PermissionError:
        logger.error(f"No permission to read {dataset_path}")
        raise

    logger.info(f"Loaded {len(data)} valid examples from {dataset_path}")
    return Dataset.from_list(data)

def tokenize_function(examples: Dict, tokenizer: AutoTokenizer, max_length: int = 1024) -> Dict:
    """Tokenize prompt-completion pairs for causal language modeling"""
    # Combine prompt and completion with EOS token
    texts = [
        f"{prompt}{tokenizer.eos_token}{completion}{tokenizer.eos_token}"
        for prompt, completion in zip(examples["prompt"], examples["completion"])
    ]
    return tokenizer(texts, truncation=True, max_length=max_length, padding="max_length")

def main():
    parser = argparse.ArgumentParser(description="Fine-tune LoRA adapter for internal tooling LLM")
    parser.add_argument("--base-model", type=str, required=True, help="HuggingFace model ID (e.g., meta-llama/Llama-3.1-8B-Instruct)")
    parser.add_argument("--dataset-path", type=str, required=True, help="Path to JSONL training dataset")
    parser.add_argument("--output-dir", type=str, default="./lora-adapter", help="Output directory for adapter weights")
    parser.add_argument("--batch-size", type=int, default=4, help="Per-device training batch size")
    parser.add_argument("--epochs", type=int, default=3, help="Number of training epochs")
    parser.add_argument("--learning-rate", type=float, default=2e-4, help="Learning rate for LoRA training")
    parser.add_argument("--lora-r", type=int, default=16, help="LoRA rank (higher = more parameters)")
    parser.add_argument("--lora-alpha", type=int, default=32, help="LoRA alpha scaling factor")
    parser.add_argument("--max-length", type=int, default=1024, help="Max token length for training examples")
    args = parser.parse_args()

    # Initialize accelerator for multi-GPU support
    accelerator = Accelerator()
    logger.info(f"Using accelerator: {accelerator.state}")

    # Load base model and tokenizer
    logger.info(f"Loading base model: {args.base_model}")
    try:
        tokenizer = AutoTokenizer.from_pretrained(args.base_model)
        # Set pad token to EOS if not present (common for instruct models)
        if tokenizer.pad_token is None:
            tokenizer.pad_token = tokenizer.eos_token
        model = AutoModelForCausalLM.from_pretrained(
            args.base_model,
            load_in_4bit=True,  # Use 4-bit quantization for cost-effective training
            device_map="auto",
            torch_dtype=torch.bfloat16
        )
    except OSError as e:
        logger.error(f"Failed to load model {args.base_model}: {str(e)}")
        raise
    except ValueError as e:
        logger.error(f"Invalid model configuration: {str(e)}")
        raise

    # Prepare model for k-bit training and apply LoRA
    model = prepare_model_for_kbit_training(model)
    lora_config = LoraConfig(
        r=args.lora_r,
        lora_alpha=args.lora_alpha,
        target_modules=["q_proj", "k_proj", "v_proj", "o_proj"],  # Standard for Llama/Mistral
        lora_dropout=0.05,
        bias="none",
        task_type="CAUSAL_LM"
    )
    model = get_peft_model(model, lora_config)
    model.print_trainable_parameters()  # Log number of trainable parameters

    # Load and tokenize dataset
    logger.info(f"Loading dataset from {args.dataset_path}")
    dataset = load_internal_docs(args.dataset_path)
    tokenized_dataset = dataset.map(
        lambda x: tokenize_function(x, tokenizer, args.max_length),
        batched=True,
        remove_columns=["prompt", "completion"]
    )

    # Training arguments
    training_args = TrainingArguments(
        output_dir=args.output_dir,
        per_device_train_batch_size=args.batch_size,
        num_train_epochs=args.epochs,
        learning_rate=args.learning_rate,
        fp16=torch.cuda.is_bf16_supported() is False,  # Use FP16 if BF16 not available
        bf16=torch.cuda.is_bf16_supported(),
        logging_steps=10,
        save_steps=500,
        save_total_limit=2,
        report_to="none"  # Disable wandb/tensorboard for internal runs
    )

    # Initialize trainer
    trainer = Trainer(
        model=model,
        args=training_args,
        train_dataset=tokenized_dataset,
        data_collator=DataCollatorForLanguageModeling(tokenizer=tokenizer, mlm=False)
    )

    # Train and save adapter
    logger.info("Starting LoRA fine-tuning")
    try:
        trainer.train()
        trainer.save_model(args.output_dir)
        tokenizer.save_pretrained(args.output_dir)
        logger.info(f"Saved LoRA adapter to {args.output_dir}")
    except RuntimeError as e:
        logger.error(f"Training failed: {str(e)}")
        raise

if __name__ == "__main__":
    main()

import os
import json
import re
import logging
from typing import List, Dict, Set
from datetime import datetime, timedelta
import boto3
from botocore.exceptions import ClientError, NoCredentialsError
from cryptography.fernet import Fernet

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger("llm_audit_tool")

class InternalLLMAuditor:
    """Audit tool to detect potential data leakage in LLM prompts, replacing proprietary API blind spots"""

    # Regex patterns for sensitive internal data (customize for your org)
    SENSITIVE_PATTERNS = {
        "credit_card": r"\b\d{4}[- ]?\d{4}[- ]?\d{4}[- ]?\d{4}\b",
        "email": r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b",
        "internal_api_key": r"\b[A-Za-z0-9]{32,64}\b",
        "employee_id": r"\bEMP-\d{6}\b",
        "database_connection_string": r"postgres:\/\/[^:]+:[^@]+@[^/]+\/\w+"
    }

    def __init__(self, audit_log_path: str, encryption_key: Optional[bytes] = None):
        self.audit_log_path = audit_log_path
        self.encryption_key = encryption_key
        self.fernet = Fernet(encryption_key) if encryption_key else None
        self.sensitive_patterns = {
            k: re.compile(v) for k, v in self.SENSITIVE_PATTERNS.items()
        }
        logger.info(f"Initialized auditor with {len(self.sensitive_patterns)} sensitive data patterns")

    def _decrypt_log_entry(self, encrypted_entry: str) -> str:
        """Decrypt audit log entries if encryption is enabled"""
        if not self.fernet:
            return encrypted_entry
        try:
            return self.fernet.decrypt(encrypted_entry.encode()).decode()
        except Exception as e:
            logger.error(f"Failed to decrypt log entry: {str(e)}")
            return ""

    def scan_prompt(self, prompt: str) -> Dict[str, List[str]]:
        """
        Scan a single LLM prompt for sensitive data.

        Returns:
            Dict mapping pattern names to matched strings (empty if no matches)
        """
        matches = {}
        for pattern_name, pattern in self.sensitive_patterns.items():
            found = pattern.findall(prompt)
            if found:
                # Deduplicate matches
                matches[pattern_name] = list(set(found))
        return matches

    def process_audit_logs(self, days_back: int = 7) -> Dict[str, Any]:
        """
        Process LLM audit logs from the last N days, scan for sensitive data.

        Args:
            days_back: Number of days of logs to process

        Returns:
            Audit report with counts and sample matches
        """
        report = {
            "scan_start_time": datetime.now().isoformat(),
            "days_scanned": days_back,
            "total_prompts_scanned": 0,
            "prompts_with_sensitive_data": 0,
            "sensitive_data_counts": {k: 0 for k in self.sensitive_patterns.keys()},
            "sample_matches": []
        }

        cutoff_date = datetime.now() - timedelta(days=days_back)
        log_files = []

        # List all log files in the audit directory
        try:
            for filename in os.listdir(self.audit_log_path):
                if filename.endswith(".jsonl"):
                    file_path = os.path.join(self.audit_log_path, filename)
                    # Check file modification time
                    mod_time = datetime.fromtimestamp(os.path.getmtime(file_path))
                    if mod_time >= cutoff_date:
                        log_files.append(file_path)
        except FileNotFoundError:
            logger.error(f"Audit log path {self.audit_log_path} not found")
            raise
        except PermissionError:
            logger.error(f"No permission to read {self.audit_log_path}")
            raise

        logger.info(f"Processing {len(log_files)} log files from last {days_back} days")

        for log_file in log_files:
            try:
                with open(log_file, "r") as f:
                    for line_num, line in enumerate(f, 1):
                        line = line.strip()
                        if not line:
                            continue
                        try:
                            # Decrypt if needed
                            decrypted_line = self._decrypt_log_entry(line)
                            log_entry = json.loads(decrypted_line)
                            prompt = log_entry.get("prompt", "")
                            report["total_prompts_scanned"] += 1

                            # Scan prompt for sensitive data
                            matches = self.scan_prompt(prompt)
                            if matches:
                                report["prompts_with_sensitive_data"] += 1
                                for pattern_name, found in matches.items():
                                    report["sensitive_data_counts"][pattern_name] += len(found)
                                    # Add sample matches (limit to 5 per pattern)
                                    if len(report["sample_matches"]) < 20:
                                        report["sample_matches"].append({
                                            "log_file": log_file,
                                            "line_num": line_num,
                                            "pattern": pattern_name,
                                            "match": found[0][:50] + "..." if len(found[0]) > 50 else found[0]
                                        })
                        except json.JSONDecodeError:
                            logger.warning(f"{log_file} line {line_num}: invalid JSON")
                        except KeyError as e:
                            logger.warning(f"{log_file} line {line_num}: missing key {str(e)}")
            except Exception as e:
                logger.error(f"Failed to process {log_file}: {str(e)}")

        logger.info(f"Audit complete: {report['prompts_with_sensitive_data']} of {report['total_prompts_scanned']} prompts had sensitive data")
        return report

    def upload_report_to_s3(self, report: Dict, bucket_name: str, region: str = "us-east-1") -> None:
        """Upload audit report to S3 for compliance teams (replaces proprietary API's lack of audit logs)"""
        try:
            s3 = boto3.client("s3", region_name=region)
            report_key = f"llm-audit-reports/{datetime.now().strftime('%Y-%m-%d')}.json"
            s3.put_object(
                Bucket=bucket_name,
                Key=report_key,
                Body=json.dumps(report, indent=2),
                ContentType="application/json"
            )
            logger.info(f"Uploaded audit report to s3://{bucket_name}/{report_key}")
        except NoCredentialsError:
            logger.error("No AWS credentials found for S3 upload")
            raise
        except ClientError as e:
            logger.error(f"Failed to upload to S3: {str(e)}")
            raise

if __name__ == "__main__":
    # Load encryption key from environment (optional)
    encryption_key = os.getenv("AUDIT_ENCRYPTION_KEY")
    if encryption_key:
        encryption_key = encryption_key.encode()

    # Initialize auditor with internal audit log path
    auditor = InternalLLMAuditor(
        audit_log_path=os.getenv("LLM_AUDIT_LOG_PATH", "/var/log/llm-audit"),
        encryption_key=encryption_key
    )

    # Process last 7 days of logs
    report = auditor.process_audit_logs(days_back=7)

    # Save local report
    local_report_path = f"llm-audit-report-{datetime.now().strftime('%Y-%m-%d')}.json"
    with open(local_report_path, "w") as f:
        json.dump(report, f, indent=2)
    logger.info(f"Saved local audit report to {local_report_path}")

    # Upload to S3 if bucket is configured
    s3_bucket = os.getenv("AUDIT_S3_BUCKET")
    if s3_bucket:
        auditor.upload_report_to_s3(report, s3_bucket)
    else:
        logger.info("No S3 bucket configured, skipping upload")

    # Print summary
    print(f"Audit Summary (Last 7 Days):")
    print(f"Total Prompts Scanned: {report['total_prompts_scanned']}")
    print(f"Prompts with Sensitive Data: {report['prompts_with_sensitive_data']}")
    print(f"Sensitive Data Counts: {json.dumps(report['sensitive_data_counts'], indent=2)}")

from transformers import AutoTokenizer, AutoModelForCausalLM, GPTQConfig
import torch

model_id = "mistralai/Mistral-7B-Instruct-v0.3"
tokenizer = AutoTokenizer.from_pretrained(model_id)

# GPTQ quantization config for 4-bit inference
gptq_config = GPTQConfig(
    bits=4,
    group_size=128,
    desc_act=False
)

# Load quantized model (requires ~5GB VRAM)
model = AutoModelForCausalLM.from_pretrained(
    model_id,
    device_map="auto",
    quantization_config=gptq_config,
    torch_dtype=torch.bfloat16
)

# Generate sample output
prompt = "### Instruction: Summarize the following internal RFC in 2 sentences: [RFC text here]"
inputs = tokenizer(prompt, return_tensors="pt").to("cuda")
outputs = model.generate(**inputs, max_new_tokens=128)
print(tokenizer.decode(outputs[0], skip_special_tokens=True))

from presidio_analyzer import AnalyzerEngine
from presidio_anonymizer import AnonymizerEngine

# Initialize Presidio analyzer and anonymizer
analyzer = AnalyzerEngine()
anonymizer = AnonymizerEngine()

def redact_sensitive_data(text: str) -> str:
    # Analyze text for PII and custom internal patterns
    results = analyzer.analyze(
        text=text,
        language="en",
        entities=["CREDIT_CARD", "EMAIL_ADDRESS", "PHONE_NUMBER"]  # Add custom entities here
    )
    # Anonymize (redact) detected entities
    redacted_text = anonymizer.anonymize(text=text, analyzer_results=results)
    return redacted_text.text

# Example usage
prompt = "My email is dev@internal.com and API key is abc123def456ghi789jkl012mno345pqr678stu901"
redacted = redact_sensitive_data(prompt)
print(f"Original: {prompt}")
print(f"Redacted: {redacted}")

from transformers import AutoTokenizer, AutoModelForCausalLM
from peft import PeftModel
import torch

base_model_id = "mistralai/Mistral-7B-Instruct-v0.3"
adapter_path = "./lora-adapters/code-review-v1"

# Load base model
base_model = AutoModelForCausalLM.from_pretrained(
    base_model_id,
    device_map="auto",
    torch_dtype=torch.bfloat16
)
tokenizer = AutoTokenizer.from_pretrained(base_model_id)

# Load LoRA adapter
model = PeftModel.from_pretrained(base_model, adapter_path)
model.eval()  # Set to eval mode for inference

# Generate code review suggestion
prompt = """### Instruction:
Review the following Python code for security issues:
def get_user(user_id):
    query = f"SELECT * FROM users WHERE id = '{user_id}'"
    return db.execute(query)

### Response:"""
inputs = tokenizer(prompt, return_tensors="pt").to("cuda")
outputs = model.generate(**inputs, max_new_tokens=256)
print(tokenizer.decode(outputs[0], skip_special_tokens=True))