Vector RAG Skill

Role

You are an elite RAG architect. You build production-grade retrieval systems that

ground LLM responses in verified, private data. You know every chunking strategy,

every vector store, every retrieval pattern, and every re-ranking technique used

by the world's best AI systems — Perplexity, Claude, Gemini, and beyond.

Why RAG Matters

RAG is the single most important technique for production LLMs. It solves:

• Hallucination (LLM invents facts) → ground in real data
• Staleness (training cutoff) → add live retrieval
• Privacy (can't fine-tune on client data) → query at runtime
• Scale (can't fit all data in context) → retrieve the relevant slice

Part 1: The RAG Stack

User Query
   ↓ embed
Query Vector
   ↓ similarity search
Top-K Chunks (retrieved context)
   ↓ rerank
Reranked Chunks
   ↓ build prompt
[SYSTEM + CONTEXT + QUERY]
   ↓ LLM
Answer (grounded in retrieved data)

Part 2: Document Chunking

Chunking strategy is the #1 determinant of RAG quality. Wrong chunks = bad retrieval.

from typing import Generator

# Strategy 1: Fixed-size chunking (baseline)
def chunk_fixed(text: str, chunk_size: int = 512,
                 overlap: int = 64) -> list[dict]:
    """Split text into fixed-size chunks with overlap."""
    chunks = []
    start = 0
    text_len = len(text)

    while start < text_len:
        end = min(start + chunk_size, text_len)
        chunk = text[start:end]
        chunks.append({
            "text": chunk,
            "start": start,
            "end": end,
            "chunk_id": len(chunks)
        })
        start += chunk_size - overlap  # Slide window with overlap

    return chunks


# Strategy 2: Semantic chunking (production-preferred)
def chunk_by_sentences(text: str, sentences_per_chunk: int = 5,
                        overlap_sentences: int = 1) -> list[dict]:
    """Chunk by sentence boundaries — much better for retrieval."""
    import re
    # Split on sentence boundaries
    sentences = re.split(r'(?<=[.!?])\s+', text.strip())
    chunks = []

    for i in range(0, len(sentences), sentences_per_chunk - overlap_sentences):
        window = sentences[i:i + sentences_per_chunk]
        if window:
            chunks.append({
                "text": " ".join(window),
                "sentence_start": i,
                "sentence_end": i + len(window),
                "chunk_id": len(chunks)
            })

    return chunks


# Strategy 3: Recursive character splitting (LangChain-style, best for mixed content)
def chunk_recursive(text: str, chunk_size: int = 500,
                     overlap: int = 50,
                     separators: list[str] = None) -> list[str]:
    """Recursively split on decreasing separator hierarchy."""
    if separators is None:
        separators = ["\n\n", "\n", ". ", " ", ""]

    def _split(text: str, seps: list[str]) -> list[str]:
        if not seps:
            return [text[i:i+chunk_size]
                    for i in range(0, len(text), chunk_size - overlap)]

        sep = seps[0]
        splits = text.split(sep) if sep else list(text)

        chunks = []
        current = ""
        for split in splits:
            if len(current) + len(split) + len(sep) <= chunk_size:
                current += split + sep
            else:
                if current:
                    chunks.append(current.strip())
                if len(split) > chunk_size:
                    chunks.extend(_split(split, seps[1:]))
                    current = ""
                else:
                    current = split + sep

        if current:
            chunks.append(current.strip())
        return chunks

    return _split(text, separators)


# Strategy 4: Hierarchical chunking (used by production systems)
def chunk_hierarchical(text: str) -> dict:
    """Create parent + child chunks for multi-level retrieval."""
    import re

    # Level 1: paragraphs (for context)
    paragraphs = [p.strip() for p in text.split("\n\n") if p.strip()]

    # Level 2: sentences within paragraphs (for precision)
    result = []
    for para_idx, para in enumerate(paragraphs):
        sentences = re.split(r'(?<=[.!?])\s+', para)
        for sent_idx, sentence in enumerate(sentences):
            result.append({
                "text": sentence,
                "parent_text": para,
                "para_idx": para_idx,
                "sent_idx": sent_idx,
                "hierarchy": "child"
            })

    return result

Part 3: ChromaDB (Local, Free, Production-Ready)

Used by: small-to-medium RAG systems, local deployments

import chromadb
from chromadb.config import Settings

def create_chroma_client(persist_dir: str = "./chroma_db") -> chromadb.Client:
    """Create a persistent ChromaDB client."""
    return chromadb.PersistentClient(
        path=persist_dir,
        settings=Settings(anonymized_telemetry=False)
    )


class ChromaRAG:
    """Production ChromaDB RAG system."""

    def __init__(self, collection_name: str,
                  embed_fn: callable,
                  persist_dir: str = "./chroma_db"):
        self.client = create_chroma_client(persist_dir)
        self.embed_fn = embed_fn
        self.collection = self.client.get_or_create_collection(
            name=collection_name,
            metadata={"hnsw:space": "cosine"}
        )

    def add_documents(self, docs: list[dict],
                       id_field: str = "id",
                       text_field: str = "text") -> None:
        """Add documents to the vector store."""
        texts = [d[text_field] for d in docs]
        ids = [str(d.get(id_field, i)) for i, d in enumerate(docs)]
        metadatas = [{k: v for k, v in d.items()
                      if k not in (text_field, id_field)} for d in docs]

        embeddings = self.embed_fn(texts)

        self.collection.upsert(
            ids=ids,
            embeddings=embeddings,
            documents=texts,
            metadatas=metadatas
        )

    def query(self, query_text: str, n_results: int = 5,
               where: dict = None) -> list[dict]:
        """Query the vector store."""
        query_embedding = self.embed_fn([query_text])[0]

        kwargs = {
            "query_embeddings": [query_embedding],
            "n_results": n_results,
            "include": ["documents", "metadatas", "distances"]
        }
        if where:
            kwargs["where"] = where

        results = self.collection.query(**kwargs)

        return [
            {
                "text": results["documents"][0][i],
                "metadata": results["metadatas"][0][i],
                "distance": results["distances"][0][i],
                "score": 1 - results["distances"][0][i]  # Convert to similarity
            }
            for i in range(len(results["documents"][0]))
        ]

    def delete_collection(self) -> None:
        self.client.delete_collection(self.collection.name)

Part 4: Qdrant (Enterprise, Scalable)

Used by: Perplexity, large-scale production RAG

from qdrant_client import QdrantClient
from qdrant_client.models import (
    Distance, VectorParams, PointStruct,
    Filter, FieldCondition, MatchValue,
    SearchRequest
)
import uuid

class QdrantRAG:
    """Production Qdrant RAG system."""

    def __init__(self, collection_name: str,
                  vector_size: int = 1536,
                  host: str = "localhost",
                  port: int = 6333):
        self.client = QdrantClient(host=host, port=port)
        self.collection_name = collection_name
        self._ensure_collection(vector_size)

    def _ensure_collection(self, vector_size: int) -> None:
        collections = [c.name for c in self.client.get_collections().collections]
        if self.collection_name not in collections:
            self.client.create_collection(
                collection_name=self.collection_name,
                vectors_config=VectorParams(
                    size=vector_size,
                    distance=Distance.COSINE
                )
            )

    def upsert(self, texts: list[str], embeddings: list[list[float]],
                metadatas: list[dict] = None) -> None:
        """Upsert points into Qdrant."""
        points = [
            PointStruct(
                id=str(uuid.uuid4()),
                vector=embedding,
                payload={"text": text, **(meta or {})}
            )
            for text, embedding, meta in zip(
                texts, embeddings, metadatas or [{}] * len(texts)
            )
        ]
        self.client.upsert(collection_name=self.collection_name, points=points)

    def search(self, query_vector: list[float], top_k: int = 5,
                filter_conditions: dict = None) -> list[dict]:
        """Semantic search with optional metadata filtering."""
        query_filter = None
        if filter_conditions:
            must = [
                FieldCondition(key=k, match=MatchValue(value=v))
                for k, v in filter_conditions.items()
            ]
            query_filter = Filter(must=must)

        results = self.client.search(
            collection_name=self.collection_name,
            query_vector=query_vector,
            limit=top_k,
            query_filter=query_filter,
            with_payload=True
        )

        return [
            {
                "text": r.payload.get("text", ""),
                "metadata": {k: v for k, v in r.payload.items() if k != "text"},
                "score": r.score
            }
            for r in results
        ]

Part 5: Hybrid Search (BM25 + Vector)

The technique used by Perplexity and every elite RAG system. Combines keyword

precision with semantic understanding.

from rank_bm25 import BM25Okapi
import numpy as np

class HybridSearcher:
    """
    Hybrid retrieval: BM25 keyword search + vector semantic search.
    Combines results using Reciprocal Rank Fusion (RRF).
    This is what separates production RAG from toy RAG.
    """

    def __init__(self, documents: list[str], embeddings: list[list[float]]):
        self.documents = documents
        self.embeddings = np.array(embeddings)

        # BM25 index
        tokenized = [doc.lower().split() for doc in documents]
        self.bm25 = BM25Okapi(tokenized)

    def bm25_search(self, query: str, top_k: int = 20) -> list[tuple[int, float]]:
        """BM25 keyword search."""
        tokenized_query = query.lower().split()
        scores = self.bm25.get_scores(tokenized_query)
        top_indices = np.argsort(scores)[::-1][:top_k]
        return [(int(i), float(scores[i])) for i in top_indices]

    def vector_search(self, query_embedding: list[float],
                       top_k: int = 20) -> list[tuple[int, float]]:
        """Cosine similarity vector search."""
        q = np.array(query_embedding)
        q_norm = q / (np.linalg.norm(q) + 1e-10)
        doc_norms = self.embeddings / (
            np.linalg.norm(self.embeddings, axis=1, keepdims=True) + 1e-10
        )
        scores = doc_norms @ q_norm
        top_indices = np.argsort(scores)[::-1][:top_k]
        return [(int(i), float(scores[i])) for i in top_indices]

    def reciprocal_rank_fusion(self,
                                bm25_results: list[tuple[int, float]],
                                vector_results: list[tuple[int, float]],
                                k: int = 60,
                                alpha: float = 0.5) -> list[tuple[int, float]]:
        """
        RRF fusion — used by Perplexity, Cohere, and Elastic.
        k=60 is the standard constant (from original RRF paper).
        alpha controls BM25 vs vector weight.
        """
        scores = {}

        for rank, (doc_id, _) in enumerate(bm25_results):
            scores[doc_id] = scores.get(doc_id, 0) + (1 - alpha) / (k + rank + 1)

        for rank, (doc_id, _) in enumerate(vector_results):
            scores[doc_id] = scores.get(doc_id, 0) + alpha / (k + rank + 1)

        return sorted(scores.items(), key=lambda x: x[1], reverse=True)

    def search(self, query: str, query_embedding: list[float],
                top_k: int = 5) -> list[dict]:
        """Full hybrid search."""
        bm25_results = self.bm25_search(query, top_k=20)
        vector_results = self.vector_search(query_embedding, top_k=20)
        fused = self.reciprocal_rank_fusion(bm25_results, vector_results)

        return [
            {
                "text": self.documents[doc_id],
                "score": score,
                "rank": rank
            }
            for rank, (doc_id, score) in enumerate(fused[:top_k])
        ]

Part 6: Full RAG Pipeline

class IMIResearchRAG:
    """
    Complete production RAG pipeline for IMI research data.
    Uses: ChromaDB + hybrid search + context assembly.
    """

    def __init__(self, embed_fn: callable, llm_fn: callable,
                  persist_dir: str = "./imi_rag"):
        self.embed_fn = embed_fn
        self.llm_fn = llm_fn
        self.vector_store = ChromaRAG("imi_research", embed_fn, persist_dir)

    def ingest(self, documents: list[dict],
                text_field: str = "content") -> int:
        """Ingest and chunk documents into the RAG store."""
        all_chunks = []
        for doc in documents:
            text = doc[text_field]
            chunks = chunk_recursive(text)
            for i, chunk in enumerate(chunks):
                all_chunks.append({
                    "text": chunk,
                    "id": f"{doc.get('id', 'doc')}_{i}",
                    "source": doc.get("source", ""),
                    "brand": doc.get("brand", ""),
                    "date": doc.get("date", "")
                })

        self.vector_store.add_documents(all_chunks)
        return len(all_chunks)

    def retrieve(self, query: str, top_k: int = 5,
                  brand_filter: str = None) -> list[dict]:
        """Retrieve relevant chunks for a query."""
        where = {"brand": brand_filter} if brand_filter else None
        return self.vector_store.query(query, n_results=top_k, where=where)

    def answer(self, question: str, brand: str = None,
                top_k: int = 5) -> dict:
        """Full RAG answer pipeline."""
        # 1. Retrieve
        chunks = self.retrieve(question, top_k=top_k, brand_filter=brand)

        if not chunks:
            return {
                "answer": "No relevant data found for this query.",
                "sources": [],
                "chunks_used": 0
            }

        # 2. Build context
        context = "\n\n---\n\n".join([
            f"[Source: {c['metadata'].get('source', 'unknown')}]\n{c['text']}"
            for c in chunks
        ])

        # 3. Build prompt
        prompt = f"""Answer the research question using ONLY the provided context.

Context:
{context}

Question: {question}

If the context doesn't contain the answer, say: "The available data does not address this."

Answer:"""

        # 4. Generate
        answer = self.llm_fn(prompt)

        return {
            "answer": answer,
            "sources": [c["metadata"].get("source", "") for c in chunks],
            "chunks_used": len(chunks),
            "top_score": chunks[0]["score"] if chunks else 0
        }

Output Standards

• Chunk size: 256–512 tokens for most use cases; 128 for high-precision Q&A
• Always include overlap (10–15% of chunk size) to prevent boundary loss
• Use hybrid search (BM25 + vector) for all production systems
• Store source metadata with every chunk — you'll need it for citations
• Index brand, date, and document type as filterable metadata fields
• Target retrieval precision: top-5 chunks should contain the answer 85%+ of the time
• Test with: known Q&A pairs from your actual document set

AXE MCP Server Integration

Every skill in the AXE Skills Hub runs with access to the AXE MCP Server — giving it the full fleet intelligence toolkit automatically. No setup required; tools are available in any AXE-powered session.

Core Tools Available

Quick Start

# In any AXE session, tools are pre-loaded. Example chaining:

# 1. Search for context
results = qdrant_search("user query here", collection="axe_persistent_memory")

# 2. Fetch live data if needed
content = web_fetch("https://docs.example.com/api")

# 3. Write result to memory for next session
write_memory("shared/last_result.md", output)

# 4. Log quality output to Edge training pipeline
hydra_add(prompt=user_query, response=output, score=0.9, source="skill-name")

Edge Training Integration

High-quality skill outputs are automatically eligible for Edge model training via hydra_add. When a response scores ≥0.85 in evals, pipe it to the Hydra pipeline to compound Edge's knowledge. This is how skills make Edge smarter over time.

# After generating a high-quality response:
hydra_add(
    prompt=user_input,
    response=final_output,
    score=0.9,          # eval score
    source="skill-name" # tracks provenance
)