288 lines
10 KiB
Python
288 lines
10 KiB
Python
"""Import Prime Verifiers v1 episode JSONL into Bench's existing scorecard.
|
|
|
|
Verifiers owns rollout execution and its trace schema. Bench reads the durable
|
|
``traces.jsonl`` artifact with the standard library, reduces one episode to one
|
|
sample outcome, and keeps its own result schema unchanged. Raw prompts, task
|
|
data, tool arguments and full conversations never enter the committed card.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from .results import QualityResult, SampleOutcome, truncate
|
|
|
|
FORMAT = "verifiers-v1-episode-jsonl"
|
|
|
|
|
|
class VerifiersImportError(ValueError):
|
|
pass
|
|
|
|
|
|
def traces_path(path: str | Path) -> Path:
|
|
candidate = Path(path)
|
|
if candidate.is_dir():
|
|
candidate = candidate / "traces.jsonl"
|
|
if not candidate.is_file():
|
|
raise VerifiersImportError(f"Verifiers traces file not found: {candidate}")
|
|
return candidate
|
|
|
|
|
|
def source_sha256(path: str | Path) -> str:
|
|
return hashlib.sha256(traces_path(path).read_bytes()).hexdigest()
|
|
|
|
|
|
def _reward_value(rewards: Any) -> tuple[float | None, list[str]]:
|
|
if not isinstance(rewards, dict):
|
|
return None, []
|
|
total = 0.0
|
|
found = False
|
|
names: list[str] = []
|
|
for name, reward in sorted(rewards.items()):
|
|
if reward is None:
|
|
continue
|
|
if isinstance(reward, (int, float)) and not isinstance(reward, bool):
|
|
score, weight = float(reward), 1.0
|
|
elif isinstance(reward, dict) and isinstance(reward.get("score"), (int, float)):
|
|
score = float(reward["score"])
|
|
weight = float(reward.get("weight", 1.0))
|
|
else:
|
|
continue
|
|
found = True
|
|
names.append(str(name))
|
|
total += score * weight
|
|
return (total if found else None), names
|
|
|
|
|
|
def _trace_value(trace: dict[str, Any], primary_metric: str | None) -> tuple[float | None, list[str]]:
|
|
if primary_metric:
|
|
metrics = trace.get("metrics") or {}
|
|
value = metrics.get(primary_metric) if isinstance(metrics, dict) else None
|
|
if isinstance(value, (int, float)) and not isinstance(value, bool):
|
|
return float(value), [primary_metric]
|
|
return None, []
|
|
return _reward_value(trace.get("rewards"))
|
|
|
|
|
|
def _content_text(content: Any) -> str:
|
|
if isinstance(content, str):
|
|
return content
|
|
if isinstance(content, list):
|
|
parts: list[str] = []
|
|
for part in content:
|
|
if isinstance(part, str):
|
|
parts.append(part)
|
|
elif isinstance(part, dict) and isinstance(part.get("text"), str):
|
|
parts.append(part["text"])
|
|
return "".join(parts)
|
|
return ""
|
|
|
|
|
|
def _last_assistant_text(trace: dict[str, Any]) -> str | None:
|
|
for node in reversed(trace.get("nodes") or []):
|
|
if not isinstance(node, dict):
|
|
continue
|
|
message = node.get("message") or {}
|
|
if isinstance(message, dict) and message.get("role") == "assistant":
|
|
text = _content_text(message.get("content")).strip()
|
|
if text:
|
|
return text
|
|
completion = trace.get("completion")
|
|
return completion.strip() if isinstance(completion, str) and completion.strip() else None
|
|
|
|
|
|
def _eligible_traces(episode: dict[str, Any]) -> list[dict[str, Any]]:
|
|
traces = episode.get("traces")
|
|
if not isinstance(traces, list):
|
|
raise VerifiersImportError("episode field 'traces' must be a list")
|
|
typed = [trace for trace in traces if isinstance(trace, dict)]
|
|
trainable = [
|
|
trace
|
|
for trace in typed
|
|
if (trace.get("agent") or {}).get("trainable", True) is not False
|
|
]
|
|
# Eval files from older v1 builds may omit trainable. If an env explicitly
|
|
# marks every trace non-trainable, keep the traces rather than manufacturing
|
|
# an empty sample.
|
|
return trainable or typed
|
|
|
|
|
|
def _task_identity(episode: dict[str, Any]) -> str:
|
|
task = episode.get("task") or {}
|
|
if not isinstance(task, dict):
|
|
task = {}
|
|
durable = task.get("key") or task.get("hash")
|
|
if durable:
|
|
return str(durable)
|
|
# Hash task data locally if an older trace lacks key/hash. The card receives
|
|
# only this digest; private task content is never copied into it.
|
|
canonical = json.dumps(task.get("data"), sort_keys=True, separators=(",", ":"))
|
|
return hashlib.sha256(canonical.encode()).hexdigest()
|
|
|
|
|
|
def _dataset_fingerprint(episodes: list[dict[str, Any]]) -> str | None:
|
|
if not episodes:
|
|
return None
|
|
identities = [
|
|
f"{(episode.get('env') or {}).get('id', '')}\x1f{_task_identity(episode)}"
|
|
for episode in episodes
|
|
]
|
|
digest = hashlib.sha256()
|
|
for identity in sorted(identities):
|
|
digest.update(identity.encode())
|
|
digest.update(b"\x1e")
|
|
return f"sha256:{digest.hexdigest()[:16]}"
|
|
|
|
|
|
def _duration(episodes: list[dict[str, Any]]) -> float | None:
|
|
total = 0.0
|
|
measured = False
|
|
for episode in episodes:
|
|
starts: list[float] = []
|
|
ends: list[float] = []
|
|
for trace in episode.get("traces") or []:
|
|
timing = trace.get("timing") if isinstance(trace, dict) else None
|
|
if not isinstance(timing, dict):
|
|
continue
|
|
for span in timing.values():
|
|
if not isinstance(span, dict):
|
|
continue
|
|
start, end = span.get("start"), span.get("end")
|
|
if isinstance(start, (int, float)) and start > 0:
|
|
starts.append(float(start))
|
|
if isinstance(end, (int, float)) and end > 0:
|
|
ends.append(float(end))
|
|
if starts and ends and max(ends) >= min(starts):
|
|
total += max(ends) - min(starts)
|
|
measured = True
|
|
return round(total, 6) if measured else None
|
|
|
|
|
|
def read_episodes(path: str | Path) -> list[dict[str, Any]]:
|
|
source = traces_path(path)
|
|
episodes: list[dict[str, Any]] = []
|
|
ids: set[str] = set()
|
|
for line_number, line in enumerate(source.read_text().splitlines(), start=1):
|
|
if not line.strip():
|
|
continue
|
|
try:
|
|
episode = json.loads(line)
|
|
except json.JSONDecodeError as exc:
|
|
raise VerifiersImportError(f"{source}:{line_number}: invalid JSON: {exc.msg}") from exc
|
|
if not isinstance(episode, dict):
|
|
raise VerifiersImportError(f"{source}:{line_number}: episode must be an object")
|
|
episode_id = str(episode.get("id") or "")
|
|
if not episode_id:
|
|
raise VerifiersImportError(f"{source}:{line_number}: episode has no id")
|
|
if episode_id in ids:
|
|
raise VerifiersImportError(f"{source}:{line_number}: duplicate episode id {episode_id!r}")
|
|
ids.add(episode_id)
|
|
_eligible_traces(episode) # validate the durable v1 boundary now
|
|
episodes.append(episode)
|
|
if not episodes:
|
|
raise VerifiersImportError(f"{source} contains no episodes")
|
|
return episodes
|
|
|
|
|
|
def import_quality(
|
|
path: str | Path,
|
|
task: str,
|
|
tier: str = "signal",
|
|
pass_threshold: float = 1.0,
|
|
primary_metric: str | None = None,
|
|
) -> QualityResult:
|
|
"""Reduce Verifiers episodes to one existing Bench QualityResult."""
|
|
if tier not in {"reference", "signal", "canary"}:
|
|
raise VerifiersImportError("tier must be reference, signal or canary")
|
|
episodes = read_episodes(path)
|
|
outcomes: list[SampleOutcome] = []
|
|
versions: set[str] = set()
|
|
|
|
for episode in episodes:
|
|
traces = _eligible_traces(episode)
|
|
values: list[float] = []
|
|
reward_names: set[str] = set()
|
|
excerpts: list[str] = []
|
|
trace_ids: list[str] = []
|
|
agents: list[str] = []
|
|
trace_ok = True
|
|
error_types = {
|
|
str(error.get("type"))
|
|
for error in (episode.get("errors") or [])
|
|
if isinstance(error, dict) and error.get("type")
|
|
}
|
|
for trace in traces:
|
|
value, names = _trace_value(trace, primary_metric)
|
|
if value is not None:
|
|
values.append(value)
|
|
reward_names.update(names)
|
|
trace_ids.append(str(trace.get("id") or ""))
|
|
agent = trace.get("agent") or {}
|
|
agents.append(str(agent.get("name") or "agent"))
|
|
trace_ok = trace_ok and bool(trace.get("ok", False))
|
|
verifiers = trace.get("verifiers") or {}
|
|
if isinstance(verifiers, dict) and verifiers.get("version"):
|
|
versions.add(str(verifiers["version"]))
|
|
error_types.update(
|
|
str(error.get("type"))
|
|
for error in (trace.get("errors") or [])
|
|
if isinstance(error, dict) and error.get("type")
|
|
)
|
|
text = _last_assistant_text(trace)
|
|
if text:
|
|
excerpts.append(f"{agents[-1]}: {text}" if len(traces) > 1 else text)
|
|
|
|
score = sum(values) / len(values) if values else 0.0
|
|
episode_ok = bool(episode.get("ok", False)) and trace_ok
|
|
outcomes.append(
|
|
SampleOutcome(
|
|
sample_id=str(episode["id"]),
|
|
score=round(score, 6),
|
|
passed=episode_ok and score >= pass_threshold,
|
|
excerpt=truncate("\n\n".join(excerpts) or None),
|
|
metadata={
|
|
"source": FORMAT,
|
|
"env": str((episode.get("env") or {}).get("id") or ""),
|
|
"task_key": _task_identity(episode),
|
|
"trace_ids": trace_ids,
|
|
"agents": agents,
|
|
"episode_ok": episode_ok,
|
|
"score_source": f"metric:{primary_metric}" if primary_metric else "weighted_rewards",
|
|
"reward_names": sorted(reward_names),
|
|
"error_types": sorted(error_types),
|
|
"ungraded": not values,
|
|
},
|
|
)
|
|
)
|
|
|
|
accuracy = sum(outcome.passed for outcome in outcomes) / len(outcomes)
|
|
mean_score = sum(outcome.score for outcome in outcomes) / len(outcomes)
|
|
metrics = {
|
|
"accuracy": round(accuracy, 6),
|
|
"mean_score": round(mean_score, 6),
|
|
"pass_threshold": float(pass_threshold),
|
|
}
|
|
return QualityResult(
|
|
task=task,
|
|
tier=tier,
|
|
dataset_version=("verifiers:" + ",".join(sorted(versions))) if versions else None,
|
|
dataset_fingerprint=_dataset_fingerprint(episodes),
|
|
n_samples=len(episodes),
|
|
metrics=metrics,
|
|
samples=outcomes,
|
|
duration_s=_duration(episodes),
|
|
)
|
|
|
|
|
|
__all__ = [
|
|
"FORMAT",
|
|
"VerifiersImportError",
|
|
"import_quality",
|
|
"read_episodes",
|
|
"source_sha256",
|
|
"traces_path",
|
|
]
|