RL-Explorer / app /envs /base.py
AdithyaSK's picture
AdithyaSK HF Staff
Deploy HF RL Explorer
da5cba1 verified
Raw History Blame Contribute Delete
13.3 kB
"""How a dataset that isn't a folder of Harbor tasks is read: a *processor* turns its rows into tasks people can browse.
A processor says whether it can read a dataset (`match`), how to show one row in a list (`card`), and how to show it
in full (`view`). The explorer picks the processor with the highest match, so a specific one (MiMo, NeMo Gym, verl,
packed Harbor tasks) wins over the generic reader, which infers each column's role from its name and its values.
Answers never leave the server: a row's answer-like fields (by name: answer, solution, ground_truth, expected, ...)
are left out of everything a processor returns, and named in `withheld` instead. Write a new processor in
`app/envs/processors.py` (see CUSTOM_ENVS.md for a walk-through).
"""
from __future__ import annotations
import json
import os
import re
from dataclasses import dataclass, field
from typing import Any
ANSWER = re.compile(r"(^|[_.\-])(answers?|solutions?|ground_?truths?|gold\w*|targets?|expected\w*|references?|oracle|correct\w*|"
r"canonical\w*|labels?|gt|pass_anchor|final_answer)($|[_.\-])", re.I)
GRADING = re.compile(r"(reward|verif|rubric|grade|grader|scor|metric|atol|rtol|toleran|tests?$|test_|check|judge|instruction_id|"
r"^kwargs$|style|eval)", re.I)
ENVIRON = re.compile(r"(docker|image|environment|^env$|setup|^files$|repo|commit|cpu|mem|disk|gpu|timeout|sandbox|tools?$|bucket)", re.I)
TASK_KEYS = ("instruction", "prompt", "question", "problem", "task", "query", "input", "description", "statement", "messages",
"conversation", "conversations", "text", "content")
ID_KEYS = ("task_id", "instance_id", "id", "uid", "uuid", "problem_id", "question_id", "label", "name", "key", "slug", "path", "idx", "index")
TITLE_KEYS = ("title", "task_name", "name", "summary", "instruction_summary", "short_description", "heading")
FACET_SKIP = re.compile(r"(^id$|_id$|uuid|path|url|license|hash|sha|date|time)", re.I)
@dataclass
class Dataset:
"""What a processor sees of a dataset: its id, tags and card, its configs and splits, and a few rows."""
spec: str
tags: list[str]
card: dict[str, Any]
splits: list[dict[str, str]]
features: list[dict[str, Any]] = field(default_factory=list) # the viewer's: [{name, type}]
sample: list[dict[str, Any]] = field(default_factory=list) # the first rows of the first split
@property
def columns(self) -> list[str]:
return [f["name"] for f in self.features]
def has(self, *names: str) -> bool:
cols = {c.lower() for c in self.columns}
return all(n.lower() in cols for n in names)
def is_messages(v: Any) -> bool:
return isinstance(v, list) and bool(v) and all(isinstance(m, dict) and isinstance(m.get("role"), str) for m in v[:20])
def parse_json(v: Any) -> Any:
"""The viewer gives JSON columns as strings sometimes: a string that is a JSON object or list, decoded."""
if isinstance(v, str) and v[:1] in "{[" and len(v) < 4_000_000:
try:
return json.loads(v)
except ValueError:
return v
return v
def withhold(v: Any, path: str = "", depth: int = 0, keep: tuple[str, ...] = (), hide: frozenset[str] | set[str] = frozenset()) -> tuple[Any, list[str]]:
"""A value with its answer-like fields left out (recursively), and their dotted names. Fields are answer-like by
name (ANSWER), or because a format says so: `hide` holds dotted names ("expected_action", "verifier.patterns")
that a reader knows hold its answers though their names don't say it."""
gone: list[str] = []
if depth > 8:
return v, gone
if isinstance(v, dict):
out = {}
for k, x in v.items():
name = f"{path}.{k}" if path else str(k)
if name in hide or (isinstance(k, str) and ANSWER.search(k) and name not in keep):
gone.append(name)
continue
out[k], g = withhold(parse_json(x), name, depth + 1, keep, hide)
gone += g
return out, gone
if isinstance(v, list):
out = []
for x in v[:2000]:
y, g = withhold(x, path, depth + 1, keep, hide)
out.append(y)
gone += g
return out, sorted(set(gone))
return v, gone
def retitle(cards: list[dict[str, Any]]) -> None:
"""When a page of tasks all open the same way (a shared preamble), title each by what comes after it."""
texts = [c.get("_text") or "" for c in cards]
if len(cards) >= 3 and all(texts) and len({c["title"] for c in cards}) < len(cards) * 0.6:
p = _common_prefix(texts) or os.path.commonprefix(texts)
cut = p.rfind("\n") + 1 if "\n" in p else 0
for c, t in zip(cards, texts):
rest = t[cut:].lstrip() if cut else t
title = first_line(rest)
if title:
c["title"] = title[:200]
body = re.sub(r"\s+", " ", rest).strip()
c["snippet"] = body[body.find(title.rstrip("…")[:40]) + len(title):][:260].strip() if title.rstrip("…")[:40] in body else body[:260]
for c in cards:
c.pop("_text", None)
def _common_prefix(texts: list[str]) -> str:
"""The opening all these texts share, cut back to a paragraph or line break; "" when it's short."""
import os
p = os.path.commonprefix([t for t in texts if t])
if len(p) < 40:
return ""
cut = max(p.rfind("\n\n"), p.rfind("\n"))
return p[:cut + 1] if cut > 30 else ""
def first_line(text: str, n: int = 140) -> str:
for line in str(text or "").splitlines():
line = re.sub(r"^[#>*\-\s`]+", "", line).strip()
if len(line) > 3:
return line[:n] + ("…" if len(line) > n else "")
return ""
def text_of(v: Any) -> str:
"""The readable text of a task field: a string, or a chat's user turns."""
v = parse_json(v)
if isinstance(v, str):
return v
if is_messages(v):
return "\n\n".join(m["content"] if isinstance(m.get("content"), str) else json.dumps(m.get("content"))[:4000]
for m in v if m.get("role") in ("user", "human") or len(v) == 1)
if isinstance(v, dict):
for k in ("input", "messages", "prompt", "text", "content"):
if k in v:
return text_of(v[k])
return ""
class Processor:
id = "generic"
name = "Rows"
about = "Each row is a task. Which column holds the task, its grading and its answer is inferred from names and values."
framework = "Rows" # the label on cards and the page header
def match(self, ds: Dataset) -> float:
return 0.1
# the columns' roles, worked out once per dataset from its features and first rows
def roles(self, ds: Dataset) -> dict[str, Any]:
cols = ds.columns
low = {c.lower(): c for c in cols}
sample = ds.sample[:20]
def filled(c: str) -> bool:
return any(parse_json(r.get(c)) not in (None, "", [], {}) for r in sample)
task = next((low[k] for k in TASK_KEYS if k in low and filled(low[k])), None)
if task is None: # the longest text column
texts = sorted(((sum(len(str(r.get(c) or "")) for r in sample), c) for c in cols
if any(isinstance(r.get(c), str) for r in sample) and not ANSWER.search(c)), reverse=True)
task = texts[0][1] if texts and texts[0][0] > 40 else None
ident = next((low[k] for k in ID_KEYS if k in low and low[k] != task and all(isinstance(r.get(low[k]), (str, int)) for r in sample if r.get(low[k]) is not None)
and len({str(r.get(low[k])) for r in sample}) == len(sample)), None)
title = next((low[k] for k in TITLE_KEYS if k in low and low[k] not in (task, ident) and filled(low[k])), None)
# every task opening with the same preamble makes a useless title: then a column that varies, and short, names it
texts = [text_of(r.get(task)) for r in sample] if task else []
prefix = _common_prefix(texts) if len(texts) >= 3 else ""
if not title and texts and len({first_line(t) for t in texts}) < max(2, len(texts) * 0.6):
for k in TASK_KEYS + TITLE_KEYS:
c = low.get(k)
if not c or c == task or ANSWER.search(c):
continue
vals = [str(r.get(c) or "") for r in sample]
if all(isinstance(r.get(c), str) for r in sample if r.get(c) is not None) and len(set(vals)) >= len(vals) * 0.6 \
and sum(map(len, vals)) / max(1, len(vals)) < 400:
title = c
break
answer = [c for c in cols if ANSWER.search(c) and c != ident]
grading = [c for c in cols if c not in answer and c not in (task, ident, title) and GRADING.search(c)]
environ = [c for c in cols if c not in answer + grading and c not in (task, ident, title) and ENVIRON.search(c)]
return {"task": task, "id": ident, "title": title, "answer": answer, "grading": grading, "environment": environ,
"_prefix": prefix}
def overview(self, ds: Dataset, roles: dict[str, Any], stats: list[dict[str, Any]], config: str, split: str) -> list[dict[str, Any]]:
"""Sections for the dataset's own page (contract.section): what a row asks, how it's graded, how to run it with
its framework. Built from the first rows (`ds.sample`) and the viewer's column statistics (`stats`, [] when the
rows come from files): no more reading than the page already did."""
return []
def card(self, row: dict[str, Any], i: int, roles: dict[str, Any]) -> dict[str, Any]:
"""One row in a list: its id, a title, a line of its task, a few short facts."""
task = text_of(row.get(roles["task"])) if roles.get("task") else ""
if roles.get("_prefix") and task.startswith(roles["_prefix"]): # the preamble every task shares says nothing here
task = task[len(roles["_prefix"]):].lstrip()
ident = row.get(roles["id"]) if roles.get("id") else None
title = str(row.get(roles["title"]) or "") if roles.get("title") else ""
title = title or first_line(task) or (str(ident) if ident is not None else f"Row {i}")
chips = []
for c, v in row.items():
if c in (roles.get("task"), roles.get("id"), roles.get("title")) or c in roles.get("answer", []) or FACET_SKIP.search(c):
continue
if isinstance(v, str) and 0 < len(v) <= 24 and "\n" not in v:
chips.append(f"{v}")
elif isinstance(v, (int, float)) and not isinstance(v, bool) and c.lower() in ("difficulty", "level", "difficulty_level"):
chips.append(f"{c} {v}")
if len(chips) >= 4:
break
# titled by a column of its own (a question beside a long instruction): the instruction adds only boilerplate
snippet = "" if roles.get("title") else task if task and first_line(task) != title else ""
return {"i": i, "id": str(ident) if ident is not None else None, "title": title[:200],
"snippet": re.sub(r"\s+", " ", snippet)[:260], "chips": chips, "_text": task if not roles.get("title") else ""}
def view(self, row: dict[str, Any], i: int, ds: Dataset, roles: dict[str, Any]) -> dict[str, Any]:
"""One row in full, as sections the page renders: the task, how it's graded, what it runs in, the rest."""
keep = tuple(c for c in (roles.get("id"), roles.get("title")) if c)
clean, gone = withhold({k: parse_json(v) for k, v in row.items()}, keep=keep)
sections = []
t = roles.get("task")
if t and t in clean:
v = clean[t]
sections.append(section("task", "The task", v, note=f"from {t}"))
grading = {c: clean[c] for c in roles.get("grading", []) if c in clean}
if grading:
sections.append(section("grading", "How it's graded", grading, kind="value", note="the row's grading fields"))
env = {c: clean[c] for c in roles.get("environment", []) if c in clean}
if env:
sections.append(section("environment", "What it runs in", env, kind="value"))
rest = {k: v for k, v in clean.items() if k not in (t, *grading, *env)}
if rest:
sections.append(section("data", "Everything else in the row", rest, kind="value"))
card = self.card(row, i, roles)
return {"title": card["title"], "id": card["id"], "chips": card["chips"], "sections": sections, "withheld": gone,
"glance": [["Row", f"{i:,}"], *([["Id", card["id"]]] if card["id"] else []),
["Task from", t or "–"], ["Graded by", ", ".join(grading) or "not in the row"]]}
def section(id: str, title: str, value: Any, kind: str | None = None, note: str = "") -> dict[str, Any]:
"""A view section: `kind` says how the page draws `body` (markdown, messages, code, value, kv, files, or blocks: a
list of contract blocks, for a reader that builds its own)."""
if kind is None:
v = parse_json(value)
kind = "messages" if is_messages(v) else "markdown" if isinstance(v, str) else "value"
value = v
return {"id": id, "title": title, "kind": kind, "body": value, "note": note}