Evaluating Claude Opus 5 vs Fable 5

pip install -U anthropic weave wandb pytest

ANTHROPIC_API_KEY=your_anthropic_key
WANDB_API_KEY=your_wandb_key
WEAVE_PROJECT=your_wandb_entity/Opus-5-vs-Fable

ANTHROPIC_MODEL_OPUS=claude-opus-5
ANTHROPIC_MODEL_FABLE=claude-fable-5

MAX_MODEL_OUTPUT_TOKENS=7000
MAX_MODEL_TURNS=35

import os

import anthropic
import weave

WEAVE_PROJECT = "your_wandb_entity/Opus-5-vs-Fable"
ANTHROPIC_MODEL_OPUS = "claude-opus-5"
ANTHROPIC_MODEL_FABLE = "claude-fable-5"
MAX_MODEL_OUTPUT_TOKENS = 7000
MAX_MODEL_TURNS = 35

client = anthropic.Anthropic(
    api_key=os.environ["ANTHROPIC_API_KEY"]
)

weave.init(WEAVE_PROJECT)

print("Environment configured")
print("Output cap:", MAX_MODEL_OUTPUT_TOKENS)
print("Turn limit:", MAX_MODEL_TURNS)


@weave.op
def claude_inference(prompt: str, model: str) -> str:
    response = client.messages.create(
        model=model,
        max_tokens=2000,
        messages=[
            {
                "role": "user",
                "content": prompt,
            }
        ],
    )

    return "".join(
        block.text
        for block in response.content
        if block.type == "text"
    )


prompt = """
Review this function and identify one correctness issue:

def divide(a, b):
    return a / b
"""

opus_result = claude_inference(
    prompt=prompt,
    model=ANTHROPIC_MODEL_OPUS,
)

fable_result = claude_inference(
    prompt=prompt,
    model=ANTHROPIC_MODEL_FABLE,
)

import json
from pathlib import Path
from typing import Any

import weave

TASKS = (
    "legacy-api-modernization",
    "file-to-sqlite-migration",
    "transactional-order-workflow",
)

TEST_RESULT_FIELDS = (
    "passed",
    "failed",
    "errors",
    "skipped",
)

PATCH_VALIDATION_FIELDS = (
    "valid",
    "candidate_compiles",
    "candidate_imports",
    "workspace_gradeable",
    "changed_files_allowed",
    "workspace_contained",
)


def require_fields(
    value: dict[str, Any],
    fields: tuple[str, ...],
    name: str,
):
    """Raise an error when required fields are missing."""
    missing = [field for field in fields if field not in value]

    if missing:
        raise ValueError(
            f"{name} is missing required fields: "
            f"{', '.join(missing)}"
        )


def build_dataset(
    root: str | Path,
    selected_task: str | None = None,
):
    root = Path(root)
    tasks = (selected_task,) if selected_task else TASKS
    rows = []

    for task in tasks:
        manifest_path = root / task / "task_manifest.json"

        manifest = json.loads(
            manifest_path.read_text(encoding="utf-8")
        )

        require_fields(
            manifest,
            ("difficulty", "category", "dataset_version"),
            f"Manifest for {task}",
        )

        rows.append({
            "task": task,
            "difficulty": manifest["difficulty"],
            "category": manifest["category"],
            "manifest_path": manifest_path.as_posix(),
            "dataset_version": manifest["dataset_version"],
        })

    return weave.Dataset(
        name="coding-agent-repair-benchmark",
        rows=rows,
    )


@weave.op()
def comprehensive_repair_scorer(output, **_):
    require_fields(
        output,
        (
            "public_tests",
            "hidden_tests",
            "patch_validation",
            "protected_hashes_before",
            "protected_hashes_after",
        ),
        "Agent output",
    )

    public = output["public_tests"]
    hidden = output["hidden_tests"]
    patch = output["patch_validation"]

    require_fields(
        public,
        TEST_RESULT_FIELDS,
        "Public test results",
    )
    require_fields(
        hidden,
        TEST_RESULT_FIELDS,
        "Hidden test results",
    )
    require_fields(
        patch,
        PATCH_VALIDATION_FIELDS,
        "Patch validation",
    )

    public_total = sum(
        public[field] for field in TEST_RESULT_FIELDS
    )
    hidden_total = sum(
        hidden[field] for field in TEST_RESULT_FIELDS
    )

    patch_valid = all(
        patch[field] for field in PATCH_VALIDATION_FIELDS
    )

    protected_unchanged = (
        output["protected_hashes_before"]
        == output["protected_hashes_after"]
    )

    security_violation_records = output.get(
        "security_violation_records",
        [],
    )
    tool_calls = output.get("tool_calls", [])
    model_calls = output.get("model_calls", [])

    costs = [
        call["weave_total_cost"]
        for call in model_calls
        if call.get("weave_total_cost") is not None
    ]

    return {
        "fully_resolved": all([
            public["failed"] == 0,
            public["errors"] == 0,
            hidden["failed"] == 0,
            hidden["errors"] == 0,
            patch_valid,
            protected_unchanged,
            len(security_violation_records) == 0,
        ]),
        "public_pass_rate": (
            public["passed"] / public_total
            if public_total
            else 0
        ),
        "hidden_pass_rate": (
            hidden["passed"] / hidden_total
            if hidden_total
            else 0
        ),
        "patch_valid": patch_valid,
        "protected_files_unchanged": protected_unchanged,
        "security_violations": len(
            security_violation_records
        ),
        "tool_calls": len(tool_calls),
        "total_tokens": sum(
            call.get("input_tokens", 0)
            + call.get("output_tokens", 0)
            for call in model_calls
        ),
        "estimated_cost": (
            sum(costs) if costs else None
        ),
    }


@weave.op()
async def run_live_evaluation(
    root: str | Path,
    model,
    trials: int = 1,
    selected_task: str | None = None,
):
    evaluation = weave.Evaluation(
        evaluation_name=f"live-{model.model_name}",
        dataset=build_dataset(root, selected_task),
        scorers=[comprehensive_repair_scorer],
        trials=trials,
    )

    return await evaluation.evaluate(model)

from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
import weave

@dataclass
class BenchmarkConfig:
    weave_project: str = "your_wandb_entity/Opus-5-vs-Fable"
    opus_model: str = "claude-opus-5"
    fable_model: str = "claude-fable-5"
    seed: int = 42
    cache_dir: Path = Path("benchmark-cache")
    run_dir: Path = Path("evaluation-runs")
    max_turns: int = 35
    max_tool_calls: int = 75
    max_file_reads: int = 40
    max_searches: int = 20
    max_patch_attempts: int = 12
    max_patch_size: int = 200_000
    max_public_test_runs: int = 4
    max_agent_test_runs: int = 4
    max_wall_seconds: int = 1800
    max_model_output_tokens: int = 7000
    context_compaction_turns: int = 12
    provider_timeout: int = 300
    subprocess_timeout: int = 300
    provider_retries: int = 2
    dataset_version: str = "1.0.0"
    harness_version: str = "2.1.0"

def create_repair_agent(config, model_name):
    settings = {
        "model": model_name,
        "seed": config.seed,
        "harness_version": config.harness_version,
        "dataset_version": config.dataset_version,
        "max_turns": config.max_turns,
        "max_tool_calls": config.max_tool_calls,
        "max_file_reads": config.max_file_reads,
        "max_searches": config.max_searches,
        "max_patch_attempts": config.max_patch_attempts,
        "max_patch_size": config.max_patch_size,
        "max_public_test_runs": config.max_public_test_runs,
        "max_agent_test_runs": config.max_agent_test_runs,
        "max_wall_seconds": config.max_wall_seconds,
        "max_model_output_tokens":
            config.max_model_output_tokens,
        "context_compaction_turns":
            config.context_compaction_turns,
        "provider_timeout": config.provider_timeout,
        "subprocess_timeout": config.subprocess_timeout,
        "provider_retries": config.provider_retries,
    }

    return RepairAgent(
        model_name=model_name,
        seed=config.seed,
        agent_config_hash=digest(settings),
        max_turns=config.max_turns,
        max_tool_calls=config.max_tool_calls,
        max_file_reads=config.max_file_reads,
        max_searches=config.max_searches,
        max_patch_attempts=config.max_patch_attempts,
        max_patch_size=config.max_patch_size,
        max_public_test_runs=config.max_public_test_runs,
        max_agent_test_runs=config.max_agent_test_runs,
        max_wall_seconds=config.max_wall_seconds,
        max_output_tokens=config.max_model_output_tokens,
        context_compaction_turns=
            config.context_compaction_turns,
        provider_timeout=config.provider_timeout,
        subprocess_timeout=config.subprocess_timeout,
        provider_retries=config.provider_retries,
        run_dir=config.run_dir.as_posix(),
        cache_dir=config.cache_dir.as_posix(),
    )

def initialize_weave(project, model_names):
    if not project:
        raise ValueError("WEAVE_PROJECT is missing")

    client = weave.init(project)

    pricing = {
        "claude-fable-5": (
            10.0,
            50.0,
            datetime(2026, 6, 9, tzinfo=timezone.utc),
        ),
        "claude-opus-5": (
            5.0,
            25.0,
            datetime(2026, 7, 24, tzinfo=timezone.utc),
        ),
    }

    selected = {
        name for name in model_names if name in pricing
    }

    existing = {
        cost.llm_id
        for cost in client.query_costs(
            llm_ids=sorted(selected)
        )
    }

    for model_name in sorted(selected - existing):
        input_price, output_price, effective_date = pricing[model_name]

        client.add_cost(
            llm_id=model_name,
            prompt_token_cost=input_price / 1_000_000,
            completion_token_cost=output_price / 1_000_000,
            effective_date=effective_date,
        )

    return client

ROOT = Path(".").resolve()
config = BenchmarkConfig()

SELECTED_TASK = "legacy-api-modernization"
RUNS_PER_MODEL = 1
MODELS = [config.opus_model, config.fable_model]

initialize_weave(
    project=config.weave_project,
    model_names=MODELS,
)

async def evaluate_models():
    results = {}

    for model_name in MODELS:
        agent = create_repair_agent(config, model_name)

        results[model_name] = await run_live_evaluation(
            ROOT,
            agent,
            RUNS_PER_MODEL,
            SELECTED_TASK,
        )

    return results

evaluation_results = await evaluate_models()
evaluation_results