V2 merge: purpose_agent/self_taught.py

purpose_agent/self_taught.py

@@ -0,0 +1,214 @@
+"""
+self_taught.py — Synthetic training data for the Purpose Function.
+
+From Self-Taught Evaluators (arxiv:2408.02666):
+  Generate synthetic preference pairs (good vs bad state evaluations)
+  from the agent's own traces, then use them to improve the Purpose
+  Function's prompts without any human labels.
+
+Adaptation for Purpose Agent (no weight updates):
+  1. Take a completed trace with Φ scores
+  2. For each step, generate a "worse" evaluation (modified instruction trick)
+  3. The correct evaluation becomes a positive example
+  4. The worse evaluation becomes a negative example
+  5. Store both as critic_calibration memories
+  6. The Purpose Function improves via in-context learning from these examples
+
+This is an automatic curriculum: as the Purpose Function improves,
+it generates harder training pairs, which further improve it.
+"""
+from __future__ import annotations
+
+import json
+import logging
+from typing import Any
+
+from purpose_agent.llm_backend import LLMBackend, ChatMessage
+from purpose_agent.trace import Trace
+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__)
+
+GENERATE_CONTRAST_PROMPT = """\
+You are generating training data for a state evaluator (critic).
+
+Given this CORRECT evaluation of a state transition:
+  State before: {state_before}
+  Action: {action}
+  State after: {state_after}
+  Purpose: {purpose}
+  Correct Φ_before: {phi_before:.1f}
+  Correct Φ_after: {phi_after:.1f}
+  Correct reasoning: {reasoning}
+
+Generate a PLAUSIBLE BUT WRONG evaluation that makes a common mistake.
+Common mistakes:
+- Giving credit for intentions rather than actual state changes
+- Inflating scores to be encouraging (sycophancy)
+- Ignoring evidence and scoring based on action name alone
+- Being inconsistent with the scoring scale
+
+Respond with JSON:
+{{
+  "wrong_phi_after": <a plausible but incorrect score>,
+  "wrong_reasoning": "<plausible but flawed reasoning>",
+  "mistake_type": "<which common mistake this represents>"
+}}
+"""
+
+CONTRAST_SCHEMA = {
+    "type": "object",
+    "properties": {
+        "wrong_phi_after": {"type": "number"},
+        "wrong_reasoning": {"type": "string"},
+        "mistake_type": {"type": "string"},
+    },
+    "required": ["wrong_phi_after", "wrong_reasoning", "mistake_type"],
+}
+
+
+class SelfTaughtEvaluator:
+    """
+    Generates synthetic training data for the Purpose Function from traces.
+
+    Usage:
+        ste = SelfTaughtEvaluator(llm=model, memory_ci=ci)
+
+        # After a trace is complete:
+        pairs = ste.generate_from_trace(trace)
+        # → Creates critic_calibration memories with good/bad examples
+
+        # Iterative: as the critic improves, it generates harder pairs
+        for iteration in range(3):
+            for trace in recent_traces:
+                ste.generate_from_trace(trace)
+    """
+
+    def __init__(
+        self,
+        llm: LLMBackend,
+        memory_ci: MemoryCI,
+        min_delta_for_training: float = 0.5,
+    ):
+        self.llm = llm
+        self.memory_ci = memory_ci
+        self.min_delta = min_delta_for_training
+        self._pairs_generated = 0
+
+    def generate_from_trace(self, trace: Trace) -> int:
+        """
+        Generate contrast pairs from a trace's score events.
+
+        Returns number of pairs generated.
+        """
+        count = 0
+        score_events = [e for e in trace.events if e.kind == "score"]
+
+        for event in score_events:
+            data = event.data
+            phi_before = data.get("phi_before", 0)
+            phi_after = data.get("phi_after", 0)
+            delta = phi_after - phi_before
+
+            # Only generate pairs for meaningful transitions
+            if abs(delta) < self.min_delta:
+                continue
+
+            try:
+                pair = self._generate_contrast_pair(
+                    state_before=data.get("state_before", ""),
+                    action=data.get("action_name", ""),
+                    state_after=data.get("state_after", ""),
+                    purpose=trace.purpose,
+                    phi_before=phi_before,
+                    phi_after=phi_after,
+                    reasoning=data.get("reasoning", ""),
+                )
+                if pair:
+                    self._store_pair(pair, trace.trace_id)
+                    count += 1
+            except Exception as e:
+                logger.warning(f"SelfTaught: Failed to generate pair: {e}")
+
+        self._pairs_generated += count
+        logger.info(f"SelfTaught: Generated {count} contrast pairs from trace {trace.trace_id}")
+        return count
+
+    def _generate_contrast_pair(
+        self,
+        state_before: str,
+        action: str,
+        state_after: str,
+        purpose: str,
+        phi_before: float,
+        phi_after: float,
+        reasoning: str,
+    ) -> dict[str, Any] | None:
+        """Generate a single (correct, wrong) evaluation pair."""
+        messages = [
+            ChatMessage(role="system", content="Generate a plausible but incorrect evaluation for training."),
+            ChatMessage(role="user", content=GENERATE_CONTRAST_PROMPT.format(
+                state_before=state_before[:200],
+                action=action,
+                state_after=state_after[:200],
+                purpose=purpose,
+                phi_before=phi_before,
+                phi_after=phi_after,
+                reasoning=reasoning[:200],
+            )),
+        ]
+
+        try:
+            result = self.llm.generate_structured(messages, schema=CONTRAST_SCHEMA)
+        except Exception:
+            raw = self.llm.generate(messages, temperature=0.7)
+            try:
+                result = json.loads(raw)
+            except Exception:
+                return None
+
+        return {
+            "correct_phi_after": phi_after,
+            "correct_reasoning": reasoning,
+            "wrong_phi_after": result.get("wrong_phi_after", phi_after + 2),
+            "wrong_reasoning": result.get("wrong_reasoning", ""),
+            "mistake_type": result.get("mistake_type", "unknown"),
+        }
+
+    def _store_pair(self, pair: dict, trace_id: str) -> None:
+        """Store a contrast pair as calibration memories."""
+        # Positive example
+        self.memory_ci.submit(MemoryCard(
+            kind=MemoryKind.CRITIC_CALIBRATION,
+            content=(
+                f"CORRECT scoring example: Φ_after={pair['correct_phi_after']:.1f}. "
+                f"Reasoning: {pair['correct_reasoning'][:200]}"
+            ),
+            pattern="When evaluating state transitions",
+            strategy=f"Score like this: {pair['correct_reasoning'][:150]}",
+            trust_score=0.7,
+            source_trace_id=trace_id,
+            created_by="self_taught",
+        ))
+
+        # Negative example (what NOT to do)
+        self.memory_ci.submit(MemoryCard(
+            kind=MemoryKind.FAILURE_PATTERN,
+            content=(
+                f"WRONG scoring example ({pair['mistake_type']}): "
+                f"Incorrectly scored Φ_after={pair['wrong_phi_after']:.1f}. "
+                f"Flawed reasoning: {pair['wrong_reasoning'][:200]}"
+            ),
+            pattern="When evaluating state transitions",
+            strategy=f"AVOID this mistake: {pair['mistake_type']}",
+            trust_score=0.7,
+            source_trace_id=trace_id,
+            created_by="self_taught",
+            scope=MemoryScope(agent_roles=["critic"]),
+        ))
+
+    @property
+    def pairs_generated(self) -> int:
+        return self._pairs_generated