Download app/envs/base.py from githubtailscaleduplicity342/RL-Explorer: direct link, hf CLI and curl.
- Browser
- Download file 13.3 kB
-
https://huggingface.co/spaces/githubtailscaleduplicity342/RL-Explorer/resolve/main/app/envs/base.py
- Command line
-
hf download hf://spaces/githubtailscaleduplicity342/RL-Explorer/app/envs/base.py
-
curl -L -o base.py https://huggingface.co/spaces/githubtailscaleduplicity342/RL-Explorer/resolve/main/app/envs/base.py
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) | |
| 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 | |
| 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} | |