⚙ AI Tools/2026-03-26Advanced

Building a RAG Pipeline with Antigravity— Unlock Your Company's Knowledge with Vector Search and LLMs

A comprehensive guide to designing and implementing RAG (Retrieval-Augmented Generation) pipelines using Antigravity. Covers embedding generation, ChromaDB integration, hybrid search with reranking, prompt optimization, and production best practices.

rag⁸ vector-search³ embedding llm³ antigravity⁴³⁰ ai-tools¹⁴ chromadb advanced²⁰

✦ Premium Article

Setup and context — What Is RAG and Why Does It Matter?

Large Language Models carry broad general knowledge, but they can't answer questions about your proprietary data — internal documentation, API specifications, knowledge bases, or customer interaction histories. Fine-tuning is expensive and requires retraining every time the data changes.

RAG (Retrieval-Augmented Generation) solves this problem elegantly. It retrieves documents relevant to a user's question via vector search, then feeds that information as context to the LLM, enabling accurate and up-to-date responses.

In this article, we'll build a complete RAG pipeline from scratch, leveraging Antigravity's AI assistance at every step. This guide is aimed at engineers with Python fundamentals who want to build LLM-powered applications over their own data. If you'd like a refresher on crafting effective prompts first, check out the Antigravity Prompt Engineering Advanced Guide — it'll make the prompt design sections of this tutorial feel much more intuitive.

RAG Architecture Overview

A RAG system consists of three major phases.

Ingestion Phase

Documents are loaded, split into appropriately-sized chunks, vectorized using an embedding model, and stored in a vector database.

# RAG ingestion pipeline generated with Antigravity
# document_ingestor.py
 
import os
from pathlib import Path
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.document_loaders import (
    DirectoryLoader,
    TextLoader,
    PyPDFLoader,
    UnstructuredMarkdownLoader,
)
from langchain_google_genai import GoogleGenerativeAIEmbeddings
import chromadb
from chromadb.config import Settings
 
# --- Configuration ---
CHROMA_PERSIST_DIR = "./chroma_db"
COLLECTION_NAME = "company_knowledge"
CHUNK_SIZE = 800       # Character count, not tokens
CHUNK_OVERLAP = 200    # Overlap between chunks
 
def load_documents(source_dir: str) -> list:
    """Load documents in multiple formats"""
    loaders = {
        "**/*.txt": TextLoader,
        "**/*.md": UnstructuredMarkdownLoader,
        "**/*.pdf": PyPDFLoader,
    }
    documents = []
    for glob_pattern, loader_cls in loaders.items():
        loader = DirectoryLoader(
            source_dir,
            glob=glob_pattern,
            loader_cls=loader_cls,
            show_progress=True,
        )
        documents.extend(loader.load())
    return documents
 
def chunk_documents(documents: list) -> list:
    """Split documents into semantically meaningful chunks"""
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=CHUNK_SIZE,
        chunk_overlap=CHUNK_OVERLAP,
        separators=["\n## ", "\n### ", "\n\n", "\n", ". ", " "],
        length_function=len,
    )
    return splitter.split_documents(documents)
 
def create_embeddings_and_store(chunks: list):
    """Generate embeddings and store in ChromaDB"""
    # Use Gemini's embedding model
    embeddings = GoogleGenerativeAIEmbeddings(
        model="models/text-embedding-004",
        google_api_key=os.getenv("GOOGLE_API_KEY"),
    )
 
    client = chromadb.PersistentClient(
        path=CHROMA_PERSIST_DIR,
        settings=Settings(anonymized_telemetry=False),
    )
 
    # Create collection (reset if exists)
    collection = client.get_or_create_collection(
        name=COLLECTION_NAME,
        metadata={"hnsw:space": "cosine"},  # Cosine similarity
    )
 
    # Batch processing for efficiency
    batch_size = 100
    for i in range(0, len(chunks), batch_size):
        batch = chunks[i:i + batch_size]
        texts = [chunk.page_content for chunk in batch]
        metadatas = [chunk.metadata for chunk in batch]
        ids = [f"doc_{i + j}" for j in range(len(batch))]
 
        # Generate embeddings
        vectors = embeddings.embed_documents(texts)
 
        collection.add(
            ids=ids,
            documents=texts,
            embeddings=vectors,
            metadatas=metadatas,
        )
        print(f"  Stored: {i + len(batch)}/{len(chunks)} chunks")
 
    print(f"✅ All {len(chunks)} chunks stored in ChromaDB")
 
# --- Execution ---
if __name__ == "__main__":
    docs = load_documents("./knowledge_base")
    print(f"📄 Loaded {len(docs)} documents")
 
    chunks = chunk_documents(docs)
    print(f"✂️  Split into {len(chunks)} chunks")
 
    create_embeddings_and_store(chunks)

Expected output:

📄 Loaded 47 documents
✂️  Split into 312 chunks
  Stored: 100/312 chunks
  Stored: 200/312 chunks
  Stored: 300/312 chunks
  Stored: 312/312 chunks
✅ All 312 chunks stored in ChromaDB

Retrieval Phase

The user's question is vectorized, and the most similar chunks are retrieved. Retrieval accuracy is the single biggest factor in overall RAG quality.

Generation Phase

Retrieved results are injected into a prompt, and the LLM generates a grounded answer.

✦

Thank you for reading this far.

Continue Reading

What follows includes implementation code, benchmarks, and practical content we hope you'll find useful. This site runs without ads — server and development costs are supported entirely by members like you. If it's been helpful, we'd be truly grateful for your support.

WHAT YOU'LL LEARN

✦Understand the design principles and key components of RAG architecture systematically

✦Build a vector search pipeline in Antigravity that instantly incorporates your own data into LLM responses

✦Master chunking strategies, reranking techniques, and evaluation metrics essential for production RAG systems

Secure payment via Stripe · Cancel anytime

✦

Unlock This Article

Get full access to the rest of this article. Buy once, read anytime. This site is ad-free — your support goes directly toward keeping it running.

Unlock all articles with Membership →

Chunking Strategy — 80% of RAG Quality Is Decided Here

The single most impactful decision in a RAG pipeline is how you split your documents into chunks. When you ask Antigravity's agent to "optimize the chunking strategy," it can propose sophisticated splitting logic like the following.

Semantic Chunking

Rather than splitting at fixed character counts, semantic chunking preserves meaningful boundaries in the text.

# semantic_chunker.py — Advanced chunking that preserves semantic coherence
 
from langchain_experimental.text_splitter import SemanticChunker
from langchain_google_genai import GoogleGenerativeAIEmbeddings
import os
 
def create_semantic_chunks(documents: list) -> list:
    """Embedding-based semantic chunking"""
    embeddings = GoogleGenerativeAIEmbeddings(
        model="models/text-embedding-004",
        google_api_key=os.getenv("GOOGLE_API_KEY"),
    )
 
    # Split where semantic distance between sentences exceeds threshold
    chunker = SemanticChunker(
        embeddings,
        breakpoint_threshold_type="percentile",
        breakpoint_threshold_amount=90,  # Split at top 10% distance
    )
 
    chunks = []
    for doc in documents:
        split = chunker.create_documents(
            [doc.page_content],
            metadatas=[doc.metadata],
        )
        chunks.extend(split)
 
    return chunks
 
# --- Parent-Child Chunking Strategy ---
class ParentChildChunker:
    """
    Use large 'parent chunks' to preserve context and
    small 'child chunks' for precise retrieval — the best of both worlds.
    """
 
    def __init__(self, parent_size=2000, child_size=400, overlap=100):
        self.parent_splitter = RecursiveCharacterTextSplitter(
            chunk_size=parent_size,
            chunk_overlap=0,
        )
        self.child_splitter = RecursiveCharacterTextSplitter(
            chunk_size=child_size,
            chunk_overlap=overlap,
        )
 
    def split(self, documents: list) -> tuple:
        """Generate parent and child chunks simultaneously"""
        parent_chunks = []
        child_chunks = []
 
        for doc in documents:
            parents = self.parent_splitter.split_documents([doc])
            for idx, parent in enumerate(parents):
                parent.metadata["parent_id"] = f"{doc.metadata.get('source', 'unknown')}_{idx}"
                parent_chunks.append(parent)
 
                # Split parent into smaller child chunks
                children = self.child_splitter.split_documents([parent])
                for child in children:
                    child.metadata["parent_id"] = parent.metadata["parent_id"]
                    child_chunks.append(child)
 
        return parent_chunks, child_chunks
 
# Expected behavior:
# - Search is performed against child chunks for precision
# - The parent chunk of each hit is passed as context to the LLM
# → Achieves both retrieval precision and rich context

Choosing the Right Chunk Size

There's no one-size-fits-all chunk size — the optimal value depends on the nature of your data.

The best approach is to have Antigravity's agent generate benchmark code for each pattern and compare against your actual data. As a general guideline, technical documentation (API references, etc.) works well with smaller chunks of 400–600 characters, since code context tends to be self-contained within short spans. Narrative documents (internal wikis, meeting notes, etc.) benefit from slightly larger chunks of 800–1,200 characters, as shorter chunks risk losing meaning in longer-form text. Legal and contract documents requiring strict precision call for large chunks of 1,500–2,000 characters to preserve the full context of each clause.

Advanced Retrieval — Hybrid Search and Reranking

Vector search alone falls short in certain scenarios. When exact keyword matching matters — searching for error codes, identifying product model numbers — combining vector search with keyword-based BM25 in a hybrid search approach delivers significantly better results.

# hybrid_retriever.py — Hybrid search combining vector search and BM25
 
from langchain.retrievers import EnsembleRetriever
from langchain_community.retrievers import BM25Retriever
from langchain_community.vectorstores import Chroma
from langchain_google_genai import GoogleGenerativeAIEmbeddings
import os
 
def create_hybrid_retriever(chunks: list, k: int = 5):
    """Hybrid retriever combining vector search and BM25"""
 
    embeddings = GoogleGenerativeAIEmbeddings(
        model="models/text-embedding-004",
        google_api_key=os.getenv("GOOGLE_API_KEY"),
    )
 
    # 1. Vector search retriever
    vectorstore = Chroma.from_documents(
        documents=chunks,
        embedding=embeddings,
        collection_name="hybrid_search",
    )
    vector_retriever = vectorstore.as_retriever(
        search_type="mmr",          # Maximal Marginal Relevance
        search_kwargs={"k": k, "fetch_k": k * 3},
    )
 
    # 2. BM25 keyword search retriever
    bm25_retriever = BM25Retriever.from_documents(
        chunks,
        k=k,
    )
 
    # 3. Ensemble (weighted combination)
    ensemble = EnsembleRetriever(
        retrievers=[vector_retriever, bm25_retriever],
        weights=[0.6, 0.4],  # Slightly favor vector search
    )
 
    return ensemble
 
# --- Reranking with Cohere Reranker ---
from langchain.retrievers import ContextualCompressionRetriever
from langchain_cohere import CohereRerank
 
def add_reranking(base_retriever, top_n: int = 3):
    """Re-score search results with a reranker"""
    reranker = CohereRerank(
        model="rerank-v3.5",
        cohere_api_key=os.getenv("COHERE_API_KEY"),
        top_n=top_n,
    )
 
    return ContextualCompressionRetriever(
        base_compressor=reranker,
        base_retriever=base_retriever,
    )
 
# Usage example:
# retriever = create_hybrid_retriever(chunks, k=10)
# reranked_retriever = add_reranking(retriever, top_n=3)
# results = reranked_retriever.invoke("How do I verify Stripe webhook signatures?")
#
# Expected behavior:
# → BM25 picks up candidates matching "Stripe", "webhook", "signatures"
# → Vector search also finds semantically similar documents
# → Reranker narrows down to the top 3 most relevant documents

MMR (Maximal Marginal Relevance) for Result Diversity

Setting search_type="mmr" in the vector retriever considers both similarity and diversity in results. This prevents the retriever from returning near-duplicate chunks and provides the LLM with a more comprehensive view of the topic.

Generation Phase — Prompt Design and Guardrails

How you structure the prompt with retrieved context is the key to answer quality.

# rag_chain.py — Building the RAG chain
 
from langchain_google_genai import ChatGoogleGenerativeAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough
 
def build_rag_chain(retriever):
    """Build a RAG chain that generates answers from retrieved context"""
 
    # --- Prompt template ---
    template = ChatPromptTemplate.from_messages([
        ("system", """You are an assistant that answers questions accurately based on internal knowledge.
 
Follow these rules strictly:
1. Base your answer ONLY on the provided context
2. If the context doesn't contain the answer, explicitly state "This information was not found in the knowledge base"
3. Never mix in speculation or external knowledge
4. Cite the source document for each claim
5. Include code examples for technical content
 
Context:
{context}"""),
        ("human", "{question}"),
    ])
 
    # --- LLM ---
    llm = ChatGoogleGenerativeAI(
        model="gemini-2.5-pro",
        temperature=0.1,        # Low temperature for factual responses
        max_output_tokens=4096,
    )
 
    # --- Chain construction ---
    def format_docs(docs):
        """Format search results into a context string"""
        formatted = []
        for i, doc in enumerate(docs, 1):
            source = doc.metadata.get("source", "unknown")
            formatted.append(
                f"[Source {i}: {source}]\n{doc.page_content}"
            )
        return "\n\n---\n\n".join(formatted)
 
    chain = (
        {
            "context": retriever | format_docs,
            "question": RunnablePassthrough(),
        }
        | template
        | llm
        | StrOutputParser()
    )
 
    return chain
 
# --- Streaming version ---
def query_with_streaming(chain, question: str):
    """Display answer with streaming"""
    print(f"\n📝 Question: {question}\n")
    print("=" * 60)
    for chunk in chain.stream(question):
        print(chunk, end="", flush=True)
    print("\n" + "=" * 60)
 
# Usage example:
# chain = build_rag_chain(reranked_retriever)
# query_with_streaming(chain, "What's the setup procedure for a new project?")
#
# Expected output:
# 📝 Question: What's the setup procedure for a new project?
# ============================================================
# Based on our internal knowledge base, new project setup follows
# these steps:
# 1. Create a repository on GitHub...
# [Source: docs/setup-guide.md]
# ============================================================

Evaluation and Monitoring — Quantifying RAG Quality

A RAG system that merely "works" isn't good enough. You need a way to measure answer quality quantitatively and improve it continuously.

# rag_evaluator.py — RAG pipeline evaluation framework
 
from dataclasses import dataclass
from langchain_google_genai import ChatGoogleGenerativeAI
import json
 
@dataclass
class EvalResult:
    question: str
    expected: str
    actual: str
    faithfulness: float    # Is it faithful to the context? (0-1)
    relevance: float       # Is it relevant to the question? (0-1)
    completeness: float    # Is the answer thorough enough? (0-1)
 
class RAGEvaluator:
    """Evaluate RAG answers using LLM-as-a-Judge pattern"""
 
    def __init__(self):
        self.judge = ChatGoogleGenerativeAI(
            model="gemini-2.5-pro",
            temperature=0,
        )
 
    def evaluate_faithfulness(
        self, context: str, answer: str
    ) -> float:
        """Evaluate whether the answer is faithful to the context"""
        prompt = f"""Evaluate whether the following answer is based solely on the provided context.
 
Context:
{context}
 
Answer:
{answer}
 
Return a score from 0.0 (completely unsupported) to 1.0 (fully faithful)
in JSON format: {{"score": 0.X, "reason": "explanation"}}"""
 
        response = self.judge.invoke(prompt)
        result = json.loads(response.content)
        return result["score"]
 
    def evaluate_relevance(
        self, question: str, answer: str
    ) -> float:
        """Evaluate whether the answer addresses the question"""
        prompt = f"""Evaluate whether the following answer appropriately addresses the question.
 
Question: {question}
Answer: {answer}
 
Return a score from 0.0 (irrelevant) to 1.0 (perfectly on point)
in JSON format: {{"score": 0.X, "reason": "explanation"}}"""
 
        response = self.judge.invoke(prompt)
        result = json.loads(response.content)
        return result["score"]
 
    def run_eval_suite(
        self, chain, retriever, test_cases: list[dict]
    ) -> list[EvalResult]:
        """Run evaluation across a test case suite"""
        results = []
        for case in test_cases:
            # Retrieve relevant documents
            docs = retriever.invoke(case["question"])
            context = "\n".join(d.page_content for d in docs)
 
            # Generate answer
            answer = chain.invoke(case["question"])
 
            # Evaluate
            faithfulness = self.evaluate_faithfulness(context, answer)
            relevance = self.evaluate_relevance(case["question"], answer)
 
            results.append(EvalResult(
                question=case["question"],
                expected=case.get("expected", ""),
                actual=answer,
                faithfulness=faithfulness,
                relevance=relevance,
                completeness=0.0,  # Evaluate separately
            ))
 
        # Print summary
        avg_faith = sum(r.faithfulness for r in results) / len(results)
        avg_rel = sum(r.relevance for r in results) / len(results)
        print(f"\n📊 Evaluation Summary ({len(results)} cases)")
        print(f"  Faithfulness: {avg_faith:.2f}")
        print(f"  Relevance:    {avg_rel:.2f}")
 
        return results
 
# Test case example:
# test_cases = [
#     {"question": "What's the deploy procedure?", "expected": "Run wrangler deploy..."},
#     {"question": "How do I generate API keys?", "expected": "From the admin panel..."},
# ]
# evaluator = RAGEvaluator()
# results = evaluator.run_eval_suite(chain, retriever, test_cases)

Production Best Practices

Here are the key considerations for running a RAG system reliably in production.

Incremental ingestion: When documents are updated, only re-embed the changed files. Record file hashes and selectively re-process modified documents — this dramatically reduces both cost and processing time.

Caching strategy: Cache answers for repeated questions to reduce latency and API costs. A "semantic cache" that considers queries with vector similarity above a threshold (e.g., 0.95) as cache hits is particularly effective.

Fallback design: When vector search returns low-scoring results (e.g., cosine similarity below 0.3), responding with "No relevant information was found" is far more trustworthy than generating a hallucinated answer.

Monitoring metrics: In production, continuously monitor these indicators: search latency (target P95 under 500ms), answer generation latency (target P95 under 3 seconds), average similarity score of search results (a declining trend signals stale data), and user feedback ratio (thumbs up/down rates).

A Note from an Indie Developer

Key Takeaways

RAG is the most practical approach to making LLMs smart about your own data. In this article, we built out the entire pipeline — ingestion, retrieval, generation, and evaluation — with Antigravity's AI assistance.

If you're interested in working with multiple LLMs, the Antigravity Multi-Model Mastery Guide covers optimal strategies for combining Gemini, Claude, and GPT. For those looking into model optimization, our AI Model Quantization Guide is also worth a read.

The key takeaways are: chunking strategy determines 80% of RAG quality, so consider semantic chunking and parent-child chunking strategies carefully; hybrid search (vector + BM25) combined with reranking dramatically improves retrieval accuracy; and the LLM-as-a-Judge pattern lets you quantitatively measure answer quality for continuous improvement.

Thank You for Reading

Antigravity Lab is ad-free, supported entirely by members like you. We publish practical guides daily with implementation code, benchmarks, and production-ready patterns. If you've found it useful, we'd love to have you on board.