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

User Query
    ↓
[Embedding Model]
    ↓
[Vector Search] → Retrieve relevant documents
    ↓
[Context + Query] → LLM
    ↓
Grounded Response

from langchain_openai import OpenAIEmbeddings, OpenAI
from langchain_community.vectorstores import Pinecone
from langchain.chains import RetrievalQA
from pinecone import Pinecone as PineconeClient

# Initialize Pinecone
pc = PineconeClient(api_key="YOUR_API_KEY")

# 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-4o", 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.invoke({"query": "What are the benefits of RAG?"})
print(response["result"])

# 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": "2026-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-v3.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: str) -> list:
    if text in embedding_cache:
        return embedding_cache[text]
    embedding = embeddings.embed_query(text)
    embedding_cache[text] = embedding
    return embedding

from openai import OpenAI
import time

client = 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
job = client.fine_tuning.jobs.create(
    training_file="file-abc123",
    model="gpt-4o-mini-2024-07-18",
    hyperparameters={
        "n_epochs": 3,
        "batch_size": 32,
        "learning_rate_multiplier": 0.1
    }
)

job_id = job.id

# Monitor progress
while True:
    job = client.fine_tuning.jobs.retrieve(job_id)
    print(f"Status: {job.status}")
    if job.status in ("succeeded", "failed"):
        model_id = job.fine_tuned_model
        break
    time.sleep(10)

# Use fine-tuned model
response = client.chat.completions.create(
    model=model_id,
    messages=[
        {"role": "user", "content": "Explain containerization"}
    ]
)
print(response.choices[0].message.content)

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

# Cache inference results
from functools import lru_cache

@lru_cache(maxsize=1000)
def cached_inference(prompt: str, model: str = "gpt-4o-mini") -> str:
    from openai import OpenAI
    client = OpenAI()
    response = client.chat.completions.create(
        model=model,
        messages=[{"role": "user", "content": prompt}]
    )
    return response.choices[0].message.content

┌─────────────────────────────────────────────────────────────┐
│                     Client Applications                      │
└────────────────────────┬────────────────────────────────────┘
                         │
┌────────────────────────▼────────────────────────────────────┐
│                    API Gateway                               │
│              (Rate Limiting, Auth, Routing)                 │
└────────────────────────┬────────────────────────────────────┘
                         │
        ┌────────────────┼────────────────┐
        │                │                │
┌───────▼──────┐  ┌──────▼──────┐  ┌─────▼──────┐
│  LLM Service │  │ RAG Service │  │ Cache Layer│
│  (OpenAI,    │  │ (Retrieval) │  │ (Redis)    │
│   Claude)    │  │             │  │            │
└───────┬──────┘  └──────┬──────┘  └─────┬──────┘
        │                │               │
        └────────────────┼───────────────┘
                         │
        ┌────────────────┼────────────────┐
        │                │                │
┌───────▼──────┐  ┌──────▼──────┐  ┌─────▼──────┐
│ Vector DB    │  │ Document    │  │ Monitoring │
│ (Pinecone)   │  │ Store       │  │ (Prometheus)
└──────────────┘  └─────────────┘  └────────────┘

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

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

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

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

# Initialize services
from openai import OpenAI
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Pinecone

openai_client = OpenAI()
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):
    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
        completion = openai_client.chat.completions.create(
            model=request.model,
            messages=[
                {"role": "user", "content": f"Context: {request.context}\n\nQuestion: {request.query}"}
            ]
        )
        response_text = completion.choices[0].message.content

        # Cache result
        redis_client.setex(cache_key, 3600, response_text)

        latency_ms = (time.time() - start_time) * 1000
        logger.info(f"Query processed in {latency_ms:.2f}ms")

        return QueryResponse(
            response=response_text,
            tokens_used=completion.usage.total_tokens,
            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"}

from openai import OpenAI
from pydantic import BaseModel
import tenacity

client = OpenAI()

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

# 2. Use structured outputs
class AnalysisResult(BaseModel):
    summary: str
    key_points: list[str]
    sentiment: str

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

# 3. Implement retry logic
@tenacity.retry(
    wait=tenacity.wait_exponential(multiplier=1, min=2, max=10),
    stop=tenacity.stop_after_attempt(3)
)
def call_llm_with_retry(prompt: str):
    return client.chat.completions.create(
        model="gpt-4o",
        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
    # gpt-4o pricing: $2.50/1M input, $10.00/1M output
    total_cost = (input_tokens * 2.50 + output_tokens * 10.00) / 1_000_000
    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

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]

for batch in batches:
    # Generate articles
    articles = [generate_article(topic) for topic in batch]

    # Quality check with separate model
    checked = [check_quality(article) for article in articles]

    # Save to database
    save_articles(checked)