import logging
import math
from pathlib import Path
import yaml
from inspect_ai import eval_async
from inspect_ai.scorer import value_to_float
from trajectory import Client
if not isinstance(model, str) or not model.startswith("openai/"):
raise TypeError("Pass an Inspect openai/... model identifier")
client = Client()
actor_options = dict(eval_options)
actor_options.update(run_samples=True, score=True, log_samples=True, log_header_only=False)
actor_options.pop("model", None)
existing_args = actor_options.pop("model_args", None) or {}
if isinstance(existing_args, str):
existing_args = yaml.safe_load(Path(existing_args).read_text()) or {}
model_args = dict(existing_args)
headers = dict(model_args.pop("default_headers", None) or {})
actor_options.pop("model_base_url", None)
tid = client.trajectories.create().tid
headers["X-Trajectory-Id"] = tid
model_args.update(api_key=client.api_key, default_headers=headers)
try:
logs = await eval_async(
task,
model=model,
model_base_url=f"{str(client.base_url).rstrip('/')}/v1",
model_args=model_args,
**actor_options,
)
if len(logs) != 1 or logs[0].status != "success":
raise RuntimeError("Inspect did not complete successfully")
samples = logs[0].samples
if (
samples is None
or len(samples) != 1
or samples[0].error is not None
or samples[0].output is None
):
raise RuntimeError("Expected one completed Inspect sample")
value = samples[0].scores["accuracy"].value
if not isinstance(value, (str, int, float)):
raise ValueError("Use the benchmark's conversion for structured scores")
if (
isinstance(value, str)
and value not in {"C", "I", "P", "N"}
and value.lower() not in {"yes", "no", "true", "false"}
):
value = float(value)
reward = value_to_float()(value)
if not math.isfinite(reward):
raise ValueError("Inspect did not produce a finite score")
termination_reason = (
"TRUNCATION"
if samples[0].output.stop_reason in {"max_tokens", "model_length"}
else "ENV_DONE"
)
client.trajectories.log_reward(tid, name="reward_accuracy", value=reward)
except Exception:
try:
client.trajectories.complete(tid, termination_reason="ERROR")
except Exception:
logging.exception("Failed to report trajectory failure")
raise
else:
client.trajectories.complete(tid, termination_reason=termination_reason)