evaluator-creation

name	evaluator-creation
description	Write, test, and upload custom evaluators for code optimization — the fitness functions that drive optimization quality, including the Modal-first GPU evaluator pattern
version	2.1.0
author	Kai Agent
metadata	{"kai":{"tags":["kai","optimization","evaluators","fitness-function","custom-evaluator","gpu","modal"]}}

Evaluator Creation

The evaluator is the most important part of code optimization. It defines what "better" means. A custom evaluator that measures real metrics always outperforms AI-generated or LLM-only evaluation.

What is an evaluator?

A Python script that scores each code variant produced by the optimization engine. It defines:

• Correctness tests (the code must still produce correct results)
• Performance metrics (what to optimize: latency, memory, throughput, gas)
• Quality thresholds (minimum acceptable behavior)

The optimization engine runs hundreds of variants against your evaluator and evolves toward higher scores.

Evaluator approach

Always write the evaluator yourself. You understand the code, the optimization goal, and what "better" means — an auto-generated evaluator cannot match that context.

Do NOT use create_ai_evaluator. Do NOT skip the evaluator (LLM-only mode). Write it, test it, upload it.

Workflow

Step 1: Activate optimization tools

activate_category("optimization")

Step 2: Analyze the target code

browse_repository_files(workspaceId, repoId)
read_repository_files(workspaceId, repoId, paths="src/hot_path.py")

Understand:

• What the code does (inputs, outputs, side effects)
• What "better" means for this code (faster? less memory? lower gas cost?)
• What inputs are realistic (production patterns, not toy examples)
• What edge cases must still pass (correctness constraints)

Step 3: Write the evaluator locally

Write a Python evaluator script. The script must define an evaluate(program_path) function.

CRITICAL RULES:

1. evaluate() receives a FILE PATH (string), not a module. OpenEvolve passes the path to the evolved program file. You must read or import it yourself.
2. Must return a dict with metric names as keys and float scores as values (higher = better). If you return a plain float, all programs score 0.
3. Include correctness checks that penalize wrong answers.
4. Test multiple input sizes/patterns to avoid overfitting.

How to load the program in the evaluator:

import importlib.util

def evaluate(program_path):
    # Load the evolved program as a module
    spec = importlib.util.spec_from_file_location("evolved", program_path)
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    
    # Now use module.function_name() to test it
    # ...
    return {"combined_score": score, "correctness": 1.0, "performance": score}

For static analysis (code quality, patterns) you can also just read the source:

def evaluate(program_path):
    source = open(program_path).read()
    # Analyze source code patterns...
    return {"combined_score": score, "correctness": 1.0, "quality": score}

Example: Performance evaluator

import importlib.util
import time
import statistics

def evaluate(program_path):
    """Score a sorting implementation on speed and correctness."""
    spec = importlib.util.spec_from_file_location("evolved", program_path)
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)

    test_cases = [
        list(range(1000, 0, -1)),      # worst case
        list(range(1000)),              # best case
        [42] * 500 + list(range(500)), # mixed
    ]
    
    # Correctness check
    for tc in test_cases:
        result = module.sort_func(tc[:])
        if result != sorted(tc):
            return {"combined_score": 0.0, "correctness": 0.0, "performance": 0.0}
    
    # Performance measurement
    times = []
    large_input = list(range(10000, 0, -1))
    for _ in range(20):
        start = time.perf_counter()
        module.sort_func(large_input[:])
        elapsed = time.perf_counter() - start
        times.append(elapsed)
    
    median_ms = statistics.median(times) * 1000
    perf_score = max(0, 1000 / (median_ms + 1))
    combined = perf_score  # correctness passed, so combined = performance
    return {"combined_score": combined, "correctness": 1.0, "performance": perf_score}

Example: Static analysis evaluator (for code quality optimization)

import ast
import re

def evaluate(program_path):
    """Score code quality and patterns."""
    source = open(program_path).read()
    
    # Check it parses
    try:
        tree = ast.parse(source)
    except SyntaxError:
        return {"combined_score": 0.0, "correctness": 0.0, "quality": 0.0}
    
    # Check required class/functions exist
    names = {node.name for node in ast.walk(tree) if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef))}
    required = {"MyClass", "my_method"}
    if not required.issubset(names):
        return {"combined_score": 0.0, "correctness": 0.0, "quality": 0.0}
    
    # Performance patterns
    score = 0.5
    if "asyncio.gather" in source:
        score += 0.2  # parallelized I/O
    if source.count("await self.connect()") <= 3:
        score += 0.2  # reduced redundant calls
    if len(source.splitlines()) < 400:
        score += 0.1  # concise
    
    quality = min(score, 1.0)
    return {"combined_score": quality, "correctness": 1.0, "quality": quality}

Step 4: Test the evaluator locally

Run it in the terminal against the actual target code before uploading:

python3 -c "
import importlib.util
spec = importlib.util.spec_from_file_location('target', 'target_code.py')
mod = importlib.util.module_from_spec(spec)
spec.loader.exec_module(mod)

from evaluator import evaluate
score = evaluate(mod)
print(f'Score: {score}')
"

Verify:

• Returns a positive number for correct, fast code
• Returns 0 or near-zero for incorrect code
• Scores vary meaningfully between good and bad implementations
• Runs in reasonable time (< 30 seconds per evaluation)
• No hardcoded paths or environment-specific dependencies

Fix any issues before uploading. A broken evaluator wastes compute on garbage runs.

Step 5: Upload the evaluator

Use the single-step create_evaluator_from_code tool — pass the Python code directly as a string:

create_evaluator_from_code(workspaceId, repoId, code=evaluator_code, name="Sort Performance Evaluator")
# → returns { evaluatorId: "..." }

Returns evaluatorId — use this when starting the optimization. No presigned URLs, no HTTP PUT, no finalize step.

Step 6: Verify the upload

list_code_evaluators(workspaceId, repoId)
get_evaluator_details(workspaceId, repoId, evaluatorId)

GPU Evaluator Pattern

When the optimization target requires GPU (ML inference, CUDA kernels, tensor operations), pick the option that matches what the workspace has connected. Default to Option A when Modal is available — it's zero-setup, zero user infrastructure, and keeps credentials out of the sandbox.

Decision order

1. Option A — Modal via workspace proxy (default if MODAL_SERVER_URL is set)
2. Option B — Modal via user tokens (if MODAL_TOKEN_ID + MODAL_TOKEN_SECRET are set but MODAL_SERVER_URL is not)
3. Option C — User's own GPU infrastructure (RunPod, AWS EC2, Lambda, on-prem)
4. Option D — CPU approximation (last resort)

Don't skip up the list. If Modal is connected, use it — don't ask the user to deploy a RunPod pod or stand up an HTTP scoring server. Those are fallback paths for when Modal isn't available.

Option A — Modal via workspace proxy (default)

Preconditions: MODAL_SERVER_URL is set in the sandbox. The kai-compute/modal skill explains how that gets there (Agent Settings → Modal integration). No user action required at runtime.

Two small files — a scorer you deploy to Modal, and a thin evaluator you upload to Kai.

1. scorer.py — a Modal GPU function. Deploy once with modal deploy scorer.py.

import modal

app = modal.App("kai-evaluator")

image = modal.Image.debian_slim(python_version="3.11").pip_install(
    "torch", "numpy"  # add whatever the target code needs
)

@app.function(gpu="A10G", image=image, timeout=120)
def score(code: str) -> dict:
    """Run the candidate code on GPU and return a Kai score dict."""
    import time
    import statistics
    import torch

    # Correctness setup — run the candidate in an isolated namespace
    ns: dict = {}
    try:
        exec(code, ns)
    except Exception:
        return {"combined_score": 0.0, "correctness": 0.0, "performance": 0.0}

    target = ns.get("run_inference")  # name the candidate must expose
    if target is None:
        return {"combined_score": 0.0, "correctness": 0.0, "performance": 0.0}

    # Correctness check against a reference output
    ref_input = torch.randn(1, 128, device="cuda")
    try:
        out = target(ref_input)
        if not torch.is_tensor(out) or out.shape[0] != 1:
            return {"combined_score": 0.0, "correctness": 0.0, "performance": 0.0}
    except Exception:
        return {"combined_score": 0.0, "correctness": 0.0, "performance": 0.0}

    # Performance: median over warm runs on GPU
    # ... warm up ...
    for _ in range(5):
        _ = target(ref_input)
    torch.cuda.synchronize()

    times = []
    for _ in range(20):
        t0 = time.perf_counter()
        _ = target(ref_input)
        torch.cuda.synchronize()
        times.append(time.perf_counter() - t0)

    median_ms = statistics.median(times) * 1000
    perf = max(0.0, 1000.0 / (median_ms + 1.0))
    return {"combined_score": perf, "correctness": 1.0, "performance": perf}

Choose the GPU type by workload — T4 or A10G for quantized / sub-13B models, A100/H100 only when memory or bandwidth demand it. Cheaper GPUs mean faster iteration loops and lower credit burn.

2. evaluator.py — a thin bridge. Upload this to Kai.

import modal

# Lookup the deployed scoring function by (app_name, function_name).
_score_fn = modal.Function.lookup("kai-evaluator", "score")

def evaluate(program_path: str) -> dict:
    """Kai evaluator: ships the candidate to Modal and returns the score dict."""
    code = open(program_path).read()
    return _score_fn.remote(code)

Self-test locally before uploading (the evaluator must be reachable from wherever Kai runs it — which requires Modal auth in that env; flag this to the user if the test fails with an auth error):

python3 -c "
from evaluator import evaluate
r = evaluate('target_code.py')
print(r)
assert isinstance(r, dict) and r.get('combined_score', 0) > 0
print('PASS')
"

Then upload:

create_evaluator_from_code(workspaceId, repoId, code=evaluator_code, name="Modal GPU Evaluator")

Each optimization iteration, Kai reads the candidate, your bridge ships it to Modal's proxy, Modal runs it on a real GPU, and the score comes back. The proxy is rate-limited (~500 req/min per workspace), so tight loops are fine but parallel .map() calls should batch candidates when you run comparison sweeps outside the optimization engine.

Option B — Modal via user tokens

Same template as Option A. The only difference: MODAL_TOKEN_ID and MODAL_TOKEN_SECRET are real Modal tokens the user pasted, not proxy credentials. Treat them as sensitive — don't echo them, don't write them to disk. Everything else is identical.

Option C — User's own GPU infrastructure

Fallback for when Modal isn't an option but the user has their own GPU (RunPod, AWS EC2, Lambda Labs, on-prem cluster). More setup: the user stands up an HTTP scoring server on their GPU; the Kai-side evaluator is an HTTP bridge. Use this only if the user explicitly opts out of Modal or their workload has a binding reason to run on their infra (data residency, custom CUDA kernels, etc.).

1. Remote HTTP scoring server (runs on user's GPU machine):

# gpu_scorer.py
from fastapi import FastAPI
import torch

app = FastAPI()

@app.post("/score")
async def score(payload: dict):
    code = payload["code"]
    # exec, benchmark, return
    # ... same logic as Option A's score() ...
    return {"combined_score": measured, "correctness": 1.0, "performance": measured}

Walk the user through deploying it and get back the endpoint URL.

2. Kai-side bridge evaluator:

import urllib.request
import json

GPU_ENDPOINT = "https://user-gpu-server.example.com/score"

def evaluate(program_path: str) -> dict:
    code = open(program_path).read()
    req = urllib.request.Request(
        GPU_ENDPOINT,
        data=json.dumps({"code": code}).encode(),
        headers={"Content-Type": "application/json"},
        method="POST",
    )
    with urllib.request.urlopen(req, timeout=120) as resp:
        return json.loads(resp.read())

Option D — CPU approximation (last resort)

If no GPU is reachable, write a CPU-only evaluator that approximates the GPU metric — algorithm complexity, allocation patterns, small-input microbenchmarks, operation-count analysis. Evolution still converges in the right direction, but the final score doesn't reflect real GPU performance. Tell the user explicitly that results are directional, not ground-truth. Always prefer Options A-C when any of them are reachable.

Anti-patterns

• No evaluator at all: LLM-only mode is a last resort, not a default
• Only testing one input: Optimization will overfit to that single case
• Missing correctness checks: If you only measure speed, evolved code might return wrong answers fast
• Unrealistic inputs: Test data should match production usage patterns
• Single timing measurement: Use statistical measures (median of N runs), never single runs
• Too-broad scope: Testing everything means nothing gets optimized well
• Uploading without local testing: A broken evaluator wastes compute and produces garbage

Saving patterns

After a successful optimization, save the evaluator pattern as a workspace learning:

workspace_learnings_add(category="optimization", content="For [repo]: custom evaluator measuring [metric] on [input pattern] produced [X]x improvement. Evaluator ID: [id]")