Spaces:
Runtime error
Runtime error
File size: 2,521 Bytes
cd5c208 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 | """Formatting, parsing, and IO-suppression helpers for inference."""
from __future__ import annotations
import io
from collections.abc import Iterable
from contextlib import contextmanager, redirect_stderr, redirect_stdout
from typing import Any, Iterator
try:
from tasks import task_ids
except ImportError: # pragma: no cover
from python_env.tasks import task_ids # type: ignore[no-redef]
def compact_text(
value: Any,
*,
default: str = "",
limit: int = 240,
preserve_newlines: bool = False,
) -> str:
"""Convert values into validator-safe text."""
if value is None:
return default
try:
text = str(value)
except Exception:
return default
if preserve_newlines:
text = text.strip()
else:
text = " ".join(text.split())
return text[:limit] if text else default
def observation_attr(observation: Any, name: str, default: Any = None, *, preserve_newlines: bool = False) -> Any:
"""Read an observation attribute without trusting the payload shape."""
if isinstance(observation, dict):
value = observation.get(name, default)
else:
value = getattr(observation, name, default)
if isinstance(value, str):
return compact_text(
value,
default=default if isinstance(default, str) else "",
preserve_newlines=preserve_newlines,
)
return value
def format_bool(value: Any) -> str:
return "true" if bool(value) else "false"
def format_reward(value: Any) -> str:
try:
reward = float(value)
except Exception:
reward = 0.0
return f"{reward:.2f}"
def format_error(value: Any) -> str:
text = compact_text(value, default="")
return text if text else "null"
def parse_task_ids() -> list[str]:
"""Load stable task names with a deterministic fallback."""
try:
values = task_ids()
if isinstance(values, Iterable):
loaded = [compact_text(item, default="") for item in values]
loaded = [item for item in loaded if item]
if loaded:
return loaded
except Exception:
pass
return [
"syntax_fix_invoice_totals",
"bug_fix_session_windows",
"optimization_rank_active_users",
]
@contextmanager
def suppress_output() -> Iterator[None]:
"""Silence libraries that write noisy logs to stdout or stderr."""
with redirect_stdout(io.StringIO()), redirect_stderr(io.StringIO()):
yield
|