Asymmetric Embeddings: Building a Personal Brand RAG Pipeline with Voyage 4 + Claude

The Idea That Started on TikTok

🎯 The thesis: You can embed your knowledge base with voyage-4-large for maximum retrieval quality, then query it at runtime with a lighter model like voyage-4-lite — and because the entire Voyage 4 family shares a single embedding space, it just works. This isn't a workaround. It's the intended design, and it's industry-first.

Why Asymmetric Embeddings Matter

The Voyage 4 Family

Real cost check: Running vai_estimate on 500 docs with 1,000 queries/month over 12 months: ingestion with voyage-4-large costs ~$0.02. Queries with voyage-4-lite run ~$0.01/month. Total 12-month embedding cost: under $0.20. At that price, the only reason not to use the best model for ingestion is impatience.

The Architecture

Three Ways to Ingest: A Teaching Moment

code-highlightLevel 1: Python SDK     — voyageai.Client, direct control, explicit batching
Level 2: REST API       — raw HTTP, portable to any language or runtime
Level 3: VAI            — single function call, chunking + embedding + storage abstracted away

Level 1: Python SDK (voyageai package)

bash code-highlightpip install voyageai pymongo

python code-highlight# ingest_sdk.py — maximum control, full visibility
import os, hashlib
from pathlib import Path
import voyageai
from pymongo import MongoClient

# Singleton clients — initialise once, reuse across calls
voyage = voyageai.Client(api_key=os.environ["VOYAGE_API_KEY"])
mongo = MongoClient(os.environ["MONGODB_URI"])
collection = mongo["personal_brand"]["rag_documents"]

def chunk_markdown(text: str, chunk_size=400, overlap=80) -> list[str]:
    """
    Sliding window chunker with overlap for context continuity.

    Why overlap? If a key sentence falls at a chunk boundary, overlap ensures
    it appears in full in at least one chunk — preserving context for the
    embedding model and avoiding meaning being split across retrieval units.
    """
    words = text.split()
    chunks = []
    for i in range(0, len(words), chunk_size - overlap):
        chunk = " ".join(words[i:i + chunk_size])
        if len(chunk.strip()) > 50:  # discard near-empty tail chunks
            chunks.append(chunk)
    return chunks

def ingest_file(filepath: Path) -> int:
    text = filepath.read_text()
    content_hash = hashlib.md5(text.encode()).hexdigest()

    # Content-hash deduplication: skip unchanged files entirely.
    # This makes re-running the script safe and cheap — use it in a cron
    # job or git hook without worrying about re-embedding costs.
    existing = collection.find_one({
        "source.filePath": str(filepath),
        "contentHash": content_hash
    })
    if existing:
        print(f"  Skipping (unchanged): {filepath.name}")
        return 0

    chunks = chunk_markdown(text)
    print(f"  Embedding {len(chunks)} chunks from {filepath.name}...")

    # voyage-4-large + input_type="document":
    # The SDK prepends an internal retrieval prompt before encoding.
    # This produces vectors optimised for being *found*, not for finding.
    # Never skip input_type — it meaningfully changes the vector geometry.
    result = voyage.embed(
        chunks,
        model="voyage-4-large",
        input_type="document"
    )

    docs = [
        {
            "content": chunk,
            "contentHash": hashlib.md5(chunk.encode()).hexdigest(),
            "embedding": embedding,
            "source": {
                "filePath": str(filepath),
                "title": filepath.stem.replace("-", " ").title(),
                "category": filepath.parent.name,
            },
            "chunk": {"index": i, "totalChunks": len(chunks)}
        }
        for i, (chunk, embedding) in enumerate(zip(chunks, result.embeddings))
    ]

    # Delete old chunks for this file, insert fresh — simple upsert pattern
    collection.delete_many({"source.filePath": str(filepath)})
    if docs:
        collection.insert_many(docs)
    return len(docs)

total = sum(ingest_file(f) for f in Path("./docs").rglob("*.md"))
print(f"\nDone. Ingested {total} chunks.")

• input_type="document" vs "query" produces meaningfully different vectors. The SDK prepends a different internal prompt for each. Using the wrong one degrades retrieval quality measurably.
• Content hashing makes ingestion idempotent. Run it in CI, in a cron job, or on every git commit to ./docs without paying twice for unchanged files.
• voyage.embed() handles API batching internally. You don't need to split 10,000 chunks manually — the SDK does it in groups of 128.

Level 2: REST API (direct HTTP)

• You might be ingesting from Node.js, a GitHub Action, or a serverless function where you can't install packages
• You want to understand what the SDK is actually doing beneath the surface
• You need the portability of a plain HTTP call

python code-highlight# ingest_rest.py — same logic, raw HTTP instead of SDK
import os, hashlib, json
from pathlib import Path
import urllib.request
from pymongo import MongoClient

VOYAGE_API = "https://api.voyageai.com/v1/embeddings"
VOYAGE_KEY = os.environ["VOYAGE_API_KEY"]

def embed_documents(texts: list[str]) -> list[list[float]]:
    """
    Direct REST call to Voyage AI embeddings endpoint.
    This is exactly what voyageai.Client.embed() calls under the hood.
    """
    payload = json.dumps({
        "input": texts,
        "model": "voyage-4-large",
        "input_type": "document"
    }).encode()

    req = urllib.request.Request(
        VOYAGE_API,
        data=payload,
        headers={
            "Content-Type": "application/json",
            "Authorization": f"Bearer {VOYAGE_KEY}"
        }
    )
    with urllib.request.urlopen(req) as response:
        data = json.loads(response.read())

    # Sort by index — the API doesn't guarantee order
    return [item["embedding"] for item in sorted(data["data"], key=lambda x: x["index"])]

json code-highlight{
  "object": "list",
  "data": [
    { "object": "embedding", "index": 0, "embedding": [0.021, -0.014, "..."] }
  ],
  "model": "voyage-4-large",
  "usage": { "total_tokens": 142 }
}

Level 3: VAI — One Function Call to Rule Them All

python code-highlightvai_ingest(
  text="...",
  model="voyage-4-large",
  chunkStrategy="markdown",
  chunkSize=512,
  collection="rag_documents",
  source="about.md",
  metadata={ "category": "bio" }
)

• Splitting the text using the strategy you specify (markdown, recursive, sentence, paragraph, or fixed)
• Calling Voyage AI to embed each chunk with input_type="document"
• Writing all chunks, embeddings, and metadata to your MongoDB collection
• Returning chunk count and token usage

The mental model: VAI is the power drill, the SDK is the hand drill, the REST API is the screwdriver. All three drive the same screw. Your job is knowing when to reach for which one.

Phase 2: Atlas Vector Search Index

json code-highlight{
  "name": "personal_brand_vector",
  "type": "vectorSearch",
  "definition": {
    "fields": [
      {
        "type": "vector",
        "path": "embedding",
        "numDimensions": 1024,
        "similarity": "cosine"
      },
      {
        "type": "filter",
        "path": "source.category"
      }
    ]
  }
}

Phase 3: The Query Pipeline (Next.js API Route)

typescript code-highlight// pages/api/chat.ts
import { NextApiRequest, NextApiResponse } from 'next';
import Anthropic from '@anthropic-ai/sdk';
import { connectToDatabase } from '@/lib/db/connection';
import { RagDocumentModel } from '@/lib/db/models/RagDocument';

const VOYAGE_API = "https://api.voyageai.com/v1/embeddings";

async function embedQuery(query: string): Promise<number[]> {
  // voyage-4-lite: $0.02/M tokens vs voyage-4-large at $0.12/M
  // Same embedding space — vectors are directly comparable with our ingested docs
  const res = await fetch(VOYAGE_API, {
    method: "POST",
    headers: {
      "Content-Type": "application/json",
      "Authorization": `Bearer ${process.env.VOYAGE_API_KEY}`
    },
    body: JSON.stringify({
      input: [query],
      model: "voyage-4-lite",
      input_type: "query"   // Prepends a different internal prompt — optimised for searching
    })
  });
  const data = await res.json();
  return data.data[0].embedding;
}

async function retrieveContext(queryEmbedding: number[], topK = 5) {
  await connectToDatabase();

  const results = await RagDocumentModel.aggregate([
    {
      $vectorSearch: {
        index: "personal_brand_vector",
        path: "embedding",
        queryVector: queryEmbedding,
        numCandidates: topK * 20,  // over-fetch, then filter by score threshold
        limit: topK,
      }
    },
    {
      $project: {
        content: 1,
        "source.title": 1,
        "source.category": 1,
        score: { $meta: "vectorSearchScore" }
      }
    }
  ]);

  return results.filter(r => r.score >= 0.72);
}

export default async function handler(req: NextApiRequest, res: NextApiResponse) {
  const { message, history = [] } = req.body;

  // Asymmetric retrieval in one place:
  // Documents were embedded once with voyage-4-large (rich, high quality)
  // Queries are embedded per-request with voyage-4-lite (fast, cheap)
  // Compatible because they share the Voyage 4 embedding space
  const queryEmbedding = await embedQuery(message);
  const context = await retrieveContext(queryEmbedding);

  const contextText = context
    .map((c, i) => `[Context ${i+1} - ${c.source.title}]\n${c.content}`)
    .join("\n\n---\n\n");

  const anthropic = new Anthropic();
  const stream = await anthropic.messages.stream({
    model: "claude-sonnet-4-20250514",
    max_tokens: 1024,
    system: `You are an AI assistant that knows [YOUR NAME] deeply. Use the retrieved context to answer questions about their work, background, and thinking. Be conversational, warm, and specific. If context doesn't cover the question, say so rather than guessing.

Retrieved context: ${contextText}`,
    messages: [
      ...history,
      { role: "user", content: message }
    ]
  });

  res.setHeader('Content-Type', 'text/event-stream');
  for await (const chunk of stream) {
    if (chunk.type === 'content_block_delta') {
      res.write(`data: ${JSON.stringify({ text: chunk.delta.text })}\n\n`);
    }
  }
  res.end();
}

The Asymmetry, Explained Simply

• Document embedding is a one-time offline cost → use voyage-4-large, pay for quality once.
• Query embedding is an online per-request cost → use voyage-4-lite, pay $0.02/M tokens instead of $0.12/M.

Local dev path: Use voyage-4-nano (open-weight, free on HuggingFace) to build and test your entire pipeline at zero API cost. When you're ready for production, change one string — "voyage-4-nano" → "voyage-4-large" — and re-run ingestion. Same embedding space. Same Atlas index. No other changes needed.

What to Put in Your Knowledge Base

Real-World Cost Breakdown

What This Demonstrates (The Real Teaching Moment)

About RAG architecture

• Asymmetric embedding is a legitimate pattern, not a hack — Voyage 4 was designed for it
• input_type="document" vs "query" matters: different internal prompts produce meaningfully different vectors, and mixing them intentionally is the whole point
• MongoDB Atlas Vector Search makes production RAG accessible without a dedicated vector database
• Shared embedding spaces eliminate re-indexing cost as you evolve your infrastructure

About API interaction patterns

• The Python SDK, the REST API, and VAI are three layers of the same abstraction stack
• Understanding all three makes you adaptable — not every environment can run Python packages
• Content hashing + idempotent writes is the correct pattern for any ingestion pipeline
• usage.total_tokens in every API response is your billing meter — log it from day one

About tooling philosophy

• VAI is the power drill: fast, declarative, right for prototyping and AI-assisted workflows
• The SDK is the hand drill: explicit, controllable, right when you need custom logic
• The REST API is the screwdriver: portable, dependency-free, right for constrained environments
• Knowing which tool to reach for — and why — is the actual skill

What to Build Next

• Add a feedback mechanism (thumbs up/down per response) — log to MongoDB, use to improve docs
• Track which chunks get retrieved most via score logging — those are your most resonant ideas
• Add a "sources" panel showing which documents informed each answer
• Experiment: use vai_query in your dev workflow to test retrieval quality before wiring up the UI — it embeds, searches, and reranks in one call
• Upgrade path: swap voyage-4-lite to voyage-4 on queries, compare scores in your logs — one string, zero re-indexing
• Cost visibility: pipe usage.total_tokens from every Voyage API call into a MongoDB collection, build a lightweight dashboard on top