warm-start patch 3/3: true_skill_ranking.py

patched/vendor/ranking_moderation/src/ranking_moderation/true_skill_ranking.py

@@ -0,0 +1,1481 @@
+"""
+Bradley-Terry ranking model for global ranking from pairwise comparisons.
+This module provides the core ranking functionality using the Bradley-Terry model.
+"""
+
+import logging
+import json
+import random
+from pathlib import Path
+from typing import Dict, List, Tuple, Any, Optional, TypeVar, Generic
+from dataclasses import dataclass, field
+from multiprocessing import Manager, Lock
+import numpy as np
+import time
+import choix
+from src.ranking_moderation.pairwise_comparison import PairwiseComparison
+from scipy.optimize import minimize
+from trueskill import Rating, rate_1vs1, quality_1vs1
+from tqdm import tqdm
+from concurrent.futures import ThreadPoolExecutor
+
+logger = logging.getLogger(__name__)
+
+T = TypeVar('T')
+
+@dataclass
+class RulerItem(Generic[T]):
+    """Represents an item in the ranking system.
+    
+    Attributes:
+        item_id: Unique identifier for the item
+        score: Current ranking score (same as mu for TrueSkill)
+        mu: TrueSkill mu value (mean of the skill estimate)
+        sigma: TrueSkill sigma value (standard deviation of the skill estimate)
+        rank: Current rank position (optional)
+        item: The original item data
+        metadata: Additional metadata about the ranking process
+    """
+    item_id: str
+    item: T
+    score: float = 0.0
+    mu: float = 0.0  # Add mu field
+    sigma: float = 8.333
+    rank: Optional[int] = None
+    metadata: Dict[str, Any] = field(default_factory=dict)
+
+    def to_dict(self) -> Dict[str, Any]:
+        """Convert to dictionary for serialization."""
+        return {
+            'item_id': self.item_id,
+            'item': self.item,
+            'score': self.score,
+            'mu': self.mu,  # Add mu to serialization
+            'sigma': self.sigma,
+            'rank': self.rank,
+            'metadata': self.metadata
+        }
+
+    @classmethod
+    def from_dict(cls, data: Dict[str, Any]) -> 'RulerItem[T]':
+        """Create from dictionary."""
+        return cls(
+            item_id=data['item_id'],
+            item=data['item'],
+            score=data.get('score', 0.0),
+            mu=data.get('mu', 0.0),  # Add mu to deserialization
+            sigma=data.get('sigma', 8.333),
+            rank=data.get('rank'),
+            metadata=data.get('metadata', {})
+        )
+
+class TrueSkillRanking:
+    """
+    TrueSkill ranking model for global ranking from pairwise comparisons.
+    Uses pairwise comparisons and TrueSkill model for global ranking.
+    Supports iterative ranking with adaptive sampling based on current estimates.
+    """
+
+    def __init__(self, 
+                 pairwise_comparison: PairwiseComparison,
+                 data_dir: str,
+                 file_name: str,
+                 convergence_threshold: float = 0.01,
+                 max_iterations: int = 50,
+                 max_workers: int = 10):
+        """Initialize TrueSkill ranking model.
+        
+        Args:
+            pairwise_comparison: PairwiseComparison instance for handling comparisons
+            data_dir: Directory to store state files
+            convergence_threshold: Threshold for score convergence
+        """
+        logger.info(f"Initialize TrueSkillRanking.")
+        self.pairwise_comparison = pairwise_comparison
+        self.convergence_threshold = convergence_threshold
+        self.data_dir = Path(data_dir)
+        self.data_dir.mkdir(parents=True, exist_ok=True)
+        self.file_name = file_name
+        self.max_iterations = max_iterations
+        self.max_workers = max_workers
+        # Initialize process manager for shared data
+        self.manager = Manager()
+
+        # Initialize state
+        self.ruler_items = self.manager.list()  # Store RulerItem instances
+
+        # Add process-safe locks
+        self._ruler_lock = Lock()  # Lock for ruler items
+
+        # Load saved state if exists
+        self.load_state()
+
+    def get_state_file_path(self, filename: str) -> Path:
+        """Get the full path for a state file."""
+        return self.data_dir / filename
+
+    def save_state(self):
+        """Save current state to disk."""
+        logger.info(f"Saving state to {self.data_dir}")
+
+        try:
+            # Save ruler items with their scores
+            with self._ruler_lock:
+                items_data = [item.to_dict() for item in self.ruler_items]
+                with self.get_state_file_path(self.file_name).open("w") as f:
+                    json.dump(items_data, f, indent=2, ensure_ascii=False)
+
+            logger.info("Successfully saved state")
+        except Exception as e:
+            logger.error(f"Failed to save state: {str(e)}")
+            raise
+
+    def load_state(self):
+        """Load state from disk if available."""
+        logger.info(f"Loading state from {self.data_dir}")
+
+        try:
+            # Load ruler items with their scores
+            with self._ruler_lock:
+                ruler_file = self.get_state_file_path(self.file_name)
+                if ruler_file.exists():
+                    with ruler_file.open("r") as f:
+                        items_data = json.load(f)
+                        self.ruler_items.extend([RulerItem.from_dict(item) for item in items_data])
+                    logger.info(f"Loaded {len(self.ruler_items)} ruler items")
+        except Exception as e:
+            logger.error(f"Failed to load state: {str(e)}")
+            raise
+
+    def get_current_scores(self) -> Dict[str, float]:
+        """Get current scores for all ruler items."""
+        with self._ruler_lock:
+            return {item.item_id: item.score for item in self.ruler_items}
+
+    def update_scores(self, new_scores: Dict[str, float]):
+        """Update scores for ruler items."""
+        with self._ruler_lock:
+            for item in self.ruler_items:
+                if item.item_id in new_scores:
+                    item.score = new_scores[item.item_id]
+
+    def fit(self, items: List[RulerItem]) -> List[RulerItem]:
+        """Fit TrueSkill model to the items.
+        
+        Args:
+            items: List of items to rank
+            max_iterations: Maximum number of iterations for TrueSkill
+        """
+        # Process each comparison result
+        comparison_info = self.pairwise_comparison.get_compare_information(
+            [item.item_id for item in items]
+        )
+        item_ids = set([])
+        data = []
+        for pair_key, results in comparison_info['comparison_results'].items():
+            id1, id2 = pair_key.split('_')
+            item_ids.add(id1)
+            item_ids.add(id2)
+
+            if results:
+                votes = {1: 0, -1: 0, 0: 0}
+                for result in results:
+                    orig_id1, orig_id2 = result['original_ids']
+                    if (orig_id1, orig_id2) == (id1, id2):
+                        votes[result['judgment']] += 1
+                    else:
+                        votes[-result['judgment']] += 1
+
+                final_judgment = max(votes.items(), key=lambda x: x[1])[0]
+
+                if final_judgment > 0:
+                    data.append((id1, id2, "win"))
+                elif final_judgment < 0:
+                    data.append((id2, id1, "win"))
+                else:
+                    data.append((id2, id1, "draw"))
+        ratings = {str(item.item_id): Rating(mu=50, sigma=50/3) for item in items}
+        mu_history = []
+        num_epochs = self.max_iterations
+        early_stop_threshold = 1e-3
+        logger.info(f"Fit by {len(data)} pair")
+        epoch_bar = tqdm(range(num_epochs), desc="Epochs", disable=True)
+        for epoch in epoch_bar:
+            random.shuffle(data)
+            delta_mu = []
+
+            for item_id_1, item_id_2, outcome in tqdm(data, desc=f"Epoch {epoch}", leave=False):
+                r1_old = ratings[item_id_1]
+                r2_old = ratings[item_id_2]
+
+                if outcome == "win":
+                    r1_new, r2_new = rate_1vs1(r1_old, r2_old)
+                elif outcome == "lose":
+                    r2_new, r1_new = rate_1vs1(r2_old, r1_old)
+                elif outcome == "draw":
+                    r1_new, r2_new = rate_1vs1(r1_old, r2_old, drawn=True)
+                else:
+                    continue
+
+                ratings[item_id_1] = r1_new
+                ratings[item_id_2] = r2_new
+
+                delta_mu.append(abs(r1_new.mu - r1_old.mu))
+                delta_mu.append(abs(r2_new.mu - r2_old.mu))
+
+            avg_delta = sum(delta_mu) / len(delta_mu)
+            mu_history.append(avg_delta)
+
+            epoch_bar.set_postfix(avg_delta_mu=f"{avg_delta:.5f}")
+
+            if avg_delta < early_stop_threshold:
+                logger.debug(f"Early stopped at epoch {epoch} (avg Δmu: {avg_delta:.5f})")
+                break
+            # Store results
+        for item in items:
+            item.score = ratings[str(item.item_id)].mu
+            item.mu = ratings[str(item.item_id)].mu
+            item.sigma = ratings[str(item.item_id)].sigma
+
+        return items
+
+    def ranking(self, 
+                   items: List[Dict], 
+                   num_comparisons_per_round: int = 50,
+                   num_rounds: int = 3,
+                   use_cache: bool = True) -> List[Dict]:
+        """Get global ranking of items using iterative TrueSkill model.
+        
+        Args:
+            items: List of items to rank, each item should have 'item_id' field
+            num_comparisons_per_round: Number of comparisons to make in each round
+            num_rounds: Number of ranking rounds    
+            
+        Returns:
+            List of items sorted by their final estimated scores
+        """
+        logger.info(f"Starting ranking process with {len(items)} items, "
+                    f"{num_comparisons_per_round} comparisons per round, "
+                    f"{num_rounds} rounds")
+        if num_rounds <= 0:
+            logger.info("num_rounds <= 0, return the current ruler_items")
+            return [item.item for item in self.ruler_items]
+
+        # Convert input items to RulerItems
+        all_items = [
+                RulerItem(
+                    item_id=str(item['item_id']),
+                    item=item,
+                    metadata={}
+                )
+                for item in items
+            ]
+        with self._ruler_lock:
+            self.ruler_items[:] = all_items
+
+        # Track consecutive rounds with max_rank_change = 0
+        consecutive_zero_changes = 0
+        required_consecutive_zeros = 20
+
+        for round_idx in range(num_rounds):
+            logger.info(f"Starting ranking round {round_idx + 1}/{num_rounds}")
+
+            # Sample pairs for comparison
+            pairs = self._sample_pairs(num_comparisons_per_round, all_items)
+            
+            paired_item = [(item1.item, item2.item) for item1, item2 in pairs]
+
+            # Perform comparisons
+            self.pairwise_comparison.compare_pairs(paired_item, use_cache=use_cache)
+
+            # Fit model and get new scores
+            new_items = self.fit(all_items)
+
+            # Check convergence
+            current_scores = self.get_current_scores()
+            new_scores = {item.item_id: item.score for item in new_items}
+            if current_scores and self.convergence_threshold is not None:
+                # Calculate ranks based on scores
+                current_ranked_items = sorted(
+                    [(item_id, score) for item_id, score in current_scores.items()],
+                    key=lambda x: x[1],
+                    reverse=True
+                )
+                new_ranked_items = sorted(
+                    [(item_id, score) for item_id, score in new_scores.items()],
+                    key=lambda x: x[1],
+                    reverse=True
+                )
+                
+                # Create rank mappings
+                current_ranks = {item_id: rank for rank, (item_id, _) in enumerate(current_ranked_items)}
+                new_ranks = {item_id: rank for rank, (item_id, _) in enumerate(new_ranked_items)}
+                
+                # Calculate rank changes
+                rank_changes = []
+                for item_id in current_ranks:
+                    if item_id in new_ranks:  # Only consider items that have been compared
+                        old_rank = current_ranks.get(item_id, 0)
+                        new_rank = new_ranks[item_id]
+                        rank_changes.append(abs(new_rank - old_rank))
+
+                if rank_changes:  # Only calculate mean if we have valid rank changes
+                    mean_rank_change = np.mean(rank_changes)
+                    max_rank_change = np.max(rank_changes)
+                    
+                    # Track items with their rank changes
+                    item_rank_changes = []
+                    for item_id in current_ranks:
+                        if item_id in new_ranks:
+                            old_rank = current_ranks[item_id]
+                            new_rank = new_ranks[item_id]
+                            rank_change = abs(new_rank - old_rank)
+                            item_rank_changes.append((item_id, old_rank, new_rank, rank_change))
+                    
+                    # Sort by rank change and get top 3
+                    top_changes = sorted(item_rank_changes, key=lambda x: x[3], reverse=True)[:3]
+                    
+                    logger.info(f"Round {round_idx + 1} rank changes - Mean: {mean_rank_change:.2f}, Max: {max_rank_change} "
+                              f"(based on {len(rank_changes)} compared items)")
+                    
+                    # Log top 3 changes
+                    logger.info("Top 3 largest rank changes:")
+                    for item_id, old_rank, new_rank, change in top_changes:
+                        logger.info(f"Item {item_id}: {old_rank} -> {new_rank} (change: {change})")
+                    
+                    # Update consecutive zero changes counter
+                    if max_rank_change == 0:
+                        consecutive_zero_changes += 1
+                        logger.info(f"Consecutive rounds with no rank changes: {consecutive_zero_changes}/{required_consecutive_zeros}")
+                    else:
+                        consecutive_zero_changes = 0
+                    
+                    # Consider converged if we have enough consecutive rounds with no rank changes
+                    # or if mean rank change is below threshold
+                    if consecutive_zero_changes >= required_consecutive_zeros or mean_rank_change < self.convergence_threshold:
+                        logger.info(f"Converged after {round_idx + 1} rounds - ranks stabilized")
+                        break
+                else:
+                    logger.info(f"Round {round_idx + 1}: No compared items to evaluate convergence")
+
+            # Update scores
+            with self._ruler_lock:
+                sorted_items = sorted(
+                    new_items,
+                    key=lambda x: x.score,
+                    reverse=True
+                )
+                # Update ranks
+                for rank, item in enumerate(sorted_items):
+                    item.rank = rank
+                self.ruler_items[:] = sorted_items
+
+            # Save state after each round
+            self.save_state()
+
+        self.save_state()
+        logger.info(f"Completed ranking process, returning {len(sorted_items)} sorted items")
+
+        # Convert back to dictionary format
+        return [item.item for item in self.ruler_items]
+
+    def sample_pairs_by_rejection(self, n, k):
+        seen = set()
+        pbar = tqdm(total=k, desc="Sampling pairs", unit="pair", leave=False)
+        while len(seen) < k:
+            i = random.randrange(n)
+            # j 从 [0,n-2] 里挑一个，若 >= i 就跳过 i：
+            j = random.randrange(n-1)
+            if j >= i:
+                j += 1
+            # 保证 (i,j) 有序
+            a, b = (i, j) if i < j else (j, i)
+            if (a, b) not in seen:
+                seen.add((a, b))
+                pbar.update(1)
+        pbar.close()
+        return list(seen)
+
+    def _sample_pairs_by_score_gaps(self, items: List[RulerItem], current_scores: Dict[str, float], 
+                                  comparison_counts: Dict[str, int], pair_comparison_counts: Dict[str, int],
+                                  num_comparisons: int) -> List[Tuple[RulerItem, RulerItem]]:
+        """Sample pairs based on score gaps between adjacent items in the ranking.
+        
+        Args:
+            items: List of items to sample from
+            current_scores: Current scores for all items
+            comparison_counts: Number of comparisons for each item
+            pair_comparison_counts: Number of comparisons for each pair
+            num_comparisons: Number of pairs to sample
+            
+        Returns:
+            List of (item1, item2) pairs to compare
+        """
+        # Sort items by score
+        sorted_items = sorted(items, key=lambda x: current_scores.get(x.item_id, 0), reverse=True)
+        
+        # Calculate score differences between adjacent items
+        score_diffs = []
+        for i in range(len(sorted_items) - 1):
+            item1 = sorted_items[i]
+            item2 = sorted_items[i + 1]
+            score1 = current_scores.get(item1.item_id, 0)
+            score2 = current_scores.get(item2.item_id, 0)
+            score_diff = abs(score1 - score2)
+            score_diffs.append((score_diff, item1, item2))
+        
+        if not score_diffs:
+            return []
+            
+        # Calculate mean and standard deviation of score differences
+        mean_diff = np.mean([diff for diff, _, _ in score_diffs])
+        std_diff = np.std([diff for diff, _, _ in score_diffs])
+        
+        # Select pairs with score differences significantly larger than mean
+        threshold = mean_diff + std_diff  # Items with gaps larger than mean + 1 std
+        candidate_pairs = []
+        
+        for diff, item1, item2 in score_diffs:
+            if diff > threshold:
+                pair_key = self.pairwise_comparison.get_pair_key(item1.item_id, item2.item_id)
+                # Only consider pairs that haven't been compared too many times
+                if pair_comparison_counts.get(pair_key, 0) < self.pairwise_comparison.max_comparisons_per_pair:
+                    candidate_pairs.append((diff, item1, item2))
+        
+        # Sort by score difference and take top num_comparisons
+        candidate_pairs.sort(reverse=True)
+        return [(item1, item2) for _, item1, item2 in candidate_pairs[:num_comparisons]]
+
+    def _sample_pairs_randomly(self, items: List[RulerItem], num_pairs: int, 
+                             pair_comparison_counts: Dict[str, int]) -> List[Tuple[RulerItem, RulerItem]]:
+        """Sample pairs randomly for comparison.
+        
+        Args:
+            items: List of items to sample from
+            num_pairs: Number of pairs to sample
+            pair_comparison_counts: Dictionary of pair comparison counts
+            
+        Returns:
+            List of (item1, item2) pairs to compare
+        """
+        logger.info(f"randomly sample {num_pairs} pairs")
+        n = len(items)
+        random_pairs = []
+        if n < 500:
+            all_idx_pairs = [(i, j) for i in range(n) for j in range(i + 1, n)]
+            valid_idx_pairs = []
+            for i, j in all_idx_pairs:
+                key = self.pairwise_comparison.get_pair_key(items[i].item_id, items[j].item_id)
+                if pair_comparison_counts.get(key, 0) < self.pairwise_comparison.max_comparisons_per_pair:
+                    valid_idx_pairs.append((i, j))
+            if len(valid_idx_pairs) == 0:
+                return []
+            random.shuffle(valid_idx_pairs)
+            selected_idx_pairs = valid_idx_pairs[:min(num_pairs, len(valid_idx_pairs))]
+            for i, j in selected_idx_pairs:
+                random_pairs.append((items[i], items[j]))
+        else:
+        
+            while len(random_pairs) < num_pairs:
+                i, j = random.sample(range(n), 2)
+                if i < j and (i, j) not in random_pairs:
+                    pair_key = self.pairwise_comparison.get_pair_key(
+                        items[i].item_id, 
+                        items[j].item_id
+                    )
+                    if pair_comparison_counts.get(pair_key, 0) < self.pairwise_comparison.max_comparisons_per_pair:
+                        random_pairs.append((items[i], items[j]))
+                    
+        return random_pairs
+
+    def _sample_pairs(self, num_comparisons: int, items: List[RulerItem]) -> List[Tuple[Dict, Dict]]:
+        """Sample pairs for comparison based on current scores and comparison history."""
+        n = len(items)
+        epsilon = 1e-6
+        # Get comparison counts from pairwise_comparison
+        comparison_info = self.pairwise_comparison.get_compare_information(
+            [item.item_id for item in items]
+        )
+        comparison_counts = comparison_info['comparison_counts']
+        pair_comparison_counts = comparison_info['pair_comparison_counts']
+
+        # Calculate sampling quotas
+        num_random_pairs = max(1, int(num_comparisons * 0.10))  # 20% random
+        num_thompson_pairs = max(1, int(num_comparisons * 0.30))  # 40% Thompson
+        # num_adaptive_pairs = num_comparisons - num_random_pairs - num_thompson_pairs
+
+        # Get random pairs
+        random_pairs = self._sample_pairs_randomly(items, num_random_pairs, pair_comparison_counts)
+
+        current_scores = self.get_current_scores()
+        if not current_scores:
+            logger.info("First round or no scores available: using random sampling")
+            remaining_random_pairs = self._sample_pairs_randomly(
+                items,
+                num_comparisons - len(random_pairs),
+                pair_comparison_counts
+            )
+            return random_pairs + remaining_random_pairs
+
+        # Get Thompson sampling pairs
+        thompson_pairs = self._sample_pairs_by_thompson(items, num_thompson_pairs, pair_comparison_counts)
+
+        num_adaptive_pairs = num_comparisons - num_random_pairs - len(thompson_pairs)
+
+        # Get adaptive sampling pairs
+        total_comparisons = sum(comparison_counts.values())
+        avg_comparisons_per_item = total_comparisons / len(items) if items else 0
+
+        if avg_comparisons_per_item >= 10:
+            logger.info("Using score gap based sampling")
+            adaptive_pairs = self._sample_pairs_by_score_gaps(
+                items, current_scores, comparison_counts, pair_comparison_counts, num_adaptive_pairs
+            )
+            
+            if len(adaptive_pairs) < num_adaptive_pairs:
+                logger.info(f"Score gap sampling only found {len(adaptive_pairs)} pairs, falling back to uncertainty sampling")
+                remaining_pairs = self._sample_pairs_by_uncertainty(
+                    items, current_scores, comparison_counts, pair_comparison_counts,
+                    num_adaptive_pairs - len(adaptive_pairs)
+                )
+                adaptive_pairs.extend(remaining_pairs)
+        else:
+            logger.info("Using adaptive sampling based on current scores")
+            adaptive_pairs = self._sample_pairs_by_uncertainty(
+                items, current_scores, comparison_counts, pair_comparison_counts, num_adaptive_pairs
+            )
+
+        # Combine all sampling strategies
+        return random_pairs + thompson_pairs + adaptive_pairs
+
+    def _sample_pairs_by_uncertainty(self, items: List[RulerItem], current_scores: Dict[str, float],
+                                   comparison_counts: Dict[str, int], pair_comparison_counts: Dict[str, int],
+                                   num_comparisons: int) -> List[Tuple[RulerItem, RulerItem]]:
+        """Sample pairs based on uncertainty in their relative ordering.
+        
+        Args:
+            items: List of items to sample from
+            current_scores: Current scores for all items
+            comparison_counts: Number of comparisons for each item
+            pair_comparison_counts: Number of comparisons for each pair
+            num_comparisons: Number of pairs to sample
+            
+        Returns:
+            List of (item1, item2) pairs to compare
+        """
+        n = len(items)
+        epsilon = 1e-6  # Small constant to avoid division by zero
+        
+        # Use heap to maintain top-k pairs with highest uncertainty scores
+        import heapq
+        top_pairs = []  # Heap to store (score, i, j)
+        
+        # Randomly sample a subset of pairs for evaluation
+        sample_size = min(num_comparisons * 1000, n * (n - 1) // 2)  # Sample at most 2000x num_comparisons pairs
+        logger.info(f"Sample size is {sample_size}")
+
+        if n <= 5000:
+            candidate_pairs = random.sample([(i, j) for i in range(n) for j in range(i + 1, n)], sample_size)
+        else:
+            candidate_pairs = self.sample_pairs_by_rejection(n, sample_size)
+        for i, j in candidate_pairs:
+            id1 = items[i].item_id
+            id2 = items[j].item_id
+            pair_key = self.pairwise_comparison.get_pair_key(id1, id2)
+
+            if pair_comparison_counts.get(pair_key, 0) >= self.pairwise_comparison.max_comparisons_per_pair:
+                continue
+
+            # Get comparison counts for both items
+            count_i = comparison_counts.get(id1, 0)
+            count_j = comparison_counts.get(id2, 0)
+
+            # Calculate score difference term
+            score1 = current_scores.get(id1, 0)
+            score2 = current_scores.get(id2, 0)
+            score_diff = abs(score1 - score2)
+            score_term = 1 / (score_diff + epsilon)
+
+            # Calculate comparison count penalty term
+            count_term = 1 / np.sqrt(1 + count_i + count_j)
+
+            # Calculate final uncertainty score
+            uncertainty_score = score_term * count_term
+
+            # Use heap to maintain top-k
+            if len(top_pairs) < num_comparisons:
+                heapq.heappush(top_pairs, (uncertainty_score, i, j))
+            elif uncertainty_score > top_pairs[0][0]:
+                heapq.heapreplace(top_pairs, (uncertainty_score, i, j))
+
+        # Get final pairs from heap
+        return [(items[i], items[j]) for _, i, j in sorted(top_pairs, reverse=True)]
+
+    def estimate_new_item_score(self, new_item: RulerItem, ruler_items: List[RulerItem], num_of_epoch: int = 10) -> float:
+        """Estimate score for a new item while keeping existing item scores fixed.
+        
+        Args:
+            new_item: New item to estimate score for
+            ruler_items: List of existing ruler items with fixed scores
+            
+        Returns:
+            float: Estimated score (mu) for the new item
+        """
+        # Get comparison results for the new item
+        all_item_ids = [item.item_id for item in ruler_items] + [new_item.item_id]
+        comparison_info = self.pairwise_comparison.get_compare_information(all_item_ids)
+        comparison_results = comparison_info['comparison_results']
+        
+        # Create fixed ratings for existing items
+        fixed_ratings = {
+            item.item_id: Rating(mu=item.mu, sigma=item.sigma)
+            for item in ruler_items
+        }
+        
+        # Create initial rating for new item
+        new_rating = Rating(mu=50, sigma=50/3)
+        
+        # Process each comparison result
+        item_ids = set([])
+        data = []
+        for pair_key, results in comparison_info['comparison_results'].items():
+            id1, id2 = pair_key.split('_')
+            if id1 != new_item.item_id and id2 != new_item.item_id:
+                continue
+            item_ids.add(id1)
+            item_ids.add(id2)
+
+            if results:
+                votes = {1: 0, -1: 0, 0: 0}
+                for result in results:
+                    orig_id1, orig_id2 = result['original_ids']
+                    if (orig_id1, orig_id2) == (id1, id2):
+                        votes[result['judgment']] += 1
+                    else:
+                        votes[-result['judgment']] += 1
+
+                final_judgment = max(votes.items(), key=lambda x: x[1])[0]
+
+                if id1 == new_item.item_id:
+                    new_item_id = id1
+                    old_item_id = id2
+                elif id2 == new_item.item_id:
+                    new_item_id = id2
+                    old_item_id = id1
+                    final_judgment = -final_judgment
+                else:
+                    assert False, "New item is not in this pair"
+
+                if final_judgment > 0:
+                    data.append((new_item_id, old_item_id, "win"))
+                elif final_judgment < 0:
+                    data.append((new_item_id, old_item_id, "loose"))
+                else:
+                    data.append((new_item_id, old_item_id, "draw"))
+                
+        if not data:
+            logger.warning(f"No valid comparison data for new item {new_item.item_id}; return initial rating")
+            return new_rating.mu, new_rating.sigma
+
+        epoch_bar = tqdm(range(num_of_epoch), desc="Epochs", disable=True)
+        for epoch in epoch_bar:
+            random.shuffle(data)
+            delta_mu = []
+            new_rating_old = new_rating
+            for new_item_id, old_item_id, outcome in tqdm(data, desc=f"Epoch {epoch}", leave=False, disable=True):
+                if outcome == "win":
+                    new_rating, _ = rate_1vs1(new_rating, fixed_ratings[old_item_id])
+                elif outcome == "loose":
+                    _, new_rating = rate_1vs1(fixed_ratings[old_item_id], new_rating)
+                else:
+                    new_rating, _ = rate_1vs1(new_rating, fixed_ratings[old_item_id], drawn=True)
+                
+                delta_mu.append(abs(new_rating.mu - new_rating_old.mu))
+                
+            if not delta_mu:
+                logger.warning(f"No rating updates for new item {new_item.item_id}; stop fitting")
+                break
+            avg_delta = sum(delta_mu) / len(delta_mu)
+            epoch_bar.set_postfix(avg_delta_mu=f"{avg_delta:.5f}")
+            
+            if avg_delta < self.convergence_threshold:
+                logger.debug(f"Early stopped at epoch {epoch} (avg Δmu: {avg_delta:.5f})")
+                break
+
+        return new_rating.mu, new_rating.sigma
+
+
+    def _similarity_search(self, new_item: RulerItem, num_rounds: int, initial_comparisons: int, ruler_items: List[RulerItem]) -> Tuple[int, List[Dict]]:
+        round_candidates = []
+        
+        for round_idx in range(num_rounds):
+            logger.debug(f"Starting round {round_idx + 1}/{num_rounds}")
+            round_info = {
+                'round': round_idx + 1,
+                'candidates': [],
+                'sampling_method': None
+            }
+            
+            if round_idx == 0:
+                # First round: compare with midpoint
+                midpoint_idx = len(ruler_items) // 2
+                midpoint_item = ruler_items[midpoint_idx]
+                candidates = [(midpoint_idx, midpoint_item)]
+                
+                round_info['sampling_method'] = 'midpoint_comparison'
+                round_info['candidates'] = [{'id': midpoint_item.item_id, 'position': midpoint_idx, 
+                                            'score': midpoint_item.score, 'rank': midpoint_item.rank}]
+                
+            else:
+                # Find items with most similar quality (closest to 1 in quality_1vs1)
+                quality_values = []
+                for i, item in enumerate(ruler_items):
+                    pair_key = self.pairwise_comparison.get_pair_key(new_item.item_id, item.item_id)
+                    if self.pairwise_comparison.get_comparison_count(pair_key) > 0:
+                        continue
+                        
+                    quality = quality_1vs1(new_item, item)
+                    quality_values.append((i, item, quality))
+                
+                # Sort by quality (highest first)
+                quality_values.sort(key=lambda x: -x[2])
+                candidates = [(i, item) for i, item, _ in quality_values[:initial_comparisons]]
+                
+                if not candidates:
+                    logger.info("No more pairs to compare, proceeding with final fitting")
+                    break
+                
+
+
+                round_info['sampling_method'] = 'quality_based'
+                round_info['candidates'] = [{'id': c[1].item_id, 'position': c[0], 
+                                            'score': c[1].score, 'rank': c[1].rank} for c in candidates]
+
+
+                # for debug
+                # round_info['quality_values'] = [(x[0], x[2]) for x in quality_values]
+            
+           
+            # Perform comparisons
+            pairs = [(new_item.item, candidate[1].item) for candidate in candidates]
+            self.pairwise_comparison.compare_pairs(pairs)
+            
+            # Estimate new item score
+            new_item.mu, new_item.sigma = self.estimate_new_item_score(new_item, ruler_items)
+            round_info['item_trueskill'] = {'mu': new_item.mu, 'sigma': new_item.sigma}
+            round_candidates.append(round_info)
+        return round_idx, round_candidates
+
+        
+    def _heuristic_search(self, new_item: RulerItem, num_rounds: int, initial_comparisons: int,
+                          ruler_items: List[RulerItem], start_pos: int = None) -> Tuple[int, List[Dict]]:
+        # First round: uniform sampling from different score ranges
+        round_candidates = []
+
+        for round_idx in range(num_rounds):
+            logger.debug(f"Starting round {round_idx + 1}/{num_rounds}")
+            round_info = {
+                'round': round_idx + 1,
+                'candidates': [],
+                'sampling_method': None
+            }
+
+            if round_idx == 0:
+                candidates = []
+                if start_pos is not None:
+                    # warm-start：直接拿 embedding 估的位置当首轮唯一候选
+                    idx = max(0, min(len(ruler_items) - 1, int(start_pos)))
+                    temp_item = ruler_items[idx]
+                    pair_key = self.pairwise_comparison.get_pair_key(new_item.item_id, temp_item.item_id)
+                    if self.pairwise_comparison.get_comparison_count(pair_key) == 0:
+                        candidates = [(idx, temp_item)]
+                    round_info['sampling_method'] = 'heuristic_warm_start'
+                else:
+                    # 原均匀分段采样
+                    num_comparisons = initial_comparisons
+                    segment_size = max(1, len(ruler_items) // num_comparisons)
+                    for i in range(0, len(ruler_items), segment_size):
+                        segment = ruler_items[i:i + segment_size]
+                        if segment:
+                            for temp_item in segment:
+                                pair_key = self.pairwise_comparison.get_pair_key(new_item.item_id, temp_item.item_id)
+                                if self.pairwise_comparison.get_comparison_count(pair_key) == 0:
+                                    candidates.append((i, temp_item))
+                                    break
+                    round_info['sampling_method'] = 'uniform_segments'
+
+                round_info['candidates'] = [{'id': c[1].item_id, 'position': c[0], 'score': c[1].score, 'rank': c[1].rank}
+                                            for c in candidates]
+                logger.debug(f"Round {round_idx + 1} sampling info:")
+                logger.debug(f"  Method: {round_info['sampling_method']}")
+                logger.debug(f"  Total candidates: {len(candidates)}")
+                logger.debug("  Candidates details:")
+                for i, candidate in enumerate(round_info['candidates']):
+                    logger.debug(f"    {i+1}. ID: {candidate['id']}, Position: {candidate['position']}, Score: {candidate['score']:.4f}")
+            else:
+                # Find items with similar scores
+                score_diffs = []
+                for i, item in enumerate(ruler_items):
+                    # Skip if this pair has already been compared
+                    pair_key = self.pairwise_comparison.get_pair_key(new_item.item_id, item.item_id)
+                    if self.pairwise_comparison.get_comparison_count(pair_key) > 0:
+                        continue
+                        
+                    score_diff = abs(item.score - new_item.score)
+                    score_diffs.append((i, item, score_diff))
+                
+                # Sort by score difference and take top candidates
+                score_diffs.sort(key=lambda x: x[2])
+                candidates = [(i, item) for i, item, _ in score_diffs[:initial_comparisons]]
+                
+                # If no candidates to compare, we're done
+                if not candidates:
+                    logger.info("No more pairs to compare, proceeding with final fitting")
+                    break
+                
+                round_info['sampling_method'] = 'score_based'
+                round_info['candidates'] = [{'id': c[1].item_id, 'position': c[0], 'score': c[1].score, 'rank': c[1].rank}
+                                            for c in candidates]
+                logger.debug(f"Round {round_idx + 1} sampling info:")
+                logger.debug(f"  Method: {round_info['sampling_method']}")
+                logger.debug(f"  Current new item score: {new_item.score:.4f}")
+                logger.debug(f"  Total candidates: {len(candidates)}")
+                logger.debug("  Candidates details:")
+                for i, candidate in enumerate(round_info['candidates']):
+                    logger.debug(f"    {i+1}. ID: {candidate['id']}, Position: {candidate['position']}, Score: {candidate['score']:.4f}")
+            
+            
+            
+            # Perform comparisons
+            pairs = [(new_item.item, candidate[1].item) 
+                    for candidate in candidates]
+            self.pairwise_comparison.compare_pairs(pairs)
+            
+            # Estimate new item score while keeping existing scores fixed
+            new_item.score, new_item.sigma = self.estimate_new_item_score(new_item, ruler_items)
+            round_info['item_trueskill'] = {'mu': new_item.score, 'sigma': new_item.sigma}
+            round_candidates.append(round_info)
+        return round_idx, round_candidates
+
+    def _binary_search(self, new_item: RulerItem, ruler_items: List[RulerItem],
+                       start_pos: int = None) -> Tuple[int, List[Dict]]:
+        left, right = 0, len(ruler_items) - 1
+        round_candidates = []
+        round_idx = 0
+        while left <= right:
+
+            logger.debug(f"Starting round {round_idx + 1}")
+            round_info = {
+                'round': round_idx + 1,
+                'candidates': [],
+                'sampling_method': 'binary_search'
+            }
+
+            # warm-start：第一轮 mid 用 start_pos（夹到 [left, right]）
+            if round_idx == 0 and start_pos is not None:
+                mid = max(left, min(right, int(start_pos)))
+                round_info['sampling_method'] = 'binary_search_warm_start'
+            else:
+                mid = (left + right) // 2
+            mid_item = ruler_items[mid]
+
+            pair_key = self.pairwise_comparison.get_pair_key(new_item.item_id, mid_item.item_id)
+            if self.pairwise_comparison.get_comparison_count(pair_key) == 0:
+                candidates = [(mid, mid_item)]
+                round_info['candidates'] = [{'id': mid_item.item_id, 'position': mid, 'score': mid_item.score, 'rank': mid_item.rank}]
+            else:
+                candidates = []
+            logger.debug(f"Round {round_idx + 1} sampling info:")
+            logger.debug(f"  Method: {round_info['sampling_method']}")
+            logger.debug(f"  Total candidates: {len(candidates)}")
+            logger.debug("  Candidates details:")
+            for i, candidate in enumerate(round_info['candidates']):
+                logger.debug(f"    {i+1}. ID: {candidate['id']}, Position: {candidate['position']}, Score: {candidate['score']:.4f}")
+
+            # round_candidates.append(round_info)
+
+            # Perform comparisons
+            if candidates:
+                pairs = [(new_item.item, candidate[1].item) for candidate in candidates]
+                self.pairwise_comparison.compare_pairs(pairs)
+
+                # Estimate new item score while keeping existing scores fixed
+                new_item.score, new_item.sigma = self.estimate_new_item_score(new_item, ruler_items)
+
+
+            if new_item.score <= mid_item.score:  # new_item is less violative than mid_item
+                left = mid + 1
+            else: 
+                right = mid - 1
+
+
+            round_idx += 1
+
+            round_info['item_trueskill'] = {'mu': new_item.score, 'sigma': new_item.sigma}
+            round_candidates.append(round_info)
+        return round_idx, round_candidates
+
+
+
+    def _full_traversal(self, new_item: RulerItem, ruler_items: List[RulerItem]) -> Tuple[int, List[Dict]]:
+        all_candidates = []
+        round_idx = 0
+        round_candidates = []
+        for idx, item in enumerate(ruler_items):
+            logger.debug(f"Starting round {round_idx + 1}")
+            round_info = {
+                'round': round_idx + 1,
+                'candidates': [],
+                'sampling_method': 'full_traversal'
+            }
+            candidates = []
+            pair_key = self.pairwise_comparison.get_pair_key(new_item.item_id, item.item_id)
+            if self.pairwise_comparison.get_comparison_count(pair_key) == 0:
+                all_candidates.append((idx, item))
+                candidates = [(idx, item)]
+                round_info['candidates'] = [{'id': item.item_id, 'position': idx, 'score': item.score, 'rank': item.rank}]
+
+            logger.debug(f"Round {round_idx + 1} sampling info:")
+            logger.debug(f"  Method: {round_info['sampling_method']}")
+            logger.debug("  Candidates details:")
+            for i, candidate in enumerate(round_info['candidates']):
+                logger.debug(f"    {i+1}. ID: {candidate['id']}, Position: {candidate['position']}, Score: {candidate['score']:.4f}")
+
+            round_candidates.append(round_info)
+
+        pairs = [(new_item.item, candidate[1].item) for candidate in all_candidates]
+        self.pairwise_comparison.compare_pairs(pairs)
+        return round_idx, round_candidates
+
+
+    def find_sample_index(self,
+                         new_item: Dict,
+                         initial_comparisons: int = 5,
+                         num_rounds: int = 3,
+                         search_method: str = 'binary_search',
+                         start_pos: int = None) -> Dict:
+        """Find the insertion position for a new sample in the existing ranking.
+
+        Args:
+            new_item: New item to find position for, should have 'item_id' field
+            initial_comparisons: Number of initial random comparisons
+            num_rounds: Number of comparison rounds
+            search_method: Specify search method, 'binary_search' or 'heuristic_search'
+            start_pos: Warm-start position for binary/heuristic search. If None,
+                       fall back to ruler midpoint / segment-uniform sampling.
+            
+        Returns:
+            Dict containing:
+                - position: int, insertion position (0-based index)
+                - score: float, final score of the new item
+                - comparison_results: Dict, comparison results for the new item
+                - comparison_counts: Dict, comparison counts for the new item
+                - rounds_completed: int, number of rounds completed
+                - converged: bool, whether the score has converged
+                - round_candidates: List[Dict], candidates selected in each round
+        """
+        logger.info(f"Starting position finding process for new item {new_item['item_id']}")
+        new_item = RulerItem(
+            item_id=str(new_item['item_id']),
+            item=new_item,
+            metadata={}
+        )
+        # Get ruler items
+        with self._ruler_lock:
+            ruler_items = list(self.ruler_items)
+        
+        if not ruler_items:
+            raise ValueError("No ruler items available. Please run get_ranking first.")
+        
+
+        if search_method == 'heuristic_search':
+            round_idx, round_candidates = self._heuristic_search(new_item, num_rounds, initial_comparisons, ruler_items, start_pos=start_pos)
+        elif search_method == 'binary_search':
+            round_idx, round_candidates = self._binary_search(new_item, ruler_items, start_pos=start_pos)
+        elif search_method == 'full_traversal':
+            round_idx, round_candidates = self._full_traversal(new_item, ruler_items)
+        elif search_method == 'similarity_search':
+            round_idx, round_candidates = self._similarity_search(new_item, num_rounds, initial_comparisons, ruler_items)
+        else:
+            raise ValueError(f"Unknown search method: {search_method}")
+
+
+        # Estimate new item score while keeping existing scores fixed
+        new_item.score, new_item.sigma = self.estimate_new_item_score(new_item, ruler_items)
+        # Sort items by score
+        all_items = ruler_items + [new_item]
+        sorted_items = sorted(all_items, key=lambda x: x.score, reverse=True)
+        insertion_pos = next(i for i, item in enumerate(sorted_items) 
+                            if item.item_id == new_item.item_id)
+        
+        # Get adjacent items if they exist
+        prev_item = sorted_items[insertion_pos - 1] if insertion_pos > 0 else None
+        next_item = sorted_items[insertion_pos + 1] if insertion_pos < len(sorted_items) - 1 else None
+        
+        logger.debug(f"Final insertion position: {insertion_pos}, score: {new_item.score:.4f}")
+        if prev_item:
+            logger.debug(f"Previous item: ID={prev_item.item_id}, Score={prev_item.score:.4f}")
+            logger.debug(f"Previous item details: {json.dumps(prev_item.item, ensure_ascii=False, indent=2)}")
+        if next_item:
+            logger.debug(f"Next item: ID={next_item.item_id}, Score={next_item.score:.4f}")
+            logger.debug(f"Next item details: {json.dumps(next_item.item, ensure_ascii=False, indent=2)}")
+        logger.debug(f"Current item: ID={new_item.item_id}, Score={new_item.score:.4f}")
+        logger.debug(f"Current item details: {json.dumps(new_item.item, ensure_ascii=False, indent=2)}")
+        
+        # Get comparison results for the new item
+        all_item_ids = [item.item_id for item in ruler_items] + [new_item.item_id]
+        all_comparison_info = self.pairwise_comparison.get_compare_information(all_item_ids)
+        
+        # Filter comparison results to only include pairs with new_item
+        filtered_comparison_results = {
+            pair_key: results 
+            for pair_key, results in all_comparison_info['comparison_results'].items()
+            if new_item.item_id in pair_key.split('_')
+        }
+        
+        # Filter comparison counts to only include new_item
+        filtered_comparison_counts = {
+            item_id: count 
+            for item_id, count in all_comparison_info['comparison_counts'].items()
+            if item_id == new_item.item_id
+        }
+        
+        # Filter pair comparison counts to only include pairs with new_item
+        filtered_pair_comparison_counts = {
+            pair_key: count 
+            for pair_key, count in all_comparison_info['pair_comparison_counts'].items()
+            if new_item.item_id in pair_key.split('_')
+        }
+        
+        return {
+            'position': insertion_pos,
+            'score': new_item.score,
+            'sigma': new_item.sigma,
+            'comparison_results': filtered_comparison_results,
+            'comparison_counts': filtered_comparison_counts,
+            'pair_comparison_counts': filtered_pair_comparison_counts,
+            'rounds_completed': round_idx + 1,
+            'round_candidates': round_candidates
+        }
+
+    def find_sample_index_binary_search(self, new_item: Dict, comparison_time: int = 1
+                         ) -> Dict:
+        """Find the insertion position for a new sample in the existing ranking with binary_search
+        
+        Args:
+            new_item: New item to find position for, should have 'item_id' field
+    
+        Returns:
+            Dict containing:
+                - position: int, insertion position (0-based index)
+                - comparison_results: Dict, comparison results for the new item
+                - comparison_counts: Dict, comparison counts for the new item
+        """
+        logger.info(f"Starting position finding process for new item {new_item['item_id']}")
+        new_item = RulerItem(
+            item_id=str(new_item['item_id']),
+            item=new_item,
+            metadata={}
+        )
+        
+        # Get ruler items
+        with self._ruler_lock:
+            ruler_items = list(self.ruler_items)
+        # 留一法，排出掉自身
+        temp_ruler = [item for item in ruler_items if item.item_id != new_item.item_id]
+        
+        if not temp_ruler:
+            raise ValueError("No ruler items available. Please run get_ranking first.")
+        try:
+            left = 0
+            right = len(temp_ruler)
+            comparison_pairs = []
+            while left < right:
+                mid = (left + right) // 2
+                ref_item = temp_ruler[mid]
+                pair = (new_item.item, ref_item.item)
+                # 不用cache，因为这里是二分查找，每次都要比较
+                result = self.pairwise_comparison.compare_single_pair(pair, use_cache=False)
+                judgment = result.get("judgment")
+                item_id_a = result.get("item_id_a")
+                item_id_b = result.get("item_id_b")
+                # 判断new_item_id在A还是B
+                if item_id_a == new_item.item_id:
+                    # new_item是A
+                    if judgment == 1:
+                        is_new_more_severe = True
+                    elif judgment == -1:
+                        is_new_more_severe = False
+                    else:  # judgment == 0
+                        is_new_more_severe = False  # 默认插后面
+                elif item_id_b == new_item.item_id:
+                    # new_item是B
+                    if judgment == 1:
+                        is_new_more_severe = False
+                    elif judgment == -1:
+                        is_new_more_severe = True
+                    else:  # judgment == 0
+                        is_new_more_severe = False
+                else:
+                    logger.error(f"Neither item_id_a nor item_id_b matches new_item_id: {new_item.item_id}")
+                    raise ValueError("Pairwise comparison result does not match new_item_id.")
+                comparison_pairs.append({
+                    "new_item_id": new_item.item_id,
+                    "ref_item_id": ref_item.item_id,
+                    "ref_item_rank": mid,
+                    "judgment": judgment,
+                    "item_id_a": item_id_a,
+                    "item_id_b": item_id_b,
+                    "result": result,
+                    "is_new_more_severe": is_new_more_severe
+                })
+                if is_new_more_severe:
+                    right = mid
+                else:
+                    left = mid + 1
+            logger.info(f"Final insertion position for new item {new_item.item_id}: {left}")
+            return {
+                "position": left,  # 0-based插入点
+                "comparison_pairs": comparison_pairs
+            }
+        except Exception as e:
+            logger.error(f"Error in find_sample_index_binary_search: {str(e)}")
+            raise
+
+    def _calculate_thompson_params(self, item_ids: List[str]) -> Dict[str, Dict[str, float]]:
+        """Calculate Thompson Sampling parameters for items based on their comparison history.
+        
+        Args:
+            item_ids: List of item IDs to calculate parameters for
+            
+        Returns:
+            Dictionary mapping item IDs to their Thompson parameters (alpha, beta)
+        """
+        # Get comparison results for all items
+        comparison_info = self.pairwise_comparison.get_compare_information(item_ids)
+        comparison_results = comparison_info['comparison_results']
+        
+        # Initialize parameters for all items
+        thompson_params = {item_id: {'alpha': 1.0, 'beta': 1.0} for item_id in item_ids}
+        
+        # Update parameters based on comparison results
+        for pair_key, results in comparison_results.items():
+            if not results:
+                continue
+                
+            # Parse item IDs from pair_key
+            id1, id2 = pair_key.split('_')
+            if id1 not in item_ids or id2 not in item_ids:
+                continue
+                
+            # Count votes for each judgment
+            votes = {1: 0, -1: 0, 0: 0}
+            for result in results:
+                orig_id1, orig_id2 = result['original_ids']
+                if (orig_id1, orig_id2) == (id1, id2):
+                    votes[result['judgment']] += 1
+                else:
+                    votes[-result['judgment']] += 1
+            
+            # Get the judgment with the most votes
+            final_judgment = max(votes.items(), key=lambda x: x[1])[0]
+            
+            # Update parameters based on final judgment
+            if final_judgment > 0:
+                thompson_params[id1]['alpha'] += 1.0
+                thompson_params[id2]['beta'] += 1.0
+            elif final_judgment < 0:
+                thompson_params[id2]['alpha'] += 1.0
+                thompson_params[id1]['beta'] += 1.0
+        
+        return thompson_params
+
+    def _sample_pairs_by_thompson(self, items: List[RulerItem], num_pairs: int,
+                                pair_comparison_counts: Dict[str, int]) -> List[Tuple[RulerItem, RulerItem]]:
+        """Sample pairs using Thompson Sampling.
+        
+        Args:
+            items: List of items to sample from
+            num_pairs: Number of pairs to sample
+            pair_comparison_counts: Dictionary of pair comparison counts
+            
+        Returns:
+            List of (item1, item2) pairs to compare
+        """
+        n = len(items)
+        thompson_pairs = []
+        
+        # Calculate Thompson parameters for all items
+        item_ids = [item.item_id for item in items]
+        thompson_params = self._calculate_thompson_params(item_ids)
+        
+        # Sample from Beta distributions
+        item_scores = {}
+        for item in items:
+            params = thompson_params[item.item_id]
+            # Sample from Beta distribution
+            score = np.random.beta(params['alpha'], params['beta'])
+            item_scores[item.item_id] = score
+        
+        # Sort items by Thompson scores
+        sorted_items = sorted(items, key=lambda x: item_scores[x.item_id], reverse=True)
+        
+        # Select pairs based on Thompson scores
+        for i in range(min(n-1, num_pairs)):
+            item1 = sorted_items[i]
+            item2 = sorted_items[i+1]
+            
+            pair_key = self.pairwise_comparison.get_pair_key(
+                item1.item_id,
+                item2.item_id
+            )
+            
+            if pair_comparison_counts.get(pair_key, 0) < self.pairwise_comparison.max_comparisons_per_pair:
+                thompson_pairs.append((item1, item2))
+        
+        return thompson_pairs
+
+    def understand_ruler_marks(self):
+        item_ids = [item.item_id for item in self.ruler_items]
+        compare_information = self.pairwise_comparison.get_compare_information(item_ids)
+
+        def _run_one_understanding(item, comparison_results):
+
+            if "understanding" in item.metadata:
+                return item.metadata["understanding"]
+
+            understanding = self.pairwise_comparison.understand_by_pairs(item.item, compare_information["comparison_results"])
+            return understanding
+
+        with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
+            futures = []
+            all_items = list(self.ruler_items)
+            for item in all_items:
+                future = executor.submit(_run_one_understanding, item, comparison_results=compare_information["comparison_results"])
+                futures.append((future, item))
+
+            # Get successfully compared pairs
+            successful_items = []
+            for i, (future, item) in tqdm(enumerate(futures), desc="Generating understanding", total=len(futures), leave=False):
+                try:
+                    result = future.result()
+                    if result:  # If comparison was successful
+                        item.metadata["understanding"] = result
+
+                    if i % 10 == 0:
+                        with self._ruler_lock:
+                            self.ruler_items[:] = all_items
+                        self.save_state()
+                except Exception as e:
+                    logger.error(f"Error generating understanding: {str(e)}")
+        
+        # Save the remainder of i%10
+        with self._ruler_lock:
+            self.ruler_items[:] = all_items
+        self.save_state()
+
+        #return successful_pairs
+
+        # i = 0
+        # all_items = list(self.ruler_items)
+        # for item in tqdm(all_items):
+        #     i += 1
+        #     if "understanding" in item.metadata:
+        #         continue
+        #     understanding = self.pairwise_comparison.understand_by_pairs(item.item, compare_information["comparison_results"])
+        #     item.metadata["understanding"] = understanding
+        #     if i % 10 == 0:
+        #         with self._ruler_lock:
+        #             self.ruler_items[:] = all_items
+        #         self.save_state()
+
+        self.save_state()
+
+    
+    def generate_ranking_reasoning(self):
+        """Generate ranking reasoning for all items in the ranking."""
+        # Get all items in the ranking
+        all_items = list(self.ruler_items)
+        for i in range(len(all_items)):
+            item = all_items[i]
+            prev_item = all_items[i-1] if i > 0 else None
+            next_item = all_items[i+1] if i < len(all_items) - 1 else None
+            if prev_item:
+                comparison_result = self.pairwise_comparison.compare_single_pair(
+                    (item.item, prev_item.item)
+                )
+                original_id_a,  original_id_b = comparison_result['original_ids']
+                prev_conflit = False
+                if original_id_a == item.item_id:
+                    a_str = "current"
+                    b_str = "previous"
+                    if comparison_result['judgment'] > 0:
+                        prev_conflit = True
+                else:
+                    a_str = "previous"
+                    b_str = "current"
+                    if comparison_result['judgment'] < 0:
+                        prev_conflit = True
+                item.metadata["prev_item_compare_result"] = f"""
+Comment A({a_str}): {original_id_a}
+Comment B({b_str}): {original_id_b}
+
+{comparison_result['raw_response']}
+"""
+                item.metadata["prev_item_conflict"] = prev_conflit
+            if next_item:
+                comparison_result = self.pairwise_comparison.compare_single_pair(
+                    (item.item, next_item.item)
+                )
+                original_id_a, original_id_b = comparison_result['original_ids']
+                next_conflit = False
+                if original_id_a == item.item_id:
+                    a_str = "current"
+                    b_str = "next"
+                    if comparison_result['judgment'] < 0:
+                        next_conflit = True
+                else:
+                    a_str = "next"
+                    b_str = "current"
+                    if comparison_result['judgment'] > 0:
+                        next_conflit = True
+                item.metadata["next_item_compare_result"] = f"""
+                Comment A({a_str}): {original_id_a}
+                Comment B({b_str}): {original_id_b}
+                {comparison_result['raw_response']}
+                """
+                
+                item.metadata["next_item_conflict"] = next_conflit
+        with self._ruler_lock:
+            self.ruler_items[:] = all_items
+        self.save_state()
+
+    def compare_with_anchor(self, samples: List[Dict], anchor_rank: int=None, anchor_score=None, reuse_compair=True) -> List[Dict]:
+        """Compare samples with an anchor sample at specified rank.
+        
+        Args:
+            samples: List of samples to compare with anchor
+            anchor_rank: Rank of the anchor sample (0-based)
+            
+        Returns:
+            List of dictionaries containing:
+                - sample: Original sample
+                - is_better_than_anchor: Whether sample is better than anchor
+                - comparison_result: Raw comparison result
+        """
+        # Get anchor sample
+        with self._ruler_lock:
+            anchor_item = None
+            if anchor_rank is not None:
+                anchor_item = next((item for item in self.ruler_items if item.rank == anchor_rank), None)
+                if anchor_item is None:
+                    raise ValueError(f"No item found with rank {anchor_rank}")
+            elif anchor_score is not None:
+                for item in sorted(self.ruler_items, key=lambda x: x.score, reverse=True):
+                    if item.score < anchor_score:
+                        anchor_item = item
+                        break
+                if anchor_item is None:
+                    raise ValueError(f"No item found with score less than {anchor_score}")
+            else:
+                raise ValueError("Either anchor_rank or anchor_score must be provided")
+
+        # Get history 
+        sample_ids = [str(s['item_id']) for s in samples]
+        comparison_results = self.pairwise_comparison.load_data_to_cache_from_es(sample_ids)
+
+        results = []
+        to_compare = []
+        
+
+        for sample in samples:
+            sample_id = str(sample['item_id'])
+            inferred = False
+
+            if reuse_compair:
+                for other in self.ruler_items:
+                    if other.item_id == anchor_item.item_id:
+                        continue  # 跳过锚点自身
+                    # 获取 sample vs other 的历史比对
+                    key = self.pairwise_comparison.get_pair_key(sample_id, other.item_id)
+                    history = comparison_results.get(key, [])
+                    if not history:
+                        continue
+
+                    # 统计多数票，判断 sample 相对 other 的关系
+                    votes = {1: 0, -1: 0, 0: 0}
+                    for r in history:
+                        id1, id2 = r['original_ids']
+                        # 注意方向：若顺序颠倒，需要取反
+                        if (id1, id2) == (sample_id, other.item_id):
+                            votes[r['judgment']] += 1
+                        else:
+                            votes[-r['judgment']] += 1
+                    direction, count = max(votes.items(), key=lambda x: x[1])
+
+                    # 传递性推断
+                    if other.rank > anchor_item.rank and direction < 0:
+                        # other 在锚点“之后”且 sample < other => sample < anchor
+                        # other 风险比 anchor 低，other 分数比 anchor 低。 然后同时 sample 风险比other 也低，所以 sample 风险比 anchor 低
+                        results.append({
+                            'sample': sample,
+                            'final_judgment': -1,
+                            'reason': f"Sample({sample_id}) < Other({other.item_id}) and Other({other.item_id}) < anchor({anchor_item.item_id}). Infereed from {other.item_id}. {history[0]['raw_response']}" 
+                        })
+                        inferred = True
+                    elif other.rank < anchor_item.rank and direction > 0:
+                        # other 在锚点“之前”且 sample > other => sample > anchor
+                        results.append({
+                            'sample': sample,
+                            'final_judgment': 1,
+                            'reason': f"Sample({sample_id}) > Other({other.item_id}) and Other({other.item_id}) > anchor({anchor_item.item_id}). Infereed from {other.item_id}. {history[0]['raw_response']}" 
+                        })
+                        inferred = True
+
+                    if inferred:
+                        break  # 一旦推断成功，跳出 other 循环
+
+            if not inferred:
+                to_compare.append(sample)
+
+        if len(to_compare) > 0:
+            # Prepare pairs for batch comparison
+            pairs = [(sample, anchor_item.item) for sample in to_compare]
+            
+            # Batch compare all samples with anchor
+            self.pairwise_comparison.compare_pairs(pairs)
+            
+            for sample in to_compare:
+                try:
+                    sample_id = str(sample['item_id'])
+                    pair_key = self.pairwise_comparison.get_pair_key(sample_id, anchor_item.item_id)
+                    
+                    # Get comparison results for this pair
+                    pair_results = self.pairwise_comparison.get_comparison_results(pair_key)
+                    # Get the most recent comparison result
+                    if pair_results:
+                        votes = {1: 0, -1: 0, 0: 0}
+                        for result in pair_results:
+                            orig_id1, orig_id2 = result['original_ids']
+                            if (orig_id1, orig_id2) == (sample_id, anchor_item.item_id):
+                                votes[result['judgment']] += 1
+                            else:
+                                votes[-result['judgment']] += 1
+
+                        final_judgment = max(votes.items(), key=lambda x: x[1])[0]
+                        results.append({
+                            'sample': sample,
+                            'final_judgment': final_judgment,
+                            'reason': pair_results[0]['raw_response']
+                        })
+                    
+                except Exception as e:
+                    logger.error(f"Error processing comparison result for sample {sample.get('item_id', '')}: {str(e)}")
+                    raise
+        return results
+
+            
+        
+        
+        
+