crinn-contrastive-rl-ann-search

name	crinn-contrastive-rl-ann-search
title	CRINN - Contrastive RL for HNSW Optimization
version	0.0.2
engine	skillxiv-v0.0.2-claude-opus-4.6
license	MIT
url	https://arxiv.org/abs/2508.02091
keywords	["reinforcement-learning","nearest-neighbor-search","hnsw","code-optimization"]
description	Optimize approximate nearest neighbor search via contrastive RL, learning to generate efficient code for HNSW graph construction, search, and refinement.

CRINN: Contrastive RL for ANN Search Optimization

CRINN applies contrastive reinforcement learning to optimize Hierarchical Navigable Small World (HNSW) nearest neighbor search. Rather than hand-tuning parameters, it trains LLMs to generate optimized code by comparing implementations with different performance characteristics, learning which optimizations matter most.

Core Concept

ANN search is performance-critical yet parameter-sensitive. Traditional approaches optimize via grid search, but this doesn't generalize to new datasets or hardware. CRINN treats optimization as an RL problem: LLM generates code variants, measures their query speed (QPS), and learns patterns of what makes code fast. Contrastive learning—comparing fast vs. slow implementations—teaches the model efficiency principles more effectively than absolute rewards.

Architecture Overview

• Contrastive Reward System: Compare code variants, score based on QPS at fixed recall levels
• Sequential Module Optimization: Optimize HNSW modules (graph construction, search, refinement) independently
• Speed-Based Rewards: nDCG-style metric: area under QPS-recall curve for comparing implementations
• GRPO Training: Group Relative Policy Optimization for efficient RL updates
• Multi-Benchmark Evaluation: Validate on 6 standard ANN datasets (GIST, MNIST, GloVe, etc.)

Implementation Steps

Step 1: Set Up HNSW Benchmarking

import time
import numpy as np
from typing import Tuple, Dict, List

class HNSWBenchmark:
    """Benchmark HNSW implementations on standard datasets."""

    def __init__(self, dataset_name: str, dataset: np.ndarray, queries: np.ndarray):
        self.dataset_name = dataset_name
        self.data = dataset  # (n, d)
        self.queries = queries  # (m, d)
        self.ground_truth = self._compute_ground_truth()

    def _compute_ground_truth(self):
        """Compute exact nearest neighbors via brute force."""
        distances = np.linalg.norm(
            self.queries[:, np.newaxis, :] - self.data[np.newaxis, :, :],
            axis=2
        )
        # Top-k indices for each query
        return np.argsort(distances, axis=1)[:, :100]  # Top 100

    def evaluate_hnsw(self, hnsw_impl) -> Dict[str, float]:
        """
        Evaluate HNSW implementation on recall and QPS.
        """
        # Time queries
        start = time.time()
        results = hnsw_impl.search_batch(self.queries, k=)
        qps = (.queries) / (time.time() - start)

        
        recall_sum = 
         i, result_ids  (results):
            hits = ((result_ids) & (.ground_truth[i]))
            recall_sum += hits / (.ground_truth[i])

        recall = recall_sum / (.queries)

         {
            : qps,
            : recall,
            :  / qps  
        }

     () -> :
        
        qps_at_recall = []

         ef  ef_sweep:
            result = .evaluate_hnsw(...)  
             result[] >= :  
                qps_at_recall.append(result[])

        
          qps_at_recall:
             

        reward = np.trapz(qps_at_recall)
         reward /   

Step 2: Implement Contrastive Reward

def compute_contrastive_reward(impl_a: str, impl_b: str,
                              benchmark: HNSWBenchmark) -> Tuple[float, float]:
    """
    Compare two HNSW implementations.
    Returns rewards for comparing them (preference learning).
    """
    # Build and evaluate both
    hnsw_a = build_hnsw_from_code(impl_a)
    hnsw_b = build_hnsw_from_code(impl_b)

    results_a = benchmark.evaluate_hnsw(hnsw_a)
    results_b = benchmark.evaluate_hnsw(hnsw_b)

    # Reward is comparative: which is better?
    qps_a = results_a['qps']
    qps_b = results_b['qps']

    # Normalize by maximum
    max_qps = max(qps_a, qps_b)
    reward_a = qps_a / max_qps
    reward_b = qps_b / max_qps

    return reward_a, reward_b

class ContrastiveRL:
    """
    Train LLM to generate HNSW code via contrastive learning.
    """
    def __init__(self, model, benchmark: HNSWBenchmark):
        self.model = model
        self.benchmark = benchmark

    def generate_code_variants(self, module_type: str = 'search', num_variants: int = 4) -> List[str]:
        """
        Generate multiple HNSW code variants using LLM.
        """
        prompt = f"""Generate {num_variants} different optimized implementations of HNSW {module_type} module.
Focus on efficiency: prefetching, vectorization, cache locality, etc.
Provide complete C++ code for the {module_type} operation.

Variant 1:
"""

        
        variants = []
         temp  np.linspace(, , num_variants):
            code = .model.generate(prompt, temperature=temp, max_tokens=)
            variants.append(code)

         variants

     ():
        
        
        variants = .generate_code_variants(module_type, num_variants=)

        
        rewards = []
         variant  variants:
            :
                hnsw = build_hnsw_from_code(variant)
                reward = .benchmark.compute_reward(
                    .benchmark.evaluate_hnsw(hnsw)
                )
                rewards.append(reward)
            :
                rewards.append()

        
        
         i  ((variants)):
             j  (i + , (variants)):
                 rewards[i] > rewards[j]:
                    
                    better_code = variants[i]
                    worse_code = variants[j]
                :
                    better_code = variants[j]
                    worse_code = variants[i]

                
                
                ._update_model(better_code, worse_code)

     ():
        
        
        preferred_logp = .model.compute_logp(preferred)
        dispreferred_logp = .model.compute_logp(dispreferred)

        
        loss = -torch.log(torch.sigmoid(preferred_logp - dispreferred_logp))
        loss.backward()
        .model.optimizer.step()

     () -> :
        
        templates = {
            : ,
            : ,
            : 
        }
         templates.get(module_type, )

Step 3: Sequential Module Optimization

class SequentialHNSWOptimizer:
    """Optimize HNSW modules one at a time."""

    def __init__(self, model, benchmarks: Dict[str, HNSWBenchmark]):
        self.model = model
        self.benchmarks = benchmarks
        self.optimized_modules = {}

    def optimize_construction(self, num_iterations: int = 10):
        """Optimize graph construction module."""
        print("Optimizing graph construction...")
        rl = ContrastiveRL(self.model, self.benchmarks['construction'])

        for i in range(num_iterations):
            rl.train_step(module_type='construction')
            print(f"  Iteration {i}: Training contrastive RL")

        # Extract best variant
        best_code = rl.generate_code_variants('construction', num_variants=1)[0]
        self.optimized_modules['construction'] = best_code

    def optimize_search(self, num_iterations: int = 10):
        """Optimize search module."""
        print("Optimizing search module...")
        rl = ContrastiveRL(self.model, .benchmarks[])

         i  (num_iterations):
            rl.train_step(module_type=)

        best_code = rl.generate_code_variants(, num_variants=)[]
        .optimized_modules[] = best_code

     ():
        
        ()
        rl = ContrastiveRL(.model, .benchmarks[])

         i  (num_iterations):
            rl.train_step(module_type=)

        best_code = rl.generate_code_variants(, num_variants=)[]
        .optimized_modules[] = best_code

     ():
        
        .optimize_construction(num_iterations=)
        .optimize_search(num_iterations=)
        .optimize_refinement(num_iterations=)

        
         .compile_optimized_hnsw()

     ():
        
        full_code = 
         full_code

Step 4: Evaluate on Benchmark Suite

def evaluate_crinn(optimized_code: str, datasets: Dict[str, Tuple[np.ndarray, np.ndarray]]) -> Dict:
    """
    Evaluate CRINN-optimized code on multiple benchmarks.
    """
    results = {}

    for dataset_name, (data, queries) in datasets.items():
        benchmark = HNSWBenchmark(dataset_name, data, queries)
        hnsw = build_hnsw_from_code(optimized_code)
        metrics = benchmark.evaluate_hnsw(hnsw)

        results[dataset_name] = metrics
        print(f"{dataset_name}: {metrics['qps']:.1f} QPS @ {metrics['recall']:.3f} recall")

    return results

Practical Guidance

When to Use:

• Optimizing ANN search for specific hardware/datasets
• Scenarios where baseline HNSW doesn't meet latency targets
• Applications with domain-specific distance metrics
• Cases where code generation + evaluation is feasible

When NOT To Use:

• Standard HNSW parameters are sufficient
• Real-time online optimization (training is slow)
• Proprietary hardware without benchmarking access
• Scenarios requiring formal correctness guarantees

Hyperparameters:

Parameter	Default	Impact
num_code_variants	4	More variants = better exploration, higher eval cost
rl_iterations_per_module	15	Training iterations; balance quality vs. time
ef_sweep_points	[10,20,50,100,200]	Parameter ranges for reward calculation
recall_threshold	0.85	Minimum recall for valid configurations

Reference

Paper: CRINN: Contrastive RL for Approximate Nearest Neighbor Search (2508.02091)

• Best-in-class on 3/6 benchmarks, ties on 2 others
• 3-85% QPS improvements at fixed recall
• Contrastive learning more effective than absolute rewards