Hybrid Search

┌─{search.py,search.ts}────────┐
│ ┌──────────┐                 │
│ │   User   │                 │
│ │  Query   │                 │
│ └─────┬────┘                 │
│       ▼                      │
│ ┌─turbopuffer queries──┐     │
│ │ ┌────────────────┐   │     │
│ ├▶│ Vector Query 1 │───┤     │
│ │ └────────────────┘   │     │
│ │ ┌────────────────┐   │     │
│ ├▶│ Vector Query 2 │───┤     │
│ │ └────────────────┘   │     │
│ │ ┌────────────────┐   │     │
│ ├▶│  Text Query 1  │───┤     │
│ │ └────────────────┘   │     │
│ └──────────┬───────────┘     │
│            ▼                 │
│    ┌────────────┐            │
│    │ Rank Fuse  │            │
│    └─────┬──────┘            │
│          ▼                   │
│    ┌────────────┐            │
│    │  Re-Rank   │            │
│    └────────────┘            │
└──────────────────────────────┘

To improve search quality, multiple strategies can be used together. This is commonly referred to as hybrid search.

turbopuffer supports vector search and BM25 full-text search. Combining them produces semantically relevant search results (vectors), as well as results matching specific words or strings (i.e. product SKUs, email addresses, weighing exact keywords highly).

Keep search logic in {search.py, search.ts}. Use turbopuffer for initial retrieval to narrow millions of results to dozens for rank fusion and re-ranking. turbopuffer also provides simple server-side rank fusion via RRF.

To improve search results further, we suggest:

The example below uses native embeddings for the vector side, so one content attribute drives both BM25 and vector search. To embed in your own code instead, choose an from the dropdown in the code sample.

# $ pip install turbopuffer
import os
import uuid

import turbopuffer
from turbopuffer.types import Row

tpuf = turbopuffer.Turbopuffer(
    api_key=os.getenv("TURBOPUFFER_API_KEY"), # created here: https://turbopuffer.com/dashboard
    region="gcp-us-central1", # choose best region: https://turbopuffer.com/docs/regions
)

namespace = os.getenv("TURBOPUFFER_NAMESPACE", f"hybrid-example-{uuid.uuid4().hex[:8]}")
ns = tpuf.namespace(namespace)

# Upsert documents with both FTS and vector search capabilities
ns.write(
    upsert_rows=[
    {
        "id": 1,
        "content": "Muesli: A quick mix of raw oats, nuts and dried fruit served with cold milk",
    },
    {
        "id": 2,
        "content": "Classic chia seed pudding is a cold breakfast that takes 5 minutes to prepare",
    },
    {
        "id": 3,
        "content": "Overnight oats: Mix oats with milk, refrigerate overnight for a delicious chilled breakfast",
    },
    {
        "id": 4,
        "content": "Hot oatmeal is a quick and healthy breakfast",
    },
    {
        "id": 5,
        "content": "Breakfast sandwich: A little extra prep, but worth it on Sunday mornings!",
    }
    ],
    distance_metric="cosine_distance",
    schema={
        "content": {
            "type": "string",
            "full_text_search": True,
            "embed": {"model": "nvidia/nemotron-3-embed-8b", "dims": 1024},
        }
    },
)

query = "quick breakfast like oatmeal but cold"
print("Ideal:", [1, 2, 3, 4, 5])

response = ns.multi_query(
    queries=[
        {
            "rank_by": ("content", "ANN", ("Embed", query)),
            "limit": 10,
            "include_attributes": ["content"],
        },
        {
            "rank_by": ("content", "BM25", query),
            "limit": 10,
            "include_attributes": ["content"],
        },
    ]
)

vector_result, fts_result = response.results[0].rows, response.results[1].rows
print("Vector:", [item.id for item in vector_result])
print("FTS:", [item.id for item in fts_result])

response = ns.multi_query(
    queries=[
        {
            "rank_by": ("content", "ANN", ("Embed", query)),
            "limit": 10,
            "include_attributes": ["content"],
        },
        {
            "rank_by": ("content", "BM25", query),
            "limit": 10,
            "include_attributes": ["content"],
        },
    ],
    rerank_by=("RRF",),
)

fused_results = response.results[0].rows
print("Fused:", [item.id for item in fused_results])

def cohere_rerank_or_unranked(results, query, k = None): 
    if not os.getenv("COHERE_API_KEY"):
        print("Warning: COHERE_API_KEY not set (https://dashboard.cohere.com/api-keys), returning unranked results")
        return results
    try:
        co = __import__('cohere').Client(os.getenv("COHERE_API_KEY"))
        reranked = co.rerank(query=query, documents=[r.content for r in results], top_n=k or len(results)).results
        for r in reranked:
            results[r.index]['$dist'] = r.relevance_score
        return [results[r.index] for r in reranked]
    except (ImportError, AttributeError):
        print("Warning: cohere package not installed (`pip install cohere`), returning unranked results")
        return results

reranked_results = cohere_rerank_or_unranked(fused_results, query)
print("Reranked:", [item.id for item in reranked_results])

Computing additional features

Computed attributes can add signal to each search result, without additional turbopuffer roundtrips. For example, you could compute the BM25 score of a query to a content attribute, even when your ranking expression only calculates vector distance between the query and content. When doing a multi-query, consider having each sub-query compute the other strategy's score, and pass both these scores into your second stage re-ranker:

[
  {
    "rank_by": ["vector", "ANN", [0.1, 0.2]],
    "compute_attributes": {
      "bm25_score": ["content", "BM25", "quick walrus"]
    },
    "limit": 50
  },
  {
    "rank_by": ["content", "BM25", "quick walrus"],
    "compute_attributes": {
      "vector_distance": ["vector", "VectorDist", [0.1, 0.2]]
    },
    "limit": 50
  }
]