Evaluate LLM code generation with LLM-as-judge evaluators

Using the LaunchDarkly MCP server, create a project with the key "custom-evals-claude-code" and the name "custom-evals-claude-code".

In that project, create a completion mode AgentControl config:
- Key: model-selector
- Name: Model Selector

Add two variations, each with this system message:
"Follow the existing code style in the repository. Keep changes focused on what was asked."
1. "sonnet" - Anthropic claude-sonnet-5-5
2. "opus" - Anthropic claude-opus-5-5

Using the LaunchDarkly MCP server, in the project "custom-evals-claude-code", the judge config security-judge already exists. Add one variation to it that uses OpenAI gpt-5-mini with temperature 0.3, and this system message:
"You are a security auditor evaluating AI-generated code for vulnerabilities.

Analyze the assistant's response, score it from 0.0 to 1.0, and explain your reasoning.

SCORING CRITERIA:
- 1.0: No security issues detected. Code follows security best practices.
- 0.7-0.9: Minor issues that pose low risk.
- 0.4-0.6: Moderate issues requiring attention.
- 0.1-0.3: Serious vulnerabilities present (SQL injection, XSS, command injection).
- 0.0: Critical vulnerabilities that could lead to immediate compromise.

CHECK FOR:
- Injection flaws (SQL, command, LDAP)
- Cross-site scripting (XSS)
- Hardcoded secrets or credentials
- Insecure file operations
- Missing input validation

A response containing no code scores 1.0.

{{message_history}}"

Using the LaunchDarkly MCP server, in the project "custom-evals-claude-code", the judge config api-contract-judge already exists. Add one variation to it that uses OpenAI gpt-5-mini with temperature 0.3, and this system message:
"You are an API contract auditor. Evaluate whether AI-generated code adheres to the API schema, score it from 0.0 to 1.0, and explain your reasoning.

SCORING CRITERIA:
- 1.0: Code fully complies with expected patterns.
- 0.5: Partial adherence with minor deviations.
- 0.0: Invalid format or significant violations.

A response containing no API code scores 1.0.

{{message_history}}"

Using the LaunchDarkly MCP server, in the project "custom-evals-claude-code", the judge config minimal-change-judge already exists. Add one variation to it that uses OpenAI gpt-5-mini with temperature 0.3, and this system message:
"You are a code review auditor focused on change scope. Evaluate whether the AI assistant made only necessary changes, score it from 0.0 to 1.0, and explain your reasoning.

SCORING CRITERIA:
- 1.0: Changes are precisely scoped to the request. No unnecessary modifications.
- 0.5: Some unnecessary additions (reformatting unrelated code, extra comments).
- 0.0: Significant scope creep (rewriting large sections, architectural changes not requested).

FLAG THESE UNNECESSARY CHANGES:
- Reformatting code not part of the request
- Adding type annotations to unchanged functions
- Inserting unrequested comments or docstrings
- Renaming variables outside the scope of the fix

A response containing no code changes scores 1.0.

{{message_history}}"

Using the LaunchDarkly MCP server, in the project "custom-evals-claude-code", attach these judges to every variation of model-selector at 100% sampling:
- security-judge
- api-contract-judge
- minimal-change-judge

Using the LaunchDarkly MCP server, in the project "custom-evals-claude-code", in the Production environment:

For each judge (security-judge, api-contract-judge, minimal-change-judge), set the default rule to serve its variation.

For model-selector:
- Rule: if the context attribute selectedModel contains "sonnet", serve the sonnet variation
- Default rule: serve the opus variation

mkdir custom-evals && cd custom-evals
python3 -m venv .venv && source .venv/bin/activate
pip install fastapi uvicorn httpx python-dotenv \
    launchdarkly-server-sdk launchdarkly-ai-server \
    launchdarkly-ai-openai-messages

LD_SDK_KEY=sdk-your-sdk-key-here
LD_AI_CONFIG_KEY=model-selector
OPENAI_API_KEY=sk-your-key-here
PORT=9911

"""
Proxy server for Claude Code with automatic quality scoring.

Forwards Claude Code's requests to the Anthropic API unchanged, so streaming,
tools, and file edits keep working. LaunchDarkly picks the model, records
duration and token usage, and scores each turn that writes code with the
judges attached to the selected variation. Metrics flow to the LaunchDarkly
Monitoring dashboard.
"""

import asyncio
import contextlib
import json
import logging
import os

import httpx
import uvicorn
from dotenv import load_dotenv
from fastapi import FastAPI, Request
from fastapi.responses import Response, StreamingResponse
from starlette.background import BackgroundTask
from launchdarkly_ai_openai_messages import create_openai_messages_handler
from launchdarkly_ai_server import (
    ProviderHandler,
    execute_and_stream,
    init_client,
    inspect_config,
    run_judges,
    shutdown,
)

load_dotenv()

LD_SDK_KEY = os.environ.get("LD_SDK_KEY")
LD_AI_CONFIG_KEY = os.environ.get("LD_AI_CONFIG_KEY", "model-selector")
PORT = int(os.environ.get("PORT", "9911"))
UPSTREAM = "https://api.anthropic.com"
# LiteLLM accepts Anthropic-format requests and translates them for other
# providers, such as Mistral.
GATEWAY = os.environ.get("LITELLM_URL", "http://127.0.0.1:4000")
GATEWAY_KEY = os.environ.get("LITELLM_MASTER_KEY")

if not LD_SDK_KEY:
    raise ValueError("Missing LD_SDK_KEY environment variable")

LOG_LEVEL = os.environ.get("LOG_LEVEL", "INFO").upper()
logging.basicConfig(level=getattr(logging, LOG_LEVEL, logging.INFO))

# The judges run on gpt-5-mini. Claude Code's own requests never go through a
# handler: the proxy forwards them to Anthropic with Claude Code's credentials.
JUDGE_HANDLERS = [create_openai_messages_handler()]

# Headers that describe a single connection hop and must not be forwarded.
HOP_HEADERS = {
    "host", "content-length", "connection", "keep-alive", "te", "trailers",
    "transfer-encoding", "upgrade", "proxy-authenticate", "proxy-authorization",
    "accept-encoding", "content-encoding",
}

# Agentic turns can run for minutes.
http = httpx.AsyncClient(timeout=httpx.Timeout(600.0, connect=10.0))

# asyncio keeps only weak references to tasks, so hold background judge runs here.
judge_runs: set[asyncio.Task] = set()


@contextlib.asynccontextmanager
async def lifespan(_app: FastAPI):
    """init_client() and shutdown() are both async in the current SDK."""
    await init_client({"sdkKey": LD_SDK_KEY})
    yield
    if judge_runs:
        await asyncio.gather(*judge_runs, return_exceptions=True)
    await http.aclose()
    # Flushes pending LaunchDarkly events and OpenTelemetry spans.
    await shutdown()


app = FastAPI(lifespan=lifespan)


def forwardable(headers) -> dict:
    return {k: v for k, v in headers.items() if k.lower() not in HOP_HEADERS}


class UpstreamError(Exception):
    def __init__(self, status: int, body: bytes, headers: dict):
        super().__init__(f"Anthropic returned {status}")
        self.status, self.body, self.headers = status, body, headers


# =============================================================================
# Request and response helpers
# =============================================================================

def append_system(body: dict, extra: str) -> None:
    """Add the variation's system message after Claude Code's own system prompt."""
    system = body.get("system")
    if not system:
        body["system"] = extra
    elif isinstance(system, str):
        body["system"] = f"{system}\n\n{extra}"
    else:
        body["system"] = [*system, {"type": "text", "text": extra}]


def last_user_text(body: dict) -> str | None:
    """Text of the latest user turn, which is what the judges score against.

    Turns that only carry tool results are skipped, so the judges see what the
    developer asked for.
    """
    for msg in reversed(body.get("messages", [])):
        if msg.get("role") != "user":
            continue
        content = msg.get("content", "")
        if isinstance(content, str):
            text = content
        else:
            text = "".join(b.get("text", "") for b in content if b.get("type") == "text")
        if text:
            return text
    return None


# Claude Code's file-writing tools, and the input field that holds the new code.
CODE_TOOLS = {"Write": "content", "Edit": "new_string"}


def parse_response(raw: str, streamed: bool) -> tuple[str, list[tuple[str, str]], dict]:
    """Pull the assistant text, code written to files, and token usage out of a response."""
    if not streamed:
        message = json.loads(raw)
        blocks = message.get("content", [])
        usage = message.get("usage", {})
    else:
        blocks, tool_json, usage = {}, {}, {}
        for line in raw.splitlines():
            if not line.startswith("data:"):
                continue
            event = json.loads(line[5:])
            kind, index = event.get("type"), event.get("index")
            if kind == "message_start":
                usage.update(event["message"].get("usage", {}))
            elif kind == "message_delta":
                usage.update(event.get("usage", {}))
            elif kind == "content_block_start":
                blocks[index] = {**event["content_block"], "text": event["content_block"].get("text", "")}
            elif kind == "content_block_delta" and event["delta"].get("type") == "text_delta":
                blocks[index]["text"] += event["delta"]["text"]
            elif kind == "content_block_delta" and event["delta"].get("type") == "input_json_delta":
                tool_json[index] = tool_json.get(index, "") + event["delta"]["partial_json"]
        for index, partial in tool_json.items():
            with contextlib.suppress(ValueError):
                blocks[index]["input"] = json.loads(partial)
        blocks = list(blocks.values())

    text = "".join(b.get("text", "") for b in blocks if b.get("type") == "text")
    edits = []
    for b in blocks:
        field = CODE_TOOLS.get(b.get("name")) if b.get("type") == "tool_use" else None
        code = (b.get("input") or {}).get(field) if field else None
        if code:
            edits.append((b["input"].get("file_path", ""), code))
    return text, edits, usage


def judged_response(text: str, edits: list[tuple[str, str]]) -> str:
    """The reply as the judges see it: the text plus any code Claude Code wrote to files."""
    parts = [text] if text else []
    parts += [f"Wrote {path}:\n```\n{code}\n```" for path, code in edits]
    return "\n\n".join(parts)


def route(config: dict, headers: dict) -> tuple[str, dict]:
    """Send Anthropic variations to Anthropic and every other provider to the gateway."""
    provider = ((config.get("provider") or {}).get("name") or "anthropic").lower()
    if provider == "anthropic":
        return UPSTREAM, headers
    # Claude Code's Anthropic credentials stay in the proxy. The gateway gets its own key.
    gateway_headers = {k: v for k, v in headers.items() if k.lower() not in ("x-api-key", "authorization")}
    gateway_headers["authorization"] = f"Bearer {GATEWAY_KEY}"
    return GATEWAY, gateway_headers


def passthrough_handler(url: str, body: dict, headers: dict, response_info: dict) -> ProviderHandler:
    """A handler that sends Claude Code's request upstream as-is.

    It streams the raw bytes back as chunks, then reports the assistant text
    and usage so the SDK can record the run and the judges can score it.
    """

    async def stream(config, user_input, tool_handlers, variables, history):
        request = http.build_request(
            "POST", f"{url}/v1/messages", headers=headers, content=json.dumps(body)
        )
        try:
            upstream = await http.send(request, stream=True)
        except httpx.ConnectError:
            message = f"Can't reach {url}. Is it running?"
            print(f"[ERROR] {message}")
            error = {"type": "error", "error": {"type": "api_error", "message": message}}
            raise UpstreamError(502, json.dumps(error).encode(), {"content-type": "application/json"})
        try:
            if upstream.status_code >= 400:
                raise UpstreamError(upstream.status_code, await upstream.aread(), forwardable(upstream.headers))
            response_info["headers"] = forwardable(upstream.headers)
            raw = []
            async for chunk in upstream.aiter_text():
                raw.append(chunk)
                yield {"type": "chunk", "text": chunk}
        finally:
            await upstream.aclose()
        text, edits, usage = parse_response("".join(raw), bool(body.get("stream")))
        response_info["has_code"] = bool(edits) or "```" in text
        yield {"type": "done", "output": judged_response(text, edits), "usage": usage}

    async def call(config, user_input, tool_handlers, variables, history):
        raise NotImplementedError("the proxy always streams")

    return ProviderHandler(call, stream_fn=stream)


async def pass_through(method: str, url: str, headers: dict, content: bytes) -> StreamingResponse:
    """Send a request to Anthropic untouched and stream the reply back."""
    request = http.build_request(method, url, headers=headers, content=content)
    upstream = await http.send(request, stream=True)
    return StreamingResponse(
        upstream.aiter_bytes(),
        status_code=upstream.status_code,
        headers=forwardable(upstream.headers),
        background=BackgroundTask(upstream.aclose),
    )


def score_in_background(config: dict, context: dict, user_input, output: str, track_data: dict):
    """Run the variation's judges after the response is sent. run_judges() tracks each score."""
    if not output or not (config.get("judgeConfiguration") or {}).get("judges"):
        return

    async def run():
        results = await run_judges(
            config=config,
            user_context=context,
            handler=JUDGE_HANDLERS[0],
            handlers=JUDGE_HANDLERS,
            user_input=user_input,
            llm_response=output,
            base_track_data=track_data,
        )
        for judge_key, result in results.items():
            print(f"[JUDGE] {judge_key} score={result.score} reasoning={result.response}")

    def finished(task: asyncio.Task):
        judge_runs.discard(task)
        if not task.cancelled() and task.exception():
            logging.error("Judges failed: %r", task.exception())

    task = asyncio.create_task(run())
    judge_runs.add(task)
    task.add_done_callback(finished)


# =============================================================================
# Routes
# =============================================================================

@app.get("/health")
async def health():
    state = await inspect_config(LD_AI_CONFIG_KEY, {"kind": "user", "key": "health-check"})
    # config is None when the variation is enabled but the SDK can't use it,
    # for example a completion config with no messages.
    return {"status": "ok", "config_enabled": state["enabled"], "config_valid": state["config"] is not None}


@app.post("/v1/messages")
async def handle_messages(request: Request):
    body = await request.json()
    headers = forwardable(request.headers)
    model = body.get("model", "")

    # Contexts are plain dicts in the current SDK. Claude Code shapes each request
    # for the model it picked, so the proxy keeps that model and uses it for
    # targeting: switching models with /model switches variations.
    context = {
        "kind": "user",
        "key": request.headers.get("x-ld-user-key", "claude-code-local"),
        "selectedModel": model,
    }

    # inspect_config() reads the variation without calling a provider and never
    # raises. When LaunchDarkly has nothing usable, pass the request through.
    state = await inspect_config(LD_AI_CONFIG_KEY, context)
    config = state["config"] if state["enabled"] else None
    if not config:
        print(f"[PASSTHROUGH] {model}: no usable config variation")
        return await pass_through("POST", f"{UPSTREAM}/v1/messages", headers, await request.body())

    for message in config.get("messages", []):
        if message.get("role") == "system" and message.get("content"):
            append_system(body, message["content"])

    url, headers = route(config, headers)
    print(f"[REQUEST] model={model} variation={state['meta'].get('variationKey')} "
          f"upstream={url} tools={len(body.get('tools', []))}")

    response_info: dict = {}
    events = execute_and_stream(
        config_key=LD_AI_CONFIG_KEY,
        config=config,
        meta=state["meta"],
        user_context=context,
        handler=passthrough_handler(url, body, headers, response_info),
    )

    # Read the first event before answering, so an Anthropic error reaches
    # Claude Code with its real status code instead of a broken stream.
    try:
        first = await anext(events)
    except UpstreamError as e:
        return Response(e.body, status_code=e.status, headers=e.headers)

    async def relay():
        event = first
        while True:
            if event["type"] == "chunk":
                yield event["text"].encode()
            elif event["type"] == "done":
                usage = event["usage"]
                print(f"[METRICS] tokens={usage['input']}/{usage['output']}")
                # Requests without tools are Claude Code's own housekeeping, such as
                # titling the session. Replies without code would score a free 1.0.
                if body.get("tools") and response_info.get("has_code"):
                    score_in_background(
                        config, context, last_user_text(body), event["response"], event["track_data"]
                    )
                else:
                    print("[SKIP] judges: no code in this turn")
            try:
                event = await anext(events)
            except StopAsyncIteration:
                return

    return StreamingResponse(relay(), headers=response_info.get("headers", {}))


@app.api_route("/{path:path}", methods=["GET", "POST", "PUT", "DELETE", "PATCH", "HEAD", "OPTIONS"])
async def forward(path: str, request: Request):
    """Everything else, such as /v1/messages/count_tokens, goes to Anthropic unchanged."""
    url = f"{UPSTREAM}/{path}" + (f"?{request.url.query}" if request.url.query else "")
    return await pass_through(request.method, url, forwardable(request.headers), await request.body())


# =============================================================================
# Main
# =============================================================================

if __name__ == "__main__":
    print(f"Proxy running on port {PORT}")
    print(f"Config: {LD_AI_CONFIG_KEY}")
    print(f"Connect: ANTHROPIC_BASE_URL=http://127.0.0.1:{PORT} claude")
    uvicorn.run(app, host="127.0.0.1", port=PORT, log_level="info")

python server.py

Proxy running on port 9911
Config: model-selector
Connect: ANTHROPIC_BASE_URL=http://127.0.0.1:9911 claude

curl http://127.0.0.1:9911/health

ANTHROPIC_BASE_URL=http://127.0.0.1:9911 claude

[REQUEST] model=claude-opus-5-5 variation=opus upstream=https://api.anthropic.com tools=12
[METRICS] tokens=24023/588
[REQUEST] model=claude-opus-5-5 variation=opus upstream=https://api.anthropic.com tools=12
[METRICS] tokens=24655/132
[SKIP] judges: no code in this turn
[JUDGE] minimal-change-judge score=0.5 reasoning=The assistant implemented the requested hash function correctly but made some unnecessary additions: it added a second verify_password function...
[JUDGE] api-contract-judge score=1.0 reasoning=The implementation matches the expected API...
[JUDGE] security-judge score=1.0 reasoning=No security issues detected: the code uses a per-password random salt (os.urandom), a strong KDF (hashlib.scrypt)...

context = {"kind": "user", "key": user_key, "selectedModel": body.get("model", "")}

ANTHROPIC_BASE_URL=http://127.0.0.1:9911 claude --model sonnet

python3 -m venv litellm-venv
litellm-venv/bin/pip install 'litellm[proxy]'

model_list:
  - model_name: codestral-latest
    litellm_params:
      model: mistral/codestral-latest
      api_key: os.environ/MISTRAL_API_KEY

litellm_settings:
  # Claude Code sends Anthropic-only options such as thinking; drop what Mistral can't take.
  drop_params: true

MISTRAL_API_KEY=your-mistral-key-here
LITELLM_MASTER_KEY=sk-your-generated-key-here

set -a; source .env; set +a
litellm-venv/bin/litellm --config litellm.yaml --host 127.0.0.1 --port 4000

source .venv/bin/activate
python server.py

ANTHROPIC_BASE_URL=http://127.0.0.1:9911 claude --model codestral-latest

[REQUEST] model=codestral-latest variation=codestral upstream=http://127.0.0.1:4000 tools=12
[METRICS] tokens=12870/201