5 Best AI Reasoning Models of 2025: Ranked!

import torch
from transformers import AutoTokenizer, AutoModelForCausalLM, BitsAndBytesConfig
import json
import re
import asyncio
import argparse
from typing import Dict, List, Any
from collections import Counter

def parse_evaluation_file(file_path: str) -> (str, Dict[str, str]):
    """
    Parses a text file to extract the original prompt and model responses.
    The first line must be the prompt, starting with "Prompt: ".
    Each subsequent model response should be preceded by its name ending with a colon.
    """
    file_path = "/home/rst/llm-as-a-judge/response.txt"
    try:
        with open(file_path, 'r', encoding='utf-8') as f:
            content = f.read()
    except FileNotFoundError:
        raise FileNotFoundError(f"The file was not found at the specified path: {file_path}")

    # Split the content to find the prompt
    if not content.lower().startswith('prompt:'):
        raise ValueError("File must start with 'Prompt: <your prompt>' on the first line.")
    
    prompt_section, *responses_section = content.split('\n\n', 1)
    original_prompt = prompt_section[len('Prompt:'):].strip()

    if not responses_section:
        raise ValueError("No model responses found after the prompt.")

    # Use regex to find model names followed by a colon at the beginning of a line
    pattern = re.compile(r'^(.*?):\n', re.MULTILINE)
    splits = pattern.split(responses_section[0])
    
    if len(splits) < 3:
        raise ValueError("Could not parse model responses. Ensure each model name is on its own line ending with a colon.")

    responses = {}
    # Iterate through model names and their corresponding responses
    for i in range(1, len(splits), 2):
        model_name = splits[i].strip()
        response_text = splits[i+1].strip()
        responses[model_name] = response_text
        
    return original_prompt, responses


class EnsembleJudge:
    """
    An LLM-as-a-Judge system using an ensemble of smaller models for robust evaluation.
    This implementation loads multiple models with 4-bit quantization to fit within ~15GB of VRAM.
    """
    def __init__(self, model_names: List[str]):
        self.judges = []
        self.quantization_config = BitsAndBytesConfig(
            load_in_4bit=True,
            bnb_4bit_compute_dtype=torch.bfloat16,
            bnb_4bit_use_double_quant=True,
            bnb_4bit_quant_type="nf4"
        )
        print("Loading judge models... This may take a while and consume significant VRAM.")
        for model_name in model_names:
            try:
                print(f"Loading {model_name}...")
                tokenizer = AutoTokenizer.from_pretrained(model_name)
                model = AutoModelForCausalLM.from_pretrained(
                    model_name,
                    quantization_config=self.quantization_config,
                    device_map="auto",
                    trust_remote_code=True
                )
                self.judges.append({"name": model_name, "model": model, "tokenizer": tokenizer})
                print(f"Successfully loaded {model_name}.")
            except Exception as e:
                print(f"Failed to load {model_name}: {e}")
        print("All judge models loaded successfully.")

    def _create_evaluation_prompt(self, original_prompt: str, responses_to_judge: Dict[str, str]) -> str:
        responses_formatted = ""
        for model_name, response in responses_to_judge.items():
            responses_formatted += f"### Response from {model_name}:\n{response}\n\n"

        return f"""You are a fair and impartial AI evaluator. Your task is to evaluate the quality of responses from several AI models based on the user's original question.

Please evaluate the responses on the following criteria:
- **Accuracy & Correctness:** Is the information factually correct and well-supported?
- **Reasoning & Depth:** Does the response demonstrate logical reasoning and a deep, nuanced understanding of the topic?
- **Clarity & Coherence:** Is the response well-structured, easy to read, and coherent?
- **Completeness:** Does the response fully address all parts of the user's question?

**Original Question:**
{original_prompt}

**Responses to Evaluate:**
{responses_formatted}
**Evaluation Task:**
Provide a detailed evaluation for each response, a score from 1 (worst) to 10 (best), and a final ranking. Structure your output as a single JSON object.

{{
  "evaluations": [
    {{
      "model_name": "Name of Model 1",
      "score": <score_1_to_10>,
      "reasoning": "Your detailed reasoning for this model's score."
    }},
    {{
      "model_name": "Name of Model 2",
      "score": <score_2_to_10>,
      "reasoning": "Your detailed reasoning for this model's score."
    }}
  ],
  "ranking": ["Name of Best Model", "Name of Second Best Model"],
  "winner": "Name of Best Model"
}}
"""

    def _parse_json_from_response(self, text: str) -> Dict[str, Any]:
        json_match = re.search(r'\{.*\}', text, re.DOTALL)
        if json_match:
            try:
                return json.loads(json_match.group())
            except json.JSONDecodeError:
                return {"error": "Failed to decode JSON", "raw_response": text}
        return {"error": "No JSON object found", "raw_response": text}

    async def _evaluate_single_judge(self, judge: Dict, prompt: str) -> Dict[str, Any]:
        inputs = judge["tokenizer"](prompt, return_tensors="pt").to(judge["model"].device)
        with torch.no_grad():
            outputs = await asyncio.to_thread(
                judge["model"].generate,
                **inputs,
                max_new_tokens=2048,
                temperature=0.1,
                do_sample=True,
                pad_token_id=judge["tokenizer"].eos_token_id
            )
        response_text = judge["tokenizer"].decode(outputs[0][inputs.input_ids.shape[1]:], skip_special_tokens=True)
        return self._parse_json_from_response(response_text)

    def _aggregate_evaluations(self, individual_evals: List[Dict]) -> Dict[str, Any]:
        all_scores = {}
        all_winners = []
        all_reasonings = {}
        
        for i, eval_result in enumerate(individual_evals):
            judge_name = self.judges[i]['name']
            if "evaluations" in eval_result and isinstance(eval_result["evaluations"], list):
                for item in eval_result["evaluations"]:
                    model_name = item.get("model_name")
                    if model_name:
                        if model_name not in all_scores:
                            all_scores[model_name] = []
                            all_reasonings[model_name] = []
                        all_scores[model_name].append(item.get("score", 0))
                        all_reasonings[model_name].append(f"Judge ({judge_name}): {item.get('reasoning', '')}")
            
            if "winner" in eval_result:
                all_winners.append(eval_result.get("winner"))

        final_scores = {model: round(sum(scores) / len(scores), 2) for model, scores in all_scores.items() if scores}
        
        winner_by_vote = Counter(all_winners).most_common(1)[0][0] if all_winners else "Undetermined"
        
        final_ranking = sorted(final_scores, key=final_scores.get, reverse=True)

        return {
            "winner_by_vote": winner_by_vote,
            "final_ranking_by_score": final_ranking,
            "average_scores": final_scores,
            "detailed_reasoning": all_reasonings
        }

    async def evaluate(self, original_prompt: str, responses_to_judge: Dict[str, str]) -> Dict[str, Any]:
        evaluation_prompt = self._create_evaluation_prompt(original_prompt, responses_to_judge)
        
        print("\nEvaluating responses with the judge ensemble...")
        tasks = [self._evaluate_single_judge(judge, evaluation_prompt) for judge in self.judges]
        individual_evaluations = await asyncio.gather(*tasks)
        
        print("Aggregating results...")
        final_judgment = self._aggregate_evaluations(individual_evaluations)
        return final_judgment


async def main():
    parser = argparse.ArgumentParser(description="Evaluate model responses from a file using an LLM judge ensemble.")
    parser.add_argument("file_path", type=str, help="Path to the text file containing the prompt and responses.")
    args = parser.parse_args()

    ensemble_model_names = [
        "meta-llama/Llama-3-7B-Instruct-hf",
        "deepseek-ai/deepseek-llm-7b-chat",
        "mistralai/Mistral-7B-Instruct-v0.2"
    ]

    try:
        print(f"Parsing file: {args.file_path}")
        original_prompt, responses_to_judge = parse_evaluation_file(args.file_path)
        print("File parsed successfully.")
        print(f"\nPrompt: {original_prompt}")
        print(f"\nModels to evaluate: {list(responses_to_judge.keys())}")

        ensemble_judge = EnsembleJudge(model_names=ensemble_model_names)

        final_evaluation = await ensemble_judge.evaluate(original_prompt, responses_to_judge)

        print("\n--- Final Aggregated Evaluation ---")
        print(json.dumps(final_evaluation, indent=2))

    except (ValueError, FileNotFoundError) as e:
        print(f"Error: {e}")
    except Exception as e:
        print(f"An unexpected error occurred: {e}")

if __name__ == "__main__":
    asyncio.run(main())