Best AI Models for RAG Applications 2026: Embeddings, Retrieval, and Generation

from qdrant_client import QdrantClient
from qdrant_client.models import SparseVector, SearchRequest, FusionQuery

client = QdrantClient(url="http://localhost:6333")

# Hybrid search: dense + sparse
results = client.query_points(
    collection_name="documents",
    query=FusionQuery(
        queries=[
            # Dense vector from embedding model
            SearchRequest(
                vector=query_embedding,
                limit=20
            ),
            # Sparse BM25 vector
            SearchRequest(
                vector=SparseVector(indices=bm25_indices, values=bm25_values),
                limit=20
            )
        ],
        fusion="rrf"  # Reciprocal Rank Fusion
    ),
    limit=10
)

import requests
import numpy as np
from typing import List, Dict

CRAZYROUTER_API = "https://crazyrouter.com/v1"
API_KEY = "sk-your-api-key"

headers = {
    "Authorization": f"Bearer {API_KEY}",
    "Content-Type": "application/json"
}


def get_embeddings(texts: List[str], model: str = "text-embedding-3-small") -> List[List[float]]:
    """Generate embeddings for a list of texts via Crazyrouter."""
    response = requests.post(f"{CRAZYROUTER_API}/embeddings", headers=headers, json={
        "model": model,
        "input": texts
    })
    data = response.json()
    return [item["embedding"] for item in data["data"]]


def cosine_similarity(a: List[float], b: List[float]) -> float:
    """Calculate cosine similarity between two vectors."""
    a, b = np.array(a), np.array(b)
    return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))


def retrieve(query: str, documents: List[Dict], top_k: int = 5) -> List[Dict]:
    """Retrieve most relevant documents for a query."""
    query_embedding = get_embeddings([query])[0]
    
    scored = []
    for doc in documents:
        score = cosine_similarity(query_embedding, doc["embedding"])
        scored.append({**doc, "score": score})
    
    scored.sort(key=lambda x: x["score"], reverse=True)
    return scored[:top_k]


def generate_answer(query: str, context_docs: List[Dict], model: str = "gpt-4o-mini") -> str:
    """Generate a grounded answer from retrieved context."""
    context = "\n\n---\n\n".join([
        f"[Source: {doc['source']}]\n{doc['text']}" 
        for doc in context_docs
    ])
    
    response = requests.post(f"{CRAZYROUTER_API}/chat/completions", headers=headers, json={
        "model": model,
        "messages": [
            {
                "role": "system",
                "content": (
                    "You are a helpful assistant. Answer the user's question based ONLY on "
                    "the provided context. If the context doesn't contain enough information, "
                    "say so. Cite sources using [Source: ...] format."
                )
            },
            {
                "role": "user",
                "content": f"Context:\n{context}\n\nQuestion: {query}"
            }
        ],
        "temperature": 0.1,
        "max_tokens": 1024
    })
    
    return response.json()["choices"][0]["message"]["content"]


# --- Usage Example ---

# Step 1: Index documents (do this once)
documents = [
    {"text": "Python 3.12 introduced type parameter syntax...", "source": "python-docs"},
    {"text": "FastAPI uses Pydantic for data validation...", "source": "fastapi-docs"},
    {"text": "Vector databases store high-dimensional embeddings...", "source": "qdrant-docs"},
]

# Generate and store embeddings
for doc in documents:
    doc["embedding"] = get_embeddings([doc["text"]])[0]

# Step 2: Query the RAG pipeline
query = "How does FastAPI handle data validation?"
relevant_docs = retrieve(query, documents, top_k=3)
answer = generate_answer(query, relevant_docs)

print(f"Answer: {answer}")
print(f"\nSources used: {[d['source'] for d in relevant_docs]}")

const axios = require('axios');

const API_BASE = 'https://crazyrouter.com/v1';
const API_KEY = 'sk-your-api-key';
const headers = {
  'Authorization': `Bearer ${API_KEY}`,
  'Content-Type': 'application/json'
};

async function embedTexts(texts, model = 'text-embedding-3-small') {
  const { data } = await axios.post(`${API_BASE}/embeddings`, {
    model,
    input: texts
  }, { headers });
  return data.data.map(item => item.embedding);
}

async function ragQuery(query, documents) {
  // Embed query
  const [queryVec] = await embedTexts([query]);
  
  // Simple cosine similarity retrieval
  const scored = documents.map(doc => ({
    ...doc,
    score: cosineSim(queryVec, doc.embedding)
  }));
  scored.sort((a, b) => b.score - a.score);
  const topDocs = scored.slice(0, 5);
  
  // Generate with streaming
  const context = topDocs.map(d => d.text).join('\n\n');
  
  const response = await axios.post(`${API_BASE}/chat/completions`, {
    model: 'gpt-4o-mini',
    stream: true,
    messages: [
      { role: 'system', content: 'Answer based only on the provided context.' },
      { role: 'user', content: `Context:\n${context}\n\nQuestion: ${query}` }
    ]
  }, { headers, responseType: 'stream' });

  // Process stream
  for await (const chunk of response.data) {
    const lines = chunk.toString().split('\n').filter(l => l.startsWith('data: '));
    for (const line of lines) {
      const json = line.replace('data: ', '');
      if (json === '[DONE]') return;
      const token = JSON.parse(json).choices[0]?.delta?.content || '';
      process.stdout.write(token);
    }
  }
}

function cosineSim(a, b) {
  const dot = a.reduce((sum, val, i) => sum + val * b[i], 0);
  const magA = Math.sqrt(a.reduce((sum, val) => sum + val * val, 0));
  const magB = Math.sqrt(b.reduce((sum, val) => sum + val * val, 0));
  return dot / (magA * magB);
}