Building Production LLM Applications: RAG, Fine-tuning, and Deployment

from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import Pinecone
from langchain.chains import RetrievalQA
from langchain.llms import OpenAI
import pinecone

# Initialize Pinecone
pinecone.init(api_key="YOUR_API_KEY", environment="us-west1-gcp")

# Create embeddings
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

# Connect to vector store
vector_store = Pinecone.from_existing_index(
    index_name="documents",
    embedding=embeddings
)

# Create RAG chain
llm = OpenAI(model="gpt-4", temperature=0)
qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",
    retriever=vector_store.as_retriever(
        search_kwargs={"k": 5}  # Retrieve top 5 documents
    )
)

# Query
response = qa_chain.run("What are the benefits of RAG?")
print(response)

# 1. Chunk documents appropriately
from langchain.text_splitter import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200,
    separators=["\n\n", "\n", " ", ""]
)

chunks = splitter.split_documents(documents)

# 2. Use metadata for filtering
documents_with_metadata = [
    {
        "content": chunk.page_content,
        "metadata": {
            "source": "documentation",
            "version": "2.0",
            "date": "2025-01-15"
        }
    }
    for chunk in chunks
]

# 3. Implement reranking for better results
from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import CohereRerank

compressor = CohereRerank(model="rerank-english-v2.0")
compression_retriever = ContextualCompressionRetriever(
    base_compressor=compressor,
    base_retriever=vector_store.as_retriever()
)

# 4. Cache embeddings to reduce costs
embedding_cache = {}

def get_embedding(text):
    if text in embedding_cache:
        return embedding_cache[text]
    embedding = embeddings.embed_query(text)
    embedding_cache[text] = embedding
    return embedding

# Fine-tune when:
# 1. Domain-specific language (medical, legal, technical)
# 2. Specific output format requirements
# 3. Consistent style/tone needed
# 4. Cost optimization (smaller model)

# Don't fine-tune when:
# 1. General knowledge questions
# 2. One-off customizations
# 3. Rapid iteration needed

import openai

# Prepare training data
training_data = [
    {
        "messages": [
            {"role": "system", "content": "You are a technical documentation expert."},
            {"role": "user", "content": "Explain microservices architecture"},
            {"role": "assistant", "content": "Microservices is an architectural pattern..."}
        ]
    },
    # ... more examples
]

# Create fine-tuning job
response = openai.FineTuningJob.create(
    training_file="file-abc123",
    model="gpt-3.5-turbo",
    hyperparameters={
        "n_epochs": 3,
        "batch_size": 32,
        "learning_rate_multiplier": 0.1
    }
)

job_id = response.id

# Monitor progress
import time
while True:
    job = openai.FineTuningJob.retrieve(job_id)
    print(f"Status: {job.status}")
    if job.status == "succeeded":
        model_id = job.fine_tuned_model
        break
    time.sleep(10)

# Use fine-tuned model
response = openai.ChatCompletion.create(
    model=model_id,
    messages=[
        {"role": "user", "content": "Explain containerization"}
    ]
)

# 1. Use smaller models for fine-tuning
# GPT-3.5-turbo: $0.003/1K tokens (input), $0.004/1K tokens (output)
# GPT-4: $0.03/1K tokens (input), $0.06/1K tokens (output)

# 2. Batch processing for cost reduction
batch_requests = [
    {
        "custom_id": "request-1",
        "params": {
            "model": "gpt-4",
            "messages": [{"role": "user", "content": "Query 1"}]
        }
    },
    # ... more requests
]

# 3. Implement caching
from functools import lru_cache

@lru_cache(maxsize=1000)
def cached_inference(prompt, model="gpt-3.5-turbo"):
    response = openai.ChatCompletion.create(
        model=model,
        messages=[{"role": "user", "content": prompt}]
    )
    return response.choices[0].message.content

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import asyncio
from typing import Optional
import logging

app = FastAPI()
logger = logging.getLogger(__name__)

class QueryRequest(BaseModel):
    query: str
    context: Optional[str] = None
    model: str = "gpt-4"

class QueryResponse(BaseModel):
    response: str
    tokens_used: int
    latency_ms: float

# Initialize services
from langchain.llms import OpenAI
from langchain.vectorstores import Pinecone
from langchain.embeddings import OpenAIEmbeddings

llm = OpenAI(model="gpt-4", temperature=0)
embeddings = OpenAIEmbeddings()
vector_store = Pinecone.from_existing_index("documents", embeddings)

# Caching layer
from redis import Redis
redis_client = Redis(host="localhost", port=6379)

@app.post("/query", response_model=QueryResponse)
async def query_llm(request: QueryRequest):
    import time
    start_time = time.time()
    
    try:
        # Check cache
        cache_key = f"query:{request.query}:{request.model}"
        cached = redis_client.get(cache_key)
        if cached:
            logger.info(f"Cache hit for query: {request.query}")
            return QueryResponse(
                response=cached.decode(),
                tokens_used=0,
                latency_ms=(time.time() - start_time) * 1000
            )
        
        # Retrieve context if not provided
        if not request.context:
            docs = vector_store.similarity_search(request.query, k=3)
            request.context = "\n".join([doc.page_content for doc in docs])
        
        # Generate response
        prompt = f"Context: {request.context}\n\nQuestion: {request.query}"
        response = llm.predict(text=prompt)
        
        # Cache result
        redis_client.setex(cache_key, 3600, response)
        
        latency_ms = (time.time() - start_time) * 1000
        logger.info(f"Query processed in {latency_ms:.2f}ms")
        
        return QueryResponse(
            response=response,
            tokens_used=len(response.split()),
            latency_ms=latency_ms
        )
    
    except Exception as e:
        logger.error(f"Error processing query: {str(e)}")
        raise HTTPException(status_code=500, detail=str(e))

@app.get("/health")
async def health_check():
    return {"status": "healthy"}

# 1. Implement streaming for better UX
from openai import OpenAI

client = OpenAI()

with client.chat.completions.stream(
    model="gpt-4",
    messages=[{"role": "user", "content": "Explain RAG"}]
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)

# 2. Use structured outputs
from pydantic import BaseModel

class AnalysisResult(BaseModel):
    summary: str
    key_points: list[str]
    sentiment: str

response = client.beta.chat.completions.parse(
    model="gpt-4",
    messages=[{"role": "user", "content": "Analyze this text"}],
    response_format=AnalysisResult
)

# 3. Implement retry logic
import tenacity

@tenacity.retry(
    wait=tenacity.wait_exponential(multiplier=1, min=2, max=10),
    stop=tenacity.stop_after_attempt(3)
)
def call_llm_with_retry(prompt):
    return client.chat.completions.create(
        model="gpt-4",
        messages=[{"role": "user", "content": prompt}]
    )

# 4. Monitor token usage
def track_tokens(response):
    input_tokens = response.usage.prompt_tokens
    output_tokens = response.usage.completion_tokens
    total_cost = (input_tokens * 0.03 + output_tokens * 0.06) / 1000
    print(f"Tokens: {input_tokens} + {output_tokens} = ${total_cost:.4f}")

# 1. Use streaming for perceived speed
# 2. Implement caching for common queries
# 3. Use smaller models for simple tasks
# 4. Batch requests when possible
# 5. Use edge deployment for low latency

# Example: Edge deployment with Cloudflare Workers
# Deploy LLM inference at edge for <100ms latency

# 1. Use GPT-3.5-turbo for most tasks ($0.0005/1K input)
# 2. Implement prompt caching (25% cost reduction)
# 3. Batch similar requests
# 4. Use smaller context windows
# 5. Fine-tune for specific tasks (lower per-token cost)

# Cost calculation
input_tokens = 1000
output_tokens = 500
cost = (input_tokens * 0.0005 + output_tokens * 0.0015) / 1000
print(f"Cost per query: ${cost:.4f}")

from langchain.retrievers import BM25Retriever, EnsembleRetriever
from langchain.vectorstores import Pinecone

# Dense retrieval (semantic)
dense_retriever = vector_store.as_retriever(search_kwargs={"k": 5})

# Sparse retrieval (keyword-based)
bm25_retriever = BM25Retriever.from_documents(documents)

# Ensemble retriever combines both
ensemble_retriever = EnsembleRetriever(
    retrievers=[dense_retriever, bm25_retriever],
    weights=[0.6, 0.4]  # 60% dense, 40% sparse
)

# Use ensemble for better results
docs = ensemble_retriever.get_relevant_documents("query")

class MultiStageRetriever:
    """Multi-stage retrieval for better accuracy"""
    
    def __init__(self, vector_store, reranker):
        self.vector_store = vector_store
        self.reranker = reranker
    
    def retrieve(self, query: str, k: int = 10) -> list:
        # Stage 1: Broad retrieval
        candidates = self.vector_store.similarity_search(query, k=k*2)
        
        # Stage 2: Reranking
        reranked = self.reranker.rerank(query, candidates, k=k)
        
        # Stage 3: Filtering
        filtered = [doc for doc in reranked if doc.score > 0.5]
        
        return filtered[:k]

class QueryExpander:
    """Expand queries for better retrieval"""
    
    def __init__(self, llm):
        self.llm = llm
    
    def expand_query(self, query: str) -> list[str]:
        """Generate alternative queries"""
        
        prompt = f"""
        Generate 3 alternative ways to ask this question:
        Original: {query}
        
        Return only the questions, one per line.
        """
        
        response = self.llm.predict(text=prompt)
        alternatives = response.strip().split('\n')
        
        return [query] + alternatives
    
    def retrieve_with_expansion(self, query: str, vector_store):
        """Retrieve using expanded queries"""
        
        expanded_queries = self.expand_query(query)
        all_docs = []
        
        for q in expanded_queries:
            docs = vector_store.similarity_search(q, k=3)
            all_docs.extend(docs)
        
        # Deduplicate and rank
        unique_docs = {doc.metadata['id']: doc for doc in all_docs}
        return list(unique_docs.values())

# LoRA reduces fine-tuning parameters by 99%
# Instead of updating all weights, update small rank matrices

from peft import LoraConfig, get_peft_model
from transformers import AutoModelForCausalLM

model = AutoModelForCausalLM.from_pretrained("meta-llama/Llama-2-7b")

# Configure LoRA
lora_config = LoraConfig(
    r=8,  # Rank
    lora_alpha=16,
    target_modules=["q_proj", "v_proj"],
    lora_dropout=0.05,
    bias="none",
    task_type="CAUSAL_LM"
)

# Apply LoRA
model = get_peft_model(model, lora_config)

# Now fine-tune with much fewer parameters
# Original: 7B parameters
# LoRA: ~1M parameters (0.01% of original)

# Fine-tune for specific domain with minimal data

training_data = [
    {
        "instruction": "Explain this medical term",
        "input": "Myocardial infarction",
        "output": "A heart attack caused by blocked blood flow..."
    },
    # ... more examples
]

# Use instruction-tuning format
formatted_data = [
    f"Instruction: {item['instruction']}\nInput: {item['input']}\nOutput: {item['output']}"
    for item in training_data
]

# Fine-tune with small learning rate
# Prevents catastrophic forgetting of general knowledge

import logging
from datetime import datetime
from typing import Dict

class LLMMonitor:
    """Monitor LLM application performance"""
    
    def __init__(self):
        self.metrics = {
            'total_queries': 0,
            'total_tokens': 0,
            'total_cost': 0,
            'avg_latency': 0,
            'error_count': 0,
            'cache_hits': 0
        }
        self.logger = logging.getLogger(__name__)
    
    def log_query(self, query: str, response: str, 
                  tokens_used: int, latency_ms: float, 
                  cost: float, cached: bool = False):
        """Log query metrics"""
        
        self.metrics['total_queries'] += 1
        self.metrics['total_tokens'] += tokens_used
        self.metrics['total_cost'] += cost
        
        if cached:
            self.metrics['cache_hits'] += 1
        
        # Update average latency
        n = self.metrics['total_queries']
        self.metrics['avg_latency'] = (
            (self.metrics['avg_latency'] * (n-1) + latency_ms) / n
        )
        
        # Log details
        self.logger.info(
            f"Query: {query[:50]}... | "
            f"Tokens: {tokens_used} | "
            f"Latency: {latency_ms:.2f}ms | "
            f"Cost: ${cost:.4f} | "
            f"Cached: {cached}"
        )
    
    def log_error(self, error: str, query: str):
        """Log errors"""
        self.metrics['error_count'] += 1
        self.logger.error(f"Error for query '{query}': {error}")
    
    def get_report(self) -> Dict:
        """Get monitoring report"""
        return {
            'total_queries': self.metrics['total_queries'],
            'total_tokens': self.metrics['total_tokens'],
            'total_cost': f"${self.metrics['total_cost']:.2f}",
            'avg_latency_ms': f"{self.metrics['avg_latency']:.2f}",
            'error_rate': f"{(self.metrics['error_count'] / max(1, self.metrics['total_queries'])) * 100:.2f}%",
            'cache_hit_rate': f"{(self.metrics['cache_hits'] / max(1, self.metrics['total_queries'])) * 100:.2f}%"
        }

class PromptInjectionDetector:
    """Detect and prevent prompt injection attacks"""
    
    def __init__(self):
        self.suspicious_patterns = [
            "ignore previous instructions",
            "forget everything",
            "system prompt",
            "administrator",
            "execute code",
            "run command"
        ]
    
    def is_suspicious(self, text: str) -> bool:
        """Check if text contains suspicious patterns"""
        text_lower = text.lower()
        return any(pattern in text_lower for pattern in self.suspicious_patterns)
    
    def sanitize_input(self, text: str) -> str:
        """Sanitize user input"""
        if self.is_suspicious(text):
            raise ValueError("Suspicious input detected")
        return text.strip()

from fastapi import FastAPI
from fastapi_limiter import FastAPILimiter
from fastapi_limiter.backends.redis import RedisBackend
from redis import asyncio as aioredis

app = FastAPI()

@app.on_event("startup")
async def startup():
    redis = aioredis.from_url("redis://localhost")
    await FastAPILimiter.init(RedisBackend(redis), key_func=lambda: "global")

@app.post("/query")
@FastAPILimiter.limit("100/minute")
async def query_llm(request: QueryRequest):
    # Handle request
    pass

# Use multiple LLM providers for redundancy
providers = [
    {"name": "openai", "model": "gpt-4", "weight": 0.5},
    {"name": "anthropic", "model": "claude-3", "weight": 0.3},
    {"name": "azure", "model": "gpt-4", "weight": 0.2}
]

import random

def select_provider():
    """Select provider based on weights"""
    return random.choices(
        providers,
        weights=[p['weight'] for p in providers],
        k=1
    )[0]

# Requirements:
# - 10,000 queries/day
# - <2s response time
# - 99.9% uptime
# - $5,000/month budget

# Solution:
# 1. Use GPT-3.5-turbo ($0.0005/1K input tokens)
# 2. Implement RAG with company documentation
# 3. Cache common questions (80/20 rule)
# 4. Use streaming for perceived speed
# 5. Implement fallback to human agents

# Cost breakdown:
# - 10,000 queries/day * 30 days = 300,000 queries
# - Avg 500 input tokens, 200 output tokens
# - Cost: (300,000 * 500 * 0.0005 + 300,000 * 200 * 0.0015) / 1000
# - = $75 + $90 = $165/month (well under budget)

# Requirements:
# - Generate 1,000 articles/day
# - Maintain brand voice
# - SEO optimized
# - Cost-effective

# Solution:
# 1. Fine-tune GPT-3.5-turbo on brand content
# 2. Use structured prompts for consistency
# 3. Implement quality checks with Claude
# 4. Batch process for cost savings
# 5. Use caching for common sections

# Implementation:
batch_size = 100
articles_per_batch = 10

for batch in batches:
    # Generate articles
    articles = [generate_article(topic) for topic in batch]
    
    # Quality check
    checked = [check_quality(article) for article in articles]
    
    # Save to database
    save_articles(checked)