Self-Evolving Agents: A Developer's Guide

Level 1 — Prompt tuning              (minutes, free)
     │  still failing after 3 rounds?
     ▼
Level 2 — Add/improve skills         (hours, cheap)
     │  still failing on reasoning/architecture?
     ▼
Level 3 — Code & Harness evolution   (hours, cheap — runs overnight)
     │  still failing on knowledge?
     ▼
Level 4 — RAG                        (hours, medium cost)
     │  still failing on reasoning style/pattern?
     ▼
Level 5 — LLM Fine-tuning            (days, expensive)

Agent runs task
      │
      ▼
Evaluator scores output
      │
      ▼
Failure classifier diagnoses root cause
      │
      ▼
Improvement dispatcher triggers the right track
      │
      ▼
Updated agent reruns

import json
from anthropic import Anthropic

client = Anthropic()

# --- Versioned prompt store ---
prompt_versions = []

def save_prompt(prompt: str, score: float):
    prompt_versions.append({"prompt": prompt, "score": score})
    prompt_versions.sort(key=lambda x: x["score"], reverse=True)

def best_prompt() -> str:
    return prompt_versions[0]["prompt"] if prompt_versions else INITIAL_PROMPT

# --- Agent runner ---
INITIAL_PROMPT = "You are a helpful assistant that answers math word problems."

def run_agent(system_prompt: str, user_task: str) -> str:
    response = client.messages.create(
        model="claude-opus-4-6-20260205",
        max_tokens=1024,
        system=system_prompt,
        messages=[{"role": "user", "content": user_task}]
    )
    return response.content[0].text

# --- LLM-as-judge evaluator ---
def evaluate_response(task: str, response: str, expected: str) -> float:
    judge_prompt = f"""
    Task: {task}
    Expected answer: {expected}
    Agent response: {response}

    Score the response from 0.0 to 1.0 based on correctness and clarity.
    Reply with JSON only: {{"score": 0.0, "reason": "..."}}
    """
    result = client.messages.create(
        model="claude-opus-4-6-20260205",
        max_tokens=256,
        messages=[{"role": "user", "content": judge_prompt}]
    )
    return json.loads(result.content[0].text)["score"]

# --- Prompt rewriter ---
def rewrite_prompt(current_prompt: str, task: str, failed_response: str, reason: str) -> str:
    rewrite_request = f"""
    The current system prompt failed on this task.

    System prompt: {current_prompt}
    Task: {task}
    Bad response: {failed_response}
    Failure reason: {reason}

    Rewrite the system prompt to handle this better.
    Reply with the new prompt text only.
    """
    result = client.messages.create(
        model="claude-opus-4-6-20260205",
        max_tokens=512,
        messages=[{"role": "user", "content": rewrite_request}]
    )
    return result.content[0].text

# --- Evolution loop ---
def evolution_loop(tasks: list[dict], threshold=0.7, max_rounds=3):
    current_prompt = INITIAL_PROMPT
    save_prompt(current_prompt, score=0.0)

    for round in range(max_rounds):
        print(f"\n=== Round {round + 1} | Prompt: {current_prompt[:60]}... ===")
        round_scores = []

        for t in tasks:
            response = run_agent(current_prompt, t["task"])
            score = evaluate_response(t["task"], response, t["expected"])
            round_scores.append(score)
            print(f"  Task: {t['task'][:50]} | Score: {score:.2f}")

            if score < threshold:
                current_prompt = rewrite_prompt(
                    current_prompt, t["task"], response, "Low score"
                )

        avg_score = sum(round_scores) / len(round_scores)
        save_prompt(current_prompt, avg_score)
        print(f"  Avg score: {avg_score:.2f}")

        if avg_score >= threshold:
            print("✅ Prompt converged.")
            break

    return best_prompt()

# Example usage
tasks = [
    {"task": "If a train travels 60mph for 2.5 hours, how far does it go?", "expected": "150 miles"},
    {"task": "A store has 240 apples. 1/3 are sold. How many remain?",       "expected": "160 apples"},
]

final_prompt = evolution_loop(tasks)
print(f"\nFinal best prompt:\n{final_prompt}")

import json
from anthropic import Anthropic

client = Anthropic()

# --- Skill registry ---
class SkillRegistry:
    def __init__(self):
        self.skills: dict[str, dict] = {}  # name -> {code, description, stats}

    def register(self, name: str, description: str, code: str):
        self.skills[name] = {
            "description": description,
            "code": code,
            "usage_count": 0,
            "success_rate": 1.0
        }
        print(f"✅ Skill registered: {name}")

    def retrieve(self, task: str, top_k=2) -> list[dict]:
        """Keyword overlap retrieval — swap for vector search in prod."""
        scored = []
        for name, skill in self.skills.items():
            overlap = len(
                set(task.lower().split()) & set(skill["description"].lower().split())
            )
            scored.append((overlap, name, skill))
        scored.sort(reverse=True)
        return [{"name": n, **s} for _, n, s in scored[:top_k]]

    def update_stats(self, name: str, success: bool):
        if name in self.skills:
            skill = self.skills[name]
            skill["usage_count"] += 1
            skill["success_rate"] = (
                skill["success_rate"] * (skill["usage_count"] - 1) + int(success)
            ) / skill["usage_count"]

    def prune(self, min_success_rate=0.4, min_uses=3):
        """Remove underperforming skills."""
        to_remove = [
            name for name, s in self.skills.items()
            if s["usage_count"] >= min_uses and s["success_rate"] < min_success_rate
        ]
        for name in to_remove:
            del self.skills[name]
            print(f"🗑️  Pruned skill: {name}")

registry = SkillRegistry()

# --- Seed with initial skills ---
registry.register(
    name="calculate_percentage",
    description="calculate percentage proportion ratio",
    code="def calculate_percentage(part, whole): return round((part / whole) * 100, 2)"
)
registry.register(
    name="days_between_dates",
    description="date difference calendar days between two dates",
    code="""
from datetime import datetime
def days_between_dates(d1: str, d2: str) -> int:
    fmt = "%Y-%m-%d"
    return abs((datetime.strptime(d2, fmt) - datetime.strptime(d1, fmt)).days)
"""
)

# --- Skill generator: agent writes new skills on demand ---
def generate_skill(task_description: str) -> dict:
    prompt = f"""
    A user needs help with: "{task_description}"
    No existing skill covers this. Write a new Python skill.

    Reply with JSON only:
    {{
        "name": "snake_case_name",
        "description": "keywords describing when to use this skill",
        "code": "def skill_name(...):\\n    ..."
    }}
    """
    result = client.messages.create(
        model="claude-opus-4-6-20260205",
        max_tokens=512,
        messages=[{"role": "user", "content": prompt}]
    )
    raw = result.content[0].text.strip().strip("```

json").strip("

```")
    return json.loads(raw)

# --- Agent that uses the skill registry ---
def skill_aware_agent(user_task: str):
    relevant_skills = registry.retrieve(user_task)
    skill_context = "\n\n".join(
        [f"Skill `{s['name']}`:\n```
{% endraw %}
python\n{s['code']}\n
{% raw %}
```" for s in relevant_skills]
    )

    system = f"""You are a Python agent. Use available skills when helpful.
Available skills:
{skill_context}

If no skill fits, say NEED_NEW_SKILL: <description of what's needed>."""

    response = client.messages.create(
        model="claude-opus-4-6-20260205",
        max_tokens=1024,
        system=system,
        messages=[{"role": "user", "content": user_task}]
    )
    answer = response.content[0].text

    # Auto-generate missing skill if flagged
    if "NEED_NEW_SKILL:" in answer:
        needed = answer.split("NEED_NEW_SKILL:")[1].strip()
        print(f"🔧 Generating new skill for: {needed}")
        new_skill = generate_skill(needed)
        registry.register(**new_skill)
        return skill_aware_agent(user_task)  # Retry with new skill

    success = "error" not in answer.lower() and "sorry" not in answer.lower()
    for s in relevant_skills:
        registry.update_stats(s["name"], success)

    return answer

# Example usage
print(skill_aware_agent("What percentage is 45 out of 180?"))
print(skill_aware_agent("How many days between 2024-01-15 and 2024-07-04?"))
print(skill_aware_agent("Convert 100 USD to EUR at a rate of 0.92"))  # triggers new skill

import json
from dataclasses import dataclass, field
from anthropic import Anthropic

client = Anthropic()

@dataclass
class EvalResult:
    score: float
    passed: bool
    feedback: str

@dataclass
class EvalSuite:
    name: str
    cases: list[dict] = field(default_factory=list)
    pass_threshold: float = 0.75

    def add_case(self, input: str, expected: str, tags: list[str] | None = None):
        self.cases.append({"input": input, "expected": expected, "tags": tags or []})

def llm_judge(task: str, expected: str, actual: str) -> EvalResult:
    prompt = f"""Evaluate this agent response.
Task: {task}
Expected: {expected}
Actual: {actual}

Reply with JSON only:
{{"score": 0.0-1.0, "passed": true/false, "feedback": "brief reason"}}"""

    result = client.messages.create(
        model="claude-opus-4-6-20260205",
        max_tokens=256,
        messages=[{"role": "user", "content": prompt}]
    )
    data = json.loads(result.content[0].text)
    return EvalResult(**data)

def run_eval_suite(suite: EvalSuite, system_prompt: str) -> dict:
    results = []
    tag_scores: dict[str, list] = {}

    for case in suite.cases:
        response = client.messages.create(
            model="claude-opus-4-6-20260205",
            max_tokens=512,
            system=system_prompt,
            messages=[{"role": "user", "content": case["input"]}]
        )
        actual = response.content[0].text
        result = llm_judge(case["input"], case["expected"], actual)
        results.append(result)

        for tag in case.get("tags", []):
            tag_scores.setdefault(tag, []).append(result.score)

        status = "✅" if result.passed else "❌"
        print(f"  {status} [{case['input'][:45]}] score={result.score:.2f} | {result.feedback}")

    avg_score = sum(r.score for r in results) / len(results)
    tag_summary = {tag: round(sum(s)/len(s), 2) for tag, s in tag_scores.items()}

    return {
        "avg_score": round(avg_score, 3),
        "passed": avg_score >= suite.pass_threshold,
        "tag_breakdown": tag_summary,
        "total_cases": len(results),
        "passed_cases": sum(1 for r in results if r.passed)
    }

def promote_prompt(candidate: str, current: str, suite: EvalSuite) -> tuple[str, dict]:
    """Only promote candidate if it beats the current prompt."""
    print("\n📊 Evaluating CURRENT prompt...")
    current_report = run_eval_suite(suite, current)

    print("\n📊 Evaluating CANDIDATE prompt...")
    candidate_report = run_eval_suite(suite, candidate)

    if candidate_report["avg_score"] > current_report["avg_score"]:
        print(f"\n🚀 Promoting ({candidate_report['avg_score']:.2f} > {current_report['avg_score']:.2f})")
        return candidate, candidate_report
    else:
        print(f"\n⏪ Keeping current ({current_report['avg_score']:.2f} >= {candidate_report['avg_score']:.2f})")
        return current, current_report

# Example usage
suite = EvalSuite(name="math_agent_v1", pass_threshold=0.75)
suite.add_case("What is 15% of 200?",                  "30",           tags=["percentage"])
suite.add_case("A rectangle is 8x5. What's its area?", "40 sq units",  tags=["geometry"])
suite.add_case("Train goes 90mph for 3 hours. Distance?", "270 miles", tags=["word_problem"])
suite.add_case("Factor 12 into primes.",                "2 × 2 × 3",   tags=["number_theory"])

current_prompt   = "You are a helpful assistant that solves math problems."
candidate_prompt = (
    "You are a precise math tutor. Always show step-by-step reasoning, "
    "state the formula used, then give a clean final answer."
)

best_prompt, report = promote_prompt(candidate_prompt, current_prompt, suite)
print(f"\nTag breakdown: {report['tag_breakdown']}")
print(f"Final: {report['passed_cases']}/{report['total_cases']} cases passed")

from datetime import datetime, timezone
from dataclasses import dataclass, field

@dataclass
class PromptVersion:
    version: int
    prompt: str
    model: str
    score: float
    timestamp: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
    metadata: dict = field(default_factory=dict)

class VersionedPrompt:
    def __init__(self, initial_prompt: str, model: str = "claude-opus-4-6-20260205"):
        self._versions = [PromptVersion(version=0, prompt=initial_prompt, model=model, score=0.0)]

    def update(self, new_prompt: str, score: float, model: str = None, **metadata) -> PromptVersion:
        v = PromptVersion(
            version=self._versions[-1].version + 1,
            prompt=new_prompt,
            model=model or self._versions[-1].model,
            score=score,
            metadata=metadata,
        )
        self._versions.append(v)
        return v

    def current(self) -> PromptVersion:
        return self._versions[-1]

    def best(self) -> PromptVersion:
        return max(self._versions, key=lambda v: v.score)

    def rollback(self, version: int) -> PromptVersion:
        self._versions = [v for v in self._versions if v.version <= version]
        return self._versions[-1]

import gepa
from gepa import EvaluationBatch

seed_candidate = {
    "system_prompt": "You are a summarization assistant. Given a section of text, produce a summary."
}

result = gepa.optimize(
    seed_candidate=seed_candidate,
    trainset=train_data,
    valset=val_data,
    adapter=your_eval_adapter,   # bridges your graders to GEPA's interface
    reflection_lm="gpt-5",
    max_metric_calls=20,
    track_best_outputs=True,
)

best_prompt = result.best_candidate["system_prompt"]

import json
from enum import Enum
from anthropic import Anthropic

client = Anthropic()

class ImprovementTrack(Enum):
    PROMPT = "prompt"
    SKILL  = "skill"
    BOTH   = "both"

def classify_failure(
    task: str,
    agent_response: str,
    expected: str,
    current_system_prompt: str
) -> dict:
    classifier_prompt = f"""
You are an AI agent debugging expert. Analyze this agent failure.

System prompt: {current_system_prompt}
Task: {task}
Expected: {expected}
Actual response: {agent_response}

Diagnose the root cause and classify it. Consider:
- PROMPT: the agent has the capability but wrong behavior/tone/reasoning style
- SKILL: the agent is missing a tool, lookup, or computation it cannot reliably do in its head
- BOTH: the prompt misdirects AND a skill is missing

Reply with JSON only:
{{
    "track": "prompt" | "skill" | "both",
    "root_cause": "one sentence explanation",
    "evidence": "specific part of the response that reveals the problem",
    "suggested_action": "concrete next step"
}}
"""
    result = client.messages.create(
        model="claude-opus-4-6-20260205",
        max_tokens=512,
        messages=[{"role": "user", "content": classifier_prompt}]
    )
    diagnosis = json.loads(result.content[0].text)
    diagnosis["track"] = ImprovementTrack(diagnosis["track"])
    return diagnosis

# Example usage
failures = [
    {
        "task": "What is the compound interest on $5000 at 4.5% for 3 years?",
        "expected": "$706.06",
        "actual": "The compound interest would be approximately $700.",
        "prompt": "You are a helpful financial assistant."
    },
    {
        "task": "Explain the steps to solve a quadratic equation.",
        "expected": "Step-by-step: factoring, completing the square, quadratic formula",
        "actual": "Just use the quadratic formula: x = (-b ± √(b²-4ac)) / 2a",
        "prompt": "You are a helpful math assistant."
    },
]

for f in failures:
    print(f"\nTask: {f['task'][:60]}...")
    diagnosis = classify_failure(f["task"], f["actual"], f["expected"], f["prompt"])
    print(f"  Track     : {diagnosis['track'].value.upper()}")
    print(f"  Root cause: {diagnosis['root_cause']}")
    print(f"  Action    : {diagnosis['suggested_action']}")

┌─────────────────────────────────────────────┐
│  Human writes program.md                    │
│  (instructions, constraints, goals)         │
│                                             │
│         ┌──────────────┐                    │
│         │  Meta-agent   │                   │
│         │  reads        │                   │
│         │  program.md   │                   │
│         └──────┬───────┘                    │
│                │                            │
│         ┌──────▼───────┐                    │
│         │  Modifies     │                   │
│         │  train.py or  │                   │
│         │  agent.py     │                   │
│         └──────┬───────┘                    │
│                │                            │
│         ┌──────▼───────┐                    │
│         │  Runs eval    │                   │
│         │  (metric)     │                   │
│         └──────┬───────┘                    │
│                │                            │
│         ┌──────▼───────┐                    │
│         │  Score better?│                   │
│         │  Keep : Revert│                   │
│         └──────────────┘                    │
└─────────────────────────────────────────────┘

import json
from anthropic import Anthropic

client = Anthropic()

# --- Toy in-memory store (swap for Chroma/Pinecone in prod) ---
knowledge_base = [
    {"id": 1, "text": "Q1 2025 audit found 3 critical gaps in access control policies."},
    {"id": 2, "text": "Revenue for Q1 2025 was $4.2M, up 18% YoY."},
    {"id": 3, "text": "The compound XR-47 showed hepatotoxicity in Phase 2 trials."},
]

def simple_retrieve(query: str, top_k=2) -> list[str]:
    """Keyword overlap retrieval — replace with embedding search in prod."""
    query_words = set(query.lower().split())
    scored = []
    for doc in knowledge_base:
        doc_words = set(doc["text"].lower().split())
        overlap = len(query_words & doc_words)
        scored.append((overlap, doc["text"]))
    scored.sort(reverse=True)
    return [text for _, text in scored[:top_k] if _ > 0]

def rag_agent(user_query: str) -> str:
    context_chunks = simple_retrieve(user_query)

    if context_chunks:
        context_block = "\n".join(f"- {c}" for c in context_chunks)
        system = f"""You are a helpful enterprise assistant.
Use ONLY the retrieved context below to answer.
If the context doesn't cover the question, say so.

Retrieved context:
{context_block}"""
    else:
        system = "You are a helpful enterprise assistant."

    response = client.messages.create(
        model="claude-opus-4-6-20260205",
        max_tokens=512,
        system=system,
        messages=[{"role": "user", "content": user_query}]
    )
    return response.content[0].text

# Example usage
queries = [
    "What did the Q1 2025 audit find?",
    "What were Q1 revenues?",
    "Tell me about XR-47 safety.",
    "What is our HR vacation policy?",  # not in KB → honest fallback
]

for q in queries:
    print(f"\nQ: {q}")
    print(f"A: {rag_agent(q)}")

Deploy → collect trajectories → filter (score ≥ 0.8) → fine-tune → redeploy → repeat

Agent runs → Failures logged
     ↓
Judge 1: Per-run evaluator (scores 0–1)
     ↓
Judge 2: Signal extractor (persistence, skill gap, knowledge gap, data volume)
     ↓
Judge 3: Track recommender (LLM synthesizes signals → verdict)
     ↓
Judge 4: Action dispatcher → calls evolution_loop() / rag_agent() / fine-tune export

import json
from dataclasses import dataclass, field
from enum import Enum
from anthropic import Anthropic

client = Anthropic()


# ── Data models ──────────────────────────────────────────────

class Track(Enum):
    PROMPT_SKILL   = "prompt_skill"
    CODE_EVOLUTION = "code_evolution"
    RAG            = "rag"
    FINE_TUNE      = "fine_tune"
    RAG_FINE_TUNE  = "rag+fine_tune"

@dataclass
class AgentRun:
    task: str
    expected: str
    actual: str
    task_type: str
    prompt_version: str
    prompt_round: int
    correct_skill_invoked: bool = False
    score: float = 0.0

@dataclass
class JudgeVerdict:
    track: Track
    confidence: float
    signals: dict
    rationale: str
    next_steps: list[str]
    estimated_effort: str
    risk: str


# ── Judge 1: Per-run evaluator ────────────────────────────────

def evaluate_run(run: AgentRun) -> AgentRun:
    """Scores a single agent run 0.0–1.0."""
    prompt = f"""
Evaluate this agent response.

Task     : {run.task}
Expected : {run.expected}
Actual   : {run.actual}

Reply with JSON only:
{{"score": 0.0-1.0, "passed": true/false, "reason": "one sentence"}}
"""
    result = client.messages.create(
        model="claude-opus-4-6-20260205",
        max_tokens=256,
        messages=[{"role": "user", "content": prompt}]
    )
    data = json.loads(result.content[0].text.strip().strip("```

json").strip("

```"))
    run.score = data["score"]
    return run


# ── Judge 2: Signal extractor ─────────────────────────────────

def extract_signals(runs: list[AgentRun], corpus_exists: bool, example_count: int) -> dict:
    """Derives quantitative signals from a batch of runs."""
    total = len(runs)
    failed = [r for r in runs if r.score < 0.7]

    if not failed:
        return {"all_passing": True}

    f = len(failed)

    # Signal 1: Prompt plateau — failures persisting after 3+ prompt rounds
    persistence_rate = len([r for r in failed if r.prompt_round >= 3]) / f

    # Signal 2: Skill bottleneck — skill fired but still failed
    skill_failure_rate = len([r for r in failed if r.correct_skill_invoked]) / f

    # Signal 3: Domain concentration — one task type dominating failures
    type_counts = {}
    for r in failed:
        type_counts[r.task_type] = type_counts.get(r.task_type, 0) + 1
    dominant_rate = max(type_counts.values()) / f if type_counts else 0
    dominant_type = max(type_counts, key=type_counts.get) if type_counts else "unknown"

    # Signal 4: Knowledge gap — failed despite no skill gap → likely needs retrieval
    knowledge_gap_rate = len([
        r for r in failed if not r.correct_skill_invoked and r.prompt_round >= 2
    ]) / f

    return {
        "total_runs"         : total,
        "failure_rate"       : round(f / total, 2),
        "persistence_rate"   : round(persistence_rate, 2),   # > 0.4 → fine-tune
        "skill_failure_rate" : round(skill_failure_rate, 2), # > 0.3 → fine-tune
        "knowledge_gap_rate" : round(knowledge_gap_rate, 2), # > 0.4 → RAG
        "dominant_type"      : dominant_type,
        "dominant_type_rate" : round(dominant_rate, 2),      # > 0.5 → systematic gap
        "corpus_exists"      : corpus_exists,
        "example_count"      : example_count,
        "data_sufficient"    : example_count >= 500
    }


# ── Judge 3: Track recommender ────────────────────────────────

def recommend_track(
    signals: dict,
    current_prompt: str,
    sample_failures: list[AgentRun]
) -> JudgeVerdict:
    """LLM judge: reads signals + failure samples → recommends track."""

    sample_text = json.dumps([
        {
            "task": r.task, "expected": r.expected,
            "actual": r.actual, "score": r.score,
            "prompt_round": r.prompt_round,
            "correct_skill_invoked": r.correct_skill_invoked
        }
        for r in sample_failures[:5]
    ], indent=2)

    judge_prompt = f"""
You are a senior AI systems architect. Decide the best improvement track
for an underperforming agent based on signals and failure samples.

## Quantitative Signals
{json.dumps(signals, indent=2)}

## Signal Thresholds
- persistence_rate > 0.4     → prompt iteration plateauing → consider fine_tune
- skill_failure_rate > 0.3   → model reasoning is bottleneck → consider fine_tune
- knowledge_gap_rate > 0.4   → facts/docs missing → consider rag
- dominant_type_rate > 0.5   → systematic domain gap
- data_sufficient = false    → BLOCK fine_tune, default to rag or prompt_skill

## Available Tracks
- prompt_skill   : Rewrite system prompt and/or add/fix tools. Fast, cheap, reversible.
- code_evolution : Let a meta-agent modify code/config against a clear metric.
                   Use when the problem has a single file to optimize and a measurable goal.
- rag            : Index a knowledge corpus and retrieve at query time.
                   Prefer over fine-tuning when knowledge changes or data < 500.
- fine_tune      : Train on trajectories. Use when reasoning style is systematically
                   wrong AND 500+ examples exist AND prompt iteration has plateaued.
- rag+fine_tune  : Both. Use when knowledge AND reasoning style are both gaps.

## Current System Prompt
{current_prompt}

## Sample Failures
{sample_text}

Be conservative — recommend fine_tune only when signals clearly justify it.

Reply with JSON only:
{{
    "track": "prompt_skill" | "code_evolution" | "rag" | "fine_tune" | "rag+fine_tune",
    "confidence": 0.0-1.0,
    "signals_fired": {{
        "prompt_plateau"   : true/false,
        "skill_bottleneck" : true/false,
        "knowledge_gap"    : true/false,
        "systematic_domain": true/false,
        "data_sufficient"  : true/false
    }},
    "rationale": "2-3 sentence explanation referencing specific signals",
    "next_steps": ["step 1", "step 2", "step 3"],
    "estimated_effort": "e.g. 2hrs prompt iteration vs 4 days fine-tuning",
    "risk": "main risk of this recommendation"
}}
"""
    result = client.messages.create(
        model="claude-opus-4-6-20260205",
        max_tokens=768,
        messages=[{"role": "user", "content": judge_prompt}]
    )
    raw = result.content[0].text.strip().strip("```

json").strip("

```")
    data = json.loads(raw)

    return JudgeVerdict(
        track=Track(data["track"]),
        confidence=data["confidence"],
        signals=data["signals_fired"],
        rationale=data["rationale"],
        next_steps=data["next_steps"],
        estimated_effort=data["estimated_effort"],
        risk=data["risk"]
    )


# ── Judge 4: Action dispatcher ────────────────────────────────

def dispatch(verdict: JudgeVerdict):
    print(f"\n{'='*60}")
    print(f"  TRACK       : {verdict.track.value.upper()}")
    print(f"  CONFIDENCE  : {verdict.confidence:.0%}")
    print(f"  RATIONALE   : {verdict.rationale}")
    print(f"  EFFORT      : {verdict.estimated_effort}")
    print(f"  RISK        : {verdict.risk}")
    print(f"  SIGNALS     : {verdict.signals}")
    print(f"\n  NEXT STEPS:")
    for i, step in enumerate(verdict.next_steps, 1):
        print(f"    {i}. {step}")
    print(f"{'='*60}")

    actions = {
        Track.PROMPT_SKILL: lambda: (
            print("\n→ Calling evolution_loop() to rewrite system prompt"),
            print("→ Calling classify_failure() to split prompt vs skill fixes")
        ),
        Track.CODE_EVOLUTION: lambda: (
            print("\n→ Set up program.md with constraints and goals"),
            print("→ Point meta-agent at the repo (autoresearch or autoagent pattern)"),
            print("→ Let it hill-climb overnight; review results in the morning")
        ),
        Track.RAG: lambda: (
            print("\n→ Chunk and embed your knowledge corpus"),
            print("→ Register retrieval as a new skill in SkillRegistry"),
            print("→ Re-run eval suite to confirm improvement")
        ),
        Track.FINE_TUNE: lambda: (
            print("\n→ Export high-scoring runs as training trajectories"),
            print("→ Filter: keep only runs with score >= 0.8"),
            print("→ Submit fine-tune job (OpenAI / HuggingFace / Anthropic)")
        ),
        Track.RAG_FINE_TUNE: lambda: (
            print("\n→ Step 1: Build RAG pipeline first (faster win)"),
            print("→ Step 2: Validate RAG improves knowledge gaps"),
            print("→ Step 3: Fine-tune on reasoning style gaps in parallel")
        )
    }
    actions[verdict.track]()


# ── Master pipeline ───────────────────────────────────────────

def run_judge_pipeline(
    runs: list[AgentRun],
    current_prompt: str,
    corpus_exists: bool = False,
    example_count: int = 0
):
    print("⏳ Step 1: Evaluating all runs...")
    evaluated = [evaluate_run(r) for r in runs]

    avg_score = sum(r.score for r in evaluated) / len(evaluated)
    failed_count = sum(1 for r in evaluated if r.score < 0.7)
    print(f"   Avg score: {avg_score:.2f} | Failed: {failed_count}/{len(evaluated)}")

    if avg_score >= 0.85:
        print("✅ Agent is performing well. No improvement needed.")
        return

    print("\n⏳ Step 2: Extracting signals...")
    signals = extract_signals(evaluated, corpus_exists, example_count)
    print(f"   Signals: {signals}")

    failed_runs = [r for r in evaluated if r.score < 0.7]

    print("\n⏳ Step 3: LLM judge recommending track...")
    verdict = recommend_track(signals, current_prompt, failed_runs)

    print("\n⏳ Step 4: Dispatching recommendation...")
    dispatch(verdict)

    return verdict


# ── Example usage ─────────────────────────────────────────────

runs = [
    AgentRun(
        task="Summarize the Q1 2025 earnings report",
        expected="Revenue $4.2M, up 18% YoY, 3 audit gaps found",
        actual="I don't have access to Q1 2025 earnings data.",
        task_type="finance", prompt_version="v3",
        prompt_round=4, correct_skill_invoked=False
    ),
    AgentRun(
        task="What were the audit findings for access control?",
        expected="3 critical gaps found in access control policies",
        actual="I cannot find specific audit findings in my knowledge.",
        task_type="finance", prompt_version="v3",
        prompt_round=4, correct_skill_invoked=False
    ),
    AgentRun(
        task="Calculate compound interest $5000 at 4.5% for 3 years",
        expected="$706.06",
        actual="Approximately $700 using compound interest formula.",
        task_type="finance", prompt_version="v3",
        prompt_round=3, correct_skill_invoked=True
    ),
    AgentRun(
        task="Analyze revenue trend from last 4 quarters",
        expected="Structured YoY trend with % changes",
        actual="Revenue seems to be going up based on general trends.",
        task_type="finance", prompt_version="v3",
        prompt_round=4, correct_skill_invoked=False
    ),
] * 10  # scale to 40 runs

current_prompt = "You are a financial analysis assistant. Be thorough and precise."

verdict = run_judge_pipeline(
    runs=runs,
    current_prompt=current_prompt,
    corpus_exists=True,   # financial docs available to index
    example_count=350     # below the 500 fine-tuning threshold
)

⏳ Step 1: Evaluating all runs...
   Avg score: 0.31 | Failed: 37/40

⏳ Step 2: Extracting signals...
   Signals: {failure_rate: 0.93, persistence_rate: 0.89,
             knowledge_gap_rate: 0.76, dominant_type: finance,
             corpus_exists: True, data_sufficient: False}

⏳ Step 3: LLM judge recommending track...

⏳ Step 4: Dispatching recommendation...
============================================================
  TRACK       : RAG
  CONFIDENCE  : 91%
  RATIONALE   : High knowledge_gap_rate (0.76) with corpus_exists=True
                and data_sufficient=False clearly points to RAG. Agent
                is failing on factual retrieval, not reasoning style.
  EFFORT      : 4–6 hours to chunk, embed, and integrate corpus
  RISK        : Retrieval quality depends on chunking strategy
  SIGNALS     : {prompt_plateau: True, skill_bottleneck: False,
                 knowledge_gap: True, systematic_domain: True,
                 data_sufficient: False}

  NEXT STEPS:
    1. Chunk Q1 earnings report and audit docs into 512-token segments
    2. Embed with text-embedding-3-small and store in Chroma/Pinecone
    3. Register retrieval as a skill and re-run eval suite

→ Chunk and embed your knowledge corpus
→ Register retrieval as a new skill in SkillRegistry
→ Re-run eval suite to confirm improvement
============================================================

Level 1 — Prompt tuning          (minutes, free)
     │  still failing after 3 rounds?
     ▼
Level 2 — Add/improve skills     (hours, cheap)
     │  still failing on reasoning/architecture?
     ▼
Level 3 — Code/harness evolution (hours, cheap — runs overnight)
     │  still failing on knowledge?
     ▼
Level 4 — RAG                    (hours, medium cost)
     │  still failing on reasoning style/pattern?
     ▼
Level 5 — Fine-tuning            (days, expensive)

import time

def continuous_monitor(
    agent,
    eval_suite,
    versioned_prompt,
    check_interval_hours=24,
    regression_threshold=0.70,
):
    """Re-evaluate the agent periodically and trigger evolution if scores regress."""
    while True:
        new_tasks = collect_recent_tasks()  # returns list[{"task": ..., "expected": ...}]
        if not new_tasks:
            time.sleep(check_interval_hours * 3600)
            continue

        report = run_eval_suite(eval_suite, versioned_prompt.current().prompt)

        if report["avg_score"] < regression_threshold:
            print(f"Score regressed to {report['avg_score']:.2f} — triggering evolution loop")
            new_prompt = evolution_loop(new_tasks, threshold=regression_threshold)
            versioned_prompt.update(new_prompt, score=report["avg_score"], trigger="auto_regression")
        else:
            print(f"Score healthy: {report['avg_score']:.2f}")

        time.sleep(check_interval_hours * 3600)