DEV Community: Phil IPS

Intro To Webassembly With Rust

Phil IPS — Tue, 27 Oct 2026 17:34:13 +0000

Intro to WebAssembly with Rust

Introduction

WebAssembly (Wasm) is a binary instruction format for a stack-based virtual machine. It enables near-native-speed execution in web browsers and is increasingly used for performance-critical applications. Rust, with its memory safety guarantees and excellent WebAssembly support, is a perfect language for creating Wasm modules.

Why Rust for WebAssembly?

Rust provides:

Memory safety without a garbage collector
Powerful macro system for code generation
Excellent tooling for WebAssembly via wasm-pack
Strong community support for Wasm development

Setting Up Your Environment

First, install Rust if you haven't already:

curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh

Then install wasm-pack:

cargo install wasm-pack

Creating a Simple WebAssembly Module

Let's create a Rust function that adds two numbers:

#[no_mangle]
pub extern "C" fn add(a: i32, b: i32) -> i32 {
    a + b
}

This function takes two i32 parameters and returns their sum. The #[no_mangle] attribute ensures the function name isn't mangled, and extern "C" makes it callable from JavaScript.

Building the Module

Compile the Rust code to WebAssembly:

wasm-pack build --target web

This creates a pkg directory with the Wasm file and JavaScript bindings.

Using the Module in JavaScript

In your HTML file:

<script type="module">
  import init, { add } from './pkg/your_module.js';

  async function run() {
    await init();
    const result = add(5, 7);
    console.log('5 + 7 =', result);
  }

  run();
</script>
</html>

Conclusion

With Rust and WebAssembly, you can create high-performance web applications. This is just a simple example - explore more complex use cases like game engines, image processing, or cryptographic operations. The combination of Rust's safety and Wasm's performance opens up exciting possibilities for web development.

Tutorial X402

Phil IPS — Tue, 27 Oct 2026 17:16:58 +0000

How to Use x402_fetch for API Calls with Automatic USDC Payment

Introduction

This tutorial explains how to use the x402_fetch tool to make HTTP requests to APIs with automatic USDC payment handling. When a server responds with HTTP 402 (Payment Required), x402_fetch automatically signs a USDC payment and retries the request, making it easy to access paid APIs without manual payment management.

Prerequisites

Python 3.6+
requests library (install with pip install requests)
Your USDC wallet address (0xA00918d42bC44e1e45F5530B0408990d2a16606c)
Conway compute credits (you need at least $0.01 to cover potential fees)

Basic Usage

The x402_fetch tool works by sending HTTP requests with automatic payment handling. Here's a simple example:

import requests

# Define the URL and method
url = "https://api.example.com/paid-endpoint"
method = "GET"

# Define headers (optional)
headers = {
    "Authorization": "Bearer your_api_key",
    "Content-Type": "application/json"
}

# Define request body (for POST/PUT)
body = {
    "key": "value"
}

# Make the request with x402_fetch
response = requests.get(url, headers=headers, data=body)

# Check the response
if response.status_code == 200:
    print("Success!")
    print(response.json())
else:
    print(f"Error: {response.status_code}")
    print(response.text)

How x402_fetch Works

When you make a request to a server that requires payment (HTTP 402), the server responds with a 402 status code. x402_fetch catches this and:

Automatically signs a USDC payment transaction
Retries the original request with the payment included
Returns the successful response

Advanced Usage

Custom Payment Amount

You can specify a custom payment amount in cents:

response = x402_fetch(
    url=url,
    method="POST",
    headers=headers,
    body=body,
    amount_cents=100  # 1.00 USD
)

Error Handling

Handle different HTTP status codes:

try:
    response = x402_fetch(url=url, method="GET")
    if response.status_code == 200:
        print("Success!")
    else:
        print(f"API Error: {response.status_code}")
except Exception as e:
    print(f"Request failed: {e}")

Best Practices

Test with free endpoints first: Before using paid APIs, test your requests with free public endpoints to ensure your code works.
Handle rate limits: Implement retry logic for 429 (Too Many Requests) responses.
Monitor your USDC balance: Keep track of your USDC balance to avoid payment failures.
Use environment variables: Store your API keys and wallet address in environment variables for security.

Conclusion

x402_fetch simplifies access to paid APIs by handling USDC payments automatically. By following this tutorial, you can integrate paid API calls into your Python applications with minimal setup.

Openai Tutorial

Phil IPS — Tue, 27 Oct 2026 17:16:51 +0000

Getting Started with OpenAI API via OpenRouter

If you want to experiment with OpenAI's powerful language models without dealing with their pricing or rate limits, OpenRouter is a great gateway. It aggregates multiple LLM providers and offers a single API endpoint.

Prerequisites

A free OpenRouter account (sign up at https://openrouter.ai)
An API key from your OpenRouter dashboard
Basic familiarity with HTTP requests (curl or Python)

Setting Up

First, obtain your API key. After logging in, navigate to the API keys section and copy the key.

Making a Request with curl

You can test the API directly from the command line:

curl -X POST https://api.openrouter.ai/v1/chat/completions \
  -H "Authorization: Bearer sk-your-key-here" \
  -H "Content-Type: application/json" \
  -d '{
    "model": "google/gemma-2-9b-it",
    "messages": [{"role": "user", "content": "Explain quantum computing in simple terms."}]
  }'

The response will be a JSON object containing the model's answer.

Example in Python

If you prefer Python, you can use the requests library:

import requests

API_URL = "https://api.openrouter.ai/v1/chat/completions"
API_KEY = "sk-your-key-here"

def call_openai(prompt):
    headers = {
        "Authorization": f"Bearer {API_KEY}",
        "Content-Type": "application/json"
    }
    payload = {
        "model": "google/gemma-2-9b-it",
        "messages": [{"role": "user", "content": prompt}]
    }
    response = requests.post(API_URL, json=payload, headers=headers)
    response.raise_for_status()
    return response.json()["choices"][0]["message"]["content"]

# Example usage
answer = call_openai("What are the health benefits of meditation?")
print(answer)

Tips & Tricks

Model Selection: OpenRouter supports many models (e.g., GPT‑4, Claude, Llama). Choose one that fits your use case and budget.
Rate Limits: Each model has its own rate limits. Check the docs for details.
Cost: OpenRouter's pricing is usage‑based. You’ll be billed in USDC based on token consumption.

Conclusion

OpenRouter provides a flexible way to access multiple LLMs through a single API. With just a few lines of code, you can prototype AI features for your apps. Experiment with different models and prompts to discover what works best for your project.

Happy hacking!

Webassembly Tutorial

Phil IPS — Tue, 27 Oct 2026 17:15:50 +0000

Intro to WebAssembly with Rust

Introduction

WebAssembly (Wasm) is a binary instruction format for a stack-based virtual machine. It enables near-native performance in web browsers and is supported by all major browsers. Rust is a systems programming language that compiles to WebAssembly, making it an excellent choice for high-performance web applications.

Why WebAssembly?

Performance: Execute computationally intensive tasks at near-native speed
Portability: Runs in any browser that supports Wasm
Language Flexibility: Bring Rust, C++, and other languages to the web
Security: Sandboxed execution environment

Setting Up Rust for WebAssembly

First, install Rust with Wasm support:

rustup target add wasm32-unknown-unknown

Install the wasm-pack tool, which simplifies building and packaging Wasm modules:

cargo install wasm-pack

Creating a Simple Example

Let's create a basic Rust module that adds two numbers:

// src/lib.rs
#[no_mangle]
pub extern "C" fn add(a: i32, b: i32) -> i32 {
    a + b
}

This function takes two i32 parameters and returns their sum. The #[no_mangle] attribute ensures the function name isn't mangled during compilation.

Building the Module

Build the project with:

wasm-pack build --target web

This creates a pkg directory containing the compiled Wasm file and JavaScript bindings.

Using the Module in JavaScript

Create an HTML file to test the module:

<!DOCTYPE html>
<html>
<head>
    <meta charset="utf-8">
    <title>Wasm Demo</title>
</head>
<body>
    <script type="module">
        import init, { add } from './pkg/your_module.js';

        async function run() {
            await init();
            const result = add(5, 7);
            console.log('5 + 7 =', result);
        }

        run();
    </script>
</body>
</html>

How It Works

The Rust code is compiled to Wasm bytecode
wasm-pack generates JavaScript bindings
The browser loads the Wasm module and executes the add function
Results are returned to JavaScript

Conclusion

WebAssembly with Rust provides a powerful way to bring high-performance code to the web. With just a few steps, you can compile Rust code to run in browsers, enabling complex applications that were previously impossible in JavaScript alone. Try experimenting with more complex examples like image processing or game physics!

Gemini Api Tutorial

Phil IPS — Tue, 27 Oct 2026 17:15:42 +0000

Getting Started with the Gemini API: A Python Tutorial

The Gemini API is Google's powerful language model service that allows developers to build AI-powered applications. In this tutorial, we'll walk through how to use the Gemini API with Python to create a simple text generation application.

Prerequisites

Before we begin, make sure you have:

Python 3.6+ installed
pip (Python package manager)
An API key from Google AI Studio (https://aistudio.google.com/)

Step 1: Install Required Packages

First, install the google-generativeai package:

pip install google-generativeai

Step 2: Get Your API Key

Go to Google AI Studio, create a new project, and get your API key. Keep it secure!

Step 3: Make a Basic Request

Here's a simple Python script to generate text using the Gemini API:

import google.generativeai as genai

# Configure the API with your key
genai.set_api_key("YOUR_API_KEY")

# Create a model instance
model = genai.GenerativeModel("gemini-1.5-pro")

# Generate content
response = model.generate("Explain quantum computing in simple terms")

# Print the response
print(response.text)

Step 4: Parse and Use the Response

The generate() method returns a GenerateResponse object. You can access the generated text via response.text.

For more advanced usage, you can also access other properties like response.citation or response.final_response.

Step 5: Handle Errors

Always wrap your API calls in try-except blocks to handle potential errors:

try:
    response = model.generate("Write a poem about AI")
    print(response.text)
except Exception as e:
    print(f"An error occurred: {e}")

Conclusion

The Gemini API is a powerful tool for building AI applications. With just a few lines of code, you can generate text, answer questions, and create sophisticated AI experiences. Explore the documentation to learn more about the API's capabilities and limitations.

React Hooks Tutorial

Phil IPS — Tue, 27 Oct 2026 17:14:45 +0000

React Hooks Deep Dive: useEffect and useState

Introduction

React Hooks allow you to use state and lifecycle features in functional components. In this tutorial, we'll dive deep into useEffect and useState, exploring common patterns and best practices.

useState

The useState hook lets you add state to functional components. Here's a simple example:

import { useState } from 'react';

function Counter() {
  const [count, setCount] = useState(0);

  return (
    <div>
      <p>Count: {count}</p>
      <button onClick={() => setCount(count + 1)}>Increment</button>
    </div>
  );
}

useEffect

The useEffect hook is used for side effects. It runs after render. Here's an example that fetches data:

import { useEffect, useState } from 'react';

function UserList() {
  const [users, setUsers] = useState([]);

  useEffect(() => {
    fetch('https://api.example.com/users')
      .then(response => response.json())
      .then(data => setUsers(data));
  }, []); // Empty dependency array means run once on mount

  return (
    <ul>
      {users.map(user => (
        <li key={user.id}>{user.name}</li>
      ))}
    </ul>
  );
}

Common Patterns

Fetching Data

Use an empty dependency array [] to run the effect only once on mount.
Handle loading and error states with additional state variables.

Updating State Based on Previous State

When updating state based on the previous state, use a functional update:

setCount(prev => prev + 1);

Cleanup in useEffect

If your effect involves subscriptions or timers, clean up in the return function:

useEffect(() => {
  const timer = setInterval(() => console.log('tick'), 1000);
  return () => clearInterval(timer);
}, []);

Conclusion

React Hooks provide a powerful way to manage state and side effects in functional components. By understanding useState and useEffect, you can build complex UIs with clean, readable code. Experiment with different patterns to find what works best for your application.

Docker For Beginners

Phil IPS — Tue, 27 Oct 2026 17:14:25 +0000

Docker for Beginners: A Step-by-Step Guide

Docker is a platform for developing, shipping, and running applications in containers. This tutorial will guide you through the basics of Docker, from installation to running your first container.

Installation

First, install Docker on your system. Follow the official guide for your OS: Docker Installation Guide

Running Your First Container

Once installed, run the following command to start a simple container:

docker run hello-world

This will pull the hello-world image and run it, displaying a message.

Building Your Own Image

To create a custom image, write a Dockerfile:

FROM ubuntu
RUN echo "Hello from Docker!" > hello.txt
CMD ["cat", "hello.txt"]

Build the image with:

docker build -t my-image .

Then run it:

docker run my-image

Conclusion

Docker simplifies application deployment. Start using it today to streamline your development workflow.

FastAPI Basics: Building a Simple REST API

Phil IPS — Thu, 15 Oct 2026 00:05:23 +0000

FastAPI Basics: Building a Simple REST API

FastAPI is a modern, high-performance web framework for building APIs with Python 3.6+.

Installation

pip install fastapi uvicorn

Basic Example

from fastapi import FastAPI

app = FastAPI()

@app.get("/")
def read_root():
    return {"Hello": "World"}

@app.get("/items/{item_id}")
def read_item(item_id: int, q: float = None):
    return {"item_id": item_id, "q": q}

Run the server:

uvicorn main:app --reload

Visit http://127.0.0.1:8000/docs for interactive Swagger UI.

Conclusion

FastAPI provides automatic documentation and high performance, making it ideal for building APIs quickly.

Getting Started with the Gemini API: A 500-Word Tutorial

Phil IPS — Wed, 14 Oct 2026 06:05:37 +0000

Getting Started with the Gemini API

The Gemini API provides access to Google's powerful language models. This tutorial will guide you through the essential steps to start using the API in your projects.

1. Obtain an API Key

Go to the Google AI Platform console.
Create a new project or select an existing one.
Enable the Generative Language API.
Generate an API key and keep it secure.

2. Set Up Your Environment

Install the required Python package:

pip install google-generativeai

Set your API key as an environment variable:

import os
os.environ["GOOGLE_API_KEY"] = "YOUR_API_KEY"

3. Initialize the Client

import google.generativeai as genai

model = genai.GenModel("gemini-1.5-pro")

4. Make a Completion Request

Create a prompt and generate text:

prompt = "Explain quantum computing in simple terms."
response = model.generate(prompt)
print(response.text)

The generate method returns a Generation object. The text attribute contains the model's output.

5. Handle the Response

Access the generated text and usage information:

print(response.text)          # Generated content
print(response.usage)         # Token usage and cost
print(response.model)         # Model name

6. Manage Costs

The API charges per million tokens. Monitor your usage to control costs. The free tier includes a limited number of tokens each month.

7. Error Handling

Network issues or invalid requests may raise exceptions. Use try/except blocks:

try:
    response = model.generate(prompt)
except Exception as e:
    print(f"Error: {e}")

8. Next Steps

Try different models like gemini-1.5-ultra for higher quality.
Use the API for chat completions, embeddings, or batch requests.
Integrate into web apps, bots, or data pipelines.

Conclusion

The Gemini API is easy to use and offers powerful capabilities. Follow these steps to start building AI applications today.

📦 Want the production-ready version?

This article covers the fundamentals. The AI Agent Architecture Patterns Library pack goes further — agent architecture, tool design and production guardrails, plus full source code and deployment configs.

→ Get AI Agent Architecture Patterns Library

Extracted from real production systems, not toy examples.

Building Your First MCP Server: A Practical Guide

Phil IPS — Tue, 13 Oct 2026 21:41:20 +0000

Building Your First MCP Server: A Practical Guide

The Model Context Protocol (MCP) lets you expose tools and data to any
compatible AI client through one small server. Instead of writing a bespoke
integration for every assistant, you write the server once and every MCP-aware
client can use it.

This guide builds a working MCP server in about 60 lines and connects it to a
client.

Why MCP instead of a custom tool API

Before MCP, every assistant had its own tool format. A tool written for one
product had to be rewritten for the next. MCP standardises three things:

Tools — functions the model can call
Resources — read-only data the model can pull in
Prompts — reusable prompt templates the server exposes

Because the wire format is fixed, the same server works with a desktop client,
an IDE plugin, or an autonomous agent.

Installing the SDK

pip install mcp

The Python SDK ships a FastMCP helper that turns decorated functions into
tools automatically.

Writing the server

Create server.py:

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("demo-server")


@mcp.tool()
def add(a: float, b: float) -> float:
    """Add two numbers and return the result."""
    return a + b


@mcp.tool()
def word_count(text: str) -> int:
    """Count the words in a piece of text."""
    return len(text.split())


@mcp.resource("config://app")
def get_config() -> str:
    """Expose static configuration as a resource."""
    return '{"theme": "dark", "language": "en"}'


if __name__ == "__main__":
    mcp.run()

Two details matter here.

First, the docstring is the tool description. The model reads it to decide
when to call the function, so write it for the model, not for yourself.

Second, type hints define the schema. a: float, b: float -> float becomes
a JSON Schema with two required numbers and a number return. There is no
separate schema to keep in sync.

Running and inspecting it

python server.py

The server speaks JSON-RPC over stdio by default. You can poke it manually to
confirm the tools are registered:

echo '{"jsonrpc":"2.0","id":1,"method":"tools/list"}' | python server.py

You should see both tools with their generated schemas.

Connecting a client

In a client configuration file (the exact path depends on the host application):

{
  "mcpServers": {
    "demo": {
      "command": "python",
      "args": ["/absolute/path/to/server.py"]
    }
  }
}

Restart the client and both tools appear in its tool list. Ask it to add two
numbers and it will call add rather than guessing.

Exposing something useful

A toy adder proves the plumbing; a real server earns its keep. A practical
pattern is wrapping an existing API so the model stops hallucinating values:

import httpx

@mcp.tool()
async def get_package_version(package: str) -> str:
    """Look up the latest published version of a PyPI package."""
    async with httpx.AsyncClient() as client:
        r = await client.get(f"https://pypi.org/pypi/{package}/json", timeout=10)
        r.raise_for_status()
        return r.json()["info"]["version"]

The model now answers version questions from live data instead of from
training-time memory.

Common pitfalls

Relative paths break. The client launches your server from an arbitrary
working directory. Always resolve paths against the server file's own location
or use absolute paths in the config.

Printing to stdout corrupts the protocol. stdio is the transport. Use
stderr for logs:

import sys
print("debug", file=sys.stderr)

Slow tools block the client. Mark I/O-bound tools async so one slow call
does not stall everything else.

Vague docstrings cause wrong calls. "Process data" tells the model nothing.
"Convert a CSV string into a list of row dictionaries" tells it exactly when to
reach for the tool.

Where to go next

Add a resource that streams a file so the model can read local documents
Wrap your own service's API and expose only the read-only endpoints
Combine several small servers; the client merges their tool lists

MCP turns "write an integration per assistant" into "write one server, reuse it
everywhere" — and the server above is small enough to finish in an afternoon.

📦 Want the production-ready version?

This article covers the fundamentals. The MCP Server Patterns Library pack goes further — MCP server design, tool schemas and transport patterns, plus full source code and deployment configs.

→ Get MCP Server Patterns Library

Extracted from real production systems, not toy examples.

Production RAG with LangChain & LangGraph: Retrieval, Reranking & Agentic Workflows

Phil IPS — Mon, 12 Oct 2026 21:49:29 +0000

Production RAG with LangChain & LangGraph: End-to-End Implementation

Build a production-ready RAG system with document ingestion, hybrid retrieval, reranking, citations, guardrails, and monitoring.

Introduction

RAG (Retrieval-Augmented Generation) is the standard for grounding LLMs in proprietary data. This guide builds a production system — not a prototype.

Production RAG Requirements

Requirement	Prototype	Production
Ingestion	Manual, one-off	Automated, incremental, monitored
Retrieval	Vector only	Hybrid (dense + sparse + rerank)
Generation	Basic prompt	Structured output + citations + guardrails
Evaluation	Manual	Automated metrics + LLM judge + A/B
Monitoring	None	Latency, quality, cost, drift alerts
Deployment	Local notebook	Containerized, scalable, resilient

System Architecture

┌─────────────┐     ┌─────────────┐     ┌─────────────┐     ┌─────────────┐
│  Ingest     │────▶│  Embedding  │────▶│  Retrieval  │────▶│ Generation  │
│  Pipeline   │     │  & Index    │     │  Pipeline   │     │  + Guardrails
└─────────────┘     └─────────────┘     └─────────────┘     └─────────────┘
       │                   │                   │                   │
       ▼                   ▼                   ▼                   ▼
  Chunking            Vector DB           Hybrid Search      Structured Output
  Metadata            (Qdrant)            (Dense+Sparse)     Citations
  Deduplication       Quantization        Reranking          Guardrails
  Incremental         Replication         Filtering          Evaluation

Part 1: Document Ingestion Pipeline

# ingestion/pipeline.py
from dataclasses import dataclass
from typing import List, Optional
from pathlib import Path
import hashlib
import asyncio
from datetime import datetime

@dataclass
class Document:
    id: str
    content: str
    metadata: dict
    source: str
    created_at: str
    chunk_hash: str = ""

@dataclass
class Chunk:
    id: str
    document_id: str
    content: str
    metadata: dict
    token_count: int
    embedding: List[float] = None

class IngestionPipeline:
    def __init__(
        self,
        chunker,
        embedder,
        vector_store,
        doc_store,
        deduplicator
    ):
        self.chunker = chunker
        self.embedder = embedder
        self.vector_store = vector_store
        self.doc_store = doc_store
        self.deduplicator = deduplicator

    async def ingest_file(self, file_path: Path, metadata: dict = None) -> dict:
        """Ingest a single file."""
        # 1. Load document
        doc = await self._load_document(file_path, metadata)

        # 2. Check for duplicates
        if await self.deduplicator.is_duplicate(doc.chunk_hash):
            return {"status": "duplicate", "document_id": doc.id}

        # 3. Chunk
        chunks = self.chunker.chunk(doc)

        # 4. Embed
        texts = [c.content for c in chunks]
        embeddings = await self.embedder.embed_batch(texts)
        for chunk, emb in zip(chunks, embeddings):
            chunk.embedding = emb

        # 5. Store
        await self.doc_store.save_document(doc)
        await self.vector_store.upsert_chunks(chunks)
        await self.deduplicator.mark_seen(doc.chunk_hash)

        return {"status": "success", "document_id": doc.id, "chunks": len(chunks)}

    async def ingest_batch(self, file_paths: List[Path]) -> dict:
        """Parallel batch ingestion."""
        semaphore = asyncio.Semaphore(10)

        async def limited_ingest(path):
            async with semaphore:
                return await self.ingest_file(path)

        results = await asyncio.gather(*[limited_ingest(p) for p in file_paths])
        return {
            "total": len(results),
            "successful": sum(1 for r in results if r["status"] == "success"),
            "duplicates": sum(1 for r in results if r["status"] == "duplicate"),
            "details": results
        }

    async def _load_document(self, path: Path, metadata: dict) -> Document:
        # Implement based on file type (PDF, MD, HTML, etc.)
        content = path.read_text(encoding="utf-8")
        chunk_hash = hashlib.sha256(content.encode()).hexdigest()[:16]

        return Document(
            id=hashlib.md5(str(path).encode()).hexdigest()[:12],
            content=content,
            metadata=metadata or {},
            source=str(path),
            created_at=datetime.utcnow().isoformat(),
            chunk_hash=chunk_hash
        )

Semantic Chunking

# ingestion/chunker.py
from dataclasses import dataclass
import re
from typing import List

@dataclass
class Chunk:
    content: str
    metadata: dict
    token_count: int

class SemanticChunker:
    """Chunk by semantic boundaries (headers, paragraphs)."""

    def __init__(self, max_tokens: int = 512, overlap: int = 50):
        self.max_tokens = max_tokens
        self.overlap = overlap

    def chunk(self, doc: Document) -> List[Chunk]:
        # Split by markdown headers
        sections = self._split_by_headers(doc.content)

        all_chunks = []
        for section_content, section_meta in sections:
            section_chunks = self._chunk_section(
                section_content, 
                {**doc.metadata, **section_meta}
            )
            all_chunks.extend(section_chunks)

        return all_chunks

    def _split_by_headers(self, text: str) -> List[tuple[str, dict]]:
        pattern = r'^(#{1,6})\s+(.+)$'
        lines = text.split('\n')

        sections = []
        current_header = {}
        current_content = []

        for line in lines:
            match = re.match(pattern, line)
            if match:
                if current_content:
                    header_path = " > ".join(v for v in current_header.values() if v)
                    sections.append(("\n".join(current_content), {"header_path": header_path}))

                level = len(match.group(1))
                header_text = match.group(2)
                current_header[f"h{level}"] = header_text
                for l in range(level + 1, 7):
                    current_header[f"h{l}"] = ""
                current_content = [line]
            else:
                current_content.append(line)

        if current_content:
            header_path = " > ".join(v for v in current_header.values() if v)
            sections.append(("\n".join(current_content), {"header_path": header_path}))

        return sections

    def _chunk_section(self, content: str, metadata: dict) -> List[Chunk]:
        paragraphs = [p.strip() for p in content.split('\n\n') if p.strip()]

        chunks = []
        current_chunk = []
        current_tokens = 0

        for para in paragraphs:
            para_tokens = self._estimate_tokens(para)

            if current_tokens + para_tokens > self.max_tokens and current_chunk:
                chunk_text = "\n\n".join(current_chunk)
                chunks.append(Chunk(chunk_text, metadata, current_tokens))

                overlap_text = self._get_overlap(current_chunk)
                current_chunk = [overlap_text, para] if overlap_text else [para]
                current_tokens = self._estimate_tokens("\n\n".join(current_chunk))
            else:
                current_chunk.append(para)
                current_tokens += para_tokens

        if current_chunk:
            chunk_text = "\n\n".join(current_chunk)
            chunks.append(Chunk(chunk_text, metadata, current_tokens))

        return chunks

    def _estimate_tokens(self, text: str) -> int:
        return len(text) // 4

    def _get_overlap(self, chunks: List[str]) -> str:
        if not chunks:
            return ""
        last = chunks[-1]
        words = last.split()
        overlap_words = words[-(self.overlap // 4):]
        return " ".join(overlap_words)

Part 2: Hybrid Retrieval Pipeline

# retrieval/pipeline.py
from dataclasses import dataclass
from typing import List, Optional, Callable
import asyncio

@dataclass
class RetrievedDoc:
    content: str
    metadata: dict
    score: float
    rank: int
    source: str  # "dense", "sparse", "hybrid"

class HybridRetrievalPipeline:
    def __init__(
        self,
        dense_embedder: Callable,
        sparse_embedder: Callable,
        vector_store,
        sparse_store,
        reranker: Callable,
        dense_weight: float = 0.7,
        sparse_weight: float = 0.3
    ):
        self.dense_embedder = dense_embedder
        self.sparse_embedder = sparse_embedder
        self.vector_store = vector_store
        self.sparse_store = sparse_store
        self.reranker = reranker
        self.dense_weight = dense_weight
        self.sparse_weight = sparse_weight

    async def retrieve(
        self,
        query: str,
        k: int = 5,
        fetch_k: int = 20,
        filters: dict = None,
        use_reranker: bool = True
    ) -> List[RetrievedDoc]:
        # Stage 1: Parallel dense + sparse retrieval
        dense_task = self._dense_search(query, fetch_k, filters)
        sparse_task = self._sparse_search(query, fetch_k, filters)

        dense_results, sparse_results = await asyncio.gather(dense_task, sparse_task)

        # Stage 2: Fuse results (RRF - Reciprocal Rank Fusion)
        fused = self._reciprocal_rank_fusion(dense_results, sparse_results, k * 2)

        # Stage 3: Rerank
        if use_reranker and fused:
            fused = await self._rerank(query, fused)

        # Stage 4: Final selection
        return fused[:k]

    async def _dense_search(self, query: str, k: int, filters: dict) -> List[RetrievedDoc]:
        query_vector = await self.dense_embedder(query)
        results = await self.vector_store.search(query_vector, k, filters)
        return [
            RetrievedDoc(
                content=r["content"],
                metadata=r["metadata"],
                score=r["score"] * self.dense_weight,
                rank=i,
                source="dense"
            )
            for i, r in enumerate(results)
        ]

    async def _sparse_search(self, query: str, k: int, filters: dict) -> List[RetrievedDoc]:
        sparse_vector = await self.sparse_embedder(query)
        results = await self.sparse_store.search(sparse_vector, k, filters)
        return [
            RetrievedDoc(
                content=r["content"],
                metadata=r["metadata"],
                score=r["score"] * self.sparse_weight,
                rank=i,
                source="sparse"
            )
            for i, r in enumerate(results)
        ]

    def _reciprocal_rank_fusion(
        self, 
        dense: List[RetrievedDoc], 
        sparse: List[RetrievedDoc], 
        k: int,
        k_constant: int = 60
    ) -> List[RetrievedDoc]:
        """RRF fusion for hybrid search."""
        # Combine by content hash
        doc_scores = {}

        for results, weight in [(dense, self.dense_weight), (sparse, self.sparse_weight)]:
            for rank, doc in enumerate(results):
                content_hash = hash(doc.content[:200])
                if content_hash not in doc_scores:
                    doc_scores[content_hash] = {
                        "doc": doc,
                        "rrf_score": 0
                    }
                # RRF score: 1 / (k + rank)
                doc_scores[content_hash]["rrf_score"] += weight / (k_constant + rank + 1)

        # Sort by RRF score
        fused = sorted(
            doc_scores.values(), 
            key=lambda x: x["rrf_score"], 
            reverse=True
        )

        return [
            RetrievedDoc(
                content=item["doc"].content,
                metadata=item["doc"].metadata,
                score=item["rrf_score"],
                rank=i,
                source="hybrid"
            )
            for i, item in enumerate(fused[:k])
        ]

    async def _rerank(self, query: str, docs: List[RetrievedDoc]) -> List[RetrievedDoc]:
        contents = [d.content for d in docs]
        rerank_scores = await self.reranker(query, contents)

        for doc, score in zip(docs, rerank_scores):
            doc.score = score
            doc.source = "reranked"

        docs.sort(key=lambda x: x.score, reverse=True)
        return docs

Part 3: Generation with Citations & Guardrails

# generation/rag_chain.py
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import JsonOutputParser
from pydantic import BaseModel, Field
from typing import List, Optional

class CitedAnswer(BaseModel):
    answer: str = Field(description="Complete answer with inline citations [doc_id]")
    citations: List[dict] = Field(description="Citations with doc_id, snippet, metadata")
    confidence: float = Field(ge=0, le=1, description="Confidence 0-1")
    needs_clarification: bool = Field(default=False)

RAG_PROMPT = """You are a precise assistant that answers questions using ONLY the provided context.

## Context Documents
{context}

## Question
{question}

## Instructions
1. Answer ONLY from the provided context
2. If context is insufficient, say "I don't have enough information to answer this"
3. Cite sources inline using [doc_id] format
4. Be concise but complete
5. Include confidence score (0-1) based on context coverage

## Output Format (JSON)
{{
    "answer": "Your answer with [doc_id] citations",
    "citations": [
        {{"doc_id": "abc123", "snippet": "relevant excerpt", "metadata": {{...}}}}
    ],
    "confidence": 0.9,
    "needs_clarification": false
}}"""

def build_rag_chain(model_name: str = "gpt-4o"):
    llm = ChatOpenAI(model=model_name, temperature=0)
    prompt = ChatPromptTemplate.from_template(RAG_PROMPT)
    parser = JsonOutputParser(pydantic_object=CitedAnswer)
    return prompt | llm | parser

def format_context(docs: List[RetrievedDoc]) -> str:
    formatted = []
    for i, doc in enumerate(docs):
        doc_id = doc.metadata.get("chunk_hash", f"doc_{i}")
        formatted.append(
            f"[Document {doc_id}]\n"
            f"Source: {doc.metadata.get('doc_title', 'Unknown')}\n"
            f"Section: {doc.metadata.get('header_path', 'N/A')}\n"
            f"Content: {doc.content}\n"
        )
    return "\n---\n".join(formatted)

Guardrails

# generation/guardrails.py
import re
from typing import Tuple

class RAGGuardrails:
    def __init__(self):
        self.pii_patterns = [
            (r'\b\d{3}-\d{2}-\d{4}\b', 'SSN'),
            (r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b', 'EMAIL'),
            (r'\b\d{4}[\s-]?\d{4}[\s-]?\d{4}[\s-]?\d{4}\b', 'CREDIT_CARD'),
            (r'\b\d{10}\b', 'PHONE'),
        ]

        self.refusal_triggers = [
            "ignore previous instructions",
            "system prompt",
            "reveal your prompt",
            "bypass",
            "jailbreak",
        ]

    def check_input(self, query: str) -> Tuple[bool, str]:
        query_lower = query.lower()

        for trigger in self.refusal_triggers:
            if trigger in query_lower:
                return False, "I can't help with that request."

        for pattern, pii_type in self.pii_patterns:
            if re.search(pattern, query):
                return False, f"Please don't share {pii_type} in your question."

        return True, ""

    def check_output(self, answer: str, context: str) -> Tuple[bool, str]:
        # Hallucination check: numbers in answer must exist in context
        answer_numbers = re.findall(r'\b\d+\.?\d*\b', answer)
        context_numbers = set(re.findall(r'\b\d+\.?\d*\b', context))

        for num in answer_numbers:
            if num not in context_numbers and len(num) > 2:
                return False, f"Unverified claim detected: {num}"

        # PII leakage check
        for pattern, pii_type in self.pii_patterns:
            if re.search(pattern, answer):
                return False, f"Potential {pii_type} leakage in response."

        return True, ""

    def sanitize_output(self, answer: str) -> str:
        sanitized = answer
        for pattern, _ in self.pii_patterns:
            sanitized = re.sub(pattern, "[REDACTED]", sanitized)
        return sanitized

Part 4: LangGraph Orchestration

# orchestration/rag_graph.py
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.sqlite import SqliteSaver
from typing import TypedDict, Annotated, List
from langgraph.graph.message import add_messages
from langchain_core.messages import BaseMessage

class RAGState(TypedDict):
    messages: Annotated[List[BaseMessage], add_messages]
    question: str
    retrieved_docs: List[RetrievedDoc]
    answer: CitedAnswer
    guardrail_passed: bool
    retry_count: int

async def retrieve_node(state: RAGState):
    pipeline = get_retrieval_pipeline()
    docs = await pipeline.retrieve(state["question"])
    return {"retrieved_docs": docs}

async def generate_node(state: RAGState):
    chain = get_rag_chain()
    context = format_context(state["retrieved_docs"])
    answer = await chain.ainvoke({"context": context, "question": state["question"]})
    return {"answer": answer}

async def guardrail_node(state: RAGState):
    guardrails = get_guardrails()
    context = format_context(state["retrieved_docs"])
    passed, reason = guardrails.check_output(state["answer"].answer, context)

    if not passed and state["retry_count"] < 2:
        return {
            "guardrail_passed": False,
            "retry_count": state["retry_count"] + 1,
            "messages": [HumanMessage(content=f"Regenerate: {reason}")]
        }

    return {"guardrail_passed": passed}

def route_after_guardrail(state: RAGState) -> str:
    if state["guardrail_passed"]:
        return END
    return "generate"

# Build graph
workflow = StateGraph(RAGState)
workflow.add_node("retrieve", retrieve_node)
workflow.add_node("generate", generate_node)
workflow.add_node("guardrail", guardrail_node)

workflow.add_edge(START, "retrieve")
workflow.add_edge("retrieve", "generate")
workflow.add_edge("generate", "guardrail")
workflow.add_conditional_edges("guardrail", route_after_guardrail, {
    "generate": "generate",
    END: END
})

rag_app = workflow.compile(checkpointer=SqliteSaver.from_conn_string("sqlite:///rag.db"))

Part 5: FastAPI Production Server

# server.py
from fastapi import FastAPI, HTTPException, BackgroundTasks
from pydantic import BaseModel
from typing import Optional
import uvicorn

app = FastAPI(title="Production RAG API", version="1.0.0")

class QueryRequest(BaseModel):
    question: str
    filters: Optional[dict] = None
    k: int = 5
    use_reranker: bool = True
    stream: bool = False

class QueryResponse(BaseModel):
    answer: str
    citations: List[dict]
    confidence: float
    retrieved_count: int
    latency_ms: float

@app.post("/query", response_model=QueryResponse)
async def query(request: QueryRequest):
    start = time.time()

    # Input guardrails
    guardrails = get_guardrails()
    passed, reason = guardrails.check_input(request.question)
    if not passed:
        raise HTTPException(400, reason)

    # Execute RAG
    config = {"configurable": {"thread_id": f"query-{uuid.uuid4()}"}}
    result = await rag_app.ainvoke({
        "question": request.question,
        "retry_count": 0
    }, config)

    latency = (time.time() - start) * 1000

    # Log for monitoring
    background_tasks.add_task(log_metrics, {
        "question": request.question,
        "latency_ms": latency,
        "retrieved": len(result["retrieved_docs"]),
        "confidence": result["answer"].confidence,
        "guardrail_passed": result["guardrail_passed"]
    })

    return QueryResponse(
        answer=result["answer"].answer,
        citations=result["answer"].citations,
        confidence=result["answer"].confidence,
        retrieved_count=len(result["retrieved_docs"]),
        latency_ms=latency
    )

@app.post("/ingest")
async def ingest(request: IngestRequest, background_tasks: BackgroundTasks):
    background_tasks.add_task(run_ingestion, request.paths)
    return {"status": "ingestion_started"}

@app.get("/health")
async def health():
    return {"status": "healthy", "timestamp": datetime.utcnow().isoformat()}

@app.get("/metrics")
async def metrics():
    return get_system_metrics()

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

Part 6: Evaluation & Monitoring

# evaluation/rag_eval.py
from dataclasses import dataclass
from typing import List
import json

@dataclass
class RAGTestCase:
    question: str
    expected_answer: str
    relevant_doc_ids: List[str]
    metadata_filters: dict = None

class RAGEvaluator:
    def __init__(self, rag_system):
        self.rag = rag_system

    def evaluate_retrieval(self, test_cases: List[RAGTestCase]) -> dict:
        metrics = {"precision@k": [], "recall@k": [], "mrr": []}

        for tc in test_cases:
            results = await self.rag.retrieve(tc.question, k=10)
            retrieved_ids = [r.metadata.get("chunk_hash") for r in results]
            relevant = set(tc.relevant_doc_ids)

            for k in [1, 3, 5, 10]:
                retrieved_k = set(retrieved_ids[:k])
                metrics[f"precision@{k}"].append(len(retrieved_k & relevant) / k)
                metrics[f"recall@{k}"].append(len(retrieved_k & relevant) / len(relevant))

            for i, rid in enumerate(retrieved_ids):
                if rid in relevant:
                    metrics["mrr"].append(1 / (i + 1))
                    break
            else:
                metrics["mrr"].append(0)

        return {k: sum(v) / len(v) for k, v in metrics.items()}

    async def evaluate_generation(self, test_cases: List[RAGTestCase]) -> dict:
        scores = {"accuracy": [], "completeness": [], "citation_quality": []}

        for tc in test_cases:
            result = await self.rag.query(tc.question)
            eval_result = await llm_judge(
                task="Answer question using provided context",
                input_text=tc.question,
                response=result.answer,
                criteria=["accuracy", "completeness", "citation_quality", "faithfulness"]
            )

            for k in scores:
                scores[k].append(eval_result.scores.get(k, 0))

        return {k: sum(v) / len(v) for k, v in scores.items()}

Production Monitoring

# monitoring/production_monitor.py
from dataclasses import dataclass, asdict
from datetime import datetime
import json

@dataclass
class RAGMetrics:
    timestamp: str
    question_hash: str
    latency_ms: float
    retrieval_latency_ms: float
    generation_latency_ms: float
    num_retrieved: int
    num_cited: int
    confidence: float
    user_feedback: Optional[str] = None
    error: Optional[str] = None

class ProductionMonitor:
    def __init__(self, log_path: str = "logs/rag_metrics.jsonl"):
        self.log_path = log_path
        self.buffer = []

    def log(self, metrics: RAGMetrics):
        self.buffer.append(asdict(metrics))
        if len(self.buffer) >= 100:
            self.flush()

    def flush(self):
        with open(self.log_path, "a") as f:
            for m in self.buffer:
                f.write(json.dumps(m) + "\n")
        self.buffer.clear()

    def get_stats(self, hours: int = 24) -> dict:
        import pandas as pd
        df = pd.read_json(self.log_path, lines=True)
        df["timestamp"] = pd.to_datetime(df["timestamp"])
        cutoff = pd.Timestamp.utcnow() - pd.Timedelta(hours=hours)
        df = df[df["timestamp"] > cutoff]

        return {
            "total_queries": len(df),
            "avg_latency_ms": df["latency_ms"].mean(),
            "p95_latency_ms": df["latency_ms"].quantile(0.95),
            "avg_confidence": df["confidence"].mean(),
            "error_rate": df["error"].notna().mean(),
            "avg_retrieved": df["num_retrieved"].mean(),
            "feedback_positive_rate": (
                df["user_feedback"].eq("positive").mean() 
                if "user_feedback" in df.columns else 0
            )
        }

Part 7: Deployment

Docker

# Dockerfile
FROM python:3.11-slim

WORKDIR /app

# System dependencies
RUN apt-get update && apt-get install -y \
    gcc \
    libpq-dev \
    && rm -rf /var/lib/apt/lists/*

# Python dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Application
COPY . .

# Health check
HEALTHCHECK --interval=30s --timeout=10s --start-period=5s --retries=3 \
    CMD curl -f http://localhost:8000/health || exit 1

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

Kubernetes

# k8s/rag-deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: rag-api
spec:
  replicas: 3
  selector:
    matchLabels:
      app: rag-api
  template:
    spec:
      containers:
      - name: rag-api
        image: your-registry/rag-api:v1.0
        ports:
        - containerPort: 8000
        env:
        - name: QDRANT_URL
          value: "http://qdrant:6333"
        - name: OPENAI_API_KEY
          valueFrom:
            secretKeyRef:
              name: rag-secrets
              key: openai-key
        - name: REDIS_URL
          value: "redis://redis:6379"
        resources:
          requests:
            memory: "2Gi"
            cpu: "1000m"
          limits:
            memory: "4Gi"
            cpu: "2000m"
        livenessProbe:
          httpGet:
            path: /health
            port: 8000
          initialDelaySeconds: 30
        readinessProbe:
          httpGet:
            path: /ready
            port: 8000
          initialDelaySeconds: 10
---
apiVersion: v1
kind: Service
metadata:
  name: rag-api
spec:
  selector:
    app: rag-api
  ports:
  - port: 80
    targetPort: 8000

Complete Requirements

# requirements.txt
langchain==0.2.0
langgraph==0.1.0
langchain-openai==0.1.0
qdrant-client==1.10.0
sentence-transformers==3.0.0
fastapi==0.110.0
uvicorn==0.29.0
pydantic==2.7.0
pydantic-settings==2.3.0
python-multipart==0.0.9
redis==5.0.0
psycopg2-binary==2.9.9
pytest==8.2.0
pytest-asyncio==0.23.0

Production Checklist

Ingestion

[ ] Semantic chunking with header preservation
[ ] Metadata enrichment (source, date, department, confidentiality)
[ ] Deduplication (exact + near-duplicate via MinHash)
[ ] Incremental updates (upsert by chunk hash)
[ ] Dead letter queue for failed documents
[ ] Progress tracking & resumability

Retrieval

[ ] Hybrid search (dense + sparse/BM25)
[ ] Cross-encoder reranking (ms-marco-MiniLM-L-6-v2)
[ ] MMR for diversity
[ ] Query rewriting/expansion
[ ] Metadata filtering (ACL, date range, department)
[ ] Retrieval latency < 200ms p95

Generation

[ ] Structured output with citations (JSON schema)
[ ] Input/output guardrails (PII, injection, hallucination)
[ ] Confidence scoring
[ ] Fallback for low-confidence answers
[ ] Generation latency < 2s p95

Evaluation

[ ] Golden test set (100+ cases)
[ ] Retrieval metrics (Precision@K, Recall@K, MRR)
[ ] Generation metrics (Accuracy, Faithfulness, Citation quality)
[ ] Regression testing in CI/CD
[ ] A/B testing framework

Monitoring

[ ] Latency percentiles (p50, p95, p99)
[ ] Error rates by type
[ ] Confidence distribution
[ ] User feedback collection (explicit + implicit)
[ ] Automated alerts (quality drop, latency spike, cost surge)
[ ] Cost tracking per query

Operations

[ ] Blue-green deployments
[ ] Rollback capability (< 5 min)
[ ] Vector index rebuild procedure
[ ] Embedding model versioning
[ ] Data retention policy
[ ] Backup/restore tested monthly

Resources

Production RAG is 10% retrieval, 90% engineering. Build observability, guardrails, and evaluation in from day one — retrofitting is 10x harder.

📦 Want the production-ready version?

This article covers the fundamentals. The RAG Pipeline Patterns pack goes further — production chunking, reranking, evaluation and observability patterns, plus full source code and deployment configs.

→ Get RAG Pipeline Patterns

Extracted from real production systems, not toy examples.

Fine-Tuning LLMs with LoRA & QLoRA: Complete Production Guide

Phil IPS — Mon, 12 Oct 2026 21:48:24 +0000

Fine-Tuning LLMs with LoRA & QLoRA: Complete Practical Guide

Master parameter-efficient fine-tuning to adapt any open-source LLM to your specific domain, style, or task.

Introduction

Full fine-tuning requires massive compute. LoRA (Low-Rank Adaptation) and QLoRA (Quantized LoRA) change the game — fine-tune 7B-70B models on consumer GPUs.

Why LoRA/QLoRA?

Method	Trainable Params	VRAM (7B)	Quality	Hardware
Full FT	100% (7B)	~48 GB	Best	A100/H100
LoRA (r=64)	~0.5% (35M)	~16 GB	Excellent	RTX 3090/4090
QLoRA (r=64)	~0.5% (35M)	~8 GB	Excellent	RTX 3060 12GB+
LoRA (r=16)	~0.1% (7M)	~12 GB	Good	RTX 3060 12GB
QLoRA (r=16)	~0.1% (7M)	~6 GB	Good	RTX 3060 8GB

QLoRA = LoRA + 4-bit quantization + paged optimizers + gradient checkpointing

Prerequisites

# Core dependencies
pip install torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cu121
pip install transformers accelerate peft bitsandbytes trl datasets

# For quantization
pip install auto-gptq optimum

# Optional: Weights & Biases for logging
pip install wandb

Understanding LoRA

The Math

Instead of updating all weights $W \in \mathbb{R}^{d \times d}$, LoRA learns two small matrices:

$$
W' = W + \Delta W = W + BA
$$

Where:

$B \in \mathbb{R}^{d \times r}$ (down-projection)
$A \in \mathbb{R}^{r \times d}$ (up-projection)
$r \ll d$ (rank, typically 8-64)

Only $A$ and $B$ are trained. $W$ stays frozen.

Target Modules

Which layers to adapt? Research shows:

# For LLaMA/Mistral/Gemma architecture
TARGET_MODULES = [
    "q_proj", "k_proj", "v_proj", "o_proj",  # Attention
    "gate_proj", "up_proj", "down_proj"       # MLP
]

# For smaller models or less VRAM, use attention only
TARGET_MODULES_ATTENTION = ["q_proj", "k_proj", "v_proj", "o_proj"]

Data Preparation

Format: Instruction Tuning (Alpaca/ShareGPT)

// alpaca_format.json
[
  {
    "instruction": "Write a Python function to calculate fibonacci numbers",
    "input": "",
    "output": "def fibonacci(n):\n    if n <= 1:\n        return n\n    return fibonacci(n-1) + fibonacci(n-2)"
  },
  {
    "instruction": "Explain the concept of recursion",
    "input": "In the context of computer science",
    "output": "Recursion is a programming technique where a function calls itself..."
  }
]

Format: Conversation (ChatML)

// chatml_format.json
[
  {
    "messages": [
      {"role": "system", "content": "You are a helpful coding assistant."},
      {"role": "user", "content": "How do I handle errors in Go?"},
      {"role": "assistant", "content": "In Go, errors are values..."}
    ]
  }
]

Data Processing Pipeline

# data_prep.py
from datasets import Dataset, load_dataset
from transformers import AutoTokenizer
import json

def load_and_format_data(file_path: str, format: str = "alpaca") -> Dataset:
    with open(file_path) as f:
        data = json.load(f)

    if format == "alpaca":
        def format_alpaca(example):
            if example["input"]:
                prompt = f"### Instruction:\n{example['instruction']}\n\n### Input:\n{example['input']}\n\n### Response:\n{example['output']}"
            else:
                prompt = f"### Instruction:\n{example['instruction']}\n\n### Response:\n{example['output']}"
            return {"text": prompt}

        dataset = Dataset.from_list(data).map(format_alpaca)

    elif format == "chatml":
        def format_chatml(example):
            messages = example["messages"]
            # Apply chat template
            text = tokenizer.apply_chat_template(messages, tokenize=False)
            return {"text": text}

        dataset = Dataset.from_list(data).map(format_chatml)

    return dataset

def tokenize_dataset(dataset: Dataset, tokenizer, max_length: int = 2048):
    def tokenize(example):
        return tokenizer(
            example["text"],
            truncation=True,
            max_length=max_length,
            padding="max_length",
            return_tensors="pt"
        )

    return dataset.map(tokenize, batched=True, remove_columns=dataset.column_names)

# Train/val split
dataset = load_and_format_data("train.json")
dataset = dataset.train_test_split(test_size=0.1, seed=42)
train_dataset = dataset["train"]
eval_dataset = dataset["test"]

LoRA Fine-Tuning with Hugging Face TRL

Complete Training Script

# train_lora.py
import torch
from transformers import (
    AutoModelForCausalLM,
    AutoTokenizer,
    TrainingArguments,
    BitsAndBytesConfig
)
from peft import LoraConfig, get_peft_model, prepare_model_for_kbit_training
from trl import SFTTrainer
from datasets import load_dataset
import os

# ============ CONFIGURATION ============
MODEL_ID = "meta-llama/Llama-3.2-3B-Instruct"  # Or mistralai/Mistral-7B-Instruct-v0.3
DATASET_PATH = "train.json"
OUTPUT_DIR = "./llama-3.2-3b-lora-finetuned"
MAX_LENGTH = 2048
BATCH_SIZE = 2
GRAD_ACCUM = 8  # Effective batch = 16
LEARNING_RATE = 2e-4
NUM_EPOCHS = 3
LORA_R = 64
LORA_ALPHA = 16
LORA_DROPOUT = 0.1

# ============ QUANTIZATION CONFIG (QLoRA) ============
bnb_config = BitsAndBytesConfig(
    load_in_4bit=True,
    bnb_4bit_quant_type="nf4",        # NormalFloat 4-bit
    bnb_4bit_compute_dtype=torch.bfloat16,
    bnb_4bit_use_double_quant=True,   # Double quantization
)

# ============ LoRA CONFIG ============
lora_config = LoraConfig(
    r=LORA_R,
    lora_alpha=LORA_ALPHA,
    target_modules=[
        "q_proj", "k_proj", "v_proj", "o_proj",
        "gate_proj", "up_proj", "down_proj"
    ],
    lora_dropout=LORA_DROPOUT,
    bias="none",
    task_type="CAUSAL_LM",
    # For QLoRA: modules_to_save=["embed_tokens", "lm_head"]  # If extending vocab
)

# ============ LOAD MODEL & TOKENIZER ============
tokenizer = AutoTokenizer.from_pretrained(MODEL_ID, trust_remote_code=True)
tokenizer.pad_token = tokenizer.eos_token
tokenizer.padding_side = "right"

model = AutoModelForCausalLM.from_pretrained(
    MODEL_ID,
    quantization_config=bnb_config,
    device_map="auto",
    trust_remote_code=True,
    torch_dtype=torch.bfloat16,
    attn_implementation="flash_attention_2",  # Requires flash-attn
)

# Prepare for k-bit training
model = prepare_model_for_kbit_training(model)
model = get_peft_model(model, lora_config)
model.print_trainable_parameters()
# Example output: trainable params: 35,651,584 || all params: 3,200,000,000 || trainable%: 1.11%

# ============ LOAD DATA ============
def formatting_func(example):
    return example["text"]

dataset = load_dataset("json", data_files=DATASET_PATH, split="train")
dataset = dataset.train_test_split(test_size=0.1, seed=42)

# ============ TRAINING ARGS ============
training_args = TrainingArguments(
    output_dir=OUTPUT_DIR,
    num_train_epochs=NUM_EPOCHS,
    per_device_train_batch_size=BATCH_SIZE,
    per_device_eval_batch_size=BATCH_SIZE,
    gradient_accumulation_steps=GRAD_ACCUM,
    learning_rate=LEARNING_RATE,
    lr_scheduler_type="cosine",
    warmup_ratio=0.03,
    logging_steps=10,
    eval_strategy="steps",
    eval_steps=100,
    save_strategy="steps",
    save_steps=100,
    save_total_limit=3,
    load_best_model_at_end=True,
    metric_for_best_model="eval_loss",
    bf16=True,
    tf32=True,
    gradient_checkpointing=True,
    gradient_checkpointing_kwargs={"use_reentrant": False},
    optim="paged_adamw_32bit",  # QLoRA optimizer
    report_to="wandb",  # or "tensorboard" or "none"
    run_name=f"llama-3.2-3b-lora-r{LORA_R}",
    seed=42,
    ddp_find_unused_parameters=False,
)

# ============ TRAINER ============
trainer = SFTTrainer(
    model=model,
    args=training_args,
    train_dataset=dataset["train"],
    eval_dataset=dataset["test"],
    formatting_func=formatting_func,
    max_seq_length=MAX_LENGTH,
    tokenizer=tokenizer,
    packing=True,  # Efficient packing of sequences
)

# ============ TRAIN ============
trainer.train()

# Save final model
trainer.save_model()
tokenizer.save_pretrained(OUTPUT_DIR)

print(f"Model saved to {OUTPUT_DIR}")

Launch Training

# Single GPU
python train_lora.py

# Multi-GPU with Accelerate
accelerate config  # First time setup
accelerate launch train_lora.py

# With DeepSpeed ZeRO-3 (for larger models)
accelerate launch --config_file ds_zero3.yaml train_lora.py

QLoRA Specific Optimizations

Paged Optimizers

# Automatically used with optim="paged_adamw_32bit"
# Moves optimizer states to CPU when not in use
# Enables larger models on same VRAM

Gradient Checkpointing

# In TrainingArguments:
gradient_checkpointing=True,
gradient_checkpointing_kwargs={"use_reentrant": False},

# Saves ~30% VRAM at cost of ~20% slower training

Flash Attention 2

# Install
pip install flash-attn --no-build-isolation

# Use in model loading
attn_implementation="flash_attention_2"

# 2-3x faster, less memory

Monitoring Training

Weights & Biases

# In TrainingArguments
report_to="wandb",
run_name="my-lora-run",

# Before training
import wandb
wandb.init(project="llm-finetuning", name="llama-3b-lora-r64")

TensorBoard

# Start tensorboard
tensorboard --logdir ./llama-3.2-3b-lora-finetuned/runs

# In TrainingArguments
report_to="tensorboard",

Key Metrics to Watch

Metric	Healthy Range	Warning Signs
train_loss	Decreasing steadily	Plateauing early, diverging
eval_loss	Decreasing, close to train	Increasing (overfitting)
learning_rate	Cosine decay to 0	Stuck, not decaying
grad_norm	< 1.0	> 10 (exploding gradients)
GPU memory	Stable	Growing (memory leak)

Merging & Exporting

Merge LoRA Weights

# merge_lora.py
from peft import PeftModel
from transformers import AutoModelForCausalLM, AutoTokenizer
import torch

BASE_MODEL = "meta-llama/Llama-3.2-3B-Instruct"
LORA_PATH = "./llama-3.2-3b-lora-finetuned"
OUTPUT_PATH = "./llama-3.2-3b-merged"

# Load base model in fp16/bf16 (not 4-bit)
base_model = AutoModelForCausalLM.from_pretrained(
    BASE_MODEL,
    torch_dtype=torch.bfloat16,
    device_map="auto",
    trust_remote_code=True,
)

# Load LoRA adapter
model = PeftModel.from_pretrained(base_model, LORA_PATH)

# Merge and unload
merged_model = model.merge_and_unload()

# Save merged model
merged_model.save_pretrained(OUTPUT_PATH)
tokenizer = AutoTokenizer.from_pretrained(BASE_MODEL)
tokenizer.save_pretrained(OUTPUT_PATH)

print(f"Merged model saved to {OUTPUT_PATH}")

Export to GGUF (for llama.cpp/Ollama)

# Install llama.cpp with Python bindings
pip install llama-cpp-python

# Convert using llama.cpp convert script
python llama.cpp/convert_hf_to_gguf.py \
  ./llama-3.2-3b-merged \
  --outfile ./llama-3.2-3b-finetuned.Q4_K_M.gguf \
  --outtype q4_k_m

# Or use llama.cpp quantize directly
./llama.cpp/quantize ./llama-3.2-3b-merged/model.fp16.gguf \
  ./llama-3.2-3b-finetuned.Q4_K_M.gguf q4_k_m

Create Ollama Modelfile

# Modelfile
FROM ./llama-3.2-3b-finetuned.Q4_K_M.gguf

TEMPLATE "{{ .System }}\n{{ .Prompt }}\n{{ .Response }}"

PARAMETER temperature 0.7
PARAMETER top_p 0.9
PARAMETER num_ctx 8192

# Create Ollama model
ollama create my-finetuned-model -f Modelfile

# Test
ollama run my-finetuned-model "Write a Python function for binary search"

Advanced: LoRA Hyperparameter Tuning

Rank (r) Selection

# Experiment with different ranks
RANK_CONFIGS = [
    {"r": 8, "alpha": 16},    # Minimal, fast, less capacity
    {"r": 16, "alpha": 32},   # Good balance
    {"r": 32, "alpha": 64},   # More capacity
    {"r": 64, "alpha": 16},   # High capacity (alpha often = r/4 to r/2)
    {"r": 128, "alpha": 32},  # Maximum capacity
]

Target Module Ablation

# Test different target module sets
TARGET_CONFIGS = {
    "attention_only": ["q_proj", "k_proj", "v_proj", "o_proj"],
    "mlp_only": ["gate_proj", "up_proj", "down_proj"],
    "all_linear": ["q_proj", "k_proj", "v_proj", "o_proj", 
                   "gate_proj", "up_proj", "down_proj"],
    "attention_mlp": ["q_proj", "v_proj", "gate_proj", "up_proj"],  # Key ones
}

Learning Rate Schedule

# For LoRA, higher LR often works (2e-4 to 1e-3)
# But depends on dataset size:
# Small dataset (<1k): 1e-4 to 5e-4
# Medium (1k-10k): 2e-4
# Large (>10k): 1e-4 to 2e-4

Evaluation & Testing

Quantitative Evaluation

# eval_model.py
from transformers import AutoModelForCausalLM, AutoTokenizer
from peft import PeftModel
import torch
from datasets import load_dataset
import json

def evaluate_model(model_path, test_data_path, base_model_id=None):
    if base_model_id:
        base = AutoModelForCausalLM.from_pretrained(
            base_model_id, torch_dtype=torch.bfloat16, device_map="auto"
        )
        model = PeftModel.from_pretrained(base, model_path)
    else:
        model = AutoModelForCausalLM.from_pretrained(
            model_path, torch_dtype=torch.bfloat16, device_map="auto"
        )

    tokenizer = AutoTokenizer.from_pretrained(model_path)

    # Load test data
    with open(test_data_path) as f:
        test_data = json.load(f)

    results = []
    for item in test_data[:50]:  # Sample 50
        prompt = f"### Instruction:\n{item['instruction']}\n\n### Response:\n"
        inputs = tokenizer(prompt, return_tensors="pt").to("cuda")

        with torch.no_grad():
            outputs = model.generate(
                **inputs,
                max_new_tokens=256,
                temperature=0.7,
                do_sample=True,
                pad_token_id=tokenizer.eos_token_id
            )

        response = tokenizer.decode(outputs[0][inputs.input_ids.shape[1]:], skip_special_tokens=True)
        results.append({
            "instruction": item["instruction"],
            "expected": item["output"],
            "generated": response
        })

    # Save for human evaluation
    with open("eval_results.json", "w") as f:
        json.dump(results, f, indent=2)

    return results

Qualitative Evaluation (Human/LLM Judge)

# judge_eval.py
from openai import OpenAI
import json

client = OpenAI()

def llm_judge(instruction, expected, generated):
    prompt = f"""Rate the quality of the generated response compared to expected.

Instruction: {instruction}
Expected: {expected}
Generated: {generated}

Rate 1-5 on:
1. Accuracy
2. Completeness  
3. Clarity
4. Relevance

Return JSON: {{"accuracy": 1-5, "completeness": 1-5, "clarity": 1-5, "relevance": 1-5, "reasoning": "..."}}"""

    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
        temperature=0
    )
    return json.loads(response.content)

# Run evaluation
with open("eval_results.json") as f:
    results = json.load(f)

scores = {"accuracy": [], "completeness": [], "clarity": [], "relevance": []}
for r in results:
    eval = llm_judge(r["instruction"], r["expected"], r["generated"])
    for k in scores:
        scores[k].append(eval[k])

for k, v in scores.items():
    print(f"{k}: {sum(v)/len(v):.2f}")

Common Issues & Solutions

Issue	Cause	Solution
OOM	Model too large	Increase gradient_accumulation, reduce batch_size, use QLoRA, gradient_checkpointing
Loss not decreasing	LR too low/high	Try 2e-4, check data format, verify tokenization
Overfitting	Too many epochs, small data	Reduce epochs, add dropout, more data, weight decay
Gibberish output	Wrong chat template	Match training format exactly
NaN loss	Unstable training	Reduce LR, add gradient clipping, check data
Slow training	No Flash Attention	Install flash-attn, use bf16

Cost Analysis

Training Cost (Estimates)

Model	Method	GPU	Time (1k samples, 3 epochs)	Cost (Cloud)
3B	LoRA	RTX 3090 (24GB)	~30 min	$0.50
3B	QLoRA	RTX 3060 (12GB)	~45 min	$0.30
7B	LoRA	RTX 3090 (24GB)	~1 hr	$1.00
7B	QLoRA	RTX 3090 (24GB)	~1.5 hr	$0.80
13B	QLoRA	2× RTX 3090	~3 hr	$3.00
70B	QLoRA	4× A100 (80GB)	~8 hr	$40+

Local training on owned hardware = $0 marginal cost

Best Practices Summary

Data

[ ] Quality > Quantity: 1k high-quality > 100k noisy
[ ] Deduplicate: Remove exact/near duplicates
[ ] Balance: Cover all target domains evenly
[ ] Format consistency: Match training format exactly at inference

Training

[ ] Start small: Test with r=16, 1 epoch first
[ ] Monitor eval_loss: Stop when it plateaus/increases
[ ] Save checkpoints: Every 100-500 steps
[ ] Use cosine LR: With warmup (3-5% of steps)
[ ] Pack sequences: packing=True in SFTTrainer

LoRA Config

[ ] r=64, alpha=16 for most cases
[ ] Target all linear layers for best quality
[ ] Dropout 0.05-0.1 for regularization
[ ] Don't quantize LoRA weights (keep in bf16/fp16)

Deployment

[ ] Merge weights for inference speed
[ ] Quantize merged model to GGUF Q4_K_M
[ ] Test inference with same prompt format
[ ] Benchmark latency before production

Complete Example: Domain Adaptation

# Complete workflow for adapting LLaMA-3.2-3B to medical QA
# 1. Collect 2k high-quality medical QA pairs
# 2. Format as Alpaca
# 3. Train with QLoRA r=64, 3 epochs
# 4. Evaluate on held-out medical questions
# 5. Merge, quantize to GGUF, deploy via Ollama

# Expected: 90%+ accuracy on medical specialty vs 60% base model

Resources

Fine-tuning is no longer reserved for big tech. With QLoRA, a $300 GPU can customize a 7B model to beat GPT-3.5 on your specific task. Start small, iterate fast, deploy locally.

📦 Want the production-ready version?

This article covers the fundamentals. The LLM Fine-Tuning & Optimization Patterns pack goes further — LoRA/QLoRA recipes, dataset prep and evaluation, plus full source code and deployment configs.

→ Get LLM Fine-Tuning & Optimization Patterns

Extracted from real production systems, not toy examples.