How to Configure an RRF Search Pipeline in OpenSearch for Hybrid Search

To configure an RRF search pipeline in OpenSearch for hybrid search, define a JSON pipeline descriptor using the score-ranker-processor with the rrf technique, register it via the Search Pipeline API, and reference it in your search requests to fuse BM25 and vector results automatically.

The jamwithai/production-agentic-rag-course repository implements a production-ready hybrid retrieval system that leverages OpenSearch's native Reciprocal Rank Fusion (RRF) to combine keyword matching and dense vector similarity. Setting up an RRF search pipeline in OpenSearch for hybrid search requires coordinating three components: a pipeline definition stored in your application code, a registration step that pushes the configuration to the OpenSearch cluster, and search-time parameters that activate the fusion processor.

Step 1: Define the RRF Pipeline

Start by creating the pipeline configuration in your application constants. In src/services/opensearch/index_config_hybrid.py (lines 72‑85), the repository defines the HYBRID_RRF_PIPELINE constant:

HYBRID_RRF_PIPELINE = {
    "id": "hybrid-rrf-pipeline",
    "description": "Post processor for hybrid RRF search",
    "phase_results_processors": [
        {
            "score-ranker-processor": {
                "combination": {
                    "technique": "rrf",
                    "rank_constant": 60
                }
            }
        }
    ],
}

This configuration tells OpenSearch to apply the score‑ranker‑processor after the query phase completes. The technique field specifies rrf, while rank_constant sets the $k$ parameter to 60, which controls how aggressively lower-ranked results are penalized using the formula $1/(k + \text{rank})$.

Step 2: Register the Pipeline with OpenSearch

Once defined, you must register the pipeline with your OpenSearch cluster. The OpenSearchClient class in src/services/opensearch/client.py implements this via the _create_rrf_pipeline() method (lines 92‑124):

• The method extracts the pipeline_id from HYBRID_RRF_PIPELINE["id"].
• If force=True, it removes any existing pipeline with that ID to prevent conflicts.
• It checks for existing pipelines using self.client.ingest.get_pipeline.
• If absent, it sends a PUT /_search/pipeline/{pipeline_id} request with the JSON body defined in Step 1.

This registration typically occurs during application bootstrap inside setup_indices(), which ensures the pipeline exists before serving traffic.

Step 3: Execute Hybrid Search with RRF

With the pipeline registered, invoke it during search by passing the pipeline ID in the request parameters. The _search_hybrid_native() method in src/services/opensearch/client.py (lines 44‑68) constructs a hybrid query combining BM25 and KNN:

hybrid_query = {
    "hybrid": {
        "queries": [
            bm25_query,
            {"knn": {"embedding": {"vector": query_embedding, "k": size * 2}}}
        ]
    }
}

Crucially, the search call includes params={"search_pipeline": HYBRID_RRF_PIPELINE["id"]} to trigger the RRF post-processor. OpenSearch executes both sub-queries, then applies Reciprocal Rank Fusion to the combined result set before returning the final ranked list.

Customizing RRF Parameters

You can tune the fusion behavior by modifying the HYBRID_RRF_PIPELINE constant in src/services/opensearch/index_config_hybrid.py:

• rank_constant: Lower values (e.g., 30) penalize lower ranks more aggressively, while higher values (e.g., 100) create a softer blend between the two result sets.
• pipeline ID: Changing the "id" field allows multiple pipeline variants for A/B testing different fusion strategies.

To apply changes, call _create_rrf_pipeline(force=True) to overwrite the existing pipeline definition in the cluster.

Complete Implementation Example

The following example demonstrates bootstrapping the client and executing a hybrid search:

from src.services.opensearch.client import OpenSearchClient
from src.config import Settings

# Initialize client

settings = Settings()
client = OpenSearchClient(
    host=settings.opensearch.host,
    settings=settings
)

# Register indices and pipeline (run once at startup)

client.setup_indices(force=True)

# Execute hybrid search

query = "transformer based retrieval"
embedding = get_embedding(query)  # Your embedding function

results = client.search_unified(
    query=query,
    query_embedding=embedding,
    size=10,
    use_hybrid=True,  # Triggers native hybrid + RRF

    min_score=0.1
)

for hit in results["hits"]:
    print(f"Score: {hit['score']:.2f} | Title: {hit['title']}")

Summary

• Define the pipeline in src/services/opensearch/index_config_hybrid.py using the score-ranker-processor with technique: "rrf" and a rank_constant (typically 60).
• Register the pipeline via _create_rrf_pipeline() in src/services/opensearch/client.py, which sends a PUT request to /_search/pipeline/{id}.
• Execute hybrid queries by passing params={"search_pipeline": "hybrid-rrf-pipeline"} with queries that include both BM25 and KNN clauses wrapped in a hybrid object.
• Tune fusion behavior by adjusting the rank_constant; lower values increase the penalty for lower-ranked items.

Frequently Asked Questions

What is the formula used by OpenSearch's RRF implementation?

OpenSearch uses the standard Reciprocal Rank Fusion formula $1/(k + \text{rank})$, where $k$ corresponds to the rank_constant parameter (default 60 in the repository). A result ranked first in one query receives a score of $1/(60+1)$, while a result ranked tenth receives $1/(60+10)$. These scores are summed across the BM25 and KNN result sets to produce the final ranking.

Can I use multiple search pipelines for different query types?

Yes. You can define multiple pipeline constants with unique "id" values in src/services/opensearch/index_config_hybrid.py and register each via separate calls to _create_rrf_pipeline(). When searching, specify the desired pipeline ID in the search_pipeline parameter. This approach enables A/B testing between different rank_constant values or even different fusion techniques if you define alternative processors.

Why does the hybrid query multiply the KNN size by 2?

The repository sets "k": size * 2 for the KNN clause (where size is the final requested result count) to ensure sufficient candidates enter the fusion pool. Since RRF operates on the ranked lists from each sub-query, retrieving more vector candidates than the final size prevents high-scoring BM25 results from dominating when the KNN result set is truncated too early. This oversampling strategy improves the quality of the fused ranking.

How do I disable RRF and fall back to a single search method?

Remove the params={"search_pipeline": ...} argument from the search() call in src/services/opensearch/client.py, or set use_hybrid=False when calling search_unified(). Without the pipeline parameter, OpenSearch returns raw BM25 or KNN results depending on which query type you submit, bypassing the fusion step entirely.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}