Spaces:
Sleeping
Sleeping
File size: 5,802 Bytes
5db30ac | 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 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 | from __future__ import annotations
import hashlib
import json
from itertools import product
from pathlib import Path
from typing import Any, Iterator
import yaml
from elbench.loaders.base import BaseLoader
from elbench.registry import ResolvedRegistryItem
from elbench.schemas.evaluation import RawRecord, Sample
class BasicEducationYamlLoader(BaseLoader):
def iter_raw_records(self, item: ResolvedRegistryItem) -> Iterator[RawRecord]:
template = _read_yaml(item.path)
task_defs = _expand_tasks(template.get("tasks", {}))
for task_index, task_payload in enumerate(task_defs, start=1):
yield RawRecord(
source_path=str(item.path),
source_file=item.entry.canonical_name,
row_index=task_index,
record={
"task_payload": task_payload,
"prompt": _extract_prompt(template, task_payload),
"task_mode": str(template.get("tasks", {}).get("mode", "iter")).lower(),
"multi_turn": _is_multi_turn(template),
"agent_count": len(template.get("agents", {})),
},
)
def iter_samples(self, item: ResolvedRegistryItem) -> Iterator[Sample]:
for raw_record in self.iter_raw_records(item):
task_payload = raw_record.record["task_payload"]
yield Sample(
sample_id=_build_sample_id(item.entry.canonical_name, raw_record.row_index, task_payload),
source_file=item.entry.canonical_name,
source_path=str(item.path),
module=item.entry.module,
subset=item.entry.subset,
task=item.entry.task,
dimension=item.entry.subset,
prompt=raw_record.record["prompt"],
reference=None,
metadata={
"row_index": raw_record.row_index,
"task_payload": task_payload,
"task_mode": raw_record.record["task_mode"],
"multi_turn": raw_record.record["multi_turn"],
"agent_count": raw_record.record["agent_count"],
},
raw_record=task_payload,
)
def _read_yaml(path: Path) -> dict[str, Any]:
with path.open("r", encoding="utf-8") as handle:
obj = yaml.safe_load(handle) or {}
if not isinstance(obj, dict):
raise ValueError(f"Expected YAML object at {path}, got {type(obj).__name__}")
return obj
def _expand_tasks(tasks_obj: Any) -> list[dict[str, Any]]:
if not isinstance(tasks_obj, dict):
return []
mode = str(tasks_obj.get("mode", "iter")).lower()
content = tasks_obj.get("content")
if mode == "iter":
if not isinstance(content, list):
return []
return [item for item in content if isinstance(item, dict)]
if mode == "union":
if not isinstance(content, dict) or not content:
return []
keys = list(content.keys())
values = [value for value in content.values()]
if not all(isinstance(value, list) for value in values):
return []
return [
{key: combo[index] for index, key in enumerate(keys)}
for combo in product(*values)
]
return []
def _extract_prompt(template: dict[str, Any], task_payload: dict[str, Any]) -> str:
start_prompt = template.get("tasks", {}).get("start_prompt")
prompt = _extract_prompt_from_message(start_prompt, task_payload)
if prompt:
return prompt
agents = template.get("agents", {})
if isinstance(agents, dict):
for agent_config in agents.values():
if not isinstance(agent_config, dict):
continue
for message in agent_config.get("prompt", []) or []:
prompt = _extract_prompt_from_message(message, task_payload)
if prompt:
return prompt
for key in ("question", "query", "prompt", "instruction"):
value = task_payload.get(key)
if value not in (None, ""):
return str(value)
return json.dumps(task_payload, ensure_ascii=False)
def _extract_prompt_from_message(message: Any, task_payload: dict[str, Any]) -> str:
if not isinstance(message, dict):
return ""
if message.get("role") != "user":
return ""
content = message.get("content")
if not isinstance(content, str) or not content.strip():
return ""
return _render_task_placeholders(content, task_payload).strip()
def _render_task_placeholders(text: str, task_payload: dict[str, Any]) -> str:
rendered = text
for key, value in task_payload.items():
string_value = _stringify(value)
rendered = rendered.replace(f"{{{key}}}", string_value)
rendered = rendered.replace(f"{{task.{key}}}", string_value)
return rendered
def _is_multi_turn(template: dict[str, Any]) -> bool:
directions = template.get("directions")
if isinstance(directions, list) and len(directions) > 2:
return True
return isinstance(template.get("tasks", {}).get("start_prompt"), dict)
def _build_sample_id(canonical_name: str, row_index: int, task_payload: dict[str, Any]) -> str:
payload = json.dumps(task_payload, ensure_ascii=False, sort_keys=True)
digest = hashlib.sha1(f"{canonical_name}::{row_index}::{payload}".encode("utf-8")).hexdigest()[:16]
safe_stem = canonical_name.replace(".", "_").replace("/", "_").replace("\\", "_")
return f"{safe_stem}-{digest}"
def _stringify(value: Any) -> str:
if isinstance(value, str):
return value
if isinstance(value, (dict, list)):
return json.dumps(value, ensure_ascii=False)
return str(value)
|