Halving LLM Costs: Building a Production-Ready RAG Pipeline with Serverless & pgvector

The Problem: Expensive, Stale, and Slow LLM Responses

In the rapidly evolving landscape of AI, Large Language Models (LLMs) have become indispensable for building intelligent applications. From chatbots to content generation, their capabilities are transformative. However, integrating LLMs into production environments presents significant challenges: prohibitive inference costs, the inability to access real-time or proprietary domain-specific information, and latency issues that degrade user experience. Relying solely on a base LLM means constant fine-tuning for new data, which is expensive and time-consuming, or accepting outdated and generalized responses.

Imagine a customer support chatbot that frequently hallucinates or provides generic answers because it lacks access to your company's latest product documentation. Or a legal research assistant that incurs massive API costs with every query because it processes entire documents for context. These issues lead to higher operational expenses, diminished user trust, and a slow, inefficient development cycle.

The core problem is simple: LLMs are powerful but costly knowledge black boxes that are often unaware of your specific, evolving data. We need a way to inject precise, up-to-date context into LLM prompts efficiently and affordably.

The Solution Concept: Retrieval Augmented Generation (RAG) with Serverless Efficiency

Retrieval Augmented Generation (RAG) offers an elegant solution to these challenges. RAG enhances LLMs by retrieving relevant information from an external knowledge base and feeding it as context into the LLM's prompt. This approach drastically reduces hallucination, grounds the LLM in specific data, and eliminates the need for expensive, frequent fine-tuning. But how do we make RAG itself cost-effective and scalable for production?

Our solution combines RAG with a serverless architecture and a powerful, yet affordable, vector database: PostgreSQL with the pgvector extension. Serverless functions (like AWS Lambda, Google Cloud Functions, or Azure Functions) provide a pay-per-execution model, automatically scaling to demand and drastically cutting idle costs. pgvector, on the other hand, allows us to store and query high-dimensional embeddings directly within our trusted relational database, avoiding the overhead and potential complexity of dedicated vector database services for many use cases.

Architectural Overview:

Our RAG pipeline consists of two main parts:

1. Ingestion Pipeline:
• Data Sources: Your proprietary documents (PDFs, Markdown, databases, web pages).
• Chunking: Splitting documents into smaller, semantically meaningful chunks.
• Embedding Generation: Converting text chunks into numerical vector embeddings using an embedding model (e.g., OpenAI's text-embedding-3-small).
• Storage: Storing these embeddings and their corresponding text chunks in a PostgreSQL database with pgvector.
2. Query Pipeline:
• User Query: The natural language question from the user.
• Serverless Function: An API endpoint (e.g., AWS Lambda) that receives the query.
• Query Embedding: The user's query is converted into a vector embedding.
• Vector Search: The serverless function queries pgvector for the top-k most similar text chunks.
• Context Construction: The retrieved text chunks are combined with the user's original query into a detailed prompt.
• LLM Call: The crafted prompt is sent to an LLM (e.g., OpenAI GPT-4, Anthropic Claude).
• Response: The LLM's generated answer is returned to the user.

Step-by-Step Implementation: Building a Documentation Chatbot

Let's build a practical example: a chatbot that answers questions based on a collection of Markdown documentation files. We'll use Python for both ingestion and the serverless query function for consistency.

Prerequisites:

• A PostgreSQL database with the pgvector extension enabled. For cloud-managed options, Supabase or Aiven are great choices as they offer pgvector out-of-the-box.
• Python 3.8+
• pip install openai psycopg2-binary tiktoken python-dotenv

1. Database Setup

First, connect to your PostgreSQL database and create a table to store your document chunks and their embeddings. Replace YOUR_DB_CONNECTION_STRING with your actual connection string.

CREATE EXTENSION IF NOT EXISTS vector;

CREATE TABLE document_chunks (
    id BIGSERIAL PRIMARY KEY,
    content TEXT NOT NULL,
    embedding VECTOR(1536) NOT NULL
);

2. Data Ingestion Pipeline (Python)

This script will load your documentation files, chunk them, generate embeddings, and store them in the document_chunks table.

import os
import psycopg2
from dotenv import load_dotenv
from openai import OpenAI
import tiktoken

load_dotenv()

# Configuration
DB_CONNECTION_STRING = os.getenv("DB_CONNECTION_STRING")
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
DOCS_DIR = "./documentation"
EMBEDDING_MODEL = "text-embedding-3-small" # Cost-effective embedding model
EMBEDDING_DIM = 1536 # Dimension for text-embedding-3-small
CHUNK_SIZE = 1000
CHUNK_OVERLAP = 100

client = OpenAI(api_key=OPENAI_API_KEY)

def get_embedding(text: str) -> list[float]:
    response = client.embeddings.create(input=text, model=EMBEDDING_MODEL)
    return response.data[0].embedding

def count_tokens(text: str) -> int:
    enc = tiktoken.encoding_for_model("gpt-4") # Use a common tokenizer
    return len(enc.encode(text))

def chunk_text(text: str, chunk_size: int, chunk_overlap: int) -> list[str]:
    # A simple recursive character text splitter. For production, consider Langchain's TextSplitter.
    if not text: return []

    chunks = []
    tokens = count_tokens(text)

    if tokens <= chunk_size: return [text]

    # Find a split point near the middle to prevent cutting sentences awkwardly
    split_index = len(text) // 2
    left_chunk = text[:split_index + chunk_overlap // 2]
    right_chunk = text[split_index - chunk_overlap // 2:]

    if count_tokens(left_chunk) > chunk_size:
        chunks.extend(chunk_text(left_chunk, chunk_size, chunk_overlap))
    else:
        chunks.append(left_chunk)

    if count_tokens(right_chunk) > chunk_size:
        chunks.extend(chunk_text(right_chunk, chunk_size, chunk_overlap))
    elif count_tokens(right_chunk) > 0: # Ensure we don't add empty chunks
        chunks.append(right_chunk)

    # Simple fall back for very long texts or complex structures
    if not chunks or any(count_tokens(c) > chunk_size for c in chunks): 
        # Fallback to character splitting if token splitting gets complicated
        # A more robust solution would be needed for truly complex documents
        words = text.split()
        current_chunk = []
        for word in words:
            current_chunk.append(word)
            if count_tokens(" ".join(current_chunk)) > chunk_size:
                chunks.append(" ".join(current_chunk[:-1]))
                current_chunk = [current_chunk[-1]]
        if current_chunk: chunks.append(" ".join(current_chunk))

    return list(set(chunks)) # Remove potential duplicates from simple splitting

def ingest_documents():
    conn = None
    try:
        conn = psycopg2.connect(DB_CONNECTION_STRING)
        cur = conn.cursor()

        for filename in os.listdir(DOCS_DIR):
            if filename.endswith(".md"):
                filepath = os.path.join(DOCS_DIR, filename)
                with open(filepath, "r", encoding="utf-8") as f:
                    doc_content = f.read()

                print(f"Processing {filename}...")
                chunks = chunk_text(doc_content, CHUNK_SIZE, CHUNK_OVERLAP)
                
                for i, chunk in enumerate(chunks):
                    if not chunk.strip(): continue
                    try:
                        embedding = get_embedding(chunk)
                        cur.execute(
                            "INSERT INTO document_chunks (content, embedding) VALUES (%s, %s)",
                            (chunk, embedding)
                        )
                        print(f"  Chunk {i+1}/{len(chunks)} embedded and stored.")
                    except Exception as e:
                        print(f"Error processing chunk {i+1} of {filename}: {e}")

        conn.commit()
        print("Ingestion complete.")

    except Exception as e:
        print(f"Database or ingestion error: {e}")
    finally:
        if conn: cur.close(); conn.close()

if __name__ == "__main__":
    # Create a dummy documentation directory and file for testing
    os.makedirs(DOCS_DIR, exist_ok=True)
    with open(os.path.join(DOCS_DIR, "example_product.md"), "w", encoding="utf-8") as f:
        f.write("# Product Overview\n\nOur revolutionary product leverages cutting-edge AI for enhanced user experience. It features real-time data processing capabilities and a modular architecture. This allows for flexible deployments across various cloud providers. Security is a top priority, with end-to-end encryption and compliance with industry standards.\n\n## Key Features\n\n*   **AI-Powered Analytics:** Gain insights with advanced machine learning models.\n*   **Scalable Infrastructure:** Built on serverless technologies for limitless scalability.\n*   **Secure Data Handling:** Industry-leading encryption and data privacy controls.\n\n## Getting Started\n\nRefer to the installation guide for setting up your first project. Documentation is continuously updated with new features and best practices. For support, visit our community forums or contact our premium support team. We're committed to providing the best tools for your business needs.")

    ingest_documents()

3. Serverless Query Function (Python)

This script represents the core logic for your serverless function. It takes a query, finds relevant chunks, and passes them to an LLM.

import os
import json
import psycopg2
from dotenv import load_dotenv
from openai import OpenAI
import numpy as np

load_dotenv()

# Configuration (should match ingestion script)
DB_CONNECTION_STRING = os.getenv("DB_CONNECTION_STRING")
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
EMBEDDING_MODEL = "text-embedding-3-small"
LLM_MODEL = "gpt-3.5-turbo" # Cost-effective LLM for responses

client = OpenAI(api_key=OPENAI_API_KEY)

def get_embedding(text: str) -> list[float]:
    response = client.embeddings.create(input=text, model=EMBEDDING_MODEL)
    return response.data[0].embedding

def query_rag(user_query: str) -> str:
    conn = None
    try:
        conn = psycopg2.connect(DB_CONNECTION_STRING)
        cur = conn.cursor()

        # 1. Embed the user query
        query_vector = get_embedding(user_query)

        # 2. Perform similarity search in pgvector
        # Use '<->' operator for L2 distance (Euclidean distance) for vector search
        cur.execute(
            "SELECT content FROM document_chunks ORDER BY embedding <-> %s LIMIT 5",
            (np.array(query_vector),)
        )
        retrieved_chunks = [row[0] for row in cur.fetchall()]

        # 3. Construct the prompt with retrieved context
        context = "\n\n".join(retrieved_chunks)
        if not context:
            return "I couldn't find relevant information in the knowledge base. Please try rephrasing your question."
        
        system_prompt = "You are a helpful assistant providing information based on the provided context. If the answer is not in the context, state that you don't know."
        full_prompt = f"{system_prompt}\n\nContext:\n{context}\n\nQuestion: {user_query}\n\nAnswer:"

        # 4. Call the LLM
        chat_completion = client.chat.completions.create(
            model=LLM_MODEL,
            messages=[
                {"role": "system", "content": system_prompt},
                {"role": "user", "content": f"Context:\n{context}\n\nQuestion: {user_query}"}
            ],
            temperature=0.7 # Adjust for creativity vs. factual accuracy
        )

        return chat_completion.choices[0].message.content

    except Exception as e:
        print(f"Error during RAG query: {e}")
        return f"An error occurred while processing your request: {e}"
    finally:
        if conn: cur.close(); conn.close()

# Example of how this might be wrapped in a serverless function handler
def lambda_handler(event, context):
    try:
        body = json.loads(event['body'])
        user_query = body.get('query')

        if not user_query:
            return {
                'statusCode': 400,
                'body': json.dumps({'error': 'Query parameter is missing'})
            }
        
        response_text = query_rag(user_query)

        return {
            'statusCode': 200,
            'body': json.dumps({'answer': response_text})
        }
    except Exception as e:
        print(f"Lambda handler error: {e}")
        return {
            'statusCode': 500,
            'body': json.dumps({'error': 'Internal server error'})
        }

if __name__ == '__main__':
    # Test the query function locally
    sample_query = "What are the key features of the product?"
    print(f"\nQuery: {sample_query}")
    answer = query_rag(sample_query)
    print(f"Answer: {answer}")

    sample_query_2 = "How do I integrate with the product?"
    print(f"\nQuery: {sample_query_2}")
    answer_2 = query_rag(sample_query_2)
    print(f"Answer: {answer_2}")

    sample_query_3 = "What is the capital of France?" # Outside context
    print(f"\nQuery: {sample_query_3}")
    answer_3 = query_rag(sample_query_3)
    print(f"Answer: {answer_3}")

Optimization & Best Practices

1. Smart Chunking Strategies: The simple character splitter works for basic cases. For complex documents (code, tables, nested headings), consider more advanced strategies like semantic chunking (splitting based on meaning), document-aware chunking (respecting headings, paragraphs), or using libraries like Langchain's RecursiveCharacterTextSplitter which attempts to keep semantically related parts together. Experiment with CHUNK_SIZE and CHUNK_OVERLAP.
2. Embedding Model Selection: text-embedding-3-small is an excellent balance of cost and performance. For higher accuracy, text-embedding-3-large might be considered, but it's more expensive. For extreme cost-saving or privacy, self-hosting open-source models (e.g., from HuggingFace Transformers) could be an option, though it introduces infrastructure overhead.
3. Vector Database Indexing: For larger datasets in pgvector, creating appropriate indexes is crucial for performance. The IVFFlat index is a common choice for approximate nearest neighbor search. Example: CREATE INDEX ON document_chunks USING ivfflat (embedding vector_l2_ops) WITH (lists = 100); Adjust lists based on your dataset size and desired recall/speed trade-off.
4. Prompt Engineering: Craft clear, concise system prompts. Explicitly tell the LLM to use only the provided context and to state if it cannot find the answer. Experiment with temperature settings (0.0-1.0) to control creativity.
5. Caching: Cache embedding results for frequently asked questions or common query patterns. For static knowledge bases, cache LLM responses if the context is identical.
6. Error Handling & Retry Mechanisms: Implement robust error handling and retry logic for external API calls (LLMs, embedding services) to enhance reliability.
7. Monitoring & Observability: Monitor serverless function invocations, execution times, and LLM API costs. Set up alerts for anomalies.

Business Impact & ROI

Implementing a RAG system with serverless and pgvector delivers tangible business value and a strong return on investment:

• Significant Cost Reduction:
• LLM Token Usage: By providing precise context, you drastically reduce the number of tokens the LLM needs to process and generate, often leading to 30-60% savings on LLM API calls compared to sending large, unoptimized prompts.
• Infrastructure Costs: Serverless functions mean you only pay for compute when your RAG system is actively processing queries. This eliminates idle server costs, translating to 50-80% infrastructure savings for sporadic or variable workloads. pgvector leverages your existing PostgreSQL instance, avoiding additional dedicated vector database subscription fees for many applications.
• Improved Accuracy & Relevance: By grounding the LLM in your specific, up-to-date data, hallucination is minimized, and responses become highly accurate and relevant to your domain. This boosts user trust and satisfaction.
• Enhanced User Experience: Faster, more accurate responses lead to better user engagement, reduced bounce rates, and a more seamless interactive experience with your AI-powered applications.
• Faster Iteration & Maintenance: Updating your knowledge base is as simple as re-ingesting documents – no expensive, time-consuming LLM fine-tuning required. This accelerates product development and allows for rapid adaptation to new information.
• Scalability: The serverless architecture ensures that your RAG system automatically scales to handle fluctuating loads, from a few queries per day to thousands per second, without manual intervention.

By making your LLM applications smarter and more affordable, you empower your business to leverage AI capabilities more broadly and competitively.

Conclusion

The marriage of Retrieval Augmented Generation with serverless architecture and PostgreSQL's pgvector extension offers a powerful, cost-effective, and scalable solution for building intelligent applications. This approach addresses the critical challenges of high LLM costs, data freshness, and accuracy, enabling developers to build robust AI features that deliver real business value. By carefully architecting your RAG pipelines with these tools, you can unlock the full potential of LLMs, providing users with accurate, context-rich responses while keeping your operational expenditures firmly in check.