| import json |
| import logging |
| import os |
| import re |
|
|
| import httpx |
|
|
| from config import MINIMAX_BASE_URL |
|
|
| logger = logging.getLogger(__name__) |
|
|
| ALIGNMENT_RUBRIC_VERSION = "alignment-v1" |
| ALIGNMENT_DIMENSIONS = ( |
| "intent_fulfillment", |
| "constraint_adherence", |
| "scope_discipline", |
| "output_format_fit", |
| "groundedness_no_fabrication", |
| ) |
|
|
| ALIGNMENT_SYSTEM_PROMPT = ( |
| "You are an evaluation judge for coding-agent benchmark runs.\n" |
| "Evaluate how well the run outcome matches the user's original request.\n" |
| "Return strict JSON with keys:\n" |
| "{" |
| '"rubric_version":"alignment-v1",' |
| '"dimension_scores":{"intent_fulfillment":0|0.5|1,"constraint_adherence":0|0.5|1,' |
| '"scope_discipline":0|0.5|1,"output_format_fit":0|0.5|1,' |
| '"groundedness_no_fabrication":0|0.5|1},' |
| '"alignment_score":0.0-1.0,' |
| '"judge_confidence":0.0-1.0,' |
| '"summary":"short",' |
| '"failure_mode":"short or empty"' |
| "}\n" |
| "Use only 0, 0.5, or 1 for dimension scores.\n" |
| "alignment_score must be the arithmetic mean of the five dimension scores." |
| ) |
|
|
|
|
| def build_alignment_prompt(run_data: dict) -> str: |
| parts = [ |
| f"Task title: {run_data.get('task_title', 'unknown')}", |
| f"User prompt: {run_data.get('task_description', '')[:2500]}", |
| f"Functional success: {run_data.get('functional_success', 0)}", |
| f"Tests passed: {run_data.get('tests_passed', 0)}/{run_data.get('tests_total', 0)}", |
| f"Functional notes: {run_data.get('functional_notes', '')[:1000]}", |
| f"Judge notes: {run_data.get('judge_notes', '')[:1500]}", |
| ] |
|
|
| diff_text = (run_data.get("diff_text") or "")[:3500] |
| if diff_text: |
| parts.append(f"Diff:\n```\n{diff_text}\n```") |
|
|
| raw_output = (run_data.get("raw_output") or "")[:1500] |
| if raw_output: |
| parts.append(f"Agent raw output excerpt:\n```\n{raw_output}\n```") |
|
|
| return "\n\n".join(parts) |
|
|
|
|
| def normalize_alignment_result(result: dict) -> dict: |
| dimensions = result.get("dimension_scores", {}) if isinstance(result, dict) else {} |
| clean = {} |
| for key in ALIGNMENT_DIMENSIONS: |
| value = dimensions.get(key, 0) |
| try: |
| value = float(value) |
| except (TypeError, ValueError): |
| value = 0.0 |
| if value not in (0.0, 0.5, 1.0): |
| value = 0.0 |
| clean[key] = value |
|
|
| alignment_score = sum(clean.values()) / len(ALIGNMENT_DIMENSIONS) |
| try: |
| judge_confidence = float(result.get("judge_confidence", 0.0)) |
| except (TypeError, ValueError): |
| judge_confidence = 0.0 |
|
|
| return { |
| "rubric_version": result.get("rubric_version") or ALIGNMENT_RUBRIC_VERSION, |
| "dimension_scores": clean, |
| "alignment_score": round(alignment_score, 4), |
| "judge_confidence": max(0.0, min(1.0, judge_confidence)), |
| "summary": (result.get("summary") or "")[:500], |
| "failure_mode": (result.get("failure_mode") or "")[:500], |
| } |
|
|
|
|
| async def llm_alignment_judge(run_data: dict) -> dict: |
| api_key = os.environ.get("MINIMAX_API_KEY", "") |
| if not api_key: |
| return {} |
|
|
| prompt = build_alignment_prompt(run_data) |
|
|
| try: |
| async with httpx.AsyncClient(timeout=30) as client: |
| resp = await client.post( |
| f"{MINIMAX_BASE_URL}/chat/completions", |
| headers={ |
| "Authorization": f"Bearer {api_key}", |
| "Content-Type": "application/json", |
| }, |
| json={ |
| "model": "MiniMax-M2.5", |
| "messages": [ |
| {"role": "system", "content": ALIGNMENT_SYSTEM_PROMPT}, |
| {"role": "user", "content": prompt}, |
| ], |
| "response_format": {"type": "json_object"}, |
| }, |
| ) |
| resp.raise_for_status() |
| content = resp.json()["choices"][0]["message"]["content"] |
| content = re.sub(r"<think>.*?</think>\s*", "", content, flags=re.DOTALL) |
| result = json.loads(content) |
| normalized = normalize_alignment_result(result) |
| normalized["provider"] = "minimax" |
| return normalized |
| except Exception as exc: |
| logger.warning("Alignment judge failed: %s", exc) |
| return {} |
|
|
