Spaces:
Sleeping
Sleeping
File size: 3,469 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 | from __future__ import annotations
import hashlib
import json
from typing import Any
from elbench.loaders.resolvers import resolve_dimension
from elbench.registry import ResolvedRegistryItem
from elbench.schemas.evaluation import RawRecord, Sample
class RecordNormalizer:
def normalize(self, item: ResolvedRegistryItem, raw_record: RawRecord) -> Sample:
record = raw_record.record
mapping = item.mapping
prompt = self._build_prompt(mapping, record)
sample_id = self._build_sample_id(item, raw_record)
dimension = self._extract_dimension(mapping, record)
reference = self._extract_reference(mapping.reference_fields, record)
metadata = self._extract_metadata(mapping.metadata_fields, raw_record)
return Sample(
sample_id=sample_id,
source_file=item.entry.canonical_name,
source_path=str(item.path),
module=item.entry.module,
subset=item.entry.subset,
task=item.entry.task,
dimension=dimension,
prompt=prompt,
reference=reference,
metadata=metadata,
raw_record=record,
)
def _build_prompt(self, mapping: Any, record: dict[str, Any]) -> str:
if mapping.prompt_template:
rendered = mapping.prompt_template
for key, value in record.items():
rendered = rendered.replace(f"{{{{ {key} }}}}", self._stringify(value))
return rendered.strip()
for key in mapping.prompt_fields:
value = record.get(key)
if value not in (None, ""):
return self._stringify(value)
raise ValueError("Prompt fields are empty for record.")
def _build_sample_id(self, item: ResolvedRegistryItem, raw_record: RawRecord) -> str:
parts = []
for key in item.mapping.sample_id_fields:
value = raw_record.record.get(key)
if value not in (None, ""):
parts.append(self._stringify(value))
if not parts:
parts = [item.entry.canonical_name, str(raw_record.row_index), json.dumps(raw_record.record, ensure_ascii=False)]
digest = hashlib.sha1("||".join(parts).encode("utf-8")).hexdigest()[:16]
return f"{_safe_stem(item.entry.canonical_name)}-{digest}"
def _extract_dimension(self, mapping: Any, record: dict[str, Any]) -> str | None:
for key in mapping.dimension_fields:
value = record.get(key)
if value not in (None, ""):
return self._stringify(value)
return resolve_dimension(mapping.dimension_resolver, record)
def _extract_reference(self, fields: list[str], record: dict[str, Any]) -> dict[str, Any] | None:
reference = {key: record.get(key) for key in fields if key in record}
return reference or None
def _extract_metadata(self, fields: list[str], raw_record: RawRecord) -> dict[str, Any]:
metadata = {key: raw_record.record.get(key) for key in fields if key in raw_record.record}
metadata["row_index"] = raw_record.row_index
return metadata
def _stringify(self, value: Any) -> str:
if isinstance(value, str):
return value
return json.dumps(value, ensure_ascii=False) if isinstance(value, (dict, list)) else str(value)
def _safe_stem(filename: str) -> str:
return filename.replace(".", "_").replace("/", "_").replace("\\", "_")
|