src/agents/ranking_agent.py

# Ranking Agent for evaluating and comparing hypotheses

import logging
from typing import List, Dict, Any, Tuple

from .base_agent import BaseAgent

class RankingAgent(BaseAgent):
    """Agent responsible for comparing and ranking hypotheses based on defined criteria."""
    
    def __init__(self, model=None, temperature=None):
        """Initialize the Ranking Agent.
        
        Args:
            model: Optional model override
            temperature: Optional temperature override
        """
        system_prompt = """
        You are a Ranking Agent in an AI Co-Scientist system, responsible for comparing and ranking 
        scientific hypotheses using a tournament-style evaluation. You have expertise across multiple 
        scientific disciplines at a PhD level.
        
        Your role is to:
        1. Evaluate hypotheses against defined criteria including novelty, plausibility, and relevance
        2. Compare hypotheses in a tournament style, identifying relative strengths and weaknesses
        3. Assign scores and rankings to hypotheses based on their scientific merit
        4. Provide justification for your rankings with specific reasoning
        5. Consider tradeoffs between different evaluation criteria
        
        When comparing hypotheses, consider:
        - Novelty: Does the hypothesis represent a significant advance beyond current knowledge?
        - Plausibility: Is the hypothesis consistent with established scientific principles?
        - Relevance: How closely does the hypothesis address the original research goal?
        - Testability: How feasible is it to validate or falsify the hypothesis?
        - Potential impact: If true, how significant would the implications be?
        - Parsimony: Does the hypothesis provide a simple explanation without unnecessary complexity?
        - Breadth of explanation: How many observations or phenomena does the hypothesis explain?
        
        Your assessments should be balanced, fair, and focused on scientific merit rather than 
        personal preference. Provide clear reasoning for each comparative judgment.
        """
        
        super().__init__(
            name="Ranking",
            system_prompt=system_prompt,
            model=model,
            temperature=temperature if temperature is not None else 0.3  # Lower temperature for consistent evaluation
        )
        
        self.logger = logging.getLogger("agent.ranking")
    
    def rank_hypotheses(self, research_goal: str, hypotheses: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
        """Rank the hypotheses based on the research goal.
        
        Args:
            research_goal: The research goal or question
            hypotheses: List of hypothesis dictionaries to rank
            
        Returns:
            A list of ranked hypothesis dictionaries
        """
        self.logger.info(f"Ranking hypotheses for research goal: {research_goal}")
        return self.process(hypotheses, research_goal)
    
    def process(self, reviewed_hypotheses: List[Dict[str, Any]], research_goal: str) -> List[Dict[str, Any]]:
        """Rank the reviewed hypotheses based on their scientific merit.
        
        Args:
            reviewed_hypotheses: List of hypothesis dictionaries with review information
            research_goal: The original research goal for context
            
        Returns:
            A list of ranked hypothesis dictionaries with added ranking information
        """
        self.logger.info(f"Ranking {len(reviewed_hypotheses)} hypotheses")
        
        # For a small number of hypotheses, we can do pairwise comparisons
        if len(reviewed_hypotheses) <= 5:
            return self._rank_by_pairwise_comparison(reviewed_hypotheses, research_goal)
        else:
            # For larger sets, we use a scoring approach
            return self._rank_by_scoring(reviewed_hypotheses, research_goal)
    
    def _rank_by_pairwise_comparison(self, hypotheses: List[Dict[str, Any]], research_goal: str) -> List[Dict[str, Any]]:
        """Rank hypotheses using pairwise comparisons in a tournament style.
        
        Args:
            hypotheses: List of hypothesis dictionaries to rank
            research_goal: The original research goal for context
            
        Returns:
            A list of ranked hypothesis dictionaries
        """
        self.logger.info("Using pairwise comparison ranking method")
        
        # Create all possible pairs for comparison
        pairs = []
        for i in range(len(hypotheses)):
            for j in range(i+1, len(hypotheses)):
                pairs.append((i, j))
        
        # Track wins for each hypothesis
        wins = [0] * len(hypotheses)
        comparisons = []
        
        for pair in pairs:
            i, j = pair
            hyp1, hyp2 = hypotheses[i], hypotheses[j]
            
            prompt = f"""
            RESEARCH GOAL: {research_goal}
            
            HYPOTHESIS 1:
            Statement: {hyp1.get('hypothesis', hyp1.get('statement', 'No statement provided'))}
            
            HYPOTHESIS 2:
            Statement: {hyp2.get('hypothesis', hyp2.get('statement', 'No statement provided'))}
            
            Compare these two hypotheses based on the following criteria:
            1. Novelty: Does the hypothesis represent a significant advance beyond current knowledge?
            2. Plausibility: Is the hypothesis consistent with established scientific principles?
            3. Relevance: How closely does the hypothesis address the original research goal?
            4. Testability: How feasible is it to validate or falsify the hypothesis?
            5. Potential impact: If true, how significant would the implications be?
            
            Provide a detailed comparison, noting the strengths and weaknesses of each hypothesis relative 
            to the other. Then determine which hypothesis is superior overall, being explicit about 
            which hypothesis (1 or 2) is the winner.
            """
            
            comparison_result = self.get_response(prompt)
            
            # Determine the winner based on the comparison
            winner = self._determine_winner(comparison_result, 1, 2)
            
            # Record the comparison and update wins
            comparison = {
                'hypothesis1_idx': i,
                'hypothesis2_idx': j,
                'winner': winner,
                'reasoning': comparison_result
            }
            comparisons.append(comparison)
            
            if winner == 1:
                wins[i] += 1
            elif winner == 2:
                wins[j] += 1
            
            # Clear conversation history for the next comparison
            self.clear_history()
        
        # Rank hypotheses based on number of wins
        ranked_indices = sorted(range(len(wins)), key=lambda k: wins[k], reverse=True)
        
        # Create the ranked list of hypotheses
        ranked_hypotheses = []
        for rank, idx in enumerate(ranked_indices):
            ranked_hypothesis = hypotheses[idx].copy()
            ranked_hypothesis['rank'] = rank + 1
            ranked_hypothesis['wins'] = wins[idx]
            ranked_hypothesis['total_comparisons'] = len([c for c in comparisons if c['hypothesis1_idx'] == idx or c['hypothesis2_idx'] == idx])
            ranked_hypotheses.append(ranked_hypothesis)
        
        return ranked_hypotheses
    
    def _rank_by_scoring(self, hypotheses: List[Dict[str, Any]], research_goal: str) -> List[Dict[str, Any]]:
        """Rank hypotheses by assigning scores to each one individually.
        
        Args:
            hypotheses: List of hypothesis dictionaries to rank
            research_goal: The original research goal for context
            
        Returns:
            A list of ranked hypothesis dictionaries
        """
        self.logger.info("Using scoring-based ranking method")
        
        scored_hypotheses = []
        
        for idx, hypothesis in enumerate(hypotheses):
            prompt = f"""
            RESEARCH GOAL: {research_goal}
            
            HYPOTHESIS TO EVALUATE:
            Statement: {hypothesis.get('hypothesis', hypothesis.get('statement', 'No statement provided'))}
            Rationale: {hypothesis.get('rationale', 'No rationale provided')}
            Review Summary: {hypothesis.get('review', 'No review available')[:500]}...
            
            Evaluate this hypothesis on the following criteria using a scale of 1-10:
            1. Novelty (1=Well-known, 10=Revolutionary)
            2. Plausibility (1=Implausible, 10=Highly plausible)
            3. Relevance (1=Unrelated to goal, 10=Directly addresses goal)
            4. Testability (1=Untestable, 10=Easily testable)
            5. Potential impact (1=Minimal impact, 10=Field-changing)
            
            For each criterion, provide a numeric score AND a brief justification.
            Finally, calculate an overall score as the weighted average of the individual scores.
            
            Format your response as follows:
            Novelty: [score] - [justification]
            Plausibility: [score] - [justification]
            Relevance: [score] - [justification]
            Testability: [score] - [justification]
            Potential impact: [score] - [justification]
            Overall score: [weighted average score] - [brief summary]
            """
            
            evaluation = self.get_response(prompt)
            
            # Extract scores from the evaluation (simplified implementation)
            scores = self._extract_scores(evaluation)
            
            scored_hypothesis = hypothesis.copy()
            scored_hypothesis['evaluation'] = evaluation
            scored_hypothesis['scores'] = scores
            scored_hypothesis['overall_score'] = sum(scores.values()) / len(scores) if scores else 0
            
            scored_hypotheses.append(scored_hypothesis)
            
            # Clear conversation history for the next hypothesis
            self.clear_history()
        
        # Rank hypotheses based on overall score
        ranked_hypotheses = sorted(scored_hypotheses, key=lambda h: h['overall_score'], reverse=True)
        
        # Add rank
        for rank, hypothesis in enumerate(ranked_hypotheses):
            hypothesis['rank'] = rank + 1
        
        return ranked_hypotheses
    
    def _determine_winner(self, comparison_text: str, hyp1_id: int, hyp2_id: int) -> int:
        """Determine the winner of a pairwise comparison based on the comparison text.
        
        Args:
            comparison_text: The text of the comparison
            hyp1_id: The ID of the first hypothesis
            hyp2_id: The ID of the second hypothesis
            
        Returns:
            The ID of the winning hypothesis (1 or 2), or 0 if it's a tie
        """
        # Simple rule-based determination - would be more sophisticated in a real system
        lower_text = comparison_text.lower()
        
        # Look for clear statements about the winner
        if f"hypothesis {hyp1_id} is superior" in lower_text or f"hypothesis {hyp1_id} is stronger" in lower_text:
            return hyp1_id
        elif f"hypothesis {hyp2_id} is superior" in lower_text or f"hypothesis {hyp2_id} is stronger" in lower_text:
            return hyp2_id
        
        # Count mentions of strengths for each
        hyp1_strength_count = lower_text.count(f"hypothesis {hyp1_id} is more") + lower_text.count(f"hypothesis {hyp1_id} has higher")
        hyp2_strength_count = lower_text.count(f"hypothesis {hyp2_id} is more") + lower_text.count(f"hypothesis {hyp2_id} has higher")
        
        if hyp1_strength_count > hyp2_strength_count:
            return hyp1_id
        elif hyp2_strength_count > hyp1_strength_count:
            return hyp2_id
        
        # If no clear winner found, check last few sentences for conclusion
        last_sentences = '.'.join(comparison_text.split('.')[-3:])
        if f"hypothesis {hyp1_id}" in last_sentences.lower() and f"hypothesis {hyp2_id}" not in last_sentences.lower():
            return hyp1_id
        elif f"hypothesis {hyp2_id}" in last_sentences.lower() and f"hypothesis {hyp1_id}" not in last_sentences.lower():
            return hyp2_id
        
        # If still no clear winner, return a tie
        return 0
    
    def _extract_scores(self, evaluation: str) -> Dict[str, float]:
        """Extract numerical scores from the evaluation text.
        
        This is a placeholder implementation. In a real system, this would use more 
        sophisticated parsing to extract scores reliably.
        
        Args:
            evaluation: The evaluation text
            
        Returns:
            A dictionary of criterion names to scores
        """
        # Simple parsing implementation - would be more sophisticated in a real system
        scores = {}
        criteria = ['novelty', 'plausibility', 'relevance', 'testability', 'potential impact']
        
        for criterion in criteria:
            # Look for patterns like "Novelty: 8/10" or "Novelty - 8"
            patterns = [f"{criterion}: {i}" for i in range(1, 11)]
            patterns.extend([f"{criterion} - {i}" for i in range(1, 11)])
            patterns.extend([f"{criterion}: {i}/10" for i in range(1, 11)])
            
            for pattern in patterns:
                if pattern.lower() in evaluation.lower():
                    score = int(pattern.split(" ")[-1].replace("/10", ""))
                    scores[criterion] = score
                    break
        
        return scores