1413 lines
26 KiB
Python
1413 lines
26 KiB
Python
from __future__ import annotations
|
|
|
|
import argparse
|
|
import csv
|
|
import json
|
|
import sqlite3
|
|
import sys
|
|
from collections import defaultdict
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
PROJECT_ROOT = Path(__file__).resolve().parents[1]
|
|
|
|
if str(PROJECT_ROOT) not in sys.path:
|
|
sys.path.insert(
|
|
0,
|
|
str(PROJECT_ROOT),
|
|
)
|
|
|
|
|
|
from scripts.common import DB_FILE
|
|
from scripts.search_utils import (
|
|
DEFAULT_CANDIDATE_MULTIPLIER,
|
|
MIN_CANDIDATES,
|
|
add_fts_metadata,
|
|
add_vector_metadata,
|
|
build_match_queries,
|
|
diversify_results,
|
|
fuse_hybrid_results,
|
|
run_fts_query,
|
|
run_vector_query,
|
|
verify_search_schema,
|
|
)
|
|
|
|
|
|
EVALUATION_K_VALUES = (
|
|
1,
|
|
3,
|
|
5,
|
|
)
|
|
|
|
EVALUATION_MODES = (
|
|
"fts",
|
|
"vector",
|
|
"hybrid",
|
|
)
|
|
|
|
VALID_SPLITS = (
|
|
"dev",
|
|
"test",
|
|
)
|
|
|
|
|
|
def load_questions(
|
|
path: Path,
|
|
) -> list[dict[str, Any]]:
|
|
if not path.exists():
|
|
raise FileNotFoundError(
|
|
f"Evaluačný dataset neexistuje: {path}"
|
|
)
|
|
|
|
with path.open(
|
|
"r",
|
|
encoding="utf-8",
|
|
) as file:
|
|
data = json.load(
|
|
file
|
|
)
|
|
|
|
if not isinstance(
|
|
data,
|
|
list,
|
|
):
|
|
raise ValueError(
|
|
"questions.json musí obsahovať JSON pole"
|
|
)
|
|
|
|
questions: list[
|
|
dict[str, Any]
|
|
] = []
|
|
|
|
seen_ids: set[str] = set()
|
|
|
|
for index, item in enumerate(
|
|
data,
|
|
start=1,
|
|
):
|
|
if not isinstance(
|
|
item,
|
|
dict,
|
|
):
|
|
raise ValueError(
|
|
"Každá evaluačná otázka musí byť "
|
|
f"JSON objekt. Chyba pri položke {index}."
|
|
)
|
|
|
|
question_id = item.get(
|
|
"id"
|
|
)
|
|
|
|
question = item.get(
|
|
"question"
|
|
)
|
|
|
|
split = item.get(
|
|
"split"
|
|
)
|
|
|
|
expected_documents = item.get(
|
|
"expected_documents"
|
|
)
|
|
|
|
if (
|
|
not isinstance(
|
|
question_id,
|
|
str,
|
|
)
|
|
or not question_id.strip()
|
|
):
|
|
raise ValueError(
|
|
f"Položka {index} nemá platné id"
|
|
)
|
|
|
|
if question_id in seen_ids:
|
|
raise ValueError(
|
|
f"Dataset obsahuje duplicitné id: {question_id}"
|
|
)
|
|
|
|
seen_ids.add(
|
|
question_id
|
|
)
|
|
|
|
if (
|
|
not isinstance(
|
|
question,
|
|
str,
|
|
)
|
|
or not question.strip()
|
|
):
|
|
raise ValueError(
|
|
f"{question_id}: chýba otázka"
|
|
)
|
|
|
|
if split not in VALID_SPLITS:
|
|
raise ValueError(
|
|
f"{question_id}: split musí byť "
|
|
"'dev' alebo 'test'"
|
|
)
|
|
|
|
if (
|
|
not isinstance(
|
|
expected_documents,
|
|
list,
|
|
)
|
|
or not expected_documents
|
|
):
|
|
raise ValueError(
|
|
f"{question_id}: chýba expected_documents"
|
|
)
|
|
|
|
if not all(
|
|
isinstance(
|
|
value,
|
|
str,
|
|
)
|
|
and value.strip()
|
|
for value in expected_documents
|
|
):
|
|
raise ValueError(
|
|
f"{question_id}: expected_documents "
|
|
"musí obsahovať neprázdne reťazce"
|
|
)
|
|
|
|
questions.append(
|
|
item
|
|
)
|
|
|
|
return questions
|
|
|
|
|
|
def filter_questions_by_split(
|
|
questions: list[
|
|
dict[str, Any]
|
|
],
|
|
split: str,
|
|
) -> list[dict[str, Any]]:
|
|
if split == "all":
|
|
return questions
|
|
|
|
return [
|
|
item
|
|
for item in questions
|
|
if item.get("split") == split
|
|
]
|
|
|
|
|
|
def count_splits(
|
|
questions: list[
|
|
dict[str, Any]
|
|
],
|
|
) -> dict[str, int]:
|
|
counts = {
|
|
"dev": 0,
|
|
"test": 0,
|
|
}
|
|
|
|
for item in questions:
|
|
split = item.get(
|
|
"split"
|
|
)
|
|
|
|
if split in counts:
|
|
counts[
|
|
split
|
|
] += 1
|
|
|
|
return counts
|
|
|
|
|
|
def load_index_document_paths(
|
|
db_file: Path,
|
|
) -> set[str]:
|
|
if not db_file.exists():
|
|
raise FileNotFoundError(
|
|
f"Databáza neexistuje: {db_file}"
|
|
)
|
|
|
|
with sqlite3.connect(
|
|
db_file,
|
|
timeout=5.0,
|
|
) as conn:
|
|
rows = conn.execute(
|
|
"""
|
|
SELECT DISTINCT
|
|
document_path
|
|
FROM chunks
|
|
ORDER BY document_path
|
|
"""
|
|
).fetchall()
|
|
|
|
return {
|
|
str(row[0])
|
|
for row in rows
|
|
}
|
|
|
|
|
|
def validate_dataset(
|
|
questions: list[
|
|
dict[str, Any]
|
|
],
|
|
indexed_documents: set[str],
|
|
) -> list[dict[str, str]]:
|
|
missing: list[
|
|
dict[str, str]
|
|
] = []
|
|
|
|
for item in questions:
|
|
question_id = str(
|
|
item["id"]
|
|
)
|
|
|
|
for document_path in item[
|
|
"expected_documents"
|
|
]:
|
|
if (
|
|
document_path
|
|
not in indexed_documents
|
|
):
|
|
missing.append(
|
|
{
|
|
"question_id": (
|
|
question_id
|
|
),
|
|
"document_path": (
|
|
document_path
|
|
),
|
|
}
|
|
)
|
|
|
|
return missing
|
|
|
|
|
|
def retrieve_all_modes(
|
|
db_file: Path,
|
|
query: str,
|
|
*,
|
|
limit: int,
|
|
published_only: bool,
|
|
max_per_document: int,
|
|
) -> dict[
|
|
str,
|
|
list[dict[str, Any]],
|
|
]:
|
|
clean_query = query.strip()
|
|
|
|
if not clean_query:
|
|
return {
|
|
mode: []
|
|
for mode in EVALUATION_MODES
|
|
}
|
|
|
|
match_queries = build_match_queries(
|
|
clean_query
|
|
)
|
|
|
|
candidate_limit = max(
|
|
MIN_CANDIDATES,
|
|
(
|
|
limit
|
|
* DEFAULT_CANDIDATE_MULTIPLIER
|
|
),
|
|
)
|
|
|
|
with sqlite3.connect(
|
|
db_file,
|
|
timeout=5.0,
|
|
) as conn:
|
|
conn.row_factory = (
|
|
sqlite3.Row
|
|
)
|
|
|
|
conn.execute(
|
|
"PRAGMA query_only = ON"
|
|
)
|
|
|
|
verify_search_schema(
|
|
conn
|
|
)
|
|
|
|
# -------------------------
|
|
# FTS5
|
|
# -------------------------
|
|
|
|
fts_candidates: list[
|
|
dict[str, Any]
|
|
] = []
|
|
|
|
used_strategies: list[
|
|
str
|
|
] = []
|
|
|
|
for (
|
|
strategy,
|
|
match_query,
|
|
) in match_queries:
|
|
rows = run_fts_query(
|
|
conn,
|
|
match_query,
|
|
candidate_limit,
|
|
published_only,
|
|
)
|
|
|
|
if not rows:
|
|
continue
|
|
|
|
for row in rows:
|
|
row[
|
|
"strategy"
|
|
] = strategy
|
|
|
|
fts_candidates = rows
|
|
|
|
used_strategies = [
|
|
strategy
|
|
]
|
|
|
|
break
|
|
|
|
fts_results = add_fts_metadata(
|
|
conn,
|
|
clean_query,
|
|
fts_candidates,
|
|
)
|
|
|
|
# -------------------------
|
|
# Embeddings
|
|
# -------------------------
|
|
|
|
vector_candidates = (
|
|
run_vector_query(
|
|
conn,
|
|
clean_query,
|
|
candidate_limit,
|
|
published_only,
|
|
)
|
|
)
|
|
|
|
vector_results = (
|
|
add_vector_metadata(
|
|
conn,
|
|
vector_candidates,
|
|
)
|
|
)
|
|
|
|
# -------------------------
|
|
# Hybrid
|
|
# -------------------------
|
|
|
|
hybrid_vector_results = (
|
|
vector_results
|
|
)
|
|
|
|
if (
|
|
used_strategies
|
|
and used_strategies[0]
|
|
in {
|
|
"all_terms",
|
|
"prefix_terms",
|
|
}
|
|
):
|
|
fts_chunk_ids = {
|
|
item["chunk_id"]
|
|
for item in fts_results
|
|
}
|
|
|
|
hybrid_vector_results = [
|
|
item
|
|
for item in vector_results
|
|
if item["chunk_id"]
|
|
in fts_chunk_ids
|
|
]
|
|
|
|
hybrid_results = (
|
|
fuse_hybrid_results(
|
|
fts_results,
|
|
hybrid_vector_results,
|
|
)
|
|
)
|
|
|
|
return {
|
|
"fts": diversify_results(
|
|
fts_results,
|
|
limit,
|
|
max_per_document,
|
|
),
|
|
"vector": diversify_results(
|
|
vector_results,
|
|
limit,
|
|
max_per_document,
|
|
),
|
|
"hybrid": diversify_results(
|
|
hybrid_results,
|
|
limit,
|
|
max_per_document,
|
|
),
|
|
}
|
|
|
|
|
|
def unique_document_ranking(
|
|
results: list[
|
|
dict[str, Any]
|
|
],
|
|
) -> list[dict[str, Any]]:
|
|
selected: list[
|
|
dict[str, Any]
|
|
] = []
|
|
|
|
seen: set[str] = set()
|
|
|
|
for item in results:
|
|
document_path = str(
|
|
item["document_path"]
|
|
)
|
|
|
|
if document_path in seen:
|
|
continue
|
|
|
|
seen.add(
|
|
document_path
|
|
)
|
|
|
|
selected.append(
|
|
item
|
|
)
|
|
|
|
return selected
|
|
|
|
|
|
def first_relevant_rank(
|
|
ranked_documents: list[
|
|
dict[str, Any]
|
|
],
|
|
expected_documents: set[str],
|
|
) -> int | None:
|
|
for rank, item in enumerate(
|
|
ranked_documents,
|
|
start=1,
|
|
):
|
|
if (
|
|
item["document_path"]
|
|
in expected_documents
|
|
):
|
|
return rank
|
|
|
|
return None
|
|
|
|
|
|
def recall_at_k(
|
|
ranked_documents: list[
|
|
dict[str, Any]
|
|
],
|
|
expected_documents: set[str],
|
|
k: int,
|
|
) -> float:
|
|
if not expected_documents:
|
|
return 0.0
|
|
|
|
retrieved = {
|
|
str(
|
|
item["document_path"]
|
|
)
|
|
for item in ranked_documents[
|
|
:k
|
|
]
|
|
}
|
|
|
|
relevant_retrieved = (
|
|
retrieved
|
|
& expected_documents
|
|
)
|
|
|
|
return (
|
|
len(
|
|
relevant_retrieved
|
|
)
|
|
/ len(
|
|
expected_documents
|
|
)
|
|
)
|
|
|
|
|
|
def evaluate_question_mode(
|
|
question: dict[str, Any],
|
|
mode: str,
|
|
results: list[
|
|
dict[str, Any]
|
|
],
|
|
) -> dict[str, Any]:
|
|
ranked_documents = (
|
|
unique_document_ranking(
|
|
results
|
|
)
|
|
)
|
|
|
|
expected_documents = {
|
|
str(value)
|
|
for value in question[
|
|
"expected_documents"
|
|
]
|
|
}
|
|
|
|
rank = first_relevant_rank(
|
|
ranked_documents,
|
|
expected_documents,
|
|
)
|
|
|
|
reciprocal_rank = (
|
|
0.0
|
|
if rank is None
|
|
else 1.0 / rank
|
|
)
|
|
|
|
row: dict[str, Any] = {
|
|
"id": question[
|
|
"id"
|
|
],
|
|
"split": question.get(
|
|
"split"
|
|
),
|
|
"category": question.get(
|
|
"category",
|
|
"unknown",
|
|
),
|
|
"difficulty": question.get(
|
|
"difficulty",
|
|
"unknown",
|
|
),
|
|
"question": question[
|
|
"question"
|
|
],
|
|
"mode": mode,
|
|
"expected_documents": sorted(
|
|
expected_documents
|
|
),
|
|
"first_relevant_rank": rank,
|
|
"reciprocal_rank": round(
|
|
reciprocal_rank,
|
|
6,
|
|
),
|
|
"top_documents": [
|
|
item[
|
|
"document_path"
|
|
]
|
|
for item in ranked_documents
|
|
],
|
|
"top_source_urls": [
|
|
item.get(
|
|
"source_url"
|
|
)
|
|
for item in ranked_documents
|
|
],
|
|
}
|
|
|
|
for k in EVALUATION_K_VALUES:
|
|
row[
|
|
f"hit_at_{k}"
|
|
] = (
|
|
1
|
|
if (
|
|
rank is not None
|
|
and rank <= k
|
|
)
|
|
else 0
|
|
)
|
|
|
|
row[
|
|
f"recall_at_{k}"
|
|
] = round(
|
|
recall_at_k(
|
|
ranked_documents,
|
|
expected_documents,
|
|
k,
|
|
),
|
|
6,
|
|
)
|
|
|
|
return row
|
|
|
|
|
|
def average(
|
|
values: list[
|
|
float
|
|
],
|
|
) -> float:
|
|
if not values:
|
|
return 0.0
|
|
|
|
return (
|
|
sum(values)
|
|
/ len(values)
|
|
)
|
|
|
|
|
|
def aggregate_metrics(
|
|
rows: list[
|
|
dict[str, Any]
|
|
],
|
|
) -> dict[str, Any]:
|
|
if not rows:
|
|
return {
|
|
"questions": 0,
|
|
"hit_at_1": 0.0,
|
|
"hit_at_3": 0.0,
|
|
"hit_at_5": 0.0,
|
|
"mrr": 0.0,
|
|
"recall_at_5": 0.0,
|
|
}
|
|
|
|
return {
|
|
"questions": len(
|
|
rows
|
|
),
|
|
"hit_at_1": round(
|
|
average(
|
|
[
|
|
float(
|
|
row[
|
|
"hit_at_1"
|
|
]
|
|
)
|
|
for row in rows
|
|
]
|
|
),
|
|
6,
|
|
),
|
|
"hit_at_3": round(
|
|
average(
|
|
[
|
|
float(
|
|
row[
|
|
"hit_at_3"
|
|
]
|
|
)
|
|
for row in rows
|
|
]
|
|
),
|
|
6,
|
|
),
|
|
"hit_at_5": round(
|
|
average(
|
|
[
|
|
float(
|
|
row[
|
|
"hit_at_5"
|
|
]
|
|
)
|
|
for row in rows
|
|
]
|
|
),
|
|
6,
|
|
),
|
|
"mrr": round(
|
|
average(
|
|
[
|
|
float(
|
|
row[
|
|
"reciprocal_rank"
|
|
]
|
|
)
|
|
for row in rows
|
|
]
|
|
),
|
|
6,
|
|
),
|
|
"recall_at_5": round(
|
|
average(
|
|
[
|
|
float(
|
|
row[
|
|
"recall_at_5"
|
|
]
|
|
)
|
|
for row in rows
|
|
]
|
|
),
|
|
6,
|
|
),
|
|
}
|
|
|
|
|
|
def aggregate_by_category(
|
|
rows: list[
|
|
dict[str, Any]
|
|
],
|
|
) -> dict[
|
|
str,
|
|
dict[str, Any],
|
|
]:
|
|
grouped: dict[
|
|
str,
|
|
list[dict[str, Any]],
|
|
] = defaultdict(
|
|
list
|
|
)
|
|
|
|
for row in rows:
|
|
category = str(
|
|
row.get(
|
|
"category",
|
|
"unknown",
|
|
)
|
|
)
|
|
|
|
grouped[
|
|
category
|
|
].append(
|
|
row
|
|
)
|
|
|
|
return {
|
|
category: aggregate_metrics(
|
|
category_rows
|
|
)
|
|
for (
|
|
category,
|
|
category_rows,
|
|
) in sorted(
|
|
grouped.items()
|
|
)
|
|
}
|
|
|
|
|
|
def aggregate_by_difficulty(
|
|
rows: list[
|
|
dict[str, Any]
|
|
],
|
|
) -> dict[
|
|
str,
|
|
dict[str, Any],
|
|
]:
|
|
grouped: dict[
|
|
str,
|
|
list[dict[str, Any]],
|
|
] = defaultdict(
|
|
list
|
|
)
|
|
|
|
for row in rows:
|
|
difficulty = str(
|
|
row.get(
|
|
"difficulty",
|
|
"unknown",
|
|
)
|
|
)
|
|
|
|
grouped[
|
|
difficulty
|
|
].append(
|
|
row
|
|
)
|
|
|
|
return {
|
|
difficulty: aggregate_metrics(
|
|
difficulty_rows
|
|
)
|
|
for (
|
|
difficulty,
|
|
difficulty_rows,
|
|
) in sorted(
|
|
grouped.items()
|
|
)
|
|
}
|
|
|
|
|
|
def print_summary(
|
|
summary: dict[
|
|
str,
|
|
dict[str, Any],
|
|
],
|
|
*,
|
|
split: str,
|
|
selected_count: int,
|
|
total_count: int,
|
|
) -> None:
|
|
print()
|
|
print(
|
|
"Retrieval evaluation"
|
|
)
|
|
print(
|
|
"=" * 78
|
|
)
|
|
|
|
print(
|
|
f"Split: {split}"
|
|
)
|
|
|
|
print(
|
|
f"Questions: "
|
|
f"{selected_count}/{total_count}"
|
|
)
|
|
|
|
print(
|
|
"-" * 78
|
|
)
|
|
|
|
header = (
|
|
f"{'Mode':<10}"
|
|
f"{'Questions':>10}"
|
|
f"{'Hit@1':>10}"
|
|
f"{'Hit@3':>10}"
|
|
f"{'Hit@5':>10}"
|
|
f"{'MRR':>10}"
|
|
f"{'Recall@5':>12}"
|
|
)
|
|
|
|
print(
|
|
header
|
|
)
|
|
|
|
print(
|
|
"-" * 78
|
|
)
|
|
|
|
for mode in EVALUATION_MODES:
|
|
metrics = summary[
|
|
mode
|
|
]
|
|
|
|
print(
|
|
f"{mode:<10}"
|
|
f"{metrics['questions']:>10}"
|
|
f"{metrics['hit_at_1']:>10.3f}"
|
|
f"{metrics['hit_at_3']:>10.3f}"
|
|
f"{metrics['hit_at_5']:>10.3f}"
|
|
f"{metrics['mrr']:>10.3f}"
|
|
f"{metrics['recall_at_5']:>12.3f}"
|
|
)
|
|
|
|
print(
|
|
"=" * 78
|
|
)
|
|
print()
|
|
|
|
|
|
def save_json_results(
|
|
path: Path,
|
|
payload: dict[str, Any],
|
|
) -> None:
|
|
path.parent.mkdir(
|
|
parents=True,
|
|
exist_ok=True,
|
|
)
|
|
|
|
with path.open(
|
|
"w",
|
|
encoding="utf-8",
|
|
) as file:
|
|
json.dump(
|
|
payload,
|
|
file,
|
|
ensure_ascii=False,
|
|
indent=2,
|
|
)
|
|
|
|
file.write(
|
|
"\n"
|
|
)
|
|
|
|
|
|
def save_csv_results(
|
|
path: Path,
|
|
rows: list[
|
|
dict[str, Any]
|
|
],
|
|
) -> None:
|
|
path.parent.mkdir(
|
|
parents=True,
|
|
exist_ok=True,
|
|
)
|
|
|
|
fieldnames = [
|
|
"id",
|
|
"split",
|
|
"category",
|
|
"difficulty",
|
|
"mode",
|
|
"question",
|
|
"first_relevant_rank",
|
|
"reciprocal_rank",
|
|
"hit_at_1",
|
|
"hit_at_3",
|
|
"hit_at_5",
|
|
"recall_at_1",
|
|
"recall_at_3",
|
|
"recall_at_5",
|
|
"expected_documents",
|
|
"top_documents",
|
|
"top_source_urls",
|
|
]
|
|
|
|
with path.open(
|
|
"w",
|
|
encoding="utf-8",
|
|
newline="",
|
|
) as file:
|
|
writer = csv.DictWriter(
|
|
file,
|
|
fieldnames=fieldnames,
|
|
)
|
|
|
|
writer.writeheader()
|
|
|
|
for row in rows:
|
|
csv_row = dict(
|
|
row
|
|
)
|
|
|
|
for key in (
|
|
"expected_documents",
|
|
"top_documents",
|
|
"top_source_urls",
|
|
):
|
|
csv_row[
|
|
key
|
|
] = json.dumps(
|
|
csv_row.get(
|
|
key,
|
|
[],
|
|
),
|
|
ensure_ascii=False,
|
|
)
|
|
|
|
writer.writerow(
|
|
{
|
|
key: csv_row.get(
|
|
key
|
|
)
|
|
for key in fieldnames
|
|
}
|
|
)
|
|
|
|
|
|
def parse_args() -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(
|
|
description=(
|
|
"Porovnanie FTS5, embeddingového "
|
|
"a hybridného retrievalu."
|
|
)
|
|
)
|
|
|
|
parser.add_argument(
|
|
"--questions",
|
|
type=Path,
|
|
default=(
|
|
PROJECT_ROOT
|
|
/ "evaluation"
|
|
/ "questions.json"
|
|
),
|
|
help=(
|
|
"Cesta k questions.json"
|
|
),
|
|
)
|
|
|
|
parser.add_argument(
|
|
"--db",
|
|
type=Path,
|
|
default=DB_FILE,
|
|
help=(
|
|
"Cesta k SQLite indexu"
|
|
),
|
|
)
|
|
|
|
parser.add_argument(
|
|
"--output-dir",
|
|
type=Path,
|
|
default=(
|
|
PROJECT_ROOT
|
|
/ "evaluation"
|
|
/ "results"
|
|
),
|
|
help=(
|
|
"Adresár pre výsledky"
|
|
),
|
|
)
|
|
|
|
parser.add_argument(
|
|
"--split",
|
|
choices=(
|
|
"dev",
|
|
"test",
|
|
"all",
|
|
),
|
|
default="dev",
|
|
help=(
|
|
"Časť datasetu: "
|
|
"dev = ladenie, "
|
|
"test = finálne hodnotenie, "
|
|
"all = celý dataset. "
|
|
"Predvolené je dev."
|
|
),
|
|
)
|
|
|
|
parser.add_argument(
|
|
"--limit",
|
|
type=int,
|
|
default=5,
|
|
help=(
|
|
"Počet dokumentov použitých "
|
|
"pri evaluácii. Minimum je 5."
|
|
),
|
|
)
|
|
|
|
parser.add_argument(
|
|
"--published-only",
|
|
action="store_true",
|
|
help=(
|
|
"Vyhodnocovať iba publikované dokumenty"
|
|
),
|
|
)
|
|
|
|
parser.add_argument(
|
|
"--max-per-document",
|
|
type=int,
|
|
default=1,
|
|
help=(
|
|
"Maximálny počet chunkov "
|
|
"z jedného dokumentu"
|
|
),
|
|
)
|
|
|
|
parser.add_argument(
|
|
"--strict-dataset",
|
|
action="store_true",
|
|
help=(
|
|
"Ukončiť evaluáciu chybou, "
|
|
"ak expected document nie je v indexe."
|
|
),
|
|
)
|
|
|
|
args = parser.parse_args()
|
|
|
|
if args.limit < 5:
|
|
parser.error(
|
|
"--limit musí byť aspoň 5, "
|
|
"pretože meriame Hit@5"
|
|
)
|
|
|
|
if args.max_per_document < 0:
|
|
parser.error(
|
|
"--max-per-document nesmie byť záporné"
|
|
)
|
|
|
|
return args
|
|
|
|
|
|
def main() -> None:
|
|
args = parse_args()
|
|
|
|
all_questions = load_questions(
|
|
args.questions
|
|
)
|
|
|
|
split_counts = count_splits(
|
|
all_questions
|
|
)
|
|
|
|
questions = (
|
|
filter_questions_by_split(
|
|
all_questions,
|
|
args.split,
|
|
)
|
|
)
|
|
|
|
if not questions:
|
|
raise RuntimeError(
|
|
f"Pre split '{args.split}' "
|
|
"sa nenašli žiadne otázky."
|
|
)
|
|
|
|
print()
|
|
print(
|
|
"Dataset"
|
|
)
|
|
print(
|
|
"=" * 60
|
|
)
|
|
|
|
print(
|
|
f"Total: {len(all_questions)}"
|
|
)
|
|
|
|
print(
|
|
f"Dev: {split_counts['dev']}"
|
|
)
|
|
|
|
print(
|
|
f"Test: {split_counts['test']}"
|
|
)
|
|
|
|
print(
|
|
f"Selected split: {args.split}"
|
|
)
|
|
|
|
print(
|
|
f"Selected questions: {len(questions)}"
|
|
)
|
|
|
|
print(
|
|
"=" * 60
|
|
)
|
|
print()
|
|
|
|
indexed_documents = (
|
|
load_index_document_paths(
|
|
args.db
|
|
)
|
|
)
|
|
|
|
missing_expected = (
|
|
validate_dataset(
|
|
questions,
|
|
indexed_documents,
|
|
)
|
|
)
|
|
|
|
if missing_expected:
|
|
print()
|
|
print(
|
|
"POZOR: niektoré expected_documents "
|
|
"sa nenachádzajú v indexe:"
|
|
)
|
|
|
|
for item in missing_expected:
|
|
print(
|
|
f" {item['question_id']}: "
|
|
f"{item['document_path']}"
|
|
)
|
|
|
|
print()
|
|
|
|
if args.strict_dataset:
|
|
raise RuntimeError(
|
|
"Evaluačný dataset obsahuje "
|
|
"neexistujúce expected_documents."
|
|
)
|
|
|
|
evaluation_rows: list[
|
|
dict[str, Any]
|
|
] = []
|
|
|
|
total = len(
|
|
questions
|
|
)
|
|
|
|
for index, question in enumerate(
|
|
questions,
|
|
start=1,
|
|
):
|
|
print(
|
|
f"[{index:04d}/{total:04d}] "
|
|
f"{question['id']}: "
|
|
f"{question['question']}"
|
|
)
|
|
|
|
mode_results = (
|
|
retrieve_all_modes(
|
|
args.db,
|
|
question[
|
|
"question"
|
|
],
|
|
limit=args.limit,
|
|
published_only=(
|
|
args.published_only
|
|
),
|
|
max_per_document=(
|
|
args.max_per_document
|
|
),
|
|
)
|
|
)
|
|
|
|
for mode in EVALUATION_MODES:
|
|
evaluation_rows.append(
|
|
evaluate_question_mode(
|
|
question,
|
|
mode,
|
|
mode_results[
|
|
mode
|
|
],
|
|
)
|
|
)
|
|
|
|
summary: dict[
|
|
str,
|
|
dict[str, Any],
|
|
] = {}
|
|
|
|
by_category: dict[
|
|
str,
|
|
dict[
|
|
str,
|
|
dict[str, Any],
|
|
],
|
|
] = {}
|
|
|
|
by_difficulty: dict[
|
|
str,
|
|
dict[
|
|
str,
|
|
dict[str, Any],
|
|
],
|
|
] = {}
|
|
|
|
for mode in EVALUATION_MODES:
|
|
mode_rows = [
|
|
row
|
|
for row in evaluation_rows
|
|
if row[
|
|
"mode"
|
|
] == mode
|
|
]
|
|
|
|
summary[
|
|
mode
|
|
] = aggregate_metrics(
|
|
mode_rows
|
|
)
|
|
|
|
by_category[
|
|
mode
|
|
] = aggregate_by_category(
|
|
mode_rows
|
|
)
|
|
|
|
by_difficulty[
|
|
mode
|
|
] = aggregate_by_difficulty(
|
|
mode_rows
|
|
)
|
|
|
|
payload = {
|
|
"configuration": {
|
|
"questions_file": str(
|
|
args.questions
|
|
),
|
|
"database": str(
|
|
args.db
|
|
),
|
|
"split": (
|
|
args.split
|
|
),
|
|
"dataset_question_count": len(
|
|
all_questions
|
|
),
|
|
"selected_question_count": len(
|
|
questions
|
|
),
|
|
"dev_question_count": (
|
|
split_counts[
|
|
"dev"
|
|
]
|
|
),
|
|
"test_question_count": (
|
|
split_counts[
|
|
"test"
|
|
]
|
|
),
|
|
"limit": args.limit,
|
|
"published_only": (
|
|
args.published_only
|
|
),
|
|
"max_per_document": (
|
|
args.max_per_document
|
|
),
|
|
"modes": list(
|
|
EVALUATION_MODES
|
|
),
|
|
"metrics": [
|
|
"Hit@1",
|
|
"Hit@3",
|
|
"Hit@5",
|
|
"MRR",
|
|
"Recall@5",
|
|
],
|
|
},
|
|
"dataset_validation": {
|
|
"indexed_document_count": len(
|
|
indexed_documents
|
|
),
|
|
"missing_expected_document_count": len(
|
|
missing_expected
|
|
),
|
|
"missing_expected_documents": (
|
|
missing_expected
|
|
),
|
|
},
|
|
"summary": summary,
|
|
"by_category": (
|
|
by_category
|
|
),
|
|
"by_difficulty": (
|
|
by_difficulty
|
|
),
|
|
"questions": (
|
|
evaluation_rows
|
|
),
|
|
}
|
|
|
|
filename_suffix = (
|
|
args.split
|
|
)
|
|
|
|
json_path = (
|
|
args.output_dir
|
|
/ (
|
|
"retrieval_results_"
|
|
f"{filename_suffix}.json"
|
|
)
|
|
)
|
|
|
|
csv_path = (
|
|
args.output_dir
|
|
/ (
|
|
"retrieval_results_"
|
|
f"{filename_suffix}.csv"
|
|
)
|
|
)
|
|
|
|
save_json_results(
|
|
json_path,
|
|
payload,
|
|
)
|
|
|
|
save_csv_results(
|
|
csv_path,
|
|
evaluation_rows,
|
|
)
|
|
|
|
print_summary(
|
|
summary,
|
|
split=args.split,
|
|
selected_count=len(
|
|
questions
|
|
),
|
|
total_count=len(
|
|
all_questions
|
|
),
|
|
)
|
|
|
|
print(
|
|
"Výsledky:"
|
|
)
|
|
|
|
print(
|
|
f" JSON: {json_path}"
|
|
)
|
|
|
|
print(
|
|
f" CSV: {csv_path}"
|
|
)
|
|
|
|
if missing_expected:
|
|
print()
|
|
print(
|
|
"POZOR: pred použitím metrík "
|
|
"v diplomovej práci oprav "
|
|
"missing expected documents."
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|