跑和记：每条样本都能回放、能复现
baseline 跑完只打印一个正确率：想知道错在哪、两次结果为什么不一样、崩了该从哪接着跑，都没有依据。Evals 把一次运行记成一串按样本归档的事件，最终分数也是从这些事件算出来的；随机数跟着样本走，所以并发跑也能复现。本篇只读了源码，没有实跑，每节只贴源码模版。
按样本记事件
问题：1000 条跑完只拿到 0.62，不知道是哪几条错了，模型的原话是什么，判的时候比的是哪个标准答案。
功能：记录器（recorder）把一次运行记成一串事件（event）。每个事件都带运行 id、序号、样本 id、类型和数据：
模版 · evals/record.py:43-51
@dataclasses.dataclass
class Event:
run_id: str
event_id: int
sample_id: Optional[str]
type: str
data: dict
created_by: str
created_at: str
常用的事件类型有一份清单：采样、判对错、选项、指标、报错，其余的随便起名：
模版 · evals/record.py:54-73
class RecorderBase:
"""
The standard events for which recording methods are provided are:
- `match`: A match or non match, as specified by the `correct` bool, between
the `expected` and `picked` results.
- `embedding`: An embedding of the `prompt` of type `embedding_type`.
- `sampling`: What was `sampled` from the model given the input `prompt`.
- `cond_logp`: The conditional log probability, as `logp`, of the
`completion` from the model given the input `prompt`.
- `pick_option`: The option `picked` by the model out of the valid `options`
given the input `prompt`.
- `raw`: A raw sample specified by the `data`.
- `metrics`: A set of metrics specified by the `kwargs`.
- `error`: An `error` along with an accompanying `msg`.
- `extra`: Any extra `data` of interest to be recorded.
For these events, helper methods are defined at the bottom of this file.
More generally, you can record any event by calling `record_event` with the
event `type` and `data`.
Finally, you can also record a final report using `record_final_report`.
"""
eval 代码不用一层层把记录器传下去。跑每条样本前，外层把「当前记录器」和「当前样本 id」放进 ContextVar；模型封装深处直接调模块级的 record_sampling，就记到了对的样本名下：
模版 · evals/record.py:34-40
_default_recorder: ContextVar[Optional["RecorderBase"]] = ContextVar(
"default_recorder", default=None
)
def default_recorder() -> Optional["RecorderBase"]:
return _default_recorder.get()
模版 · evals/record.py:90-96
@contextlib.contextmanager
def as_default_recorder(self, sample_id: str):
sample_id_token = self._sample_id.set(sample_id)
default_recorder_token = _default_recorder.set(self)
yield
_default_recorder.reset(default_recorder_token)
self._sample_id.reset(sample_id_token)
模版 · evals/record.py:593-602
def record_match(correct: bool, *, expected=None, picked=None, **extra):
return default_recorder().record_match(correct, expected=expected, picked=picked, **extra)
def record_embedding(prompt, embedding_type, **extra):
return default_recorder().record_embedding(prompt, embedding_type, **extra)
def record_sampling(prompt, sampled, **extra):
return default_recorder().record_sampling(prompt, sampled, **extra)
事件先攒在内存里，攒够 100 条、并且离上次写盘过了 10 秒，才批量写一次；进程退出时再写一次：
模版 · evals/record.py:157-185
def record_event(self, type, data=None, sample_id=None):
if sample_id is None:
sample_id = self.current_sample_id()
if sample_id is None:
raise ValueError("No sample_id set! Either pass it in or use as_default_recorder!")
if self.is_paused(sample_id):
return
with self._event_lock:
event = Event(
run_id=self.run_spec.run_id,
event_id=len(self._events),
type=type,
sample_id=sample_id,
data=data,
created_by=self.run_spec.created_by,
created_at=str(datetime.now(timezone.utc)),
)
self._events.append(event)
if (
self._flushes_done < self._flushes_started
or len(self._events) < self._written_events + MIN_FLUSH_EVENTS
or time.time() < self._last_flush_time + MIN_FLUSH_SECONDS
):
return
events_to_write = self._events[self._written_events :]
self._written_events = len(self._events)
self._flushes_started += 1
self._flush_events_internal(events_to_write)
最终分数也从事件里算：Match 的 run 取全部 match 事件算正确率（见登记）。分数和逐条记录出自同一份数据，报告里每个数都能追到具体的样本。
日志首行写运行规格，末行写报告
问题：几个月后翻出两份结果文件，说不清各自是拿什么命令、什么种子、哪个版本的 eval 跑的。
功能：每次运行先生成一份运行规格：被测接口、eval 全名、split、完整命令行、种子、eval 参数；运行 id 是 UTC 时间戳加一段随机后缀：
模版 · evals/base.py:74-89
@dataclass
class RunSpec:
completion_fns: list[str]
eval_name: str
base_eval: str
split: str
run_config: Dict[str, Any]
created_by: str
run_id: str = None
created_at: str = None
def __post_init__(self):
now = datetime.datetime.utcnow()
rand_suffix = base64.b32encode(os.urandom(5)).decode("ascii")
self.run_id = now.strftime("%y%m%d%H%M%S") + rand_suffix
self.created_at = str(now)
模版 · evals/cli/oaieval.py:173-201
run_config = {
"completion_fns": completion_fns,
"eval_spec": eval_spec,
"seed": args.seed,
"max_samples": args.max_samples,
"command": " ".join(map(shlex.quote, sys.argv)),
"initial_settings": {
"visible": visible,
},
}
eval_name = eval_spec.key
if eval_name is None:
raise Exception("you must provide a eval name")
run_spec = evals.base.RunSpec(
completion_fns=completion_fns,
eval_name=eval_name,
base_eval=eval_name.split(".")[0],
split=eval_name.split(".")[1],
run_config=run_config,
created_by=args.user,
)
record_path = (
f"/tmp/evallogs/{run_spec.run_id}_{args.completion_fn}_{args.eval}.jsonl"
if args.record_path is None
else args.record_path
)
本地记录器把规格写成第一行，事件一行一条，最终报告写成最后一行：
模版 · evals/record.py:322-371
def __init__(
self, log_path: Optional[str], run_spec: RunSpec, hidden_data_fields: Sequence[Text] = []
):
"""
Initializes a LocalRecorder.
Args:
log_path (Optional[str]): Path to which the LocalRecorder will
record events. Currently accepts local paths, google cloud
storage paths, or Azure blob paths.
run_spec (RunSpec): Passed to the superclass to provide metadata
about the current evals run.
hidden_data_fields (Sequence[Text]): Fields to avoid writing in the
output. This is particularly useful when using a language model
as an evaluator of sensitive customer data which should not be
written to disc.
"""
super().__init__(run_spec)
self.event_file_path = log_path
self.hidden_data_fields = hidden_data_fields
if log_path is not None:
with bf.BlobFile(log_path, "wb") as f:
f.write((jsondumps({"spec": dataclasses.asdict(run_spec)}) + "\n").encode("utf-8"))
def _flush_events_internal(self, events_to_write: Sequence[Event]):
start = time.time()
try:
lines = [
jsondumps(event, exclude_keys=self.hidden_data_fields) + "\n"
for event in events_to_write
]
except TypeError as e:
logger.error(f"Failed to serialize events: {events_to_write}")
raise e
with bf.BlobFile(self.event_file_path, "ab") as f:
f.write(b"".join([line.encode("utf-8") for line in lines]))
logger.info(
f"Logged {len(lines)} rows of events to {self.event_file_path}: insert_time={t(time.time()-start)}"
)
self._last_flush_time = time.time()
self._flushes_done += 1
def record_final_report(self, final_report: Any):
with bf.BlobFile(self.event_file_path, "ab") as f:
f.write((jsondumps({"final_report": final_report, "run_id": self.run_spec.run_id}) + "\n").encode("utf-8"))
logging.info(f"Final report: {final_report}. Logged to {self.event_file_path}")
token 用量不用另外统计：跑完把所有 sampling 事件里的 usage 加总，以 usage_ 开头写进报告：
模版 · evals/cli/oaieval.py:269-294
def add_token_usage_to_result(result: dict[str, Any], recorder: RecorderBase) -> None:
"""
Add token usage from logged sampling events to the result dictionary from the recorder.
"""
usage_events = []
sampling_events = recorder.get_events("sampling")
for event in sampling_events:
if "usage" in event.data:
usage_events.append(dict(event.data["usage"]))
logger.info(f"Found {len(usage_events)}/{len(sampling_events)} sampling events with usage data")
if usage_events:
# Sum up the usage of all samples (assumes the usage is the same for all samples)
total_usage = {
key: sum(u[key] if u[key] is not None else 0 for u in usage_events)
for key in usage_events[0]
}
total_usage_str = "\n".join(f"{key}: {value:,}" for key, value in total_usage.items())
logger.info(f"Token usage from {len(usage_events)} sampling events:\n{total_usage_str}")
for key, value in total_usage.items():
keyname = f"usage_{key}"
if keyname not in result:
result[keyname] = value
else:
logger.warning(
f"Usage key {keyname} already exists in result, not adding {keyname}"
)
一个 JSONL 文件自带复现它所需的全部信息，拿第一行的命令就能重跑。
随机数跟样本走，不跟线程走
问题：十个线程并发跑，哪条先跑完是乱的。要是 few-shot 抽样共用一个随机数生成器，每条样本抽到哪几个示例，取决于线程调度，两次结果没法比。
功能：样本先用固定种子洗牌，--max_samples 取洗牌后的前 N 条；每条样本保留它在原题库里的序号：
模版 · evals/eval.py:26-38
SHUFFLE_SEED = 123
_MAX_SAMPLES = None
def _index_samples(samples: List[Any]) -> List[Tuple[Any, int]]:
"""Shuffle `samples` and pair each sample with its index."""
indices = list(range(len(samples)))
random.Random(SHUFFLE_SEED).shuffle(indices)
if _MAX_SAMPLES is not None:
indices = indices[:_MAX_SAMPLES]
logger.info(f"Evaluating {len(indices)} samples")
work_items = [(samples[i], i) for i in indices]
return work_items
样本 id 是 <名>.<split>.<原序号>。每条样本的随机数生成器用「样本 id 加全局种子」做种子，跑完再按原序号排回去：
模版 · evals/eval.py:112-147
def eval_all_samples(
self,
recorder: RecorderBase,
samples,
show_progress=True,
record_raw_sample=True,
**_kwargs: Any,
):
"""
Evaluate all provided samples in parallel.
"""
work_items = _index_samples(samples)
threads = int(os.environ.get("EVALS_THREADS", "10"))
show_progress = bool(os.environ.get("EVALS_SHOW_EVAL_PROGRESS", show_progress))
def eval_sample(args):
"""
Evaluate a single sample.
"""
sample, idx = args
base_name, split = self.name.split(".")[0:2]
sample_id = f"{base_name}.{split}.{idx}"
with recorder.as_default_recorder(sample_id):
seed = f"{sample_id}:{self.seed}".encode("utf-8")
rng = random.Random(seed)
return idx, self.eval_sample(sample, rng)
with ThreadPool(threads) as pool:
if os.environ.get("EVALS_SEQUENTIAL", "0") in {"1", "true", "yes"}:
logger.info("Running in sequential mode!")
iter = map(eval_sample, work_items)
else:
logger.info(f"Running in threaded mode with {threads} threads!")
iter = pool.imap_unordered(eval_sample, work_items)
idx_and_result = list(tqdm(iter, total=len(work_items), disable=not show_progress))
return [r for _, r in sorted(idx_and_result)]
同一条样本不管被哪个线程、第几个跑到，拿到的随机数都一样；改线程数、改 --max_samples，eval 这一侧给每条样本的随机数都不变，剩下的差异只来自模型自己的采样。样本 id 不含版本号，同一道题在新旧两个版本里 id 相同，方便逐条对比。
并发、重试和中途叫停
问题：串行调 API，1000 条要跑很久；碰上限流或者超时，脚本就挂了。
功能：线程池并发，线程数取环境变量 EVALS_THREADS，默认 10；调试时 EVALS_SEQUENTIAL=1 改成串行（见上一节代码）。API 调用遇到限流、连接错误、超时、服务端错误就按指数退避重试，单次最长等 60 秒：
模版 · evals/completion_fns/openai.py:19-24
OPENAI_TIMEOUT_EXCEPTIONS = (
openai.RateLimitError,
openai.APIConnectionError,
openai.APITimeoutError,
openai.InternalServerError,
)
模版 · evals/utils/api_utils.py:10-22
@backoff.on_predicate(
wait_gen=backoff.expo,
max_value=60,
factor=1.5,
)
def create_retrying(func: callable, retry_exceptions: tuple[Exception], *args, **kwargs):
"""
Retries given function if one of given exceptions is raised
"""
try:
return func(*args, **kwargs)
except retry_exceptions:
return False
SolverEval 还接受温和中断：设了 EVALS_GENTLE_INTERRUPT，按 Ctrl-C 不丢结果，报告按已经跑完的样本出：
模版 · evals/eval.py:236-253
idx_and_result = []
try:
for result in tqdm(iter, total=len(work_items), disable=not show_progress):
idx_and_result.append(result)
except KeyboardInterrupt:
# "Gentle interrupt" allows us to stop early and still get results
gentle_interrupt = os.environ.get("EVALS_GENTLE_INTERRUPT", "0") in {
"1",
"true",
"yes",
}
if gentle_interrupt:
logger.info("Evaluation stopped because of KeyboardInterrupt")
logger.info(
f"Report will be based on {len(idx_and_result)} out of the planned {len(work_items)} samples"
)
else:
raise
评测集续跑
问题：评测集里 30 个 eval，跑到第 20 个崩了，重跑一遍前 19 个既费时间又费钱。
功能：评测集运行器给每个 eval 起一个子进程。每跑完一个，就把那条命令追加进进度文件；重跑时先读进度文件，跳过已经完成的：
模版 · evals/cli/oaievalset.py:17-40
class Progress:
def __init__(self, file: str) -> None:
self.file = Path(file)
self.completed: list[Task] = []
def load(self) -> bool:
if not self.file.exists():
return False
with self.file.open() as f:
for line in f:
self.completed.append(json.loads(line))
return len(self.completed) > 0
def add(self, item: Task) -> None:
self.completed.append(item)
self.save()
def save(self) -> None:
self.file.parent.mkdir(parents=True, exist_ok=True)
with self.file.open("w") as f:
for item in self.completed:
f.write(json.dumps(item) + "\n")
print(highlight(f"Saved progress to {self.file}"))
模版 · evals/cli/oaievalset.py:109-129
num_evals = len(commands)
progress = Progress(f"/tmp/oaievalset/{args.model}.{args.eval_set}.progress.txt")
if args.resume and progress.load():
print(f"Loaded progress from {progress.file}")
print(f"{len(progress.completed)}/{len(commands)} evals already completed:")
for item in progress.completed:
print(" " + " ".join(item))
commands = [c for c in commands if c not in progress.completed]
command_strs = [" ".join(cmd) for cmd in commands]
print("Going to run the following commands:")
for command_str in command_strs:
print(" " + command_str)
num_already_completed = num_evals - len(commands)
for idx, command in enumerate(commands):
real_idx = idx + num_already_completed
print(highlight("Running command: " + " ".join(command) + f" ({real_idx+1}/{num_evals})"))
subprocess.run(command, stdout=subprocess.PIPE, check=args.exit_on_error)
progress.add(command)
续跑的粒度是整个 eval，项目自己说明了单个 eval 不行1：
Unfortunately, you can't resume a single eval from the middle. You'll have to restart from the beginning, so try to keep your individual evals quick to run.
记录去哪：本地、HTTP、数据库、只打印
问题：调 prompt 时只想看几条结果，不想留文件；团队要把结果汇总到一个库；评的是客户的敏感数据，有些字段不能落盘。
功能：记录器换实现就换去处，eval 代码一行不动。命令行按开关选：--dry-run 只打印，--local-run（默认）写本地 JSONL，--http-run POST 到一个地址，--no-local-run 写 Snowflake：
模版 · evals/cli/oaieval.py:242-266
def build_recorder(
args: OaiEvalArguments, run_spec: evals.base.RunSpec, record_path: str
) -> RecorderBase:
if args.dry_run:
return evals.record.DummyRecorder(run_spec=run_spec, log=args.dry_run_logging)
if args.local_run:
return evals.record.LocalRecorder(record_path, run_spec=run_spec)
if args.http_run:
if args.http_run_url is None:
raise ValueError("URL must be specified when using http-run mode")
return evals.record.HttpRecorder(
url=args.http_run_url,
run_spec=run_spec,
batch_size=args.http_batch_size,
fail_percent_threshold=args.http_fail_percent_threshold,
local_fallback_path=record_path,
)
return evals.record.Recorder(
record_path,
run_spec=run_spec,
)
只打印的记录器按主指标给 match 事件上色，对的绿、错的红，调 prompt 时扫一眼就够：
模版 · evals/record.py:274-313
class DummyRecorder(RecorderBase):
"""
A "recorder" which only logs certain events to the console.
Can be used by passing `--dry-run` when invoking `oaieval`.
"""
def __init__(self, run_spec: RunSpec, log: bool = True):
super().__init__(run_spec)
self.log = log
def record_event(self, type, data, sample_id=None):
from evals.registry import registry
if self.run_spec is None:
return
base_eval_spec = registry.get_base_eval(self.run_spec.base_eval)
if base_eval_spec and len(base_eval_spec.metrics) >= 1:
primary_metric = base_eval_spec.metrics[0]
else:
primary_metric = "accuracy"
with self._event_lock:
event = self._create_event(type, data)
self._events.append(event)
msg = f"Not recording event: {event}"
if type == "match":
accuracy_good = (
primary_metric == "accuracy" or primary_metric.startswith("pass@")
) and (data.get("correct", False) or data.get("accuracy", 0) > 0.5)
f1_score_good = primary_metric == "f1_score" and data.get("f1_score", 0) > 0.5
if accuracy_good or f1_score_good:
msg = _green(msg)
else:
msg = _red(msg)
if self.log:
logging.info(msg)
HTTP 上报失败的比例超过阈值（默认 5%），这一批改写进本地文件，再抛错提示：
模版 · evals/record.py:392-403
def _flush_events_internal(self, events_to_write: Sequence[Event]):
batch_size = self.batch_size
for i in range(0, len(events_to_write), batch_size):
batch = list(events_to_write[i : i + batch_size])
try:
self._send_event(batch)
except RuntimeError as e:
logger.error(f"Falling back to LocalRecorder due to error: {str(e)}")
self.local_fallback_recorder._flush_events_internal(batch)
raise RuntimeError(
"An error occurred when sending events. Your events have been saved locally using the Local recorder."
)
本地记录器还能指定不落盘的字段（hidden_data_fields），写盘时从每个事件里剔掉，见第二节的 LocalRecorder。源码注释说的场景是「用语言模型评客户的敏感数据」。这个参数命令行上没有开口，要在代码里构造记录器时传。
谁管什么
- 程序：事件编号，挂上样本 id；攒批写盘；生成运行 id 和运行规格，写首行和末行；从 sampling 事件加总 token 用量；固定种子洗牌，按样本派生随机数，结果按原序号排回；线程池、退避重试、温和中断；评测集进度文件；HTTP 失败回落本地
- 模型：不参与这一层，它的每次输出都被记成 sampling 事件
- 用户：选记录去处；定线程数、
--max_samples、种子；决定续跑还是删掉进度文件从头来；要隐藏字段时自己在代码里构造记录器
附录：源码与文档
evals/record.py 全文
"""
This file defines the recorder classes which log eval results in different ways,
such as to a local JSON file or to a remote Snowflake database.
If you would like to implement a custom recorder, you can see how the
`LocalRecorder` and `Recorder` classes inherit from the `RecorderBase` class and
override certain methods.
"""
import atexit
import contextlib
import dataclasses
import logging
import threading
import time
from contextvars import ContextVar
from datetime import datetime, timezone
from typing import Any, List, Optional, Sequence, Text
import blobfile as bf
import requests
import evals
from evals.base import RunSpec
from evals.data import jsondumps
from evals.utils.misc import t
from evals.utils.snowflake import SnowflakeConnection
logger = logging.getLogger(__name__)
MIN_FLUSH_EVENTS = 100
MAX_SNOWFLAKE_BYTES = 16 * 10**6
MIN_FLUSH_SECONDS = 10
_default_recorder: ContextVar[Optional["RecorderBase"]] = ContextVar(
"default_recorder", default=None
)
def default_recorder() -> Optional["RecorderBase"]:
return _default_recorder.get()
@dataclasses.dataclass
class Event:
run_id: str
event_id: int
sample_id: Optional[str]
type: str
data: dict
created_by: str
created_at: str
class RecorderBase:
"""
The standard events for which recording methods are provided are:
- `match`: A match or non match, as specified by the `correct` bool, between
the `expected` and `picked` results.
- `embedding`: An embedding of the `prompt` of type `embedding_type`.
- `sampling`: What was `sampled` from the model given the input `prompt`.
- `cond_logp`: The conditional log probability, as `logp`, of the
`completion` from the model given the input `prompt`.
- `pick_option`: The option `picked` by the model out of the valid `options`
given the input `prompt`.
- `raw`: A raw sample specified by the `data`.
- `metrics`: A set of metrics specified by the `kwargs`.
- `error`: An `error` along with an accompanying `msg`.
- `extra`: Any extra `data` of interest to be recorded.
For these events, helper methods are defined at the bottom of this file.
More generally, you can record any event by calling `record_event` with the
event `type` and `data`.
Finally, you can also record a final report using `record_final_report`.
"""
def __init__(
self,
run_spec: evals.base.RunSpec,
) -> None:
self._sample_id: ContextVar[Optional[int]] = ContextVar("_sample_id", default=None)
self.run_spec = run_spec
self._events: List[Event] = []
self._last_flush_time = time.time()
self._flushes_done = 0
self._written_events = 0
self._flushes_started = 0
self._event_lock = threading.Lock()
self._paused_ids: List[str] = []
atexit.register(self.flush_events)
@contextlib.contextmanager
def as_default_recorder(self, sample_id: str):
sample_id_token = self._sample_id.set(sample_id)
default_recorder_token = _default_recorder.set(self)
yield
_default_recorder.reset(default_recorder_token)
self._sample_id.reset(sample_id_token)
def current_sample_id(self) -> Optional[str]:
return self._sample_id.get()
def pause(self):
sample_id = self.current_sample_id()
with self._event_lock:
if sample_id not in self._paused_ids:
self._paused_ids.append(sample_id)
def unpause(self):
sample_id = self.current_sample_id()
with self._event_lock:
if sample_id in self._paused_ids:
self._paused_ids.remove(sample_id)
def is_paused(self, sample_id: str = None):
if sample_id is None:
sample_id = self.current_sample_id()
with self._event_lock:
return sample_id in self._paused_ids
def get_events(self, type: str) -> Sequence[Event]:
with self._event_lock:
return [event for event in self._events if event.type == type]
def get_metrics(self):
return list(map(lambda x: x.data, self.get_events("metrics")))
def get_scores(self, key: str):
return list(map(lambda e: e.data[key], self.get_events("metrics")))
def _create_event(self, type, data=None, sample_id=None):
if sample_id is None:
sample_id = self.current_sample_id()
if sample_id is None:
raise ValueError("No sample_id set! Either pass it in or use as_default_recorder!")
return Event(
run_id=self.run_spec.run_id,
event_id=len(self._events),
type=type,
sample_id=sample_id,
data=data,
created_by=self.run_spec.created_by,
created_at=str(datetime.now(timezone.utc)),
)
def _flush_events_internal(self, events_to_write: Sequence[Event]):
pass
def flush_events(self):
with self._event_lock:
if len(self._events) == self._written_events:
return
events_to_write = self._events[self._written_events :]
self._written_events = len(self._events)
self._flushes_started += 1
self._flush_events_internal(events_to_write)
def record_event(self, type, data=None, sample_id=None):
if sample_id is None:
sample_id = self.current_sample_id()
if sample_id is None:
raise ValueError("No sample_id set! Either pass it in or use as_default_recorder!")
if self.is_paused(sample_id):
return
with self._event_lock:
event = Event(
run_id=self.run_spec.run_id,
event_id=len(self._events),
type=type,
sample_id=sample_id,
data=data,
created_by=self.run_spec.created_by,
created_at=str(datetime.now(timezone.utc)),
)
self._events.append(event)
if (
self._flushes_done < self._flushes_started
or len(self._events) < self._written_events + MIN_FLUSH_EVENTS
or time.time() < self._last_flush_time + MIN_FLUSH_SECONDS
):
return
events_to_write = self._events[self._written_events :]
self._written_events = len(self._events)
self._flushes_started += 1
self._flush_events_internal(events_to_write)
def record_match(self, correct: bool, *, expected=None, picked=None, sample_id=None, **extra):
assert isinstance(
correct, bool
), f"correct must be a bool, but was a {type(correct)}: {correct}"
if isinstance(expected, list) and len(expected) == 1:
expected = expected[0]
data = {
"correct": bool(correct),
"expected": expected,
"picked": picked,
**extra,
}
self.record_event("match", data, sample_id=sample_id)
def record_embedding(self, prompt, embedding_type, sample_id=None, **extra):
data = {
"prompt": prompt,
"embedding_type": embedding_type,
**extra,
}
self.record_event("embedding", data, sample_id=sample_id)
def record_sampling(self, prompt, sampled, sample_id=None, **extra):
data = {
"prompt": prompt,
"sampled": sampled,
**extra,
}
self.record_event("sampling", data, sample_id=sample_id)
def record_function_call(self, name, arguments, return_value, sample_id=None, **extra):
data = {
"name": name,
"arguments": arguments,
"return_value": return_value,
**extra,
}
self.record_event("function_call", data, sample_id=sample_id)
def record_cond_logp(self, prompt, completion, logp, sample_id=None, **extra):
data = {
"prompt": prompt,
"completion": completion,
"logp": logp,
**extra,
}
self.record_event("cond_logp", data, sample_id=sample_id)
def record_pick_option(self, prompt, options, picked, sample_id=None, **extra):
data = {
"prompt": prompt,
"options": options,
"picked": picked,
**extra,
}
self.record_event("pick_option", data, sample_id=sample_id)
def record_raw(self, data):
self.record_event("raw_sample", data)
def record_metrics(self, **kwargs):
self.record_event("metrics", kwargs)
def record_error(self, msg: str, error: Exception, **kwargs):
data = {
"type": type(error).__name__,
"message": str(error),
}
data.update(kwargs)
self.record_event("error", data)
def record_extra(self, data, sample_id=None):
self.record_event("extra", data, sample_id=sample_id)
def record_final_report(self, final_report: Any):
logging.info(f"Final report: {final_report}. Not writing anywhere.")
def _green(str):
return f"\033[1;32m{str}\033[0m"
def _red(str):
return f"\033[1;31m{str}\033[0m"
class DummyRecorder(RecorderBase):
"""
A "recorder" which only logs certain events to the console.
Can be used by passing `--dry-run` when invoking `oaieval`.
"""
def __init__(self, run_spec: RunSpec, log: bool = True):
super().__init__(run_spec)
self.log = log
def record_event(self, type, data, sample_id=None):
from evals.registry import registry
if self.run_spec is None:
return
base_eval_spec = registry.get_base_eval(self.run_spec.base_eval)
if base_eval_spec and len(base_eval_spec.metrics) >= 1:
primary_metric = base_eval_spec.metrics[0]
else:
primary_metric = "accuracy"
with self._event_lock:
event = self._create_event(type, data)
self._events.append(event)
msg = f"Not recording event: {event}"
if type == "match":
accuracy_good = (
primary_metric == "accuracy" or primary_metric.startswith("pass@")
) and (data.get("correct", False) or data.get("accuracy", 0) > 0.5)
f1_score_good = primary_metric == "f1_score" and data.get("f1_score", 0) > 0.5
if accuracy_good or f1_score_good:
msg = _green(msg)
else:
msg = _red(msg)
if self.log:
logging.info(msg)
class LocalRecorder(RecorderBase):
"""
A recorder which logs events to the specified JSON file.
This is the default recorder used by `oaieval`.
"""
def __init__(
self, log_path: Optional[str], run_spec: RunSpec, hidden_data_fields: Sequence[Text] = []
):
"""
Initializes a LocalRecorder.
Args:
log_path (Optional[str]): Path to which the LocalRecorder will
record events. Currently accepts local paths, google cloud
storage paths, or Azure blob paths.
run_spec (RunSpec): Passed to the superclass to provide metadata
about the current evals run.
hidden_data_fields (Sequence[Text]): Fields to avoid writing in the
output. This is particularly useful when using a language model
as an evaluator of sensitive customer data which should not be
written to disc.
"""
super().__init__(run_spec)
self.event_file_path = log_path
self.hidden_data_fields = hidden_data_fields
if log_path is not None:
with bf.BlobFile(log_path, "wb") as f:
f.write((jsondumps({"spec": dataclasses.asdict(run_spec)}) + "\n").encode("utf-8"))
def _flush_events_internal(self, events_to_write: Sequence[Event]):
start = time.time()
try:
lines = [
jsondumps(event, exclude_keys=self.hidden_data_fields) + "\n"
for event in events_to_write
]
except TypeError as e:
logger.error(f"Failed to serialize events: {events_to_write}")
raise e
with bf.BlobFile(self.event_file_path, "ab") as f:
f.write(b"".join([line.encode("utf-8") for line in lines]))
logger.info(
f"Logged {len(lines)} rows of events to {self.event_file_path}: insert_time={t(time.time()-start)}"
)
self._last_flush_time = time.time()
self._flushes_done += 1
def record_final_report(self, final_report: Any):
with bf.BlobFile(self.event_file_path, "ab") as f:
f.write((jsondumps({"final_report": final_report, "run_id": self.run_spec.run_id}) + "\n").encode("utf-8"))
logging.info(f"Final report: {final_report}. Logged to {self.event_file_path}")
class HttpRecorder(RecorderBase):
def __init__(
self,
url: str,
run_spec: RunSpec,
local_fallback_path: str,
fail_percent_threshold: int = 5,
batch_size: int = 100,
):
super().__init__(run_spec)
self.url = url
self.batch_size = batch_size
self.fail_percent_threshold = fail_percent_threshold / 100
self.failed_requests = 0 # Add this line to track failed requests
self.local_fallback_path = local_fallback_path
self.local_fallback_recorder = LocalRecorder(local_fallback_path, run_spec)
logger.info(f"HttpRecorder initialized with URL {self.url}")
def _flush_events_internal(self, events_to_write: Sequence[Event]):
batch_size = self.batch_size
for i in range(0, len(events_to_write), batch_size):
batch = list(events_to_write[i : i + batch_size])
try:
self._send_event(batch)
except RuntimeError as e:
logger.error(f"Falling back to LocalRecorder due to error: {str(e)}")
self.local_fallback_recorder._flush_events_internal(batch)
raise RuntimeError(
"An error occurred when sending events. Your events have been saved locally using the Local recorder."
)
def _send_event(self, events: List[Event]):
# Convert the events to dictionaries
events_dict = [dataclasses.asdict(event) for event in events]
logger.debug(f"Sending events: {events_dict}")
try:
# Send the events to the specified URL
response = requests.post(self.url, json=events_dict)
# If the request succeeded, log a success message
if response.ok:
logger.debug("Events sent successfully")
# If the request failed, log a warning and increment failed_requests
else:
logger.warning(f"Failed to send events: {response.text}")
self.failed_requests += len(
events
) # Increase the count by the number of events in the failed request
except Exception as e:
logger.warning(f"Failed to send events: {str(e)}")
self.failed_requests += len(
events
) # Increase the count by the number of events in the failed request
# Check if the proportion of failed requests exceeds the threshold
fail_threshold = self.fail_percent_threshold
# Make a string for human comprehention
fail_threshold_str = str(fail_threshold * 100) + "%"
if self.failed_requests / len(self._events) > fail_threshold:
raise RuntimeError(
"The proportion of failed events has exceeded the threshold of: "
+ fail_threshold_str
+ "."
+ " Falling back to LocalRecorder. "
"You can modify this via the cli flag --http-fail-percent-threshold"
)
def record_final_report(self, final_report: Any):
# Convert the final report to a dictionary and prepare it as an event
report_event = Event(
run_id=self.run_spec.run_id,
event_id=len(self._events),
sample_id=None, # or you could use a specific id for final reports
type="final_report",
data=final_report,
created_by=self.run_spec.created_by,
created_at=str(datetime.now(timezone.utc)),
)
# Send the final report event
try:
self._send_event([report_event])
logging.info(f"Final report: {final_report}.")
logging.info(f"Data logged to: {self.url}")
except RuntimeError as e:
logger.error(f"Falling back to LocalRecorder due to error: {str(e)}")
self.local_fallback_recorder.record_final_report(final_report)
class Recorder(RecorderBase):
"""
A recorder which logs events to Snowflake.
Can be used by passing `--no-local-run` when invoking `oaieval`.
"""
def __init__(
self,
log_path: Optional[str],
run_spec: RunSpec,
snowflake_connection: Optional[SnowflakeConnection] = None,
) -> None:
super().__init__(run_spec)
self.event_file_path = log_path
self._writing_lock = threading.Lock()
if snowflake_connection is None:
snowflake_connection = SnowflakeConnection()
self._conn = snowflake_connection
if log_path is not None:
with bf.BlobFile(log_path, "wb") as f:
f.write((jsondumps({"spec": dataclasses.asdict(run_spec)}) + "\n").encode("utf-8"))
query = """
INSERT ALL INTO runs (run_id, model_name, eval_name, base_eval, split, run_config, settings, created_by, created_at)
VALUES (%(run_id)s, %(model_name)s, %(eval_name)s, %(base_eval)s, %(split)s, run_config, settings, %(created_by)s, %(created_at)s)
SELECT PARSE_JSON(%(run_config)s) AS run_config, PARSE_JSON(%(settings)s) AS settings
"""
self._conn.robust_query(
command=query,
params={
"run_id": run_spec.run_id,
# TODO: model_name -> completion_fns
"model_name": jsondumps(dict(completions=run_spec.completion_fns)),
"eval_name": run_spec.eval_name,
"base_eval": run_spec.base_eval,
"split": run_spec.split,
"run_config": jsondumps(run_spec.run_config),
"settings": jsondumps(run_spec.run_config.get("initial_settings", {})),
"created_by": run_spec.created_by,
"created_at": run_spec.created_at,
},
)
atexit.register(self.flush_events)
def _flush_events_internal(self, events_to_write: Sequence[Event]):
with self._writing_lock:
try:
lines = [jsondumps(event) + "\n" for event in events_to_write]
except TypeError as e:
logger.error(f"Failed to serialize events: {events_to_write}")
raise e
idx_l = 0
while idx_l < len(events_to_write):
total_bytes = 0
idx_r = idx_l
while (
idx_r < len(events_to_write)
and total_bytes + len(lines[idx_r]) < MAX_SNOWFLAKE_BYTES
):
total_bytes += len(lines[idx_r])
idx_r += 1
assert idx_r > idx_l
start = time.time()
buffer = [
(
event.run_id,
event.event_id,
event.sample_id,
event.type,
jsondumps(event.data),
event.created_by,
event.created_at,
)
for event in events_to_write[idx_l:idx_r]
]
query = """
INSERT INTO events (run_id, event_id, sample_id, type, data, created_by, created_at)
SELECT Column1 AS run_id, Column2 as event_id, Column3 AS sample_id, Column4 AS type, PARSE_JSON(Column5) AS data, Column6 AS created_by, Column7 AS created_at
FROM VALUES(%s, %s, %s, %s, %s, %s, %s)
"""
self._conn.robust_query(command=query, seqparams=buffer, many=True)
logger.info(
f"Logged {len(buffer)} rows of events to Snowflake: insert_time={t(time.time()-start)}"
)
idx_l = idx_r
with bf.BlobFile(self.event_file_path, "ab") as f:
f.write(b"".join([line.encode("utf-8") for line in lines]))
self._last_flush_time = time.time()
self._flushes_done += 1
def record_final_report(self, final_report: Any):
with self._writing_lock:
with bf.BlobFile(self.event_file_path, "ab") as f:
f.write((jsondumps({"final_report": final_report}) + "\n").encode("utf-8"))
query = """
UPDATE runs
SET final_report = PARSE_JSON(%(final_report)s)
WHERE run_id = %(run_id)s
"""
self._conn.robust_query(
command=query,
params={
"run_id": self.run_spec.run_id,
"final_report": jsondumps(final_report),
},
)
def record_event(self, type, data=None, sample_id=None):
# try to serialize data so we fail early!
_ = jsondumps(data)
return super().record_event(type, data, sample_id)
#########################################################################
### Helper methods which use the thread local global default recorder ###
#########################################################################
def current_sample_id() -> str:
return default_recorder().current_sample_id
def record_match(correct: bool, *, expected=None, picked=None, **extra):
return default_recorder().record_match(correct, expected=expected, picked=picked, **extra)
def record_embedding(prompt, embedding_type, **extra):
return default_recorder().record_embedding(prompt, embedding_type, **extra)
def record_sampling(prompt, sampled, **extra):
return default_recorder().record_sampling(prompt, sampled, **extra)
def record_function_call(name, arguments, return_value, **extra):
return default_recorder().record_function_call(name, arguments, return_value, **extra)
def record_cond_logp(prompt, completion, logp, **extra):
return default_recorder().record_cond_logp(prompt, completion, logp, **extra)
def record_pick_option(prompt, options, picked, **extra):
return default_recorder().record_pick_option(prompt, options, picked, **extra)
def record_raw(data):
return default_recorder().record_raw(data)
def record_metrics(**extra):
return default_recorder().record_metrics(**extra)
def record_error(msg: str, error: Exception = None, **extra):
return default_recorder().record_error(msg, error, **extra)
def record_extra(data):
return default_recorder().record_extra(data)
def record_event(type, data=None, sample_id=None):
return default_recorder().record_event(type, data, sample_id)
def pause():
return default_recorder().pause()
def unpause():
return default_recorder().unpause()
evals/eval.py 全文
"""
This file defines the base class for evals.
"""
import abc
import asyncio
import logging
import os
import random
from multiprocessing.pool import ThreadPool
from pathlib import Path
from typing import Any, Awaitable, Callable, Dict, List, Optional, Tuple, Union
from tqdm import tqdm
from evals.api import CompletionFn
from .data import get_jsonl
from .record import RecorderBase
from .registry import Registry
from .solvers.solver import Solver
from .solvers.utils import maybe_wrap_with_compl_fn, maybe_wrap_with_solver
logger = logging.getLogger(__name__)
SHUFFLE_SEED = 123
_MAX_SAMPLES = None
def _index_samples(samples: List[Any]) -> List[Tuple[Any, int]]:
"""Shuffle `samples` and pair each sample with its index."""
indices = list(range(len(samples)))
random.Random(SHUFFLE_SEED).shuffle(indices)
if _MAX_SAMPLES is not None:
indices = indices[:_MAX_SAMPLES]
logger.info(f"Evaluating {len(indices)} samples")
work_items = [(samples[i], i) for i in indices]
return work_items
def set_max_samples(max_samples: int):
global _MAX_SAMPLES
_MAX_SAMPLES = max_samples
class Eval(abc.ABC):
"""
Evaluation classes generally should override two methods:
`eval_sample`: Takes in a test sample and a random number generator and
records the metrics of interest.
`run`: Takes in a recorder and runs the evaluation. Generally, most `run`
methods will follow this same pattern: loading the data, calling
`eval_all_samples`, and aggregating the recorded results.
"""
def __init__(
self,
completion_fns: list[Union[CompletionFn, Solver]],
eval_registry_path: Path,
seed: int = 20220722,
name: str = "no_name_eval.default",
registry: Optional[Registry] = None,
samples_jsonl: Optional[str] = None,
):
splits = name.split(".")
if len(splits) < 2:
raise ValueError(f"Eval name must at least have <base_eval>.<split>. Got name {name}")
self.completion_fns = [maybe_wrap_with_compl_fn(fn) for fn in completion_fns]
self.eval_registry_path = eval_registry_path
self.seed = seed
self.name = name
self.registry = registry or Registry()
self.samples_jsonl = samples_jsonl
@abc.abstractmethod
def eval_sample(self, sample: Any, rng: random.Random):
raise NotImplementedError()
@property
def completion_fn(self) -> CompletionFn:
"""Helper for more ergonomic access to a single CompletionFn."""
return self.completion_fns[0]
@abc.abstractmethod
def run(self, recorder: RecorderBase) -> Dict[str, float]:
"""Run the evaluation with the corresponding recorder."""
raise NotImplementedError()
async def async_eval_all_samples(
self,
eval_fn: Callable[[Tuple[Any, int]], Awaitable[Tuple[int, Any]]],
samples: List[Any],
concurrency: int = 32,
show_progress: bool = True,
**_kwargs: Any,
):
work_items = _index_samples(samples)
semaphore = asyncio.Semaphore(concurrency)
async def eval_fn_with_semaphore(args):
async with semaphore:
return await eval_fn(args)
futures = [asyncio.ensure_future(eval_fn_with_semaphore(args)) for args in work_items]
for future in tqdm(
asyncio.as_completed(futures), total=len(samples), disable=not show_progress
):
await future
def eval_all_samples(
self,
recorder: RecorderBase,
samples,
show_progress=True,
record_raw_sample=True,
**_kwargs: Any,
):
"""
Evaluate all provided samples in parallel.
"""
work_items = _index_samples(samples)
threads = int(os.environ.get("EVALS_THREADS", "10"))
show_progress = bool(os.environ.get("EVALS_SHOW_EVAL_PROGRESS", show_progress))
def eval_sample(args):
"""
Evaluate a single sample.
"""
sample, idx = args
base_name, split = self.name.split(".")[0:2]
sample_id = f"{base_name}.{split}.{idx}"
with recorder.as_default_recorder(sample_id):
seed = f"{sample_id}:{self.seed}".encode("utf-8")
rng = random.Random(seed)
return idx, self.eval_sample(sample, rng)
with ThreadPool(threads) as pool:
if os.environ.get("EVALS_SEQUENTIAL", "0") in {"1", "true", "yes"}:
logger.info("Running in sequential mode!")
iter = map(eval_sample, work_items)
else:
logger.info(f"Running in threaded mode with {threads} threads!")
iter = pool.imap_unordered(eval_sample, work_items)
idx_and_result = list(tqdm(iter, total=len(work_items), disable=not show_progress))
return [r for _, r in sorted(idx_and_result)]
def get_samples(self):
if self.samples_jsonl is None:
raise ValueError(
"To use `get_samples`, you must provide a `samples_jsonl` path." "Got `None`."
)
samples_path = self._get_samples_path()
return get_jsonl(samples_path.as_posix())
def _get_samples_path(self) -> Path:
return self._prefix_registry_path(self.samples_jsonl)
def _prefix_registry_path(self, data_path: str) -> Path:
if os.path.isfile(data_path):
return Path(data_path)
return self.eval_registry_path / "data" / data_path
class SolverEval(Eval):
"""
Compared to Eval, SolverEval supports a single completion_fn which must be
a `Solver` type (see solvers/solver.py). The Solver is what we evaluate,
and Eval code should interact with the Solver instead of the CompletionFn
directly. A new Solver is created for each sample, and the Solver is passed
to eval_sample. This allows Solvers to be stateful (e.g. have a memory)
without interfering with other samples.
Otherwise, this is the same as Eval and requires the same methods to be
implemented:
`eval_sample`: Takes in a Solver, a test sample, and a random number
generator and records the metrics of interest.
`run`: Takes in a recorder and runs the evaluation. Generally, most `run`
methods will follow this same pattern: loading the data, calling
`eval_all_samples`, and aggregating the recorded results.
"""
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
assert (
len(self.completion_fns) == 1
), f"{type(self).__name__} supports exactly one completion_fn, got {len(self.completion_fns)}."
# Technically, instead of arg `completion_fns: list[CompletionFn]` we
# should just have `solver: Solver` but we keep the args unchanged for
# compatibility with the existing codebase.
self._solver = maybe_wrap_with_solver(self.completion_fns[0])
@abc.abstractmethod
def eval_sample(self, solver: Solver, sample: Any, rng: random.Random) -> None:
raise NotImplementedError()
def eval_all_samples(
self,
recorder: RecorderBase,
samples,
show_progress=True,
**_kwargs: Any,
):
"""
Evaluate all provided samples in parallel.
"""
work_items = _index_samples(samples)
threads = int(os.environ.get("EVALS_THREADS", "10"))
show_progress = bool(os.environ.get("EVALS_SHOW_EVAL_PROGRESS", show_progress))
def eval_sample(args):
"""
Evaluate a single sample.
"""
sample, idx = args
base_name, split = self.name.split(".")[0:2]
sample_id = f"{base_name}.{split}.{idx}"
with recorder.as_default_recorder(sample_id):
seed = f"{sample_id}:{self.seed}".encode("utf-8")
rng = random.Random(seed)
per_sample_solver = self._solver.copy()
return idx, self.eval_sample(per_sample_solver, sample, rng)
with ThreadPool(threads) as pool:
if os.environ.get("EVALS_SEQUENTIAL", "0") in {"1", "true", "yes"}:
logger.info("Running in sequential mode!")
iter = map(eval_sample, work_items)
else:
logger.info(f"Running in threaded mode with {threads} threads!")
iter = pool.imap_unordered(eval_sample, work_items)
idx_and_result = []
try:
for result in tqdm(iter, total=len(work_items), disable=not show_progress):
idx_and_result.append(result)
except KeyboardInterrupt:
# "Gentle interrupt" allows us to stop early and still get results
gentle_interrupt = os.environ.get("EVALS_GENTLE_INTERRUPT", "0") in {
"1",
"true",
"yes",
}
if gentle_interrupt:
logger.info("Evaluation stopped because of KeyboardInterrupt")
logger.info(
f"Report will be based on {len(idx_and_result)} out of the planned {len(work_items)} samples"
)
else:
raise
return [r for _, r in sorted(idx_and_result)]
-
openai/evals 的
docs/run-evals.md，项目对运行和日志的说明，2026-10-20 读。 ↩