Upload patchjudge/judge.py with huggingface_hub

patchjudge/judge.py

@@ -0,0 +1,441 @@
+"""PatchJudge LLM Judge — Core evaluation engine.
+
+Evaluates AI-generated code patches on 5 dimensions using an LLM,
+producing a MergeScore (0-100) that indicates merge-readiness.
+
+Uses HuggingFace Inference API with structured JSON output.
+"""
+
+import json
+import os
+import re
+import time
+import logging
+from typing import Optional
+
+from huggingface_hub import InferenceClient
+
+from patchjudge.models import (
+    PatchExample, PatchFeatures, DimensionScore, JudgeResult
+)
+from patchjudge.feature_extractor import FeatureExtractor
+
+logger = logging.getLogger(__name__)
+
+
+# ============================================================================
+# Prompt Templates
+# ============================================================================
+
+JUDGE_SYSTEM_PROMPT = """You are PatchJudge, an expert senior software engineer evaluating whether an AI-generated code patch is truly merge-worthy — not just whether it passes tests.
+
+You must be HARSH and PRECISE. A patch that "works but is bad code" should score low. A patch that is clean, complete, and genuinely solves the root cause should score high.
+
+Most AI-generated patches that pass tests are NOT merge-worthy. Average scores should be 3-5, not 7-8. A score of 7+ means genuinely good, publishable code."""
+
+
+JUDGE_USER_PROMPT = """Evaluate this AI-generated code patch.
+
+## THE ISSUE:
+{problem_statement}
+
+## THE PATCH (diff):
+```diff
+{agent_patch}
+```
+
+## REFERENCE GOLD PATCH (human-written):
+```diff
+{gold_patch}
+```
+
+## EXTRACTED FEATURES:
+{features_summary}
+
+## TEST RESULT: {test_result}
+
+---
+
+Score the patch on each of these 5 dimensions (0-10 integer each):
+
+1. **CORRECTNESS** (weight: 30%): Does the patch address the ROOT CAUSE described in the issue? Would the issue be genuinely resolved for all described scenarios, not just the test cases?
+
+2. **COMPLETENESS** (weight: 20%): Does it handle edge cases? Is error handling added where appropriate? Are there TODO comments or placeholder logic left behind?
+
+3. **CODE QUALITY** (weight: 20%): Does the code follow the project's existing style? Is it readable, well-structured? No unnecessary complexity?
+
+4. **NON-REGRESSION RISK** (weight: 15%): Is the change scope appropriate? Could it break unrelated functionality? Does it modify shared interfaces unnecessarily?
+
+5. **MERGE-READINESS** (weight: 15%): Would a senior engineer approve this PR as-is? Score 8+ = approve, 5-7 = request changes, below 5 = reject.
+
+---
+
+Respond with ONLY this JSON (no other text):
+```json
+{{
+  "correctness": {{"score": <0-10>, "reasoning": "<2-4 sentences>", "flags": ["<issue1>", ...]}},
+  "completeness": {{"score": <0-10>, "reasoning": "<2-4 sentences>", "flags": ["<issue1>", ...]}},
+  "code_quality": {{"score": <0-10>, "reasoning": "<2-4 sentences>", "flags": ["<issue1>", ...]}},
+  "non_regression_risk": {{"score": <0-10>, "reasoning": "<2-4 sentences>", "flags": ["<issue1>", ...]}},
+  "merge_readiness": {{"score": <0-10>, "reasoning": "<2-4 sentences>", "flags": ["<issue1>", ...]}}
+}}
+```"""
+
+
+FEATURES_TEMPLATE = """- Files changed: {num_files_changed}
+- Lines added: {num_lines_added}, removed: {num_lines_removed}
+- Hunks: {num_hunks}
+- Change scope: {change_scope}
+- Added functions: {added_functions}
+- Modified functions: {modified_functions}
+- Error handling present: {has_error_handling}
+- Edge case handling: {has_edge_case_handling}
+- Has TODOs/FIXMEs: {has_todos}
+- Has hardcoded values: {has_hardcoded_values}
+- Has debug statements: {has_debug_statements}
+- Modifies core files: {modifies_core_files}
+- New imports: {new_imports}
+- Issue keyword coverage: {keyword_coverage_ratio:.0%}
+- Touches test files: {touches_tests}
+- Style violations: {style_violations}"""
+
+
+# ============================================================================
+# PatchJudge Class
+# ============================================================================
+
+class PatchJudge:
+    """LLM-based judge for evaluating AI-generated code patches."""
+    
+    WEIGHTS = {
+        "correctness": 0.30,
+        "completeness": 0.20,
+        "code_quality": 0.20,
+        "non_regression_risk": 0.15,
+        "merge_readiness": 0.15,
+    }
+    
+    DIMENSIONS = list(WEIGHTS.keys())
+    
+    def __init__(
+        self,
+        model_id: str = "Qwen/Qwen2.5-Coder-32B-Instruct",
+        provider: str = "auto",
+        temperature: float = 0.1,
+        max_tokens: int = 2000,
+        max_retries: int = 3,
+        retry_delay: float = 2.0,
+        max_context_chars: int = 12000,
+    ):
+        """Initialize PatchJudge.
+        
+        Args:
+            model_id: HF model ID to use for judging.
+            provider: Inference provider ('auto', 'cerebras', 'novita', etc.)
+            temperature: Low for consistency (0.1 recommended).
+            max_tokens: Max tokens for LLM response.
+            max_retries: Retries on API/parse failures.
+            retry_delay: Seconds between retries.
+            max_context_chars: Max chars for patch/context in prompt.
+        """
+        token = os.environ.get("HF_TOKEN")
+        self.client = InferenceClient(
+            provider=provider,
+            api_key=token,
+        )
+        self.model_id = model_id
+        self.temperature = temperature
+        self.max_tokens = max_tokens
+        self.max_retries = max_retries
+        self.retry_delay = retry_delay
+        self.max_context_chars = max_context_chars
+        self.feature_extractor = FeatureExtractor()
+    
+    def judge(
+        self,
+        example: PatchExample,
+        features: Optional[PatchFeatures] = None,
+    ) -> JudgeResult:
+        """Evaluate a single patch example.
+        
+        Args:
+            example: The patch to evaluate.
+            features: Pre-extracted features (extracted automatically if None).
+            
+        Returns:
+            JudgeResult with MergeScore, dimension scores, and reasoning.
+        """
+        # Extract features if not provided
+        if features is None:
+            features = self.feature_extractor.extract(example)
+        
+        # Format the prompt
+        features_summary = self._format_features(features)
+        
+        # Truncate patches if needed
+        agent_patch = self._truncate(example.agent_patch, self.max_context_chars // 2)
+        gold_patch = self._truncate(example.gold_patch, self.max_context_chars // 4)
+        problem_stmt = self._truncate(example.problem_statement, self.max_context_chars // 4)
+        
+        user_prompt = JUDGE_USER_PROMPT.format(
+            problem_statement=problem_stmt,
+            agent_patch=agent_patch,
+            gold_patch=gold_patch,
+            features_summary=features_summary,
+            test_result="PASSED ✓" if example.test_passed else "FAILED ✗",
+        )
+        
+        # Call LLM with retries
+        raw_output = None
+        scores = None
+        
+        for attempt in range(self.max_retries):
+            try:
+                raw_output = self._call_llm(user_prompt)
+                scores = self._parse_json_output(raw_output)
+                self._validate_scores(scores)
+                break
+            except Exception as e:
+                logger.warning(
+                    f"Attempt {attempt+1}/{self.max_retries} failed: {e}"
+                )
+                if attempt < self.max_retries - 1:
+                    time.sleep(self.retry_delay * (attempt + 1))
+        
+        if scores is None:
+            # Return a failure result
+            logger.error(
+                f"Failed to judge {example.instance_id} after {self.max_retries} attempts"
+            )
+            scores = {
+                dim: {"score": 0, "reasoning": "Judge failed to produce valid output", "flags": ["JUDGE_ERROR"]}
+                for dim in self.DIMENSIONS
+            }
+            raw_output = raw_output or "ERROR: No output from LLM"
+        
+        # Compute MergeScore
+        merge_score = self._compute_merge_score(scores)
+        
+        return JudgeResult(
+            merge_score=merge_score,
+            dimension_scores=scores,
+            raw_output=raw_output,
+            features=features,
+            model_used=self.model_id,
+        )
+    
+    def judge_batch(
+        self,
+        examples: list[PatchExample],
+        features_list: Optional[list[PatchFeatures]] = None,
+        show_progress: bool = True,
+    ) -> list[JudgeResult]:
+        """Evaluate a batch of patches.
+        
+        Args:
+            examples: List of PatchExamples to evaluate.
+            features_list: Pre-extracted features (one per example). Optional.
+            show_progress: Print progress.
+            
+        Returns:
+            List of JudgeResults in same order as input.
+        """
+        results = []
+        
+        for i, example in enumerate(examples):
+            if show_progress:
+                print(f"  Judging [{i+1}/{len(examples)}] {example.instance_id} "
+                      f"({example.agent_name})...")
+            
+            features = features_list[i] if features_list else None
+            
+            try:
+                result = self.judge(example, features)
+                results.append(result)
+                
+                if show_progress:
+                    print(f"    MergeScore: {result.merge_score:.1f}/100")
+                    
+            except Exception as e:
+                logger.error(f"Failed to judge {example.instance_id}: {e}")
+                # Append error result
+                results.append(JudgeResult(
+                    merge_score=0.0,
+                    dimension_scores={
+                        dim: {"score": 0, "reasoning": f"Error: {str(e)}", "flags": ["ERROR"]}
+                        for dim in self.DIMENSIONS
+                    },
+                    raw_output=f"ERROR: {str(e)}",
+                    model_used=self.model_id,
+                ))
+            
+            # Rate limiting
+            time.sleep(0.5)
+        
+        return results
+    
+    # =========================================================================
+    # Internal methods
+    # =========================================================================
+    
+    def _call_llm(self, user_prompt: str) -> str:
+        """Call the LLM and return raw text response."""
+        response = self.client.chat_completion(
+            model=self.model_id,
+            messages=[
+                {"role": "system", "content": JUDGE_SYSTEM_PROMPT},
+                {"role": "user", "content": user_prompt},
+            ],
+            max_tokens=self.max_tokens,
+            temperature=self.temperature,
+        )
+        return response.choices[0].message.content
+    
+    def _compute_merge_score(self, scores: dict) -> float:
+        """Compute weighted MergeScore (0-100) from dimension scores."""
+        weighted_sum = 0.0
+        for dim, weight in self.WEIGHTS.items():
+            dim_score = scores.get(dim, {}).get("score", 0)
+            weighted_sum += dim_score * weight
+        return round(weighted_sum * 10, 1)  # Scale 0-10 → 0-100
+    
+    def _parse_json_output(self, raw: str) -> dict:
+        """Extract JSON from LLM output, handling markdown code blocks."""
+        # Try to find JSON in code blocks
+        json_match = re.search(r'```(?:json)?\s*([\{][\s\S]*?[\}])\s*```', raw)
+        if json_match:
+            return json.loads(json_match.group(1))
+        
+        # Try to find raw JSON object
+        json_match = re.search(r'(\{[\s\S]*\})', raw)
+        if json_match:
+            # Try parsing progressively larger substrings
+            text = json_match.group(1)
+            try:
+                return json.loads(text)
+            except json.JSONDecodeError:
+                pass
+            
+            # Try to find balanced braces
+            depth = 0
+            for i, ch in enumerate(text):
+                if ch == '{':
+                    depth += 1
+                elif ch == '}':
+                    depth -= 1
+                    if depth == 0:
+                        try:
+                            return json.loads(text[:i+1])
+                        except json.JSONDecodeError:
+                            continue
+        
+        raise ValueError(f"Could not parse JSON from LLM output: {raw[:200]}...")
+    
+    def _validate_scores(self, scores: dict) -> None:
+        """Validate that all required dimensions are present with valid scores."""
+        for dim in self.DIMENSIONS:
+            if dim not in scores:
+                raise ValueError(f"Missing dimension: {dim}")
+            if "score" not in scores[dim]:
+                raise ValueError(f"Missing score for {dim}")
+            score = scores[dim]["score"]
+            if not isinstance(score, (int, float)) or score < 0 or score > 10:
+                raise ValueError(f"Invalid score for {dim}: {score}")
+            # Ensure score is int
+            scores[dim]["score"] = int(round(score))
+            # Ensure flags is a list
+            if "flags" not in scores[dim]:
+                scores[dim]["flags"] = []
+            if isinstance(scores[dim]["flags"], str):
+                scores[dim]["flags"] = [scores[dim]["flags"]]
+            # Ensure reasoning exists
+            if "reasoning" not in scores[dim]:
+                scores[dim]["reasoning"] = ""
+    
+    def _format_features(self, features: PatchFeatures) -> str:
+        """Format features into a readable summary for the prompt."""
+        d = features.to_dict()
+        # Format lists as comma-separated
+        for key in ['added_functions', 'modified_functions', 'new_imports',
+                     'style_violations', 'issue_keywords_addressed',
+                     'issue_components_mentioned']:
+            if isinstance(d.get(key), list):
+                d[key] = ', '.join(str(x) for x in d[key][:10]) or 'none'
+        
+        return FEATURES_TEMPLATE.format(**d)
+    
+    def _truncate(self, text: str, max_chars: int) -> str:
+        """Truncate text, keeping beginning and end."""
+        if len(text) <= max_chars:
+            return text
+        half = max_chars // 2
+        return text[:half] + "\n\n... [truncated] ...\n\n" + text[-half:]
+
+
+# ============================================================================
+# Convenience functions
+# ============================================================================
+
+def quick_judge(
+    problem_statement: str,
+    agent_patch: str,
+    gold_patch: str = "",
+    test_passed: bool = True,
+    model_id: str = "Qwen/Qwen2.5-Coder-32B-Instruct",
+) -> JudgeResult:
+    """Quick one-shot evaluation of a patch.
+    
+    Args:
+        problem_statement: The GitHub issue text.
+        agent_patch: The AI-generated diff.
+        gold_patch: Optional reference patch.
+        test_passed: Whether tests passed.
+        model_id: LLM to use.
+        
+    Returns:
+        JudgeResult with MergeScore and breakdown.
+    """
+    example = PatchExample(
+        instance_id="quick-judge",
+        repo="unknown",
+        problem_statement=problem_statement,
+        gold_patch=gold_patch,
+        agent_patch=agent_patch,
+        agent_name="unknown",
+        test_passed=test_passed,
+        base_commit="",
+    )
+    
+    judge = PatchJudge(model_id=model_id)
+    return judge.judge(example)
+
+
+if __name__ == "__main__":
+    logging.basicConfig(level=logging.INFO)
+    
+    # Quick test with a sample
+    result = quick_judge(
+        problem_statement="Fix the divide by zero error in calculate_average when the list is empty",
+        agent_patch="""diff --git a/utils.py b/utils.py
+--- a/utils.py
++++ b/utils.py
+@@ -10,4 +10,6 @@
+ def calculate_average(numbers):
+-    return sum(numbers) / len(numbers)
++    if not numbers:
++        return 0.0
++    return sum(numbers) / len(numbers)
+""",
+        gold_patch="""diff --git a/utils.py b/utils.py
+--- a/utils.py
++++ b/utils.py
+@@ -10,4 +10,7 @@
+ def calculate_average(numbers):
++    if not numbers:
++        raise ValueError("Cannot calculate average of empty list")
+     return sum(numbers) / len(numbers)
+""",
+        test_passed=True,
+    )
+    
+    print(result.summary())