LLM Architect Skill

LLM Architect

Purpose

Provides expert large language model system architecture for designing, deploying, and optimizing LLM applications at scale. Specializes in model selection, RAG (Retrieval Augmented Generation) pipelines, fine-tuning strategies, serving infrastructure, cost optimization, and safety guardrails for production LLM systems.

When to Use

• Designing end-to-end LLM systems from requirements to production
• Selecting models and serving infrastructure for specific use cases
• Implementing RAG (Retrieval Augmented Generation) pipelines
• Optimizing LLM costs while maintaining quality thresholds
• Building safety guardrails and compliance mechanisms
• Planning fine-tuning vs RAG vs prompt engineering strategies
• Scaling LLM inference for high-throughput applications

Quick Start

Invoke this skill when:

• Designing end-to-end LLM systems from requirements to production
• Selecting models and serving infrastructure for specific use cases
• Implementing RAG (Retrieval Augmented Generation) pipelines
• Optimizing LLM costs while maintaining quality thresholds
• Building safety guardrails and compliance mechanisms

Do NOT invoke when:

• Simple API integration exists (use backend-developer instead)
• Only prompt engineering needed without architecture decisions
• Training foundation models from scratch (almost always wrong approach)
• Generic ML tasks unrelated to language models (use ml-engineer)

Decision Framework

Model Selection Quick Guide

RAG vs Fine-Tuning Decision Tree

Need to customize LLM behavior?
│
├─ Need domain-specific knowledge?
│  ├─ Knowledge changes frequently?
│  │  └─ RAG (Retrieval Augmented Generation)
│  └─ Knowledge is static?
│     └─ Fine-tuning OR RAG (test both)
│
├─ Need specific output format/style?
│  ├─ Can describe in prompt?
│  │  └─ Prompt engineering (try first)
│  └─ Format too complex for prompt?
│     └─ Fine-tuning
│
└─ Need latency <100ms?
   └─ Fine-tuned small model (7B-13B)

Architecture Pattern

[Client] → [API Gateway + Rate Limiting]
              ↓
         [Request Router]
          (Route by intent/complexity)
              ↓
    ┌────────┴────────┐
    ↓                 ↓
[Fast Model]    [Powerful Model]
(Haiku/Small)   (Sonnet/Large)
    ↓                 ↓
[Cache Layer] ← [Response Aggregator]
    ↓
[Logging & Monitoring]
    ↓
[Response to Client]

Core Workflow: Design LLM System

1. Requirements Gathering

Ask these questions:

• Latency: What's the P95 response time requirement?
• Scale: Expected requests/day and growth trajectory?
• Accuracy: What's the minimum acceptable quality? (measurable metric)
• Cost: Budget constraints? ($/request or $/month)
• Data: Existing datasets for evaluation? Sensitivity level?
• Compliance: Regulatory requirements? (HIPAA, GDPR, SOC2, etc.)

2. Model Selection

def select_model(requirements):
    if requirements.latency_p95 < 100:  # milliseconds
        if requirements.task_complexity == "simple":
            return "llama2-7b-finetune"
        else:
            return "mistral-7b-quantized"
    
    elif requirements.latency_p95 < 2000:
        if requirements.budget == "unlimited":
            return "claude-3-opus"
        elif requirements.domain_specific:
            return "claude-3-sonnet-finetuned"
        else:
            return "claude-3-haiku"
    
    else:  # Batch/async acceptable
        if requirements.accuracy_critical:
            return "gpt-4-with-ensemble"
        else:
            return "batch-api-cheapest-tier"

3. Prototype & Evaluate

# Run benchmark on eval dataset
python scripts/evaluate_model.py \
  --model claude-3-sonnet \
  --dataset data/eval_1000_examples.jsonl \
  --metrics accuracy,latency,cost

# Expected output:
# Accuracy: 94.3%
# P95 Latency: 1,245ms
# Cost per 1K requests: $2.15

4. Iteration Checklist

• [ ] Latency P95 meets requirement? If no → optimize serving (quantization, caching)
• [ ] Accuracy meets threshold? If no → improve prompts, fine-tune, or upgrade model
• [ ] Cost within budget? If no → aggressive caching, smaller model routing, batching
• [ ] Safety guardrails tested? If no → add content filters, PII detection
• [ ] Monitoring dashboards live? If no → set up Prometheus + Grafana
• [ ] Runbook documented? If no → document common failures and fixes

Cost Optimization Strategies

Safety Checklist

• [ ] Content filtering tested against adversarial examples
• [ ] PII detection and redaction validated
• [ ] Prompt injection defenses in place
• [ ] Output validation rules implemented
• [ ] Audit logging configured for all requests
• [ ] Compliance requirements documented and validated

Red Flags - When to Escalate

Quick Reference: Performance Targets

Additional Resources

• Detailed Technical Reference: See REFERENCE.md
• RAG implementation workflow
• Semantic caching patterns
• Deployment configurations

• Code Examples & Patterns: See EXAMPLES.md
• Anti-patterns (fine-tuning when prompting suffices, no fallback)
• Quality checklist for LLM systems
• Resilient LLM call patterns

LLM Architect - Examples & Patterns

Anti-Patterns

Anti-Pattern: Fine-Tuning When Prompting Would Suffice

What it looks like:

# User wants to change response format or style
# Jumps straight to fine-tuning a custom model
train_model(
    base_model="llama2-7b",
    dataset="responses_formatted_nicely.jsonl",
    epochs=3
)

Why it fails:

• Costs thousands in compute for what good prompting achieves
• Maintenance burden (need to retrain on updates)
• Slower iteration cycles (days vs minutes)

Correct approach:

# Try prompt engineering first (90% of cases this works)
prompt = """Output JSON in this exact format:
{
  "answer": "your response here",
  "confidence": 0.95,
  "sources": ["source1", "source2"]
}"""

# Only fine-tune if:
# - Prompting fails after extensive iteration
# - Need <50ms latency (fine-tuned model smaller/faster)
# - Highly domain-specific behavior required

---

Anti-Pattern: No Fallback Strategy

What it looks like:

# Single model, no error handling
response = claude.messages.create(
    model="claude-3-opus",
    messages=[{"role": "user", "content": prompt}]
)
# If Claude API is down → your app is down

Why it fails:

• API outages happen (99.9% uptime = 43 minutes downtime/month)
• Rate limits can be hit unexpectedly
• Single point of failure

Correct approach:

import asyncio

async def resilient_llm_call(prompt):
    # Strategy 1: Retry with exponential backoff
    for attempt in range(3):
        try:
            return await call_primary_llm(prompt)
        except RateLimitError:
            await asyncio.sleep(2 ** attempt)  # 1s, 2s, 4s
        except APIError as e:
            logger.warning(f"Attempt {attempt+1} failed: {e}")
    
    # Strategy 2: Fallback to alternative model
    try:
        return await call_fallback_llm(prompt)
    except Exception:
        pass
    
    # Strategy 3: Degrade gracefully
    return {
        "response": "Service temporarily unavailable, queued for processing",
        "queued": True
    }

---

Anti-Pattern: Ignoring Context Window Limits

What it looks like:

# Stuffing entire document into prompt
prompt = f"Summarize this: {entire_100k_document}"
# API error: context length exceeded

Why it fails:

• Models have token limits (4K-200K depending on model)
• Costs increase linearly with input size
• Quality degrades with very long contexts

Correct approach:

def chunk_and_summarize(document: str, max_chunk_tokens: int = 4000) -> str:
    chunks = split_into_chunks(document, max_chunk_tokens)
    
    # Map: Summarize each chunk
    summaries = []
    for chunk in chunks:
        summary = await call_llm(f"Summarize this section:\n{chunk}")
        summaries.append(summary)
    
    # Reduce: Combine summaries
    if len(summaries) == 1:
        return summaries[0]
    
    combined = "\n\n".join(summaries)
    return await call_llm(f"Combine these summaries into one:\n{combined}")

---

Anti-Pattern: Treating LLMs as Databases

What it looks like:

# Asking model to recall specific facts from training
prompt = "What is customer ID 12345's current order status?"

Why it fails:

• Models don't have deterministic memory
• Hallucinate plausible-sounding but false information
• Training data is outdated (months to years old)

Correct approach:

# Fetch from database, use LLM for synthesis/presentation
order_data = database.query("SELECT * FROM orders WHERE customer_id = 12345")
prompt = f"Summarize this order status for customer: {order_data}"

# LLM transforms data → user-friendly response
# Database provides facts (deterministic, accurate, up-to-date)
# LLM provides presentation (natural language, helpful)

---

Anti-Pattern: No Output Validation

What it looks like:

# Trusting LLM output blindly
response = await call_llm("Generate a JSON config")
config = json.loads(response)  # Crashes if invalid JSON

Correct approach:

import json
from pydantic import BaseModel, ValidationError

class ConfigOutput(BaseModel):
    setting_a: str
    setting_b: int
    enabled: bool

async def get_validated_config(prompt: str) -> ConfigOutput:
    for attempt in range(3):
        response = await call_llm(prompt)
        
        try:
            # Extract JSON from response
            json_match = re.search(r'\{.*\}', response, re.DOTALL)
            if json_match:
                data = json.loads(json_match.group())
                return ConfigOutput(**data)
        except (json.JSONDecodeError, ValidationError) as e:
            # Retry with error feedback
            prompt = f"{prompt}\n\nPrevious attempt failed: {e}. Please fix."
    
    raise ValueError("Failed to get valid output after 3 attempts")

---

Quality Checklist

Use this checklist before marking an LLM system implementation complete:

Architecture & Design

• [ ] Latency requirements documented and validated (P50, P95, P99)
• [ ] Cost projections calculated for expected traffic ($/1K requests)
• [ ] Model selection justified with trade-off analysis
• [ ] Fallback strategy implemented and tested
• [ ] Scaling strategy defined (horizontal/vertical, triggers)

Performance

• [ ] Benchmark results on representative dataset (>1000 examples)
• [ ] Accuracy/quality metrics meet minimum thresholds
• [ ] Latency <2x requirement across P95
• [ ] Throughput tested at 2x expected peak load
• [ ] Cache hit rate measured (if caching implemented)

Cost Optimization

• [ ] Caching strategy implemented and verified
• [ ] Prompt optimization applied (compression, templates)
• [ ] Multi-model routing configured (if applicable)
• [ ] Cost monitoring dashboards created
• [ ] Budget alerts configured (>110% expected spend)

Safety & Compliance

• [ ] Content filtering tested against adversarial examples
• [ ] PII detection and redaction validated
• [ ] Prompt injection defenses in place
• [ ] Output validation rules implemented
• [ ] Audit logging configured for all requests
• [ ] Compliance requirements documented and validated

Monitoring & Observability

• [ ] Latency metrics tracked (P50, P95, P99)
• [ ] Cost metrics tracked ($/day, $/1K requests)
• [ ] Quality metrics tracked (accuracy, user ratings)
• [ ] Error rate tracked and alerted (>5% error rate)
• [ ] Dashboards created for stakeholders

Operational Readiness

• [ ] Runbook documented with common failure scenarios
• [ ] On-call escalation paths defined
• [ ] Rollback procedure tested
• [ ] A/B testing framework configured (if needed)
• [ ] Model versioning strategy implemented

Documentation

• [ ] Architecture diagram created and reviewed
• [ ] API documentation published (if exposing APIs)
• [ ] Configuration documentation complete
• [ ] Decision log maintained (why this model, why this approach)
• [ ] Known limitations documented

---

Prompt Engineering Patterns

Chain of Thought

prompt = """Solve this step by step:

Question: If a train travels at 60 mph for 2.5 hours, how far does it travel?

Let me think through this:
1. First, I'll identify the formula: distance = speed × time
2. Speed = 60 mph
3. Time = 2.5 hours
4. Distance = 60 × 2.5 = 150 miles

The train travels 150 miles.

Now solve this:
Question: {user_question}

Let me think through this:"""

Few-Shot Examples

prompt = """Classify the sentiment of these reviews:

Review: "The product arrived broken and customer service was unhelpful."
Sentiment: Negative

Review: "Exactly what I needed! Fast shipping too."
Sentiment: Positive

Review: "It works okay, nothing special but does the job."
Sentiment: Neutral

Review: "{user_review}"
Sentiment:"""

Structured Output

prompt = """Extract information from this text and return as JSON:

Text: "John Smith, CEO of Acme Corp, announced Q3 revenue of $5.2M"

Output the following JSON structure:
{
  "person": "name of the person mentioned",
  "role": "their job title",
  "company": "company name",
  "metric": "any financial figures mentioned"
}

JSON:"""

LLM Architect - Technical Reference

RAG System Implementation

Use case: Build retrieval-augmented generation for document Q&A

1. Document Processing Pipeline

from langchain.text_splitter import RecursiveCharacterTextSplitter
from datetime import datetime

# Step 1: Load documents
documents = load_documents("./knowledge_base/")  # PDFs, HTML, MD, etc.

# Step 2: Chunking strategy (critical for retrieval quality)
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,  # Tokens, not characters
    chunk_overlap=200,  # Overlap to preserve context
    separators=["\n\n", "\n", ". ", " ", ""]  # Preserve semantic boundaries
)
chunks = text_splitter.split_documents(documents)

# Step 3: Add metadata (enables filtering)
for chunk in chunks:
    chunk.metadata["source_file"] = chunk.metadata.get("source")
    chunk.metadata["chunk_index"] = chunks.index(chunk)
    chunk.metadata["created_at"] = datetime.now().isoformat()

2. Embedding & Vector Store

from pinecone import Pinecone, ServerlessSpec
from openai import OpenAI

# Initialize Pinecone
pc = Pinecone(api_key="YOUR_API_KEY")
index = pc.Index("llm-rag-demo")

# Embedding model options:
# - OpenAI text-embedding-3-small: $0.02/1M tokens, 1536 dims, fast
# - Cohere embed-english-v3.0: $0.10/1M tokens, 1024 dims, high quality
# - Local BAAI/bge-large-en-v1.5: Free, 1024 dims, self-hosted

client = OpenAI()

def embed_chunks(chunks, batch_size=100):
    embeddings = []
    for i in range(0, len(chunks), batch_size):
        batch = chunks[i:i+batch_size]
        texts = [chunk.page_content for chunk in batch]
        
        response = client.embeddings.create(
            model="text-embedding-3-small",
            input=texts
        )
        embeddings.extend([r.embedding for r in response.data])
    
    return embeddings

# Upload to vector store
embeddings = embed_chunks(chunks)
index.upsert(vectors=zip(ids, embeddings, metadatas))

3. Retrieval with Hybrid Search

from sentence_transformers import CrossEncoder

def retrieve_context(query, top_k=5):
    # Step 1: Dense retrieval (semantic similarity)
    query_embedding = client.embeddings.create(
        model="text-embedding-3-small",
        input=query
    ).data[0].embedding
    
    dense_results = index.query(
        vector=query_embedding,
        top_k=top_k * 2,  # Get more candidates
        include_metadata=True
    )
    
    # Step 2: Rerank with cross-encoder (optional, +50% relevance)
    reranker = CrossEncoder('cross-encoder/ms-marco-MiniLM-L-6-v2')
    
    candidates = [(r.metadata['text'], r.score) for r in dense_results.matches]
    reranked = reranker.rank(query, [c[0] for c in candidates])
    
    # Return top-k after reranking
    return reranked[:top_k]

4. Generation with Retrieved Context

import anthropic

anthropic_client = anthropic.Anthropic()

def generate_answer(query, context_chunks):
    # Build prompt with retrieved context
    context = "\n\n".join([
        f"[Source {i+1}]: {chunk['text']}"
        for i, chunk in enumerate(context_chunks)
    ])
    
    prompt = f"""Answer the question using ONLY the provided sources.
If the sources don't contain enough information, say "I don't have enough information to answer this."

Sources:
{context}

Question: {query}

Answer:"""
    
    # Generate with Claude
    response = anthropic_client.messages.create(
        model="claude-3-sonnet-20240229",
        max_tokens=1024,
        messages=[{"role": "user", "content": prompt}]
    )
    
    return response.content[0].text


# Full RAG pipeline
def rag_query(query):
    context_chunks = retrieve_context(query, top_k=5)
    answer = generate_answer(query, context_chunks)
    return {
        "answer": answer,
        "sources": context_chunks,  # Return for citation
        "confidence": calculate_confidence(context_chunks)
    }

5. Evaluation & Iteration

from ragas import evaluate
from ragas.metrics import faithfulness, answer_relevancy

results = evaluate(
    dataset=eval_dataset,
    metrics=[faithfulness, answer_relevancy]
)

# Expected baselines:
# - Faithfulness (answer matches sources): >90%
# - Answer relevancy: >85%
# - Retrieval precision@5: >70%

---

Semantic Caching Layer

When to use: Repeated or similar queries (60%+ hit rate achievable)

import hashlib
import time
from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams

class SemanticCache:
    def __init__(self, embedding_model):
        self.client = QdrantClient(":memory:")
        self.client.create_collection(
            collection_name="llm_cache",
            vectors_config=VectorParams(size=1536, distance=Distance.COSINE)
        )
        self.embedding_model = embedding_model
        self.ttl_seconds = 86400  # 24 hours
    
    async def get(self, query, similarity_threshold=0.95):
        # Embed query
        query_vec = self.embedding_model.encode(query)
        
        # Search for similar cached queries
        results = self.client.search(
            collection_name="llm_cache",
            query_vector=query_vec,
            limit=1
        )
        
        if results and results[0].score >= similarity_threshold:
            # Check TTL
            cached_time = results[0].payload["timestamp"]
            if time.time() - cached_time < self.ttl_seconds:
                return results[0].payload["response"]  # CACHE HIT
        
        return None  # Cache miss
    
    async def set(self, query, response):
        query_vec = self.embedding_model.encode(query)
        cache_id = hashlib.md5(query.encode()).hexdigest()
        
        self.client.upsert(
            collection_name="llm_cache",
            points=[{
                "id": cache_id,
                "vector": query_vec,
                "payload": {
                    "query": query,
                    "response": response,
                    "timestamp": time.time()
                }
            }]
        )

Expected savings: 40-80% cost reduction if 60%+ queries are cacheable

---

Kubernetes Deployment

# kubernetes/llm-deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: llm-service
spec:
  replicas: 3
  template:
    spec:
      containers:
      - name: vllm-server
        image: vllm/vllm-openai:latest
        resources:
          limits:
            nvidia.com/gpu: 1
        env:
        - name: MODEL_NAME
          value: "mistralai/Mistral-7B-Instruct-v0.2"
        - name: TENSOR_PARALLEL_SIZE
          value: "1"
        - name: MAX_MODEL_LEN
          value: "4096"
        ports:
        - containerPort: 8000
        readinessProbe:
          httpGet:
            path: /health
            port: 8000
          initialDelaySeconds: 60
          periodSeconds: 10

---

Multi-Model Routing

from enum import Enum
from typing import Literal

class ModelTier(Enum):
    FAST = "haiku"      # Simple queries, <500ms
    BALANCED = "sonnet"  # Medium complexity
    POWERFUL = "opus"    # Complex reasoning

def classify_query_complexity(query: str) -> ModelTier:
    """Route queries to appropriate model tier"""
    
    # Simple heuristics (replace with classifier in production)
    word_count = len(query.split())
    
    if word_count < 20 and "?" in query:
        return ModelTier.FAST
    
    complex_indicators = [
        "explain", "analyze", "compare", "evaluate",
        "step by step", "reasoning", "implications"
    ]
    
    if any(indicator in query.lower() for indicator in complex_indicators):
        return ModelTier.POWERFUL
    
    return ModelTier.BALANCED


async def route_and_call(query: str) -> str:
    tier = classify_query_complexity(query)
    
    model_map = {
        ModelTier.FAST: "claude-3-haiku-20240307",
        ModelTier.BALANCED: "claude-3-sonnet-20240229",
        ModelTier.POWERFUL: "claude-3-opus-20240229"
    }
    
    response = await anthropic_client.messages.create(
        model=model_map[tier],
        max_tokens=1024,
        messages=[{"role": "user", "content": query}]
    )
    
    return response.content[0].text

---

Safety Guardrails

Content Filtering

from typing import Tuple

async def check_safety(content: str) -> Tuple[bool, str]:
    """Check content for safety issues"""
    
    # PII detection
    pii_patterns = [
        r'\b\d{3}-\d{2}-\d{4}\b',  # SSN
        r'\b\d{16}\b',             # Credit card
        r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b'  # Email
    ]
    
    import re
    for pattern in pii_patterns:
        if re.search(pattern, content):
            return False, "PII detected"
    
    # Prompt injection detection
    injection_patterns = [
        "ignore previous instructions",
        "disregard all prior",
        "you are now",
        "new instructions:"
    ]
    
    content_lower = content.lower()
    for pattern in injection_patterns:
        if pattern in content_lower:
            return False, "Potential prompt injection"
    
    return True, "OK"


async def safe_llm_call(query: str) -> str:
    # Check input
    is_safe, reason = await check_safety(query)
    if not is_safe:
        return f"Request blocked: {reason}"
    
    # Get response
    response = await call_llm(query)
    
    # Check output
    is_safe, reason = await check_safety(response)
    if not is_safe:
        return "Response filtered for safety reasons"
    
    return response

---

Monitoring Setup

Key Metrics to Track

import prometheus_client as prom

# Define metrics
llm_request_latency = prom.Histogram(
    'llm_request_latency_seconds',
    'LLM request latency',
    ['model', 'endpoint'],
    buckets=[0.1, 0.5, 1.0, 2.0, 5.0, 10.0]
)

llm_request_tokens = prom.Counter(
    'llm_request_tokens_total',
    'Total tokens processed',
    ['model', 'type']  # type: input/output
)

llm_cache_hits = prom.Counter(
    'llm_cache_hits_total',
    'Cache hit count',
    ['cache_type']  # exact/semantic
)

llm_errors = prom.Counter(
    'llm_errors_total',
    'LLM error count',
    ['model', 'error_type']
)


# Usage in request handler
async def handle_llm_request(query: str):
    with llm_request_latency.labels(model='sonnet', endpoint='/chat').time():
        response = await call_llm(query)
    
    llm_request_tokens.labels(model='sonnet', type='input').inc(len(query.split()))
    llm_request_tokens.labels(model='sonnet', type='output').inc(len(response.split()))
    
    return response

Fine-Tuning Guide

Overview

Fine-tuning adapts pre-trained models to specific tasks or domains, improving performance on specialized data.

Methods

1. Full Fine-Tuning

Updates all model parameters.

Pros:

• Maximum performance gains
• No architectural changes
• Works for any model

Cons:

• High computational cost
• Large storage requirements
• Risk of catastrophic forgetting

Use case:

• Critical tasks requiring maximum accuracy
• Sufficient compute resources
• Domain-specific models

2. LoRA (Low-Rank Adaptation)

Adds trainable rank decomposition matrices.

Pros:

• 100-1000x faster training
• Small storage (1-100MB)
• No performance degradation
• Easy to switch between adapters

Cons:

• Slightly lower performance than full fine-tuning
• Requires LoRA-supporting codebase

Use case:

• Most production use cases
• Multiple task adapters
• Resource-constrained environments

Configuration:

lora_config = LoraConfig(
    r=8,  # Rank (4-16 typical)
    lora_alpha=32,  # Scaling factor
    target_modules=["q_proj", "v_proj"],  # Target layers
    lora_dropout=0.05,
    bias="none",
    task_type=TaskType.CAUSAL_LM
)

3. Prefix Tuning

Learns continuous prompt embeddings.

Pros:

• No gradient through model
• Very efficient

Cons:

• More complex to implement
• Lower performance than LoRA

4. P-Tuning (Prompt Tuning)

Learns soft prompts.

Pros:

• Simple implementation
• Parameter efficient

Cons:

• Limited to prompt-based learning

Data Preparation

Dataset Format

[
    {
        "text": "The capital of France is Paris."
    },
    {
        "text": "Machine learning enables computers to learn from data."
    }
]

Data Quality Requirements

1. Relevance: Data must match target task 2. Diversity: Cover edge cases and variations 3. Size: 100-10,000+ examples depending on task 4. Balance: Balanced classes for classification 5. Cleanliness: Remove duplicates and errors

Data Augmentation

def augment_data(text):
    variations = []

    # Paraphrasing
    variations.append(paraphrase(text))

    # Back-translation
    variations.append(back_translate(text))

    # Synonym replacement
    variations.append(replace_synonyms(text))

    return variations

Training Process

Step 1: Prepare Data

from finetune_model import ModelFinetuner, FinetuningConfig

config = FinetuningConfig(
    model_name="gpt2",
    output_dir="./finetuned_model",
    train_file="train_data.json",
    validation_file="validation_data.json",
    use_lora=True,
    num_train_epochs=3,
    per_device_train_batch_size=4,
    learning_rate=2e-4
)

finetuner = ModelFinetuner(config)
finetuner.load_model()

Step 2: Train

finetuner.train()

Step 3: Evaluate

# Use validation set
validation_loss = finetuner.evaluate()

# Or test on held-out examples
test_results = finetuner.test(test_data)

Step 4: Save Model

finetuner.save_model()

Hyperparameter Tuning

Key Parameters

Tuning Strategy

# Grid search
learning_rates = [1e-5, 5e-5, 1e-4]
batch_sizes = [4, 8]

for lr in learning_rates:
    for bs in batch_sizes:
        config.learning_rate = lr
        config.per_device_train_batch_size = bs
        finetuner.train()
        results.append(evaluate())

Evaluation

Metrics

Language Generation:

• Perplexity
• BLEU score
• ROUGE score
• Human evaluation

Classification:

• Accuracy
• F1 score
• Precision/Recall
• Confusion matrix

Question Answering:

• Exact match
• F1 score (token-level)
• Semantic similarity

Evaluation Framework

from evaluate_model import ModelEvaluator

evaluator = ModelEvaluator()

# Add metrics
evaluator.add_metric(EvaluationMetric(
    name="perplexity",
    metric_type=MetricType.CUSTOM,
    description="Model perplexity"
))

# Run evaluation
report = evaluator.evaluate_model(
    model_name="finetuned_model",
    generate_func=generate_with_model,
    dataset=test_dataset
)

Deployment

Loading Fine-Tuned Models

finetuner.load_finetuned_model("path/to/model")

# Generate
response = finetuner.generate(
    prompt="Your prompt here",
    max_new_tokens=100,
    temperature=0.7
)

Serving Fine-Tuned Models

from serve_model import ModelServer

server = ModelServer(config)
server.load_model("finetuned_model", "path/to/model")
server.start()

Multi-Adapter Deployment

# Switch between adapters
model.set_adapter("adapter_1")
output_1 = model.generate(prompt)

model.set_adapter("adapter_2")
output_2 = model.generate(prompt)

Advanced Techniques

Instruction Tuning

# Format: instruction + input + output
data = [
    {
        "instruction": "Summarize the following text.",
        "input": "Long text here...",
        "output": "Summary here..."
    }
]

Multi-Task Learning

# Train on multiple tasks simultaneously
tasks = ["summarization", "translation", "qa"]

for task in tasks:
    task_data = load_task_data(task)
    finetuner.train(task_data)

Continual Learning

# Learn new tasks without forgetting
for task in new_tasks:
    finetuner.train(task_data, replay_buffer=old_data)

Common Issues

Catastrophic Forgetting

• Symptom: Performance on old tasks degrades
• Solution: Use replay buffer, elastic weight consolidation

Overfitting

• Symptom: Training loss decreases, validation loss increases
• Solution: Reduce epochs, increase dropout, add regularization

Underfitting

• Symptom: Both training and validation loss high
• Solution: Increase model capacity, train longer

Poor Convergence

• Symptom: Loss oscillates or doesn't decrease
• Solution: Adjust learning rate, check data quality

Best Practices

1. Start with LoRA: Most cases don't need full fine-tuning 2. Monitor validation loss: Stop before overfitting 3. Use checkpoints: Save best performing model 4. Evaluate on held-out data: Don't trust training metrics alone 5. Test on edge cases: Validate model robustness 6. Document hyperparameters: Enable reproducibility 7. Version control models: Track model iterations

Model Selection Guide

Model Comparison Matrix

Selection Framework

Task-Based Selection

Simple Tasks:

• Text completion: GPT-3.5 Turbo, Llama 2 7B
• Classification: Fine-tuned BERT, RoBERTa
• Summarization: BART, T5

Complex Tasks:

• Reasoning: GPT-4, Claude 3 Opus
• Code generation: GPT-4, Claude 3.5 Sonnet
• Multi-step tasks: GPT-4 Turbo

Long Context:

• Document analysis: Claude 3.5 Sonnet (200K)
• Codebase analysis: GPT-4 Turbo (128K)
• Research: Claude 3 Opus (200K)

Cost-Based Selection

def select_model_by_budget(task_complexity, budget_per_request):
    if budget_per_request < 0.01:
        if task_complexity == 'simple':
            return 'gpt-3.5-turbo'
        else:
            return 'llama-2-7b'
    elif budget_per_request < 0.10:
        if task_complexity == 'complex':
            return 'claude-3-5-sonnet-20241022'
        else:
            return 'gpt-4-turbo'
    else:
        return 'gpt-4'

Latency-Based Selection

• <100ms: Local models, TinyLlama
• 100-500ms: GPT-3.5 Turbo, small open-source
• 500ms-2s: GPT-4 Turbo, Claude 3.5 Sonnet
• >2s: GPT-4, Claude 3 Opus

Benchmarking

Metrics to Track

1. Accuracy: How correct are the outputs? 2. Latency: How fast is the response? 3. Cost: How much does it cost? 4. Token Usage: Efficient use of context? 5. Consistency: Reproducible results?

Benchmark Framework

from benchmark_models import ModelBenchmarker

benchmarker = ModelBenchmarker(models)

# Define evaluation function
def evaluate_task(model_name, test_case):
    result = generate_with_model(model_name, test_case)
    return evaluate_result(result)

# Run benchmarks
benchmarker.benchmark_task(
    task_name="summarization",
    task_func=evaluate_task,
    test_data=test_cases,
    ground_truth=references
)

# Get best model
best = benchmarker.get_best_model_for_task(
    "summarization",
    metric="accuracy"
)

Model Specialization

Coding Models

• Best: GPT-4, Claude 3.5 Sonnet
• Budget: GPT-3.5 Turbo, StarCoder
• Local: CodeLlama, DeepSeek Coder

Writing Models

• Creative: Claude 3 Opus, GPT-4
• Technical: Claude 3.5 Sonnet, GPT-4 Turbo
• Concise: Fine-tuned models

Analysis Models

• Data Analysis: GPT-4 Turbo, Claude 3.5 Sonnet
• Document Analysis: Claude 3.5 Sonnet (long context)
• Research: Claude 3 Opus

Multimodal Models

• Vision: GPT-4 Turbo, Claude 3.5 Sonnet
• Audio: Whisper
• Video: GPT-4 Turbo (frame-by-frame)

Hybrid Approaches

Cascade Selection

def cascade_model_selection(prompt):
    # Try cheapest first
    result = try_model('gpt-3.5-turbo', prompt)

    # If confidence low, escalate
    if result['confidence'] < 0.7:
        result = try_model('claude-3-5-sonnet-20241022', prompt)

    # If still low confidence, use best
    if result['confidence'] < 0.9:
        result = try_model('gpt-4', prompt)

    return result

Ensemble Methods

• Combine outputs from multiple models
• Use voting for classification
• Average for numerical outputs
• Best-of-N for quality

Deployment Considerations

API-Based Models

• No infrastructure overhead
• Scalable
• Requires internet
• Data privacy concerns

Self-Hosted Models

• Full control
• Privacy guaranteed
• Requires GPU resources
• Higher maintenance

Hybrid Deployment

• Simple tasks: Local models
• Complex tasks: API models
• Fallback to API when local fails

Best Practices

1. Benchmark before deploying: Always test on your specific task 2. Monitor performance: Track metrics over time 3. Budget alerts: Set cost limits and monitoring 4. Fallback models: Have backup options 5. Regular re-evaluation: New models appear frequently 6. A/B testing: Compare models on production data 7. Quality checks: Validate outputs before use

Serving Infrastructure Guide

Overview

Production model serving requires careful consideration of performance, scalability, and reliability.

Serving Options

1. API-Based Serving

Use provider APIs (OpenAI, Anthropic, etc.)

Pros:

• Zero infrastructure
• Automatic scaling
• Built-in monitoring
• Regular updates

Cons:

• Ongoing costs
• Data privacy concerns
• Rate limits
• Dependency on external services

Best for:

• Proof of concept
• Low to medium traffic
• No ML infrastructure team
• Rapid prototyping

2. Self-Hosted Serving

Deploy models on your own infrastructure

Pros:

• Full control
• Data privacy
• Predictable costs
• Custom optimizations

Cons:

• Infrastructure setup
• Maintenance overhead
• Scaling complexity
• Higher initial cost

Best for:

• High volume production
• Sensitive data
• Custom models
• Cost optimization at scale

Serving Frameworks

vLLM

High-throughput serving with PagedAttention.

Installation:

pip install vllm

Usage:

python -m vllm.entrypoints.api_server \
    --model meta-llama/Llama-2-7b-hf \
    --port 8000 \
    --tensor-parallel-size 4

Pros:

• 10-20x higher throughput
• Low latency
• Continuous batching
• OpenAI-compatible API

Cons:

• Newer, less battle-tested
• Limited model support

Text Generation WebUI (Oobabooga)

Feature-rich web interface for model serving.

Features:

• Web UI
• Multiple model support
• Extensions ecosystem
• API access

Setup:

git clone https://github.com/oobabooga/text-generation-webui
cd text-generation-webui
python server.py --model-path /path/to/model --listen

LocalAI

OpenAI-compatible API for local models.

Setup:

docker run -p 8080:8080 \
    -v /models:/models \
    localai/localai \
    --models-path /models

Use OpenAI client:

from openai import OpenAI

client = OpenAI(
    base_url="http://localhost:8080/v1",
    api_key="not-needed"
)

Deployment Strategies

Docker Deployment

Dockerfile:

FROM python:3.10-slim

WORKDIR /app

RUN pip install fastapi uvicorn transformers accelerate torch

COPY model.py .
COPY ./models ./models

CMD ["uvicorn", "model:app", "--host", "0.0.0.0", "--port", "8000"]

Build and Run:

docker build -t model-server .
docker run -p 8000:8000 --gpus all model-server

Kubernetes Deployment

Deployment YAML:

apiVersion: apps/v1
kind: Deployment
metadata:
  name: model-server
spec:
  replicas: 3
  selector:
    matchLabels:
      app: model-server
  template:
    metadata:
      labels:
        app: model-server
    spec:
      containers:
      - name: model-server
        image: model-server:latest
        ports:
        - containerPort: 8000
        resources:
          limits:
            nvidia.com/gpu: 1

Service:

apiVersion: v1
kind: Service
metadata:
  name: model-server
spec:
  selector:
    app: model-server
  ports:
  - port: 80
    targetPort: 8000
  type: LoadBalancer

Serverless Deployment

AWS Lambda:

import json

def lambda_handler(event, context):
    prompt = event['prompt']
    response = generate_with_model(prompt)
    return {
        'statusCode': 200,
        'body': json.dumps({'output': response})
    }

Performance Optimization

Batch Processing

@app.post("/batch_generate")
async def batch_generate(requests: List[GenerationRequest]):
    outputs = []
    for request in requests:
        output = generate(request)
        outputs.append(output)
    return outputs

Caching

from functools import lru_cache

@lru_cache(maxsize=1000)
def cached_generate(prompt_hash):
    return generate(original_prompt)

Quantization

from transformers import BitsAndBytesConfig

quantization_config = BitsAndBytesConfig(
    load_in_8bit=True
)

model = AutoModelForCausalLM.from_pretrained(
    model_path,
    quantization_config=quantization_config
)

Stream Responses

async def stream_generate(prompt):
    for token in model.generate_stream(prompt):
        yield token

Monitoring

Key Metrics

1. Throughput: Requests per second 2. Latency: P50, P95, P99 response times 3. Error Rate: Failed requests 4. GPU Utilization: Compute efficiency 5. Memory Usage: VRAM consumption

Prometheus Integration

from prometheus_client import Counter, Histogram

request_counter = Counter('model_requests_total', 'Total requests')
latency_histogram = Histogram('model_latency_seconds', 'Request latency')

@app.post("/generate")
async def generate(request: GenerationRequest):
    with latency_histogram.time():
        output = generate_with_model(request)
        request_counter.inc()
        return output

Health Checks

@app.get("/health")
async def health():
    return {
        "status": "healthy",
        "gpu_available": torch.cuda.is_available(),
        "memory_used": torch.cuda.memory_allocated()
    }

Scaling

Horizontal Scaling

Add more instances to handle increased load.

Kubernetes HPA:

apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: model-server-hpa
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: model-server
  minReplicas: 2
  maxReplicas: 10
  metrics:
  - type: Resource
    resource:
      name: cpu
      target:
        type: Utilization
        averageUtilization: 80

Vertical Scaling

Increase resources per instance.

GPU Types:

• A100 (40GB/80GB): Best performance
• V100 (16GB/32GB): Good balance
• T4 (16GB): Cost-effective
• L4 (24GB): Newer option

Load Balancing

upstream model_servers {
    least_conn;
    server server1:8000;
    server server2:8000;
    server server3:8000;
}

server {
    listen 80;
    location / {
        proxy_pass http://model_servers;
    }
}

Best Practices

1. Use FastAPI: Async, type-safe, automatic docs 2. Implement rate limiting: Prevent abuse 3. Add authentication: Secure endpoints 4. Log everything: Debugging and monitoring 5. Version models: Easy rollbacks 6. Graceful shutdown: Handle connections properly 7. Health checks: Kubernetes ready 8. Resource limits: Prevent memory leaks 9. Request validation: Use Pydantic models 10. Monitor continuously: Detect issues early