Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions packages/prime-runs/pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@ authors = [
# `tenacity`. Nothing is added to what verifiers and prime-rl already carry.
dependencies = [
"httpx>=0.25.0",
"orjson>=3.10.0",
"prime-traces>=0.0.3",
]
keywords = ["evals", "evaluations", "training", "runs", "traces", "observability"]
Expand Down
8 changes: 3 additions & 5 deletions packages/prime-runs/src/prime_runs/_http.py
Original file line number Diff line number Diff line change
Expand Up @@ -15,13 +15,13 @@
Error mapping and backoff come from ``prime_traces.core.client``.
"""

import json
import sys
import time
from typing import Any, Dict, Mapping, Optional, Tuple, Union
from urllib.parse import urlsplit, urlunsplit

import httpx
import orjson
from prime_traces.core.client import AMBIGUOUS_TRANSPORT_ERRORS, raise_for_response, retry_delay

from . import _fork
Expand Down Expand Up @@ -72,10 +72,8 @@ def _user_agent() -> str:


def encode_json(value: Any) -> bytes:
"""Compact UTF-8 JSON; bare ``NaN`` is refused here rather than as a 400."""
return json.dumps(value, ensure_ascii=False, allow_nan=False, separators=(",", ":")).encode(
"utf-8"
)
"""Compact UTF-8 JSON; NaN and infinity go out as ``null``, never as a bare ``NaN``."""
return orjson.dumps(value, option=orjson.OPT_SERIALIZE_NUMPY | orjson.OPT_NON_STR_KEYS)


class PlatformClient:
Expand Down
5 changes: 3 additions & 2 deletions packages/prime-runs/src/prime_runs/projection.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,11 +3,12 @@
training viewer's Parquet table (moved here from prime-rl). Duck-typed: verifiers
``Trace``/``Episode`` satisfy it structurally and are not imported."""

import json
import logging
import math
from typing import Any, Dict, Iterable, List, Optional, Sequence

import orjson

from ._http import encode_json

logger = logging.getLogger(__name__)
Expand Down Expand Up @@ -38,7 +39,7 @@ def _nullify_non_finite(value: Any) -> Any:


def _json_column(value: Any) -> str:
return json.dumps(_nullify_non_finite(value), allow_nan=False)
return orjson.dumps(value, option=orjson.OPT_SERIALIZE_NUMPY | orjson.OPT_NON_STR_KEYS).decode()


def summary_trace_index(episode: Any) -> int:
Expand Down
14 changes: 8 additions & 6 deletions packages/prime-runs/tests/test_init.py
Original file line number Diff line number Diff line change
Expand Up @@ -315,14 +315,16 @@ def test_a_forked_child_does_not_duplicate_the_parents_records_or_close_its_run(
assert handler.paths().count("POST /api/v1/evaluations/eval-abc/finalize") == 1


def test_records_reject_nonfinite_json_instead_of_sending_an_opaque_400(online):
run, _ = online(on_error="raise")
run.log_traces([{"sample_id": "bad", "reward": float("nan")}])

with pytest.raises(ValueError, match="Out of range float values"):
run.flush()
def test_records_send_nonfinite_floats_as_null_instead_of_an_opaque_400(online):
run, handler = online(on_error="raise")
run.log_traces([{"sample_id": "nan", "reward": float("nan")}])
run.flush()
run.finish()

posted = handler.bodies_for("/api/v1/evaluations/eval-abc/samples")
(sample,) = [s for body in posted for s in body["samples"]]
assert sample["sample_id"] == "nan" and sample["reward"] is None


# ----------------------------------------------------------------- training

Expand Down
21 changes: 21 additions & 0 deletions packages/prime-runs/tests/test_train_samples_sink.py
Original file line number Diff line number Diff line change
Expand Up @@ -371,6 +371,27 @@ def test_the_default_encoder_writes_non_finite_numbers_as_null():
assert math.isinf(row["advantage"]) # the float column keeps the value


def test_the_default_encoder_writes_numpy_record_streams():
np = pytest.importorskip("numpy")
pytest.importorskip("pyarrow")
import io
import json

import pyarrow.parquet as pq

from prime_runs.projection import episodes_to_parquet_bytes

episode = make_train_episode("e1", step=10)
trace = episode.traces[0]
trace.to_record = lambda: {"id": trace.id, "token_ids": np.array([1, 2, 3])}

payload = episodes_to_parquet_bytes([episode], "run-abc", 10)

(row,) = pq.read_table(io.BytesIO(payload)).to_pylist()
info = json.loads(row["info"])
assert info["native_wrapper"]["traces"][0]["token_ids"] == [1, 2, 3]


def test_the_default_encoder_numbers_rows_from_the_offset():
pytest.importorskip("pyarrow")
import io
Expand Down
1 change: 1 addition & 0 deletions packages/prime-traces/pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@ authors = [
]
dependencies = [
"httpx>=0.25.0",
"orjson>=3.10.0",
"pydantic>=2.0.0",
]
keywords = ["traces", "observability", "evals", "rollouts"]
Expand Down
2 changes: 1 addition & 1 deletion packages/prime-traces/src/prime_traces/async_traces.py
Original file line number Diff line number Diff line change
Expand Up @@ -86,7 +86,7 @@ def _record_lines_for(

``aiter_batches`` consumes a synchronous source entirely in a worker
thread, so handing it the sync generator — rather than adapting it to an
async one here — is what keeps ``json.dumps`` over large records off the
async one here — is what keeps encoding large records off the
event loop.
"""
if isinstance(records, AsyncIterable):
Expand Down
20 changes: 7 additions & 13 deletions packages/prime-traces/src/prime_traces/traces.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,6 @@
field registry).
"""

import json
import tempfile
import time
from pathlib import Path
Expand All @@ -37,6 +36,8 @@
)
from urllib.parse import quote

import orjson

from .batching import (
DEFAULT_TARGET_BATCH_BYTES,
Batch,
Expand Down Expand Up @@ -77,6 +78,8 @@ def to_record(self) -> Mapping[str, Any]: ...

TraceRecord = Union[Mapping[str, Any], SupportsToRecord]

RECORD_OPTIONS = orjson.OPT_SERIALIZE_NUMPY | orjson.OPT_NON_STR_KEYS | orjson.OPT_APPEND_NEWLINE


def _encode_record(record: TraceRecord, record_number: int) -> bytes:
"""Serialize one in-memory record as a compact UTF-8 JSONL line.
Expand All @@ -95,18 +98,9 @@ def _encode_record(record: TraceRecord, record_number: int) -> bytes:
if not isinstance(value, Mapping):
raise TypeError(f"Record {record_number} to_record() must return a mapping")

# Compact separators match common JSONL writers, ensure_ascii=False
# keeps the wire representation UTF-8, and allow_nan=False prevents
# emitting JavaScript-only NaN/Infinity values that strict JSON rejects.
return (
json.dumps(
dict(value),
ensure_ascii=False,
allow_nan=False,
separators=(",", ":"),
).encode("utf-8")
+ b"\n"
)
# orjson writes compact UTF-8 and encodes NaN/Infinity as null, which
# strict JSON parsers accept. Numpy arrays let callers skip a list copy.
return orjson.dumps(dict(value), option=RECORD_OPTIONS)


def _record_lines(records: Iterable[TraceRecord]) -> Iterator[bytes]:
Expand Down
42 changes: 42 additions & 0 deletions uv.lock

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

Loading