"""
for index, column in enumerate(columns)
)
body_rows = []
for _, row in frame.iterrows():
cells = "".join(
f'
{cell_html(row[column], column)}
'
for index, column in enumerate(columns)
)
body_rows.append(f"
{cells}
")
return f"""
{header}
{''.join(body_rows)}
"""
def load_datasets() -> list[dict[str, Any]]:
return read_json(DATASETS_PATH, [])
def load_results() -> list[dict[str, Any]]:
return read_jsonl(RESULTS_PATH)
def normalize_submission_record(record: dict[str, Any]) -> dict[str, Any]:
model_id = str(record.get("model_id", "")).strip()
if not model_id:
raise ValueError("`model_id` is required.")
scores = record.get("scores")
if not isinstance(scores, dict):
raise ValueError("`scores` must be an object.")
average_scores = scores.get("average", {})
dataset_scores = scores.get("datasets", {})
if average_scores is None:
average_scores = {}
if dataset_scores is None:
dataset_scores = {}
if not isinstance(average_scores, dict):
raise ValueError("`scores.average` must be an object.")
if not isinstance(dataset_scores, dict):
raise ValueError("`scores.datasets` must be an object.")
has_metric = any(metric_value(average_scores, metric) is not None for metric in METRICS)
if not has_metric:
for metrics in dataset_scores.values():
if isinstance(metrics, dict) and any(metric_value(metrics, metric) is not None for metric in METRICS):
has_metric = True
break
if not has_metric:
raise ValueError(f"At least one numeric metric is required: {', '.join(METRICS)}.")
normalized = dict(record)
normalized["model_id"] = model_id
normalized["model_name"] = str(record.get("model_name") or model_id.split("/")[-1]).strip()
normalized["organization"] = str(record.get("organization") or (model_id.split("/", 1)[0] if "/" in model_id else "")).strip()
normalized["type"] = str(record.get("type") or "dense").strip()
normalized["date"] = str(record.get("date") or date.today().isoformat()).strip()
normalized["verified"] = bool(record.get("verified", False))
normalized["source_url"] = str(record.get("source_url", "")).strip()
normalized["scores"] = {"average": average_scores, "datasets": dataset_scores}
return normalized
def parse_submission_records(record_text: str) -> list[dict[str, Any]]:
record_text = record_text.strip()
if not record_text:
return []
if record_text.startswith("["):
parsed = json.loads(record_text)
if not isinstance(parsed, list):
raise ValueError("JSON array submission must contain result objects.")
records = parsed
else:
records = []
for line_no, line in enumerate(record_text.splitlines(), start=1):
line = line.strip()
if not line:
continue
try:
records.append(json.loads(line))
except json.JSONDecodeError as exc:
raise ValueError(f"Invalid JSONL at line {line_no}: {exc}") from exc
normalized_records = []
for index, record in enumerate(records, start=1):
if not isinstance(record, dict):
raise ValueError(f"Submission item {index} must be a JSON object.")
normalized_records.append(normalize_submission_record(record))
return normalized_records
def uploaded_file_text(uploaded_file: Any) -> str:
if uploaded_file is None:
return ""
path = uploaded_file
if isinstance(uploaded_file, dict):
path = uploaded_file.get("path") or uploaded_file.get("name")
else:
path = getattr(uploaded_file, "path", None) or getattr(uploaded_file, "name", None) or uploaded_file
if not path:
return ""
return Path(path).read_text(encoding="utf-8-sig")
def merge_submission_records(current_records: list[dict[str, Any]], records: list[dict[str, Any]]) -> tuple[list[dict[str, Any]], int, int]:
existing = set()
for item in current_records:
try:
existing.add(json.dumps(normalize_submission_record(item), ensure_ascii=False, sort_keys=True, separators=(",", ":")))
except ValueError:
continue
merged_records = list(current_records)
skipped = 0
for record in records:
serialized = json.dumps(record, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
if serialized in existing:
skipped += 1
continue
existing.add(serialized)
merged_records.append(record)
return merged_records, len(merged_records) - len(current_records), skipped
def serialize_jsonl(records: list[dict[str, Any]]) -> bytes:
text = "".join(
f"{json.dumps(record, ensure_ascii=False, sort_keys=True, separators=(',', ':'))}\n"
for record in records
)
return text.encode("utf-8")
def write_local_results(content: bytes) -> None:
RESULTS_PATH.parent.mkdir(parents=True, exist_ok=True)
temporary_path = RESULTS_PATH.with_suffix(f"{RESULTS_PATH.suffix}.tmp")
temporary_path.write_bytes(content)
temporary_path.replace(RESULTS_PATH)
def append_submission_records(records: list[dict[str, Any]]) -> tuple[int, int, str | None]:
if not RESULTS_REPO_ID:
merged, added, skipped = merge_submission_records(load_results(), records)
if added:
write_local_results(serialize_jsonl(merged))
return added, skipped, None
if not HF_TOKEN:
raise ValueError("HF_TOKEN secret is missing. Add a fine-grained write token in the Space settings.")
api = HfApi(token=HF_TOKEN)
with RESULTS_LOCK:
for attempt in range(3):
repo_info = api.repo_info(repo_id=RESULTS_REPO_ID, repo_type=RESULTS_REPO_TYPE, revision=RESULTS_REVISION)
remote_path = hf_hub_download(repo_id=RESULTS_REPO_ID, filename=RESULTS_REPO_PATH, repo_type=RESULTS_REPO_TYPE, revision=repo_info.sha, token=HF_TOKEN)
current_records = read_jsonl(Path(remote_path))
merged, added, skipped = merge_submission_records(current_records, records)
if not added:
write_local_results(serialize_jsonl(merged))
return 0, skipped, None
content = serialize_jsonl(merged)
try:
commit = api.create_commit(
repo_id=RESULTS_REPO_ID,
repo_type=RESULTS_REPO_TYPE,
revision=RESULTS_REVISION,
parent_commit=repo_info.sha,
operations=[CommitOperationAdd(path_in_repo=RESULTS_REPO_PATH, path_or_fileobj=content)],
commit_message=f"Add {added} leaderboard result(s)")
except HfHubHTTPError:
if attempt < 2:
continue
raise
write_local_results(content)
return added, skipped, commit.commit_url
raise RuntimeError("Could not update leaderboard results after multiple attempts.")
def add_submission_record(uploaded_file: Any, metric: str, task_filter: str, verified_only: bool, model_filter: str) -> tuple[str, str, str]:
try:
file_text = uploaded_file_text(uploaded_file)
except OSError as exc:
return (
f"Submission was not added: could not read uploaded file: {exc}",
summary_html(),
leaderboard_table_html(metric, task_filter, verified_only, model_filter),
)
if not file_text.strip():
return (
"Upload a non-empty results.jsonl file first.",
summary_html(),
leaderboard_table_html(metric, task_filter, verified_only, model_filter),
)
try:
records = parse_submission_records(file_text)
added, skipped, commit_url = append_submission_records(records)
except (json.JSONDecodeError, ValueError, OSError, HfHubHTTPError) as exc:
return (
f"Submission was not added: {exc}",
summary_html(),
leaderboard_table_html(metric, task_filter, verified_only, model_filter),
)
if added == 0 and skipped > 0:
status = f"No new records were added; skipped {skipped} duplicate record(s)."
else:
status = f"Added {added} record(s) to `{RESULTS_PATH.name}`."
if skipped:
status += f" Skipped {skipped} duplicate record(s)."
if commit_url:
status += f" [Hub commit]({commit_url})"
return (
status,
summary_html(),
leaderboard_table_html(metric, task_filter, verified_only, model_filter),
)
def logo_data_uri() -> str:
if not LOGO_PATH.exists():
return ""
data = base64.b64encode(LOGO_PATH.read_bytes()).decode("ascii")
return f"data:image/png;base64,{data}"
def format_score(value: float | None) -> str:
if value is None:
return "n/a"
try:
numeric_value = float(value)
except (TypeError, ValueError):
return "n/a"
return f"{numeric_value * 100:.2f}"
def summary_html() -> str:
datasets = [dataset for dataset in load_datasets() if dataset.get("official", True)]
records = load_results()
frame = leaderboard_frame(DEFAULT_METRIC, "All", False, "")
best_model = "No results yet"
best_score = "n/a"
if not frame.empty and DEFAULT_METRIC in frame:
best_row = frame.iloc[0]
best_model = str(best_row.get("Model", "No results yet"))
best_score = format_score(best_row.get(DEFAULT_METRIC))
types = sorted({str(record.get("type", "")).strip() for record in records if record.get("type")})
type_text = ", ".join(types) if types else "n/a"
logo_uri = logo_data_uri()
logo_html = f'' if logo_uri else ""
return f"""
Russian Information Retrieval Benchmark
rusBEIR Leaderboard
Compare dense retrievers, sparse baselines, and reranker pipelines on official rusBEIR datasets.
The default ranking is the macro-average of {DEFAULT_METRIC}.
')
table = gr.HTML(value=leaderboard_table_html(DEFAULT_METRIC, "All", False, ""))
for control in [metric, task_filter, verified_only, model_filter]:
control.change(leaderboard_table_html,
inputs=[metric, task_filter, verified_only, model_filter],
outputs=table)
reload_button = gr.Button("Reload results", variant="secondary")
reload_button.click(leaderboard_table_html,
inputs=[metric, task_filter, verified_only, model_filter],
outputs=table)
with gr.Tab("Datasets"):
with gr.Column(elem_classes=["rusbeir-panel"]):
gr.HTML(
"""
Official Datasets
rusBEIR tasks used for the default macro-average ranking.
"""
)
gr.Dataframe(value=datasets_frame(), label="Datasets", interactive=False, wrap=True, max_height=720)
with gr.Tab("Submit"):
with gr.Column(elem_classes=["rusbeir-panel"]):
gr.HTML(
"""
Submit Results
Run the evaluator outside the Space and upload the generated results.jsonl.
Accepted files contain one JSON result object per line.
We strongly recommend checking your results before uploading them to the Leaderboard.
Retracting results is a manual process and can be handled only by @kaengreg.
rusBEIR is a Russian BEIR-style benchmark for zero-shot information retrieval.
The leaderboard is backed by a plain JSONL file, so every row can be reviewed or
mirrored to a Hugging Face Dataset.
Verified rows should point to reproducible logs or a commit with generated retrieval results.
@inproceedings{kovalev2025building,
title={Building Russian Benchmark for Evaluation of Information Retrieval Models},
author={Kovalev, Grigory and Tikhomirov, Mikhail and Kozhevnikov, Evgeny and Kornilov, Max and Loukachevitch, Natalia},
booktitle={Proceedings of the International Conference “Dialogue},
volume={2025},
year={2025}
}
"""
)
if __name__ == "__main__":
launch_kwargs = {
"server_name": "0.0.0.0",
"server_port": int(os.getenv("PORT", "7860")),
"show_error": True,
"css": CUSTOM_CSS,
"js": CUSTOM_JS
}
if "ssr_mode" in inspect.signature(demo.launch).parameters:
launch_kwargs["ssr_mode"] = False
demo.launch(**launch_kwargs)