mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-29 06:33:28 +00:00
feat(cookbook): capability profiling protocol, labeling and profile selection
Co-Authored-By: bot_apk <apk@cognition.ai>
This commit is contained in:
Devin AI
parent
64a286be8c
commit
b1f72a0ca1
8 changed files with 1833 additions and 4 deletions
33
cookbook/capability_routing_pilot/PROFILING_PLAN.md
Normal file
33
cookbook/capability_routing_pilot/PROFILING_PLAN.md
Normal file
|
|
@ -0,0 +1,33 @@
|
|||
# Capability profiling protocol
|
||||
|
||||
## Current phase
|
||||
|
||||
This phase labels the pinned public SWE-bench dev candidates with GPT-5.4-mini and drafts an availability-aware task manifest. The hard label budget is $5. The completed label run cost $0.109114 for 220 candidates. No solver, Jev, smoke, or other model calls are part of this phase
|
||||
|
||||
The candidate corpus is `princeton-nlp/SWE-bench` dev at revision `e48e2bd1e9fecd5bbd641e9414ac59da9f2e69f6`, excluding the five held-out IDs listed in the draft manifest. The original five training tasks remain eligible, but no earlier outcomes may be reused. The labeler sees only sanitized problem statements and uses the existing rubric, a 120-second timeout, no retries, and concurrency bounded at four. Label records are append-only and do not write to `costs_aux.jsonl`
|
||||
|
||||
Image discovery uses registry manifest requests only and does not pull images. Each supported category is walked in deterministic seed-hash order until 30 images are confirmed or its candidates are exhausted. The process records every inspected candidate and leaves the rest explicitly not inspected. It prefers HEAD before reading manifest content, paces requests at one per second, uses at most four workers, and honors bounded `Retry-After` backoff without retry loops. Only HTTP 404 is recorded as unavailable. Authentication, rate-limit, and other errors remain errors and are not selected. No local Docker credentials were present, so discovery uses the anonymous registry token flow
|
||||
|
||||
## Draft cohort
|
||||
|
||||
The deterministic selection seed is `litellm-capability-profiling-v1`. Candidates are ordered by `sha256(seed + task_id)` within each category, and the first 30 confirmed public-image tasks are selected from `repository_debugging` and `feature_implementation`. `shell_build` and `algorithms_data` are explicitly unsupported. Unknown labels are excluded from solves and reported. The manifest remains a draft until reviewed and frozen
|
||||
|
||||
The effective API ceiling is $185, consisting of a $5 labeling allowance and a $180 solver allowance. The solver limit is 180 runs, at most 60 tasks across three solver models. A future run uses $1 per run, at most 50 steps, and a 600-second timeout, with three workers. Keep at least 50 GiB free. Process images in batches of three and clean them up after validation and solving. Validate each small image batch with gold and noop runs before solving. Compile an intermediate capability map after each batch with `final=false`. Freeze the reviewed manifest before any solve
|
||||
|
||||
## Profile records and routing
|
||||
|
||||
Task-level records use tenant `public_dev_poc`. Profile keys include category, model alias, the concrete returned model version, deployment ID when reported, harness and budget versions, and a canonical harness-configuration hash. Infrastructure attempts contribute spend and latency but not graded evidence. Repeated graded attempts for one task and key are rejected
|
||||
|
||||
Profiles use Beta(1,1) posterior summaries and a 5th-percentile lower bound. At least 13 graded tasks are required before a cell can qualify. This is a necessary threshold, not a general adequacy claim. Selection has no global fallback. The map builder reads only `profiling_runs.jsonl`, never the prior `runs.jsonl`
|
||||
|
||||
## Resource estimates
|
||||
|
||||
The prior 15 recorded solver durations give a mean of 187.882 seconds and a p90 of 272.907 seconds. For 180 solver runs over three workers, the corresponding estimates are 11,272.907 seconds (about 3.13 hours) at the mean and 16,374.392 seconds (about 4.55 hours) at the p90. This excludes gold/noop validation. A 60-task cohort requires 20 gold-validation batches and 20 noop-validation batches; their duration is unestimated because the available evaluator summaries contain no timing fields
|
||||
|
||||
At the prior mean cost of $0.377 per run, 180 runs would cost $67.86. The solver worst case is $180. The effective worst-case API ceiling including the $5 labeling allowance is $185
|
||||
|
||||
Storage is calculated from the compressed layer sizes in confirmed selected manifests. Total selected storage is the sum of those sizes. Peak batch storage is the maximum sum for any three-task batch before cleanup. These values remain unavailable until registry discovery returns confirmed manifest sizes
|
||||
|
||||
## Evidence boundaries
|
||||
|
||||
Do not modify or reuse outcomes from `runs.jsonl`, `runs_raw/`, `eval_gold/`, `labels.json`, `costs_aux.jsonl`, or `tasks.json`. Keep held-out evaluator logs and gold patches local. Do not include credentials, gold patches, evaluator output, or private logs in the cookbook artifacts. The cancelled held-out commit is provenance only and must not be executed
|
||||
|
|
@ -141,6 +141,7 @@ def _response_metadata(response: httpx.Response, payload: dict[str, object], lat
|
|||
return {
|
||||
"call_id": headers.get("x-litellm-call-id"),
|
||||
"response_model": payload.get("model") or headers.get("x-litellm-response-model"),
|
||||
"deployment_id": headers.get("x-litellm-model-id") or headers.get("x-litellm-deployment-id"),
|
||||
"reported_version": payload.get("version")
|
||||
or headers.get("x-litellm-model-version")
|
||||
or headers.get("x-litellm-version"),
|
||||
|
|
@ -184,7 +185,12 @@ def _problem_statement_text(content: object) -> str:
|
|||
return ""
|
||||
|
||||
|
||||
def _offline_label(client: httpx.Client, task_id: str, problem_statement: str) -> dict[str, object]:
|
||||
def _offline_label(
|
||||
client: httpx.Client,
|
||||
task_id: str,
|
||||
problem_statement: str,
|
||||
record_auxiliary: bool = True,
|
||||
) -> dict[str, object]:
|
||||
body = {
|
||||
"model": LABEL_MODEL,
|
||||
"max_completion_tokens": 400,
|
||||
|
|
@ -208,8 +214,15 @@ def _offline_label(client: httpx.Client, task_id: str, problem_statement: str) -
|
|||
metadata = (
|
||||
_response_metadata(response, payload, latency_s)
|
||||
if response is not None
|
||||
else {"call_id": None, "response_model": None, "reported_version": None, "cost_usd": None,
|
||||
"latency_s": latency_s, "usage": {}}
|
||||
else {
|
||||
"call_id": None,
|
||||
"response_model": None,
|
||||
"deployment_id": None,
|
||||
"reported_version": None,
|
||||
"cost_usd": None,
|
||||
"latency_s": latency_s,
|
||||
"usage": {},
|
||||
}
|
||||
)
|
||||
choices = payload.get("choices")
|
||||
choice = choices[0] if isinstance(choices, list) and choices and isinstance(choices[0], dict) else {}
|
||||
|
|
@ -218,7 +231,8 @@ def _offline_label(client: httpx.Client, task_id: str, problem_statement: str) -
|
|||
label, rationale, parse_ok = parse_label_response(raw_text)
|
||||
if error is not None:
|
||||
label, rationale, parse_ok = "unknown", None, False
|
||||
_append_auxiliary_call("offline_label", task_id, LABEL_MODEL, metadata)
|
||||
if record_auxiliary:
|
||||
_append_auxiliary_call("offline_label", task_id, LABEL_MODEL, metadata)
|
||||
return {
|
||||
**metadata,
|
||||
"label": label,
|
||||
|
|
|
|||
585
cookbook/capability_routing_pilot/poc/profile_candidates.py
Normal file
585
cookbook/capability_routing_pilot/poc/profile_candidates.py
Normal file
|
|
@ -0,0 +1,585 @@
|
|||
import argparse
|
||||
from datetime import UTC, datetime
|
||||
from email.utils import parsedate_to_datetime
|
||||
import fcntl
|
||||
import hashlib
|
||||
import json
|
||||
import os
|
||||
import stat
|
||||
import threading
|
||||
import time
|
||||
from concurrent.futures import Future, ThreadPoolExecutor, as_completed
|
||||
from dataclasses import dataclass
|
||||
from functools import partial
|
||||
from pathlib import Path
|
||||
from typing import Final
|
||||
|
||||
import httpx
|
||||
|
||||
from poc import label
|
||||
from poc.guarded_model import valid_cost
|
||||
from poc.prepare import _load_rows
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
HELDOUT_IDS: Final = frozenset(
|
||||
{
|
||||
"pylint-dev__astroid-1164",
|
||||
"pylint-dev__astroid-2240",
|
||||
"pydicom__pydicom-793",
|
||||
"pvlib__pvlib-python-1191",
|
||||
"pylint-dev__astroid-927",
|
||||
}
|
||||
)
|
||||
LABEL_BUDGET_USD = 5.0
|
||||
LABEL_RESERVATION_USD = 1.0
|
||||
LABEL_WORKERS = 4
|
||||
IMAGE_WORKERS = 4
|
||||
IMAGE_REQUEST_INTERVAL_S = 1.0
|
||||
IMAGE_MAX_BACKOFF_S = 30.0
|
||||
REGISTRY_AUTH_URL = "https://auth.docker.io/token"
|
||||
REGISTRY_URL = "https://registry-1.docker.io"
|
||||
MANIFEST_ACCEPT = ", ".join(
|
||||
(
|
||||
"application/vnd.oci.image.index.v1+json",
|
||||
"application/vnd.docker.distribution.manifest.list.v2+json",
|
||||
"application/vnd.oci.image.manifest.v1+json",
|
||||
"application/vnd.docker.distribution.manifest.v2+json",
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class Candidate:
|
||||
task_id: str
|
||||
problem_statement: str
|
||||
image: str
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class LabelResult:
|
||||
task_id: str
|
||||
primary_category: str
|
||||
raw_response: str
|
||||
cost_usd: float | None
|
||||
call_id: str | None
|
||||
returned_model: str | None
|
||||
deployment_id: str | None
|
||||
error: str | None
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class ImageResult:
|
||||
task_id: str
|
||||
image: str
|
||||
exists: bool | None
|
||||
digest: str | None
|
||||
compressed_size_bytes: int | None
|
||||
error: str | None
|
||||
manifest_response_received: bool
|
||||
|
||||
|
||||
class LabelBudget:
|
||||
def __init__(self, committed_spend_usd: float) -> None:
|
||||
self._lock = threading.Lock()
|
||||
self._committed_spend_usd = committed_spend_usd
|
||||
self._reserved_usd = 0.0
|
||||
self._stopped = False
|
||||
|
||||
def reserve(self) -> bool:
|
||||
with self._lock:
|
||||
if self._stopped or self._committed_spend_usd + self._reserved_usd + LABEL_RESERVATION_USD > LABEL_BUDGET_USD:
|
||||
return False
|
||||
self._reserved_usd += LABEL_RESERVATION_USD
|
||||
return True
|
||||
|
||||
def settle(self, cost_usd: float | None, failed: bool) -> bool:
|
||||
with self._lock:
|
||||
self._reserved_usd -= LABEL_RESERVATION_USD
|
||||
if cost_usd is None or failed:
|
||||
self._stopped = True
|
||||
return False
|
||||
self._committed_spend_usd += cost_usd
|
||||
if self._committed_spend_usd > LABEL_BUDGET_USD:
|
||||
self._stopped = True
|
||||
return not self._stopped
|
||||
|
||||
def stop(self) -> None:
|
||||
with self._lock:
|
||||
self._stopped = True
|
||||
|
||||
@property
|
||||
def committed_spend_usd(self) -> float:
|
||||
with self._lock:
|
||||
return self._committed_spend_usd
|
||||
|
||||
|
||||
class RegistryPacer:
|
||||
def __init__(self) -> None:
|
||||
self._lock = threading.Lock()
|
||||
self._next_request_at = 0.0
|
||||
self._blocked = False
|
||||
|
||||
def wait(self) -> bool:
|
||||
while True:
|
||||
with self._lock:
|
||||
if self._blocked:
|
||||
return False
|
||||
now: Final = time.monotonic()
|
||||
delay: Final = max(0.0, self._next_request_at - now)
|
||||
if delay == 0.0:
|
||||
self._next_request_at = now + IMAGE_REQUEST_INTERVAL_S
|
||||
return True
|
||||
time.sleep(delay)
|
||||
|
||||
def backoff(self, delay_s: float) -> None:
|
||||
with self._lock:
|
||||
self._next_request_at = max(self._next_request_at, time.monotonic() + delay_s)
|
||||
|
||||
def block(self) -> None:
|
||||
with self._lock:
|
||||
self._blocked = True
|
||||
|
||||
@property
|
||||
def blocked(self) -> bool:
|
||||
with self._lock:
|
||||
return self._blocked
|
||||
|
||||
|
||||
def _registry_get(
|
||||
client: httpx.Client,
|
||||
pacer: RegistryPacer,
|
||||
url: str,
|
||||
params: dict[str, str] | None = None,
|
||||
headers: dict[str, str] | None = None,
|
||||
method: str = "GET",
|
||||
) -> httpx.Response | None:
|
||||
if not pacer.wait():
|
||||
return None
|
||||
first: Final = client.request(method, url, params=params, headers=headers)
|
||||
if first.status_code != 429:
|
||||
return first
|
||||
delay: Final = _backoff_seconds(first)
|
||||
if delay is None:
|
||||
pacer.block()
|
||||
return first
|
||||
pacer.backoff(delay)
|
||||
if not pacer.wait():
|
||||
return None
|
||||
second: Final = client.request(method, url, params=params, headers=headers)
|
||||
if second.status_code == 429:
|
||||
pacer.block()
|
||||
return second
|
||||
|
||||
|
||||
def _backoff_seconds(response: httpx.Response) -> float | None:
|
||||
retry_after: Final = response.headers.get("Retry-After")
|
||||
if retry_after is None:
|
||||
return 1.0
|
||||
try:
|
||||
delay = float(retry_after)
|
||||
except ValueError:
|
||||
try:
|
||||
retry_at = parsedate_to_datetime(retry_after)
|
||||
except (TypeError, ValueError, OverflowError):
|
||||
return 1.0
|
||||
if retry_at.tzinfo is None:
|
||||
retry_at = retry_at.replace(tzinfo=UTC)
|
||||
delay = (retry_at - datetime.now(UTC)).total_seconds()
|
||||
return None if delay > IMAGE_MAX_BACKOFF_S else max(1.0, delay)
|
||||
|
||||
|
||||
def _candidates() -> tuple[Candidate, ...]:
|
||||
return tuple(
|
||||
Candidate(
|
||||
task_id=task_id,
|
||||
problem_statement=label._sanitize_label_text(problem_statement),
|
||||
image=f"swebench/sweb.eval.x86_64.{task_id.replace('__', '_1776_').lower()}",
|
||||
)
|
||||
for row in _load_rows()
|
||||
if isinstance((task_id := row.get("instance_id")), str)
|
||||
and task_id not in HELDOUT_IDS
|
||||
and isinstance((problem_statement := row.get("problem_statement")), str)
|
||||
)
|
||||
|
||||
|
||||
def _jsonl_rows(path: Path) -> tuple[dict[str, object], ...]:
|
||||
if not path.exists():
|
||||
return ()
|
||||
rows = tuple(json.loads(line) for line in path.read_text().splitlines() if line.strip())
|
||||
if not all(isinstance(row, dict) for row in rows):
|
||||
raise ValueError(f"{path.name} must contain JSON objects")
|
||||
return tuple(row for row in rows if isinstance(row, dict))
|
||||
|
||||
|
||||
def _image_candidates() -> tuple[Candidate, ...]:
|
||||
rows: Final = _jsonl_rows(ROOT / "profiling_labels.jsonl")
|
||||
category_by_id: Final = {
|
||||
task_id: category
|
||||
for row in rows
|
||||
if isinstance((task_id := row.get("task_id")), str)
|
||||
and task_id not in HELDOUT_IDS
|
||||
and isinstance((category := row.get("primary_category")), str)
|
||||
}
|
||||
if len(category_by_id) != len(rows):
|
||||
raise ValueError("profiling_labels.jsonl contains duplicate or invalid image candidates")
|
||||
return tuple(
|
||||
Candidate(
|
||||
task_id=task_id,
|
||||
problem_statement="",
|
||||
image=f"swebench/sweb.eval.x86_64.{task_id.replace('__', '_1776_').lower()}",
|
||||
)
|
||||
for task_id in category_by_id
|
||||
)
|
||||
|
||||
|
||||
def _existing_label_state(path: Path) -> tuple[float, frozenset[str]]:
|
||||
rows: Final = _jsonl_rows(path)
|
||||
task_ids: Final = tuple(row.get("task_id") for row in rows)
|
||||
if not all(isinstance(task_id, str) for task_id in task_ids):
|
||||
raise ValueError("profiling_labels.jsonl contains a missing or invalid task_id")
|
||||
if len(task_ids) != len(set(task_ids)):
|
||||
raise ValueError("profiling_labels.jsonl contains duplicate task IDs")
|
||||
costs: Final = tuple(row.get("cost_usd") for row in rows)
|
||||
if not all(valid_cost(cost) for cost in costs):
|
||||
raise ValueError("profiling_labels.jsonl contains incomplete cost accounting")
|
||||
return sum(float(cost) for cost in costs), frozenset(task_id for task_id in task_ids if isinstance(task_id, str))
|
||||
|
||||
|
||||
def _label_one(client: httpx.Client, candidate: Candidate) -> LabelResult:
|
||||
result: Final = label._offline_label(
|
||||
client,
|
||||
candidate.task_id,
|
||||
candidate.problem_statement,
|
||||
record_auxiliary=False,
|
||||
)
|
||||
primary: Final = result.get("label")
|
||||
raw_response: Final = result.get("raw_label_text")
|
||||
cost: Final = result.get("cost_usd")
|
||||
call_id: Final = result.get("call_id")
|
||||
returned_model: Final = result.get("response_model")
|
||||
deployment_id: Final = result.get("deployment_id")
|
||||
error: Final = result.get("error")
|
||||
return LabelResult(
|
||||
task_id=candidate.task_id,
|
||||
primary_category=primary if isinstance(primary, str) else "unknown",
|
||||
raw_response=raw_response if isinstance(raw_response, str) else "",
|
||||
cost_usd=float(cost) if valid_cost(cost) else None,
|
||||
call_id=call_id if isinstance(call_id, str) else None,
|
||||
returned_model=returned_model if isinstance(returned_model, str) else None,
|
||||
deployment_id=deployment_id if isinstance(deployment_id, str) else None,
|
||||
error=error if isinstance(error, str) else None,
|
||||
)
|
||||
|
||||
|
||||
def _append_label(path: Path, result: LabelResult) -> None:
|
||||
row: Final = {
|
||||
"task_id": result.task_id,
|
||||
"tenant_id": "public_dev_poc",
|
||||
"primary_category": result.primary_category,
|
||||
"secondary_label": None,
|
||||
"raw_response": result.raw_response,
|
||||
"cost_usd": result.cost_usd,
|
||||
"call_id": result.call_id,
|
||||
"returned_model": result.returned_model,
|
||||
"deployment_id": result.deployment_id,
|
||||
}
|
||||
if result.error is not None:
|
||||
row["error"] = result.error
|
||||
with path.open("a") as output:
|
||||
fcntl.flock(output, fcntl.LOCK_EX)
|
||||
output.write(json.dumps(row, separators=(",", ":"), ensure_ascii=False) + "\n")
|
||||
output.flush()
|
||||
fcntl.flock(output, fcntl.LOCK_UN)
|
||||
|
||||
|
||||
def label_candidates() -> None:
|
||||
candidates: Final = _candidates()
|
||||
path: Final = ROOT / "profiling_labels.jsonl"
|
||||
committed_spend, existing_ids = _existing_label_state(path)
|
||||
pending_candidates: Final = tuple(candidate for candidate in candidates if candidate.task_id not in existing_ids)
|
||||
if not pending_candidates:
|
||||
print(f"labels_complete=true labeled={len(existing_ids)} spend_usd={committed_spend:.6f}")
|
||||
return
|
||||
key_path: Final = ROOT / ".private" / "profiling_key"
|
||||
if stat.S_IMODE(key_path.stat().st_mode) != 0o600:
|
||||
raise PermissionError("profiling key must have mode 0600")
|
||||
os.environ["LITELLM_PROXY_API_KEY"] = key_path.read_text().strip()
|
||||
if not os.environ["LITELLM_PROXY_API_KEY"]:
|
||||
raise ValueError("profiling key is empty")
|
||||
budget: Final = LabelBudget(committed_spend)
|
||||
pending: dict[Future[LabelResult], Candidate] = {}
|
||||
next_index = 0
|
||||
stopped_for_error = False
|
||||
with httpx.Client(
|
||||
transport=httpx.HTTPTransport(retries=0),
|
||||
timeout=label.REQUEST_TIMEOUT_S,
|
||||
) as client, ThreadPoolExecutor(max_workers=LABEL_WORKERS) as executor:
|
||||
while next_index < len(pending_candidates) or pending:
|
||||
while next_index < len(pending_candidates) and len(pending) < LABEL_WORKERS:
|
||||
if not budget.reserve():
|
||||
budget.stop()
|
||||
break
|
||||
candidate: Final = pending_candidates[next_index]
|
||||
pending[executor.submit(_label_one, client, candidate)] = candidate
|
||||
next_index += 1
|
||||
if not pending:
|
||||
break
|
||||
future: Final = next(as_completed(tuple(pending)))
|
||||
candidate = pending.pop(future)
|
||||
try:
|
||||
result: Final = future.result()
|
||||
except Exception as exception:
|
||||
result = LabelResult(
|
||||
candidate.task_id,
|
||||
"unknown",
|
||||
"",
|
||||
None,
|
||||
None,
|
||||
None,
|
||||
None,
|
||||
type(exception).__name__,
|
||||
)
|
||||
_append_label(path, result)
|
||||
budget.settle(result.cost_usd, result.error is not None)
|
||||
stopped_for_error = stopped_for_error or result.error is not None or result.cost_usd is None
|
||||
final_rows: Final = _jsonl_rows(path)
|
||||
complete = len(final_rows) == len(candidates)
|
||||
print(
|
||||
f"labels_complete={str(complete).lower()} labeled={len(final_rows)} candidates={len(candidates)} "
|
||||
f"spend_usd={budget.committed_spend_usd:.6f} "
|
||||
f"stop_reason={'error_or_cost_accounting' if stopped_for_error else 'budget_or_complete'}"
|
||||
)
|
||||
|
||||
|
||||
def _as_mapping(value: object) -> dict[str, object] | None:
|
||||
return value if isinstance(value, dict) else None
|
||||
|
||||
|
||||
def _manifest_digest(response: httpx.Response) -> str:
|
||||
header_digest: Final = response.headers.get("Docker-Content-Digest")
|
||||
if header_digest is not None:
|
||||
return header_digest
|
||||
return f"sha256:{hashlib.sha256(response.content).hexdigest()}"
|
||||
|
||||
|
||||
def _compressed_size(payload: dict[str, object]) -> int:
|
||||
layers: Final = payload.get("layers")
|
||||
layer_sizes: Final = (
|
||||
tuple(
|
||||
int(layer["size"])
|
||||
for item in layers
|
||||
if (layer := _as_mapping(item)) is not None
|
||||
and isinstance(layer.get("size"), int)
|
||||
and not isinstance(layer.get("size"), bool)
|
||||
)
|
||||
if isinstance(layers, list)
|
||||
else ()
|
||||
)
|
||||
return sum(layer_sizes)
|
||||
|
||||
|
||||
def _image_error(
|
||||
candidate: Candidate,
|
||||
error: str,
|
||||
manifest_response_received: bool = False,
|
||||
) -> ImageResult:
|
||||
return ImageResult(candidate.task_id, candidate.image, None, None, None, error, manifest_response_received)
|
||||
|
||||
|
||||
def _lookup_image_response(
|
||||
client: httpx.Client,
|
||||
pacer: RegistryPacer,
|
||||
candidate: Candidate,
|
||||
) -> ImageResult:
|
||||
auth: Final = _registry_get(
|
||||
client,
|
||||
pacer,
|
||||
REGISTRY_AUTH_URL,
|
||||
{"service": "registry.docker.io", "scope": f"repository:{candidate.image}:pull"},
|
||||
)
|
||||
if auth is None:
|
||||
return _image_error(candidate, "registry_rate_limited")
|
||||
if auth.status_code != 200:
|
||||
return _image_error(candidate, f"authorization_http_{auth.status_code}")
|
||||
auth_payload: Final = _as_mapping(auth.json())
|
||||
token: Final = (
|
||||
auth_payload.get("token") or auth_payload.get("access_token")
|
||||
if auth_payload is not None
|
||||
else None
|
||||
)
|
||||
if not isinstance(token, str):
|
||||
return _image_error(candidate, "authorization_response_missing_token")
|
||||
registry_headers: Final = {"Authorization": f"Bearer {token}", "Accept": MANIFEST_ACCEPT}
|
||||
head: Final = _registry_get(
|
||||
client,
|
||||
pacer,
|
||||
f"{REGISTRY_URL}/v2/{candidate.image}/manifests/latest",
|
||||
headers=registry_headers,
|
||||
method="HEAD",
|
||||
)
|
||||
if head is None:
|
||||
return _image_error(candidate, "registry_rate_limited")
|
||||
if head.status_code == 404:
|
||||
return ImageResult(candidate.task_id, candidate.image, False, None, None, None, True)
|
||||
if head.status_code != 200:
|
||||
return _image_error(candidate, f"manifest_head_http_{head.status_code}", True)
|
||||
response: Final = _registry_get(
|
||||
client,
|
||||
pacer,
|
||||
f"{REGISTRY_URL}/v2/{candidate.image}/manifests/latest",
|
||||
headers=registry_headers,
|
||||
)
|
||||
if response is None:
|
||||
return _image_error(candidate, "manifest_body_rate_limited", True)
|
||||
if response.status_code == 404:
|
||||
return ImageResult(candidate.task_id, candidate.image, False, None, None, None, True)
|
||||
if response.status_code != 200:
|
||||
return _image_error(candidate, f"manifest_body_http_{response.status_code}", True)
|
||||
payload: Final = _as_mapping(response.json())
|
||||
if payload is None:
|
||||
return _image_error(candidate, "manifest_response_not_object", True)
|
||||
media_type: Final = payload.get("mediaType")
|
||||
if isinstance(media_type, str) and ("index" in media_type or "manifest.list" in media_type):
|
||||
descriptors: Final = payload.get("manifests")
|
||||
amd64: Final = (
|
||||
next(
|
||||
(
|
||||
descriptor
|
||||
for item in descriptors
|
||||
if (descriptor := _as_mapping(item)) is not None
|
||||
and (platform := _as_mapping(descriptor.get("platform"))) is not None
|
||||
and platform.get("os") == "linux"
|
||||
and platform.get("architecture") in {"amd64", "x86_64"}
|
||||
),
|
||||
None,
|
||||
)
|
||||
if isinstance(descriptors, list)
|
||||
else None
|
||||
)
|
||||
if amd64 is None or not isinstance(amd64.get("digest"), str):
|
||||
return _image_error(candidate, "linux_amd64_manifest_missing", True)
|
||||
child: Final = _registry_get(
|
||||
client,
|
||||
pacer,
|
||||
f"{REGISTRY_URL}/v2/{candidate.image}/manifests/{amd64['digest']}",
|
||||
headers={"Authorization": f"Bearer {token}", "Accept": MANIFEST_ACCEPT},
|
||||
)
|
||||
if child is None:
|
||||
return _image_error(candidate, "registry_rate_limited")
|
||||
if child.status_code == 404:
|
||||
return ImageResult(candidate.task_id, candidate.image, False, _manifest_digest(response), None, None, True)
|
||||
if child.status_code != 200:
|
||||
return _image_error(candidate, f"platform_manifest_http_{child.status_code}", True)
|
||||
child_payload: Final = _as_mapping(child.json())
|
||||
if child_payload is None:
|
||||
return _image_error(candidate, "platform_manifest_response_not_object", True)
|
||||
return ImageResult(
|
||||
candidate.task_id,
|
||||
candidate.image,
|
||||
True,
|
||||
_manifest_digest(child),
|
||||
_compressed_size(child_payload),
|
||||
None,
|
||||
True,
|
||||
)
|
||||
return ImageResult(
|
||||
candidate.task_id,
|
||||
candidate.image,
|
||||
True,
|
||||
_manifest_digest(response),
|
||||
_compressed_size(payload),
|
||||
None,
|
||||
True,
|
||||
)
|
||||
|
||||
|
||||
def _lookup_image(client: httpx.Client, pacer: RegistryPacer, candidate: Candidate) -> ImageResult:
|
||||
try:
|
||||
return _lookup_image_response(client, pacer, candidate)
|
||||
except Exception as exception:
|
||||
return _image_error(candidate, type(exception).__name__)
|
||||
|
||||
|
||||
def discover_images() -> None:
|
||||
candidates: Final = _image_candidates()
|
||||
category_by_id: Final = {
|
||||
task_id: category
|
||||
for row in _jsonl_rows(ROOT / "profiling_labels.jsonl")
|
||||
if isinstance((task_id := row.get("task_id")), str)
|
||||
and isinstance((category := row.get("primary_category")), str)
|
||||
}
|
||||
ordered_categories: Final = tuple(
|
||||
(
|
||||
category,
|
||||
tuple(
|
||||
sorted(
|
||||
(
|
||||
candidate
|
||||
for candidate in candidates
|
||||
if category_by_id.get(candidate.task_id) == category
|
||||
),
|
||||
key=lambda candidate: (
|
||||
hashlib.sha256(f"litellm-capability-profiling-v1{candidate.task_id}".encode()).hexdigest(),
|
||||
candidate.task_id,
|
||||
),
|
||||
)
|
||||
),
|
||||
)
|
||||
for category in ("repository_debugging", "feature_implementation")
|
||||
)
|
||||
pacer: Final = RegistryPacer()
|
||||
with httpx.Client(transport=httpx.HTTPTransport(retries=0), timeout=30.0) as client:
|
||||
with ThreadPoolExecutor(max_workers=IMAGE_WORKERS) as executor:
|
||||
results: tuple[ImageResult, ...] = ()
|
||||
for _, category_candidates in ordered_categories:
|
||||
available = 0
|
||||
offset = 0
|
||||
while offset < len(category_candidates) and available < 30 and not pacer.blocked:
|
||||
batch: Final = category_candidates[offset : offset + IMAGE_WORKERS]
|
||||
batch_results: Final = tuple(
|
||||
executor.map(partial(_lookup_image, client, pacer), batch)
|
||||
)
|
||||
results += batch_results
|
||||
available += sum(result.exists is True for result in batch_results)
|
||||
offset += len(batch)
|
||||
path: Final = ROOT / "profiling_candidate_images.json"
|
||||
path.write_text(
|
||||
json.dumps(
|
||||
[
|
||||
{
|
||||
"task_id": result.task_id,
|
||||
"image": result.image,
|
||||
"exists": result.exists,
|
||||
"digest": result.digest,
|
||||
"compressed_size_bytes": result.compressed_size_bytes,
|
||||
"error": result.error,
|
||||
"manifest_response_received": result.manifest_response_received,
|
||||
}
|
||||
for result in results
|
||||
],
|
||||
indent=2,
|
||||
)
|
||||
+ "\n"
|
||||
)
|
||||
print(
|
||||
f"image_candidates={len(results)} "
|
||||
f"image_available={sum(result.exists is True for result in results)} "
|
||||
f"image_unavailable={sum(result.exists is False for result in results)} "
|
||||
f"image_errors={sum(result.exists is None for result in results)} "
|
||||
f"manifest_responses={sum(result.manifest_response_received for result in results)} "
|
||||
f"candidate_records={len(results)}"
|
||||
)
|
||||
|
||||
|
||||
def main() -> None:
|
||||
parser = argparse.ArgumentParser()
|
||||
parser.add_argument("operation", choices=("images", "labels"))
|
||||
operation: Final = parser.parse_args().operation
|
||||
if operation == "images":
|
||||
discover_images()
|
||||
else:
|
||||
label_candidates()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
279
cookbook/capability_routing_pilot/poc/profile_manifest.py
Normal file
279
cookbook/capability_routing_pilot/poc/profile_manifest.py
Normal file
|
|
@ -0,0 +1,279 @@
|
|||
import argparse
|
||||
import hashlib
|
||||
import json
|
||||
from collections.abc import Sequence
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from typing import Final
|
||||
|
||||
from poc.profiles import PUBLIC_DEV_TENANT, SUPPORTED_CATEGORIES
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
SEED: Final = "litellm-capability-profiling-v1"
|
||||
MAX_PER_CATEGORY: Final = 30
|
||||
ALL_CATEGORIES: Final = (
|
||||
"repository_debugging",
|
||||
"feature_implementation",
|
||||
"shell_build",
|
||||
"algorithms_data",
|
||||
"unknown",
|
||||
)
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class CandidateLabel:
|
||||
task_id: str
|
||||
primary_category: str
|
||||
secondary_label: str | None
|
||||
returned_model: str | None = None
|
||||
deployment_id: str | None = None
|
||||
call_id: str | None = None
|
||||
cost_usd: float | None = None
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class ImageAvailability:
|
||||
task_id: str
|
||||
image: str
|
||||
exists: bool | None
|
||||
digest: str | None
|
||||
compressed_size_bytes: int | None
|
||||
error: str | None
|
||||
manifest_response_received: bool
|
||||
|
||||
|
||||
def _hash_order(task_id: str) -> str:
|
||||
return hashlib.sha256(f"{SEED}{task_id}".encode()).hexdigest()
|
||||
|
||||
|
||||
def _candidate_record(
|
||||
task_id: str,
|
||||
label: CandidateLabel,
|
||||
image: ImageAvailability | None,
|
||||
) -> dict[str, object]:
|
||||
status: Final = (
|
||||
"not_inspected"
|
||||
if image is None
|
||||
else "available"
|
||||
if image.exists is True
|
||||
else "unavailable"
|
||||
if image.exists is False
|
||||
else "error"
|
||||
)
|
||||
return {
|
||||
"task_id": task_id,
|
||||
"tenant_id": PUBLIC_DEV_TENANT,
|
||||
"primary_category": label.primary_category,
|
||||
"secondary_label": label.secondary_label,
|
||||
"returned_model": label.returned_model,
|
||||
"deployment_id": label.deployment_id,
|
||||
"call_id": label.call_id,
|
||||
"label_cost_usd": label.cost_usd,
|
||||
"image": (
|
||||
image.image
|
||||
if image is not None
|
||||
else f"swebench/sweb.eval.x86_64.{task_id.replace('__', '_1776_').lower()}"
|
||||
),
|
||||
"image_available": image.exists if image is not None else None,
|
||||
"image_lookup_status": status,
|
||||
"image_digest": image.digest if image is not None else None,
|
||||
"compressed_size_bytes": image.compressed_size_bytes if image is not None else None,
|
||||
"image_error": image.error if image is not None else None,
|
||||
"manifest_response_received": image.manifest_response_received if image is not None else False,
|
||||
}
|
||||
|
||||
|
||||
def build_draft_manifest(
|
||||
labels: Sequence[CandidateLabel],
|
||||
images: Sequence[ImageAvailability],
|
||||
) -> dict[str, object]:
|
||||
label_by_id: Final = {label.task_id: label for label in labels}
|
||||
image_by_id: Final = {image.task_id: image for image in images}
|
||||
if len(label_by_id) != len(labels) or len(image_by_id) != len(images):
|
||||
raise ValueError("candidate labels and image results must have unique task IDs")
|
||||
if not set(image_by_id).issubset(label_by_id):
|
||||
raise ValueError("image results must belong to labeled candidates")
|
||||
records: Final = tuple(
|
||||
_candidate_record(task_id, label_by_id[task_id], image_by_id.get(task_id))
|
||||
for task_id in sorted(label_by_id)
|
||||
)
|
||||
category_counts: Final = {
|
||||
category: {
|
||||
"labeled": sum(label.primary_category == category for label in labels),
|
||||
"inspected": sum(
|
||||
label.primary_category == category and label.task_id in image_by_id
|
||||
for label in labels
|
||||
),
|
||||
"with_image": sum(
|
||||
label.primary_category == category and image_by_id[label.task_id].exists is True
|
||||
for label in labels
|
||||
),
|
||||
"image_unavailable": sum(
|
||||
label.primary_category == category and image_by_id[label.task_id].exists is False
|
||||
for label in labels
|
||||
),
|
||||
"image_lookup_errors": sum(
|
||||
label.primary_category == category
|
||||
and label.task_id in image_by_id
|
||||
and image_by_id[label.task_id].exists is None
|
||||
for label in labels
|
||||
),
|
||||
"manifest_responses": sum(
|
||||
label.primary_category == category
|
||||
and label.task_id in image_by_id
|
||||
and image_by_id[label.task_id].manifest_response_received
|
||||
for label in labels
|
||||
),
|
||||
"not_inspected": sum(
|
||||
label.primary_category == category and label.task_id not in image_by_id
|
||||
for label in labels
|
||||
),
|
||||
"selected": 0,
|
||||
"selection_status": "profiled" if category in SUPPORTED_CATEGORIES else "unsupported_category",
|
||||
"selected_ids": (),
|
||||
}
|
||||
for category in ALL_CATEGORIES
|
||||
}
|
||||
selected_ids: Final = {
|
||||
category: tuple(
|
||||
record["task_id"]
|
||||
for record in sorted(
|
||||
(
|
||||
record
|
||||
for record in records
|
||||
if record["primary_category"] == category and record["image_available"] is True
|
||||
),
|
||||
key=lambda record: (_hash_order(str(record["task_id"])), str(record["task_id"])),
|
||||
)[:MAX_PER_CATEGORY]
|
||||
)
|
||||
if category in SUPPORTED_CATEGORIES
|
||||
else ()
|
||||
for category in ALL_CATEGORIES
|
||||
}
|
||||
categories_with_selection: Final = {
|
||||
category: {
|
||||
**counts,
|
||||
"selected": len(selected_ids[category]),
|
||||
"target_shortfall": max(0, MAX_PER_CATEGORY - len(selected_ids[category]))
|
||||
if category in SUPPORTED_CATEGORIES
|
||||
else 0,
|
||||
"target_reached": len(selected_ids[category]) == MAX_PER_CATEGORY
|
||||
if category in SUPPORTED_CATEGORIES
|
||||
else False,
|
||||
"selected_ids": selected_ids[category],
|
||||
}
|
||||
for category, counts in category_counts.items()
|
||||
}
|
||||
return {
|
||||
"frozen": False,
|
||||
"tenant_id": PUBLIC_DEV_TENANT,
|
||||
"seed": SEED,
|
||||
"rule": f"Per supported category, sort public-image task IDs by sha256(seed + task_id) and select the first {MAX_PER_CATEGORY}; do not freeze this draft",
|
||||
"profiled_categories": SUPPORTED_CATEGORIES,
|
||||
"unsupported_categories": ("shell_build", "algorithms_data"),
|
||||
"heldout_exclusion_list": (
|
||||
"pylint-dev__astroid-1164",
|
||||
"pylint-dev__astroid-2240",
|
||||
"pydicom__pydicom-793",
|
||||
"pvlib__pvlib-python-1191",
|
||||
"pylint-dev__astroid-927",
|
||||
),
|
||||
"candidate_count": len(records),
|
||||
"unknown_count": category_counts["unknown"]["labeled"],
|
||||
"image_unavailable_count": sum(record["image_available"] is False for record in records),
|
||||
"image_lookup_error_count": sum(record["image_lookup_status"] == "error" for record in records),
|
||||
"image_not_inspected_count": sum(record["image_lookup_status"] == "not_inspected" for record in records),
|
||||
"selected_count": sum(len(ids) for ids in selected_ids.values()),
|
||||
"categories": categories_with_selection,
|
||||
"candidates": records,
|
||||
}
|
||||
|
||||
|
||||
def _load_jsonl(path: Path) -> tuple[dict[str, object], ...]:
|
||||
rows: Final = tuple(json.loads(line) for line in path.read_text().splitlines() if line.strip())
|
||||
if not all(isinstance(row, dict) for row in rows):
|
||||
raise ValueError(f"{path.name} must contain JSON objects")
|
||||
return tuple(row for row in rows if isinstance(row, dict))
|
||||
|
||||
|
||||
def _load_labels(path: Path) -> tuple[CandidateLabel, ...]:
|
||||
rows: Final = _load_jsonl(path)
|
||||
labels: Final = tuple(
|
||||
CandidateLabel(
|
||||
task_id=row["task_id"],
|
||||
primary_category=row["primary_category"],
|
||||
secondary_label=row.get("secondary_label"),
|
||||
returned_model=row.get("returned_model"),
|
||||
deployment_id=row.get("deployment_id"),
|
||||
call_id=row.get("call_id"),
|
||||
cost_usd=row.get("cost_usd"),
|
||||
)
|
||||
for row in rows
|
||||
if isinstance(row.get("task_id"), str)
|
||||
and isinstance(row.get("primary_category"), str)
|
||||
and (row.get("secondary_label") is None or isinstance(row.get("secondary_label"), str))
|
||||
and (row.get("returned_model") is None or isinstance(row.get("returned_model"), str))
|
||||
and (row.get("deployment_id") is None or isinstance(row.get("deployment_id"), str))
|
||||
and (row.get("call_id") is None or isinstance(row.get("call_id"), str))
|
||||
and (
|
||||
row.get("cost_usd") is None
|
||||
or isinstance(row.get("cost_usd"), (int, float))
|
||||
and not isinstance(row.get("cost_usd"), bool)
|
||||
)
|
||||
)
|
||||
if len(labels) != len(rows):
|
||||
raise ValueError(f"{path.name} contains invalid candidate label fields")
|
||||
return labels
|
||||
|
||||
|
||||
def _load_images(path: Path) -> tuple[ImageAvailability, ...]:
|
||||
values: Final = json.loads(path.read_text())
|
||||
if not isinstance(values, list) or not all(isinstance(value, dict) for value in values):
|
||||
raise ValueError(f"{path.name} must contain a JSON array of image results")
|
||||
images: Final = tuple(
|
||||
ImageAvailability(
|
||||
task_id=value["task_id"],
|
||||
image=value["image"],
|
||||
exists=value["exists"],
|
||||
digest=value.get("digest"),
|
||||
compressed_size_bytes=value.get("compressed_size_bytes"),
|
||||
error=value.get("error"),
|
||||
manifest_response_received=value.get("manifest_response_received", False),
|
||||
)
|
||||
for value in values
|
||||
if isinstance(value.get("task_id"), str)
|
||||
and isinstance(value.get("image"), str)
|
||||
and (value.get("exists") is None or isinstance(value.get("exists"), bool))
|
||||
and (value.get("digest") is None or isinstance(value.get("digest"), str))
|
||||
and (
|
||||
value.get("compressed_size_bytes") is None
|
||||
or isinstance(value.get("compressed_size_bytes"), int)
|
||||
and not isinstance(value.get("compressed_size_bytes"), bool)
|
||||
)
|
||||
and (value.get("error") is None or isinstance(value.get("error"), str))
|
||||
and isinstance(value.get("manifest_response_received", False), bool)
|
||||
)
|
||||
if len(images) != len(values):
|
||||
raise ValueError(f"{path.name} contains invalid image result fields")
|
||||
return images
|
||||
|
||||
|
||||
def main() -> None:
|
||||
parser = argparse.ArgumentParser()
|
||||
parser.add_argument("--labels", type=Path, default=ROOT / "profiling_labels.jsonl")
|
||||
parser.add_argument("--images", type=Path, default=ROOT / "profiling_candidate_images.json")
|
||||
parser.add_argument("--output", type=Path, default=ROOT / "profiling_manifest_draft.json")
|
||||
arguments = parser.parse_args()
|
||||
manifest: Final = build_draft_manifest(
|
||||
_load_labels(arguments.labels),
|
||||
_load_images(arguments.images),
|
||||
)
|
||||
arguments.output.write_text(json.dumps(manifest, indent=2) + "\n")
|
||||
print(
|
||||
f"manifest={arguments.output} candidates={manifest['candidate_count']} "
|
||||
f"selected={manifest['selected_count']} image_errors={manifest['image_lookup_error_count']}"
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
393
cookbook/capability_routing_pilot/poc/profiles.py
Normal file
393
cookbook/capability_routing_pilot/poc/profiles.py
Normal file
|
|
@ -0,0 +1,393 @@
|
|||
import hashlib
|
||||
import json
|
||||
import math
|
||||
from collections.abc import Mapping, Sequence
|
||||
from dataclasses import asdict, dataclass
|
||||
from enum import Enum
|
||||
from itertools import groupby
|
||||
from pathlib import Path
|
||||
from typing import Final
|
||||
|
||||
from poc.policy import beta_cell
|
||||
|
||||
SUPPORTED_CATEGORIES: Final = (
|
||||
"repository_debugging",
|
||||
"feature_implementation",
|
||||
)
|
||||
PUBLIC_DEV_TENANT: Final = "public_dev_poc"
|
||||
ALIAS_NOTE: Final = (
|
||||
"Model aliases can resolve to changing deployments; returned model versions and deployment IDs are isolated"
|
||||
)
|
||||
BETA_NOTE: Final = (
|
||||
"Beta(1,1) credible bound describes the exchangeable public-dev task distribution for this category, "
|
||||
"not guaranteed per-task success"
|
||||
)
|
||||
MIN_EVIDENCE_NOTE: Final = (
|
||||
"necessary not sufficient: 13/13 is the minimum perfect cell clearing 0.80; not a general adequacy claim"
|
||||
)
|
||||
|
||||
|
||||
class SupportStatus(str, Enum):
|
||||
NO_EVIDENCE = "no_evidence"
|
||||
UNSUPPORTED_ZERO_SUCCESS = "unsupported_zero_success"
|
||||
INSUFFICIENT_EVIDENCE = "insufficient_evidence"
|
||||
EVIDENCED_BELOW_TARGET = "evidenced_below_target"
|
||||
QUALIFIED = "qualified"
|
||||
|
||||
|
||||
class DecisionStatus(str, Enum):
|
||||
SELECTED = "selected"
|
||||
NO_QUALIFIED_MODEL = "no_qualified_model"
|
||||
UNSUPPORTED_CATEGORY = "unsupported_category"
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True, order=True)
|
||||
class ProfileKey:
|
||||
tenant_id: str
|
||||
category: str
|
||||
model_alias: str
|
||||
returned_model_version: str
|
||||
deployment_id: str
|
||||
harness_version: str
|
||||
budget_version: str
|
||||
harness_config_hash: str
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class ProfileVersions:
|
||||
harness_version: str
|
||||
budget_version: str
|
||||
harness_config_hash: str
|
||||
model_versions: tuple[tuple[str, str], ...]
|
||||
model_deployments: tuple[tuple[str, str], ...] = ()
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class HarnessConfig:
|
||||
harness_version: str
|
||||
budget_version: str
|
||||
per_run_budget_usd: float
|
||||
step_limit: int
|
||||
wall_timeout_s: int
|
||||
mini_swe_agent_version: str
|
||||
guard_sha256: str
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class TaskAttempt:
|
||||
task_id: str
|
||||
key: ProfileKey
|
||||
is_infra: bool
|
||||
success: bool | None
|
||||
cost_usd: float
|
||||
latency_s: float
|
||||
primary_label: str
|
||||
secondary_label: str | None
|
||||
manifest_sha256: str
|
||||
harness_sha256: str
|
||||
|
||||
def __post_init__(self) -> None:
|
||||
if not self.task_id or not self.primary_label:
|
||||
raise ValueError("task_id and primary_label must be non-empty")
|
||||
if not math.isfinite(self.cost_usd) or self.cost_usd < 0:
|
||||
raise ValueError("cost_usd must be finite and non-negative")
|
||||
if not math.isfinite(self.latency_s) or self.latency_s < 0:
|
||||
raise ValueError("latency_s must be finite and non-negative")
|
||||
if self.is_infra and self.success is not None:
|
||||
raise ValueError("infrastructure attempts cannot have a graded success value")
|
||||
if not self.is_infra and not isinstance(self.success, bool):
|
||||
raise ValueError("graded attempts require a boolean success value")
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class ProfileProvenance:
|
||||
manifest_sha256: str
|
||||
harness_sha256: str
|
||||
model_alias: str
|
||||
deployment_id: str | None
|
||||
alias_note: str
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class Profile:
|
||||
key: ProfileKey
|
||||
attempt_count: int
|
||||
graded_count: int
|
||||
infra_count: int
|
||||
successes: int
|
||||
raw_rate: float | None
|
||||
beta_mean: float
|
||||
lower_5pct: float
|
||||
total_spend_usd: float
|
||||
cost_per_graded_task_usd: float | None
|
||||
mean_latency_s: float
|
||||
primary_labels: tuple[str, ...]
|
||||
secondary_labels: tuple[str, ...]
|
||||
status: SupportStatus
|
||||
tau: float
|
||||
min_evidence: int
|
||||
provenance: ProfileProvenance
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class Selection:
|
||||
decision: DecisionStatus
|
||||
selected_profile: Profile | None
|
||||
threshold_unmet: bool
|
||||
reason: str
|
||||
|
||||
|
||||
def harness_config_sha256(config: HarnessConfig) -> str:
|
||||
canonical_json: Final = json.dumps(asdict(config), sort_keys=True, separators=(",", ":"))
|
||||
return hashlib.sha256(canonical_json.encode()).hexdigest()
|
||||
|
||||
|
||||
def _record_mapping(value: object) -> Mapping[str, object]:
|
||||
if not isinstance(value, Mapping):
|
||||
raise ValueError("each task attempt must be a JSON object")
|
||||
return value
|
||||
|
||||
|
||||
def _required_string(row: Mapping[str, object], field: str) -> str:
|
||||
value: Final = row.get(field)
|
||||
if not isinstance(value, str) or not value:
|
||||
raise ValueError(f"{field} must be a non-empty string")
|
||||
return value
|
||||
|
||||
|
||||
def _required_number(row: Mapping[str, object], field: str) -> float:
|
||||
value: Final = row.get(field)
|
||||
if isinstance(value, bool) or not isinstance(value, (int, float)) or not math.isfinite(value):
|
||||
raise ValueError(f"{field} must be a finite number")
|
||||
return float(value)
|
||||
|
||||
|
||||
def attempt_from_mapping(value: object) -> TaskAttempt:
|
||||
row: Final = _record_mapping(value)
|
||||
secondary: Final = row.get("secondary_label")
|
||||
success: Final = row.get("success")
|
||||
infra: Final = row.get("is_infra")
|
||||
deployment_id_value: Final = row.get("deployment_id")
|
||||
if not isinstance(infra, bool):
|
||||
raise ValueError("is_infra must be a boolean")
|
||||
if secondary is not None and not isinstance(secondary, str):
|
||||
raise ValueError("secondary_label must be a string or null")
|
||||
if success is not None and not isinstance(success, bool):
|
||||
raise ValueError("success must be a boolean or null")
|
||||
if deployment_id_value is not None and not isinstance(deployment_id_value, str):
|
||||
raise ValueError("deployment_id must be a string or null")
|
||||
key: Final = ProfileKey(
|
||||
tenant_id=_required_string(row, "tenant_id"),
|
||||
category=_required_string(row, "category"),
|
||||
model_alias=_required_string(row, "model_alias"),
|
||||
returned_model_version=_required_string(row, "returned_model_version"),
|
||||
deployment_id=deployment_id_value if isinstance(deployment_id_value, str) else "",
|
||||
harness_version=_required_string(row, "harness_version"),
|
||||
budget_version=_required_string(row, "budget_version"),
|
||||
harness_config_hash=_required_string(row, "harness_config_hash"),
|
||||
)
|
||||
return TaskAttempt(
|
||||
task_id=_required_string(row, "task_id"),
|
||||
key=key,
|
||||
is_infra=infra,
|
||||
success=success,
|
||||
cost_usd=_required_number(row, "cost_usd"),
|
||||
latency_s=_required_number(row, "latency_s"),
|
||||
primary_label=_required_string(row, "primary_label"),
|
||||
secondary_label=secondary,
|
||||
manifest_sha256=_required_string(row, "manifest_sha256"),
|
||||
harness_sha256=_required_string(row, "harness_sha256"),
|
||||
)
|
||||
|
||||
|
||||
def read_attempts_jsonl(path: Path) -> tuple[TaskAttempt, ...]:
|
||||
return tuple(
|
||||
attempt_from_mapping(json.loads(line))
|
||||
for line in path.read_text().splitlines()
|
||||
if line.strip()
|
||||
)
|
||||
|
||||
|
||||
def _status(
|
||||
graded_count: int,
|
||||
successes: int,
|
||||
lower_5pct: float,
|
||||
tau: float,
|
||||
min_evidence: int,
|
||||
) -> SupportStatus:
|
||||
if graded_count == 0:
|
||||
return SupportStatus.NO_EVIDENCE
|
||||
if successes == 0:
|
||||
return SupportStatus.UNSUPPORTED_ZERO_SUCCESS
|
||||
if graded_count < min_evidence:
|
||||
return SupportStatus.INSUFFICIENT_EVIDENCE
|
||||
if lower_5pct < tau:
|
||||
return SupportStatus.EVIDENCED_BELOW_TARGET
|
||||
return SupportStatus.QUALIFIED
|
||||
|
||||
|
||||
def _aggregate(key: ProfileKey, attempts: tuple[TaskAttempt, ...], tau: float, min_evidence: int) -> Profile:
|
||||
graded: Final = tuple(attempt for attempt in attempts if not attempt.is_infra)
|
||||
graded_task_ids: Final = tuple(attempt.task_id for attempt in graded)
|
||||
if len(graded_task_ids) != len(set(graded_task_ids)):
|
||||
raise ValueError(f"duplicate graded attempts for {(key, graded_task_ids)}")
|
||||
manifest_shas: Final = frozenset(attempt.manifest_sha256 for attempt in attempts)
|
||||
harness_shas: Final = frozenset(attempt.harness_sha256 for attempt in attempts)
|
||||
if len(manifest_shas) != 1 or len(harness_shas) != 1:
|
||||
raise ValueError(f"profile provenance changed for {key}")
|
||||
successes: Final = sum(attempt.success is True for attempt in graded)
|
||||
evidence: Final = beta_cell(successes, len(graded), 0.90, None)
|
||||
total_spend: Final = sum(attempt.cost_usd for attempt in attempts)
|
||||
return Profile(
|
||||
key=key,
|
||||
attempt_count=len(attempts),
|
||||
graded_count=len(graded),
|
||||
infra_count=sum(attempt.is_infra for attempt in attempts),
|
||||
successes=successes,
|
||||
raw_rate=successes / len(graded) if graded else None,
|
||||
beta_mean=evidence.posterior_mean,
|
||||
lower_5pct=evidence.ci_low,
|
||||
total_spend_usd=total_spend,
|
||||
cost_per_graded_task_usd=total_spend / len(graded) if graded else None,
|
||||
mean_latency_s=sum(attempt.latency_s for attempt in attempts) / len(attempts),
|
||||
primary_labels=tuple(sorted({attempt.primary_label for attempt in attempts})),
|
||||
secondary_labels=tuple(
|
||||
sorted({attempt.secondary_label for attempt in attempts if attempt.secondary_label is not None})
|
||||
),
|
||||
status=_status(len(graded), successes, evidence.ci_low, tau, min_evidence),
|
||||
tau=tau,
|
||||
min_evidence=min_evidence,
|
||||
provenance=ProfileProvenance(
|
||||
manifest_sha256=next(iter(manifest_shas)),
|
||||
harness_sha256=next(iter(harness_shas)),
|
||||
model_alias=key.model_alias,
|
||||
deployment_id=key.deployment_id or None,
|
||||
alias_note=ALIAS_NOTE,
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
def aggregate_profiles(
|
||||
attempts: Sequence[TaskAttempt],
|
||||
tau: float = 0.80,
|
||||
min_evidence: int = 13,
|
||||
) -> tuple[Profile, ...]:
|
||||
if not 0 <= tau <= 1 or min_evidence < 1:
|
||||
raise ValueError("tau must be within [0, 1] and min_evidence must be positive")
|
||||
ordered: Final = tuple(sorted(attempts, key=lambda attempt: attempt.key))
|
||||
return tuple(
|
||||
_aggregate(key, tuple(group), tau, min_evidence)
|
||||
for key, group in groupby(ordered, key=lambda attempt: attempt.key)
|
||||
)
|
||||
|
||||
|
||||
def _versions_match(profile: Profile, versions: ProfileVersions) -> bool:
|
||||
expected_model_version: Final = dict(versions.model_versions).get(profile.key.model_alias)
|
||||
expected_deployment_id: Final = dict(versions.model_deployments).get(profile.key.model_alias)
|
||||
return (
|
||||
profile.key.harness_version == versions.harness_version
|
||||
and profile.key.budget_version == versions.budget_version
|
||||
and profile.key.harness_config_hash == versions.harness_config_hash
|
||||
and profile.key.returned_model_version == expected_model_version
|
||||
and (expected_deployment_id is None or profile.key.deployment_id == expected_deployment_id)
|
||||
)
|
||||
|
||||
|
||||
def _cheapest(profiles: Sequence[Profile]) -> Profile:
|
||||
return min(
|
||||
profiles,
|
||||
key=lambda profile: (
|
||||
profile.cost_per_graded_task_usd
|
||||
if profile.cost_per_graded_task_usd is not None
|
||||
else math.inf,
|
||||
profile.key.model_alias,
|
||||
profile.key.returned_model_version,
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
def select(
|
||||
tenant_id: str,
|
||||
category: str,
|
||||
versions: ProfileVersions,
|
||||
profiles: Sequence[Profile],
|
||||
tau: float = 0.80,
|
||||
quality_first: bool = False,
|
||||
min_evidence: int = 13,
|
||||
) -> Selection:
|
||||
if category not in SUPPORTED_CATEGORIES:
|
||||
return Selection(DecisionStatus.UNSUPPORTED_CATEGORY, None, False, "unsupported_category")
|
||||
matching: Final = tuple(
|
||||
profile
|
||||
for profile in profiles
|
||||
if profile.key.tenant_id == tenant_id
|
||||
and profile.key.category == category
|
||||
and _versions_match(profile, versions)
|
||||
and profile.tau == tau
|
||||
and profile.min_evidence == min_evidence
|
||||
)
|
||||
qualified: Final = tuple(profile for profile in matching if profile.status is SupportStatus.QUALIFIED)
|
||||
if qualified:
|
||||
return Selection(DecisionStatus.SELECTED, _cheapest(qualified), False, "cheapest_qualified")
|
||||
if quality_first:
|
||||
evidenced: Final = tuple(
|
||||
profile
|
||||
for profile in matching
|
||||
if profile.graded_count >= min_evidence and profile.successes > 0
|
||||
)
|
||||
if evidenced:
|
||||
best_lower_bound: Final = max(profile.lower_5pct for profile in evidenced)
|
||||
best: Final = _cheapest(
|
||||
tuple(profile for profile in evidenced if profile.lower_5pct == best_lower_bound)
|
||||
)
|
||||
return Selection(DecisionStatus.SELECTED, best, True, "quality_first_threshold_unmet")
|
||||
return Selection(DecisionStatus.NO_QUALIFIED_MODEL, None, False, "no_qualified_model")
|
||||
|
||||
|
||||
def compile_map(
|
||||
tenant_id: str,
|
||||
versions: ProfileVersions,
|
||||
profiles: Sequence[Profile],
|
||||
tau: float = 0.80,
|
||||
min_evidence: int = 13,
|
||||
final: bool = False,
|
||||
) -> dict[str, object]:
|
||||
return {
|
||||
"final": final,
|
||||
"tenant_id": tenant_id,
|
||||
"thresholds": {"tau": tau, "min_evidence": min_evidence},
|
||||
"min_evidence_note": MIN_EVIDENCE_NOTE,
|
||||
"interpretation": BETA_NOTE,
|
||||
"category_decisions": {
|
||||
"repository_debugging": "profiled",
|
||||
"feature_implementation": "profiled",
|
||||
"shell_build": DecisionStatus.UNSUPPORTED_CATEGORY.value,
|
||||
"algorithms_data": DecisionStatus.UNSUPPORTED_CATEGORY.value,
|
||||
"unknown": DecisionStatus.UNSUPPORTED_CATEGORY.value,
|
||||
},
|
||||
"versions": {
|
||||
"harness_version": versions.harness_version,
|
||||
"budget_version": versions.budget_version,
|
||||
"harness_config_hash": versions.harness_config_hash,
|
||||
"model_versions": dict(versions.model_versions),
|
||||
"model_deployments": dict(versions.model_deployments),
|
||||
},
|
||||
"profiles": [
|
||||
asdict(profile)
|
||||
for profile in sorted(profiles, key=lambda item: item.key)
|
||||
if profile.key.tenant_id == tenant_id
|
||||
],
|
||||
}
|
||||
|
||||
|
||||
def compile_map_from_profiling_runs(
|
||||
path: Path,
|
||||
tenant_id: str,
|
||||
versions: ProfileVersions,
|
||||
tau: float = 0.80,
|
||||
min_evidence: int = 13,
|
||||
) -> dict[str, object]:
|
||||
if path.name != "profiling_runs.jsonl" or path.is_symlink():
|
||||
raise ValueError("capability maps may only read profiling_runs.jsonl")
|
||||
profiles: Final = aggregate_profiles(read_attempts_jsonl(path), tau, min_evidence)
|
||||
return compile_map(tenant_id, versions, profiles, tau, min_evidence, final=False)
|
||||
|
|
@ -0,0 +1,37 @@
|
|||
import httpx
|
||||
|
||||
from poc.profile_candidates import Candidate, RegistryPacer, _lookup_image
|
||||
|
||||
|
||||
def _candidate() -> Candidate:
|
||||
return Candidate(
|
||||
task_id="example__repo-1",
|
||||
problem_statement="A sanitized problem statement.",
|
||||
image="swebench/sweb.eval.x86_64.example_1776_repo-1",
|
||||
)
|
||||
|
||||
|
||||
def test_registry_404_is_the_only_manifest_status_recorded_as_unavailable() -> None:
|
||||
def handler(request: httpx.Request) -> httpx.Response:
|
||||
if request.url.host == "auth.docker.io":
|
||||
return httpx.Response(200, json={"token": "anonymous"})
|
||||
return httpx.Response(404)
|
||||
|
||||
with httpx.Client(transport=httpx.MockTransport(handler)) as client:
|
||||
result = _lookup_image(client, RegistryPacer(), _candidate())
|
||||
assert result.exists is False
|
||||
assert result.error is None
|
||||
assert result.manifest_response_received
|
||||
|
||||
|
||||
def test_registry_401_is_an_error_not_an_unavailable_image() -> None:
|
||||
def handler(request: httpx.Request) -> httpx.Response:
|
||||
if request.url.host == "auth.docker.io":
|
||||
return httpx.Response(200, json={"token": "anonymous"})
|
||||
return httpx.Response(401)
|
||||
|
||||
with httpx.Client(transport=httpx.MockTransport(handler)) as client:
|
||||
result = _lookup_image(client, RegistryPacer(), _candidate())
|
||||
assert result.exists is None
|
||||
assert result.error == "manifest_head_http_401"
|
||||
assert result.manifest_response_received
|
||||
|
|
@ -0,0 +1,77 @@
|
|||
import hashlib
|
||||
|
||||
from poc.profile_manifest import (
|
||||
CandidateLabel,
|
||||
ImageAvailability,
|
||||
MAX_PER_CATEGORY,
|
||||
SEED,
|
||||
build_draft_manifest,
|
||||
)
|
||||
|
||||
|
||||
def test_manifest_selects_deterministic_public_images_and_marks_unsupported_categories() -> None:
|
||||
labels = tuple(
|
||||
CandidateLabel(f"repo-{index}", "repository_debugging", None)
|
||||
for index in range(35)
|
||||
) + tuple(
|
||||
CandidateLabel(f"feature-{index}", "feature_implementation", None)
|
||||
for index in range(32)
|
||||
) + (
|
||||
CandidateLabel("shell-1", "shell_build", None),
|
||||
CandidateLabel("unknown-1", "unknown", None),
|
||||
)
|
||||
images = tuple(
|
||||
ImageAvailability(
|
||||
label.task_id,
|
||||
f"image/{label.task_id}",
|
||||
True,
|
||||
f"sha256:{label.task_id}",
|
||||
100,
|
||||
None,
|
||||
True,
|
||||
)
|
||||
for label in labels
|
||||
)
|
||||
manifest = build_draft_manifest(tuple(reversed(labels)), tuple(reversed(images)))
|
||||
categories = manifest["categories"]
|
||||
repo_expected = tuple(
|
||||
sorted(
|
||||
(label.task_id for label in labels if label.primary_category == "repository_debugging"),
|
||||
key=lambda task_id: (hashlib.sha256(f"{SEED}{task_id}".encode()).hexdigest(), task_id),
|
||||
)[:MAX_PER_CATEGORY]
|
||||
)
|
||||
feature_expected = tuple(
|
||||
sorted(
|
||||
(label.task_id for label in labels if label.primary_category == "feature_implementation"),
|
||||
key=lambda task_id: (hashlib.sha256(f"{SEED}{task_id}".encode()).hexdigest(), task_id),
|
||||
)[:MAX_PER_CATEGORY]
|
||||
)
|
||||
assert manifest["frozen"] is False
|
||||
assert categories["repository_debugging"]["selected_ids"] == repo_expected
|
||||
assert categories["feature_implementation"]["selected_ids"] == feature_expected
|
||||
assert categories["shell_build"]["selection_status"] == "unsupported_category"
|
||||
assert categories["shell_build"]["selected"] == 0
|
||||
assert categories["algorithms_data"]["selected"] == 0
|
||||
assert manifest["unknown_count"] == 1
|
||||
|
||||
|
||||
def test_manifest_reports_image_unavailable_and_lookup_errors_without_selecting_them() -> None:
|
||||
labels = (
|
||||
CandidateLabel("available", "repository_debugging", None),
|
||||
CandidateLabel("unavailable", "repository_debugging", None),
|
||||
CandidateLabel("error", "repository_debugging", None),
|
||||
)
|
||||
images = (
|
||||
ImageAvailability("available", "image/a", True, "sha256:a", 100, None, True),
|
||||
ImageAvailability("unavailable", "image/b", False, None, None, None, True),
|
||||
ImageAvailability("error", "image/c", None, None, None, "http_429", True),
|
||||
)
|
||||
manifest = build_draft_manifest(labels, images)
|
||||
counts = manifest["categories"]["repository_debugging"]
|
||||
assert counts["labeled"] == 3
|
||||
assert counts["with_image"] == 1
|
||||
assert counts["image_unavailable"] == 1
|
||||
assert counts["image_lookup_errors"] == 1
|
||||
assert counts["selected_ids"] == ("available",)
|
||||
assert manifest["image_unavailable_count"] == 1
|
||||
assert manifest["image_lookup_error_count"] == 1
|
||||
411
cookbook/capability_routing_pilot/tests/test_profiles.py
Normal file
411
cookbook/capability_routing_pilot/tests/test_profiles.py
Normal file
|
|
@ -0,0 +1,411 @@
|
|||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from poc.profiles import (
|
||||
PUBLIC_DEV_TENANT,
|
||||
BETA_NOTE,
|
||||
MIN_EVIDENCE_NOTE,
|
||||
DecisionStatus,
|
||||
HarnessConfig,
|
||||
Profile,
|
||||
ProfileKey,
|
||||
ProfileVersions,
|
||||
SupportStatus,
|
||||
TaskAttempt,
|
||||
aggregate_profiles,
|
||||
compile_map,
|
||||
compile_map_from_profiling_runs,
|
||||
harness_config_sha256,
|
||||
select,
|
||||
)
|
||||
|
||||
MODEL_VERSION = "anthropic/claude-haiku-4-5-20251001"
|
||||
HARNESS_VERSION = "profile_harness_v1"
|
||||
BUDGET_VERSION = "budget_1usd_50_600_v1"
|
||||
CONFIG_HASH = "config-hash"
|
||||
MANIFEST_SHA = "manifest-sha"
|
||||
HARNESS_SHA = "harness-sha"
|
||||
|
||||
|
||||
def _key(
|
||||
model_alias: str = "haiku",
|
||||
returned_model_version: str = MODEL_VERSION,
|
||||
tenant_id: str = PUBLIC_DEV_TENANT,
|
||||
category: str = "repository_debugging",
|
||||
harness_version: str = HARNESS_VERSION,
|
||||
budget_version: str = BUDGET_VERSION,
|
||||
harness_config_hash: str = CONFIG_HASH,
|
||||
deployment_id: str = "deployment-haiku",
|
||||
) -> ProfileKey:
|
||||
return ProfileKey(
|
||||
tenant_id=tenant_id,
|
||||
category=category,
|
||||
model_alias=model_alias,
|
||||
returned_model_version=returned_model_version,
|
||||
deployment_id=deployment_id,
|
||||
harness_version=harness_version,
|
||||
budget_version=budget_version,
|
||||
harness_config_hash=harness_config_hash,
|
||||
)
|
||||
|
||||
|
||||
def _attempts(
|
||||
model_alias: str,
|
||||
successes: int,
|
||||
graded_count: int,
|
||||
cost_per_attempt: float,
|
||||
**key_values: str,
|
||||
) -> tuple[TaskAttempt, ...]:
|
||||
key = _key(model_alias=model_alias, **key_values)
|
||||
return tuple(
|
||||
TaskAttempt(
|
||||
task_id=f"task-{index}",
|
||||
key=key,
|
||||
is_infra=False,
|
||||
success=index < successes,
|
||||
cost_usd=cost_per_attempt,
|
||||
latency_s=float(index + 1),
|
||||
primary_label=key.category,
|
||||
secondary_label=None,
|
||||
manifest_sha256=MANIFEST_SHA,
|
||||
harness_sha256=HARNESS_SHA,
|
||||
)
|
||||
for index in range(graded_count)
|
||||
)
|
||||
|
||||
|
||||
def _versions(
|
||||
models: tuple[tuple[str, str], ...] = (("haiku", MODEL_VERSION),),
|
||||
deployments: tuple[tuple[str, str], ...] = (("haiku", "deployment-haiku"),),
|
||||
harness_version: str = HARNESS_VERSION,
|
||||
budget_version: str = BUDGET_VERSION,
|
||||
harness_config_hash: str = CONFIG_HASH,
|
||||
) -> ProfileVersions:
|
||||
return ProfileVersions(
|
||||
harness_version=harness_version,
|
||||
budget_version=budget_version,
|
||||
harness_config_hash=harness_config_hash,
|
||||
model_versions=models,
|
||||
model_deployments=deployments,
|
||||
)
|
||||
|
||||
|
||||
def _profile(attempts: tuple[TaskAttempt, ...]) -> Profile:
|
||||
return aggregate_profiles(attempts)[0]
|
||||
|
||||
|
||||
def test_cheapest_qualified_model_wins() -> None:
|
||||
profiles = aggregate_profiles(
|
||||
_attempts("haiku", 13, 13, 1.0)
|
||||
+ _attempts(
|
||||
"sonnet",
|
||||
13,
|
||||
13,
|
||||
2.0,
|
||||
returned_model_version="anthropic/claude-sonnet-5",
|
||||
deployment_id="deployment-sonnet",
|
||||
)
|
||||
)
|
||||
decision = select(
|
||||
PUBLIC_DEV_TENANT,
|
||||
"repository_debugging",
|
||||
_versions(
|
||||
models=(("haiku", MODEL_VERSION), ("sonnet", "anthropic/claude-sonnet-5")),
|
||||
deployments=(("haiku", "deployment-haiku"), ("sonnet", "deployment-sonnet")),
|
||||
),
|
||||
profiles,
|
||||
)
|
||||
assert decision.decision is DecisionStatus.SELECTED
|
||||
assert decision.selected_profile is not None
|
||||
assert decision.selected_profile.key.model_alias == "haiku"
|
||||
|
||||
|
||||
def test_more_expensive_model_wins_when_cheaper_one_is_below_threshold() -> None:
|
||||
profiles = aggregate_profiles(
|
||||
_attempts("haiku", 12, 13, 0.1)
|
||||
+ _attempts(
|
||||
"sonnet",
|
||||
13,
|
||||
13,
|
||||
2.0,
|
||||
returned_model_version="anthropic/claude-sonnet-5",
|
||||
deployment_id="deployment-sonnet",
|
||||
)
|
||||
)
|
||||
decision = select(
|
||||
PUBLIC_DEV_TENANT,
|
||||
"repository_debugging",
|
||||
_versions(
|
||||
models=(("haiku", MODEL_VERSION), ("sonnet", "anthropic/claude-sonnet-5")),
|
||||
deployments=(("haiku", "deployment-haiku"), ("sonnet", "deployment-sonnet")),
|
||||
),
|
||||
profiles,
|
||||
)
|
||||
assert decision.decision is DecisionStatus.SELECTED
|
||||
assert decision.selected_profile is not None
|
||||
assert decision.selected_profile.key.model_alias == "sonnet"
|
||||
|
||||
|
||||
def test_missing_zero_and_insufficient_evidence_are_not_selected() -> None:
|
||||
infra_only = TaskAttempt(
|
||||
task_id="infra-task",
|
||||
key=_key(),
|
||||
is_infra=True,
|
||||
success=None,
|
||||
cost_usd=0.5,
|
||||
latency_s=8.0,
|
||||
primary_label="repository_debugging",
|
||||
secondary_label=None,
|
||||
manifest_sha256=MANIFEST_SHA,
|
||||
harness_sha256=HARNESS_SHA,
|
||||
)
|
||||
no_evidence = _profile((infra_only,))
|
||||
zero_success = _profile(_attempts("haiku", 0, 13, 0.1))
|
||||
insufficient = _profile(_attempts("haiku", 12, 12, 0.1))
|
||||
assert no_evidence.status is SupportStatus.NO_EVIDENCE
|
||||
assert zero_success.status is SupportStatus.UNSUPPORTED_ZERO_SUCCESS
|
||||
assert insufficient.status is SupportStatus.INSUFFICIENT_EVIDENCE
|
||||
assert select(PUBLIC_DEV_TENANT, "repository_debugging", _versions(), (),).decision is DecisionStatus.NO_QUALIFIED_MODEL
|
||||
assert (
|
||||
select(
|
||||
PUBLIC_DEV_TENANT,
|
||||
"repository_debugging",
|
||||
_versions(),
|
||||
(no_evidence, zero_success, insufficient),
|
||||
).decision
|
||||
is DecisionStatus.NO_QUALIFIED_MODEL
|
||||
)
|
||||
|
||||
|
||||
def test_infra_attempts_add_cost_and_latency_but_not_graded_count() -> None:
|
||||
key = _key()
|
||||
infra = TaskAttempt(
|
||||
task_id="infra-task",
|
||||
key=key,
|
||||
is_infra=True,
|
||||
success=None,
|
||||
cost_usd=0.5,
|
||||
latency_s=3.0,
|
||||
primary_label="repository_debugging",
|
||||
secondary_label=None,
|
||||
manifest_sha256=MANIFEST_SHA,
|
||||
harness_sha256=HARNESS_SHA,
|
||||
)
|
||||
failed = TaskAttempt(
|
||||
task_id="graded-task",
|
||||
key=key,
|
||||
is_infra=False,
|
||||
success=False,
|
||||
cost_usd=1.5,
|
||||
latency_s=5.0,
|
||||
primary_label="repository_debugging",
|
||||
secondary_label="shell_build",
|
||||
manifest_sha256=MANIFEST_SHA,
|
||||
harness_sha256=HARNESS_SHA,
|
||||
)
|
||||
profile = _profile((infra, failed))
|
||||
assert profile.attempt_count == 2
|
||||
assert profile.graded_count == 1
|
||||
assert profile.infra_count == 1
|
||||
assert profile.successes == 0
|
||||
assert profile.raw_rate == 0.0
|
||||
assert profile.total_spend_usd == 2.0
|
||||
assert profile.cost_per_graded_task_usd == 2.0
|
||||
assert profile.mean_latency_s == 4.0
|
||||
assert profile.status is SupportStatus.UNSUPPORTED_ZERO_SUCCESS
|
||||
|
||||
|
||||
def test_duplicate_graded_attempt_for_task_and_key_is_rejected() -> None:
|
||||
attempt = _attempts("haiku", 1, 1, 1.0)[0]
|
||||
duplicate = TaskAttempt(
|
||||
task_id=attempt.task_id,
|
||||
key=attempt.key,
|
||||
is_infra=False,
|
||||
success=False,
|
||||
cost_usd=0.2,
|
||||
latency_s=1.0,
|
||||
primary_label=attempt.primary_label,
|
||||
secondary_label=None,
|
||||
manifest_sha256=MANIFEST_SHA,
|
||||
harness_sha256=HARNESS_SHA,
|
||||
)
|
||||
with pytest.raises(ValueError, match="duplicate graded attempts"):
|
||||
aggregate_profiles((attempt, duplicate))
|
||||
|
||||
|
||||
def test_beta_bound_uses_closed_form_for_perfect_cells_and_evidence_floor() -> None:
|
||||
profile_12 = _profile(_attempts("haiku", 12, 12, 1.0))
|
||||
profile_13 = _profile(_attempts("haiku", 13, 13, 1.0))
|
||||
profile_12_of_13 = _profile(_attempts("haiku", 12, 13, 1.0))
|
||||
assert profile_12.lower_5pct == pytest.approx(0.05 ** (1 / 13))
|
||||
assert profile_12.status is SupportStatus.INSUFFICIENT_EVIDENCE
|
||||
assert profile_13.lower_5pct == pytest.approx(0.05 ** (1 / 14))
|
||||
assert profile_13.status is SupportStatus.QUALIFIED
|
||||
assert profile_12_of_13.lower_5pct < 0.80
|
||||
assert profile_12_of_13.status is SupportStatus.EVIDENCED_BELOW_TARGET
|
||||
|
||||
|
||||
def test_every_support_status_is_emitted() -> None:
|
||||
infra_only = TaskAttempt(
|
||||
task_id="infra-only",
|
||||
key=_key(),
|
||||
is_infra=True,
|
||||
success=None,
|
||||
cost_usd=0.2,
|
||||
latency_s=1.0,
|
||||
primary_label="repository_debugging",
|
||||
secondary_label=None,
|
||||
manifest_sha256=MANIFEST_SHA,
|
||||
harness_sha256=HARNESS_SHA,
|
||||
)
|
||||
profiles = (
|
||||
_profile((infra_only,)),
|
||||
_profile(_attempts("haiku", 0, 13, 1.0)),
|
||||
_profile(_attempts("haiku", 1, 12, 1.0)),
|
||||
_profile(_attempts("haiku", 12, 13, 1.0)),
|
||||
_profile(_attempts("haiku", 13, 13, 1.0)),
|
||||
)
|
||||
assert {profile.status for profile in profiles} == set(SupportStatus)
|
||||
|
||||
|
||||
def test_unsupported_categories_return_unsupported_category() -> None:
|
||||
for category in ("unknown", "shell_build", "algorithms_data"):
|
||||
decision = select(PUBLIC_DEV_TENANT, category, _versions(), ())
|
||||
assert decision.decision is DecisionStatus.UNSUPPORTED_CATEGORY
|
||||
assert decision.selected_profile is None
|
||||
|
||||
|
||||
def test_tenant_model_harness_budget_and_deployment_versions_are_isolated() -> None:
|
||||
base = _attempts("haiku", 13, 13, 2.0)
|
||||
competitors = (
|
||||
_attempts("haiku", 13, 13, 0.1, tenant_id="other"),
|
||||
_attempts("haiku", 13, 13, 0.1, returned_model_version="other-model"),
|
||||
_attempts("haiku", 13, 13, 0.1, harness_version="other-harness"),
|
||||
_attempts("haiku", 13, 13, 0.1, budget_version="other-budget"),
|
||||
_attempts("haiku", 13, 13, 0.1, harness_config_hash="other-config"),
|
||||
_attempts("haiku", 13, 13, 0.1, deployment_id="other-deployment"),
|
||||
)
|
||||
profiles = aggregate_profiles(base + tuple(item for group in competitors for item in group))
|
||||
decision = select(PUBLIC_DEV_TENANT, "repository_debugging", _versions(), profiles)
|
||||
assert decision.decision is DecisionStatus.SELECTED
|
||||
assert decision.selected_profile is not None
|
||||
assert decision.selected_profile.cost_per_graded_task_usd == 2.0
|
||||
|
||||
|
||||
def test_alias_and_returned_model_versions_split_profiles() -> None:
|
||||
attempts = (
|
||||
_attempts("haiku", 1, 1, 1.0)
|
||||
+ _attempts(
|
||||
"haiku",
|
||||
1,
|
||||
1,
|
||||
1.0,
|
||||
returned_model_version="anthropic/claude-haiku-4-5-20251002",
|
||||
)
|
||||
+ _attempts("haiku-canary", 1, 1, 1.0)
|
||||
)
|
||||
profiles = aggregate_profiles(attempts)
|
||||
assert len(profiles) == 3
|
||||
assert len({profile.key.returned_model_version for profile in profiles}) == 2
|
||||
assert len({profile.key.model_alias for profile in profiles}) == 2
|
||||
|
||||
|
||||
def test_quality_first_is_disabled_by_default_and_only_uses_evidenced_nonzero_profiles() -> None:
|
||||
twelve_successes = _attempts("haiku", 12, 13, 1.0)
|
||||
zero_successes = _attempts(
|
||||
"sonnet",
|
||||
0,
|
||||
13,
|
||||
0.01,
|
||||
returned_model_version="anthropic/claude-sonnet-5",
|
||||
deployment_id="deployment-sonnet",
|
||||
)
|
||||
insufficient = _attempts(
|
||||
"opus",
|
||||
11,
|
||||
12,
|
||||
0.02,
|
||||
returned_model_version="anthropic/claude-opus-5",
|
||||
deployment_id="deployment-opus",
|
||||
)
|
||||
profiles = aggregate_profiles(twelve_successes + zero_successes + insufficient)
|
||||
versions = _versions(
|
||||
models=(
|
||||
("haiku", MODEL_VERSION),
|
||||
("sonnet", "anthropic/claude-sonnet-5"),
|
||||
("opus", "anthropic/claude-opus-5"),
|
||||
),
|
||||
deployments=(
|
||||
("haiku", "deployment-haiku"),
|
||||
("sonnet", "deployment-sonnet"),
|
||||
("opus", "deployment-opus"),
|
||||
),
|
||||
)
|
||||
default_decision = select(PUBLIC_DEV_TENANT, "repository_debugging", versions, profiles)
|
||||
assert default_decision.decision is DecisionStatus.NO_QUALIFIED_MODEL
|
||||
quality_decision = select(
|
||||
PUBLIC_DEV_TENANT,
|
||||
"repository_debugging",
|
||||
versions,
|
||||
profiles,
|
||||
quality_first=True,
|
||||
)
|
||||
assert quality_decision.decision is DecisionStatus.SELECTED
|
||||
assert quality_decision.selected_profile is not None
|
||||
assert quality_decision.selected_profile.key.model_alias == "haiku"
|
||||
assert quality_decision.threshold_unmet
|
||||
assert quality_decision.reason == "quality_first_threshold_unmet"
|
||||
|
||||
|
||||
def test_capability_map_is_intermediate_and_contains_bound_and_version_notes() -> None:
|
||||
profiles = aggregate_profiles(_attempts("haiku", 13, 13, 1.0))
|
||||
capability_map = compile_map(PUBLIC_DEV_TENANT, _versions(), profiles)
|
||||
assert capability_map["final"] is False
|
||||
assert capability_map["min_evidence_note"] == MIN_EVIDENCE_NOTE
|
||||
assert capability_map["interpretation"] == BETA_NOTE
|
||||
assert capability_map["category_decisions"]["shell_build"] == "unsupported_category"
|
||||
assert capability_map["category_decisions"]["algorithms_data"] == "unsupported_category"
|
||||
assert capability_map["versions"]["harness_config_hash"] == CONFIG_HASH
|
||||
|
||||
|
||||
def test_map_builder_rejects_old_runs_jsonl_and_reads_only_profiling_runs(tmp_path: Path) -> None:
|
||||
versions = _versions()
|
||||
old_runs = tmp_path / "runs.jsonl"
|
||||
old_runs.write_text("not valid JSON")
|
||||
with pytest.raises(ValueError, match="only read profiling_runs.jsonl"):
|
||||
compile_map_from_profiling_runs(old_runs, PUBLIC_DEV_TENANT, versions)
|
||||
aliased_old_runs = tmp_path / "profiling_runs.jsonl"
|
||||
aliased_old_runs.symlink_to(old_runs)
|
||||
with pytest.raises(ValueError, match="only read profiling_runs.jsonl"):
|
||||
compile_map_from_profiling_runs(aliased_old_runs, PUBLIC_DEV_TENANT, versions)
|
||||
profiling_runs = tmp_path / "profiling_runs.jsonl"
|
||||
aliased_old_runs.unlink()
|
||||
profiling_runs.write_text("")
|
||||
capability_map = compile_map_from_profiling_runs(profiling_runs, PUBLIC_DEV_TENANT, versions)
|
||||
assert capability_map["final"] is False
|
||||
assert capability_map["profiles"] == []
|
||||
|
||||
|
||||
def test_harness_hash_is_canonical_json_over_the_full_budget_config() -> None:
|
||||
config = HarnessConfig(
|
||||
harness_version="run_pilot_attempt_v1",
|
||||
budget_version="budget_1usd_50_600_v1",
|
||||
per_run_budget_usd=1.0,
|
||||
step_limit=50,
|
||||
wall_timeout_s=600,
|
||||
mini_swe_agent_version="2.4.6",
|
||||
guard_sha256="guard-sha",
|
||||
)
|
||||
assert harness_config_sha256(config) == harness_config_sha256(config)
|
||||
assert harness_config_sha256(config) != harness_config_sha256(
|
||||
HarnessConfig(
|
||||
harness_version="run_pilot_attempt_v1",
|
||||
budget_version="budget_1usd_50_600_v1",
|
||||
per_run_budget_usd=1.0,
|
||||
step_limit=50,
|
||||
wall_timeout_s=601,
|
||||
mini_swe_agent_version="2.4.6",
|
||||
guard_sha256="guard-sha",
|
||||
)
|
||||
)
|
||||
Loading…
Add table
Reference in a new issue