V2 merge: purpose_agent/meta_rewarding.py

purpose_agent/meta_rewarding.py

@@ -0,0 +1,211 @@
+"""
+meta_rewarding.py — Self-improving critic via meta-judge loop.
+
+From Meta-Rewarding LLMs (arxiv:2407.19594):
+  The Purpose Function judges agent actions. But who judges the judge?
+  A meta-judge evaluates the Purpose Function's own judgments, creating
+  preference pairs (good judgment vs bad judgment) that improve the critic.
+
+Adaptation for Purpose Agent (no weight updates):
+  Instead of DPO fine-tuning, we store high-quality judgment examples
+  as critic_calibration memories. The Purpose Function's prompt gets
+  augmented with these calibration examples, improving scoring quality
+  over time through in-context learning.
+
+  Meta-judge loop:
+    1. Purpose Function scores a transition → produces (score, reasoning, evidence)
+    2. Meta-judge evaluates the judgment: was the reasoning sound? was evidence cited?
+    3. Good judgments → stored as critic_calibration memories (positive examples)
+    4. Bad judgments → stored as failure_pattern memories (negative examples)
+    5. Next time the Purpose Function runs, calibration memories are in its prompt
+
+  Result: the critic gets better at scoring without any weight updates.
+"""
+from __future__ import annotations
+
+import json
+import logging
+from typing import Any
+
+from purpose_agent.llm_backend import LLMBackend, ChatMessage
+from purpose_agent.types import PurposeScore
+from purpose_agent.memory import MemoryCard, MemoryKind, MemoryStatus
+from purpose_agent.v2_types import MemoryScope
+from purpose_agent.memory_ci import MemoryCI
+
+logger = logging.getLogger(__name__)
+
+META_JUDGE_PROMPT = """\
+You are a META-JUDGE. You evaluate the QUALITY of another AI's evaluation.
+
+You will see:
+- A state transition (before → action → after)
+- The Purpose Function's judgment (Φ scores, reasoning, evidence)
+
+Rate the judgment quality on these criteria:
+1. EVIDENCE GROUNDING: Did the judgment cite specific, verifiable state changes? (0-10)
+2. REASONING COHERENCE: Is the chain of reasoning logically sound? (0-10)
+3. CALIBRATION: Are the Φ scores proportional to actual progress? (0-10)
+4. ANTI-SYCOPHANCY: Did the judgment avoid inflating scores to be encouraging? (0-10)
+5. CONSISTENCY: Would an identical state get the same score? (0-10)
+
+Respond with JSON:
+{
+  "evidence_grounding": <0-10>,
+  "reasoning_coherence": <0-10>,
+  "calibration": <0-10>,
+  "anti_sycophancy": <0-10>,
+  "consistency": <0-10>,
+  "overall": <0-10>,
+  "feedback": "<specific feedback on what was good or bad about this judgment>"
+}
+"""
+
+META_JUDGE_SCHEMA = {
+    "type": "object",
+    "properties": {
+        "evidence_grounding": {"type": "number"},
+        "reasoning_coherence": {"type": "number"},
+        "calibration": {"type": "number"},
+        "anti_sycophancy": {"type": "number"},
+        "consistency": {"type": "number"},
+        "overall": {"type": "number"},
+        "feedback": {"type": "string"},
+    },
+    "required": ["overall", "feedback"],
+}
+
+
+class MetaRewardingLoop:
+    """
+    Evaluates and improves the Purpose Function through meta-judgment.
+
+    Usage:
+        meta = MetaRewardingLoop(meta_llm=strong_model, memory_ci=ci)
+
+        # After each Purpose Function evaluation:
+        meta.evaluate_judgment(
+            state_before_desc="Position (0,0)",
+            action_desc="move_east",
+            state_after_desc="Position (1,0)",
+            purpose="Reach (4,4)",
+            judgment=purpose_score,
+        )
+
+        # Good judgments become calibration examples in memory.
+        # Bad judgments become failure patterns.
+        # Purpose Function improves via in-context learning.
+    """
+
+    def __init__(
+        self,
+        meta_llm: LLMBackend,
+        memory_ci: MemoryCI,
+        quality_threshold: float = 7.0,
+    ):
+        self.meta_llm = meta_llm
+        self.memory_ci = memory_ci
+        self.quality_threshold = quality_threshold
+        self._eval_log: list[dict] = []
+
+    def evaluate_judgment(
+        self,
+        state_before_desc: str,
+        action_desc: str,
+        state_after_desc: str,
+        purpose: str,
+        judgment: PurposeScore,
+        trace_id: str = "",
+    ) -> dict[str, Any]:
+        """
+        Have the meta-judge evaluate a Purpose Function judgment.
+
+        If the judgment is high quality → create a positive calibration memory.
+        If low quality → create a negative calibration memory.
+        """
+        context = (
+            f"Purpose: {purpose}\n"
+            f"State before: {state_before_desc}\n"
+            f"Action: {action_desc}\n"
+            f"State after: {state_after_desc}\n\n"
+            f"Purpose Function's judgment:\n"
+            f"  Φ_before={judgment.phi_before:.1f}, Φ_after={judgment.phi_after:.1f}, Δ={judgment.delta:+.2f}\n"
+            f"  Confidence: {judgment.confidence:.2f}\n"
+            f"  Reasoning: {judgment.reasoning}\n"
+            f"  Evidence: {judgment.evidence}"
+        )
+
+        messages = [
+            ChatMessage(role="system", content=META_JUDGE_PROMPT),
+            ChatMessage(role="user", content=context),
+        ]
+
+        try:
+            result = self.meta_llm.generate_structured(messages, schema=META_JUDGE_SCHEMA)
+        except Exception as e:
+            logger.warning(f"Meta-judge failed: {e}")
+            return {"error": str(e)}
+
+        overall = float(result.get("overall", 5.0))
+        feedback = str(result.get("feedback", ""))
+
+        log_entry = {
+            "trace_id": trace_id,
+            "overall_quality": overall,
+            "feedback": feedback,
+            "components": {k: result.get(k, 0) for k in META_JUDGE_SCHEMA["properties"] if k not in ("overall", "feedback")},
+        }
+        self._eval_log.append(log_entry)
+
+        # Create calibration memory
+        if overall >= self.quality_threshold:
+            card = MemoryCard(
+                kind=MemoryKind.CRITIC_CALIBRATION,
+                status=MemoryStatus.CANDIDATE,
+                content=(
+                    f"GOOD judgment example (quality={overall:.0f}/10): "
+                    f"For Δ={judgment.delta:+.2f}, evidence was: '{judgment.evidence[:200]}'. "
+                    f"Meta-judge feedback: {feedback[:200]}"
+                ),
+                pattern=f"When scoring transitions with delta ~{judgment.delta:+.1f}",
+                strategy=f"Follow this example: {judgment.reasoning[:200]}",
+                trust_score=min(overall / 10.0, 1.0),
+                source_trace_id=trace_id,
+                created_by="meta_judge",
+            )
+            self.memory_ci.submit(card)
+            logger.info(f"MetaRewarding: Good judgment (quality={overall:.0f}) → calibration memory")
+        elif overall < 4.0:
+            card = MemoryCard(
+                kind=MemoryKind.FAILURE_PATTERN,
+                status=MemoryStatus.CANDIDATE,
+                content=(
+                    f"BAD judgment example (quality={overall:.0f}/10): "
+                    f"Avoid this pattern: {feedback[:300]}"
+                ),
+                pattern="When scoring state transitions",
+                strategy=f"Do NOT: {feedback[:200]}",
+                trust_score=0.8,
+                source_trace_id=trace_id,
+                created_by="meta_judge",
+                scope=MemoryScope(agent_roles=["critic"]),
+            )
+            self.memory_ci.submit(card)
+            logger.info(f"MetaRewarding: Bad judgment (quality={overall:.0f}) → failure pattern memory")
+
+        return log_entry
+
+    @property
+    def eval_log(self) -> list[dict]:
+        return self._eval_log
+
+    def summary(self) -> dict[str, Any]:
+        if not self._eval_log:
+            return {"evaluations": 0}
+        scores = [e["overall_quality"] for e in self._eval_log]
+        return {
+            "evaluations": len(scores),
+            "avg_quality": round(sum(scores) / len(scores), 2),
+            "min_quality": min(scores),
+            "max_quality": max(scores),
+        }