From f0f78e8d17edefe9aad35a2e7ec4cf00af21b425 Mon Sep 17 00:00:00 2001 From: jp170na Date: Sat, 15 Aug 2026 23:48:56 +0200 Subject: [PATCH] rag_control --- evaluation/evaluate_rag_answers.py | 2426 ++++++++-------------------- 1 file changed, 712 insertions(+), 1714 deletions(-) diff --git a/evaluation/evaluate_rag_answers.py b/evaluation/evaluate_rag_answers.py index 3fcf122..7f9113b 100644 --- a/evaluation/evaluate_rag_answers.py +++ b/evaluation/evaluate_rag_answers.py @@ -3,15 +3,8 @@ from __future__ import annotations import argparse import csv import json -import os -import re -import statistics import sys import time -import unicodedata -import urllib.error -import urllib.request -from collections import defaultdict from datetime import datetime, timezone from pathlib import Path from typing import Any @@ -19,57 +12,90 @@ from typing import Any PROJECT_ROOT = Path(__file__).resolve().parents[1] -QUESTIONS_PATH = ( +if str(PROJECT_ROOT) not in sys.path: + sys.path.insert( + 0, + str(PROJECT_ROOT), + ) + + +from evaluation.rag_metrics import ( + evaluate_answer, + summarize_results, +) +from evaluation.rag_runner import ( + DEFAULT_MODEL, + DEFAULT_TIMEOUT, + LOCAL_OPENAPI_URL, + LOCAL_RAG_URL, + OPENWEBUI_URL, + build_rag_tool, + load_env_value, + request_json, + run_question, +) +from evaluation.retrieval_runner import ( + filter_questions_by_split, + load_questions, +) + + +EVALUATION_DIR = ( PROJECT_ROOT / "evaluation" +) + +JSON_FILES_DIR = ( + EVALUATION_DIR + / "json_files" +) + +QUESTIONS_PATH = ( + JSON_FILES_DIR / "questions.json" ) +RAG_ANSWER_OVERRIDES_PATH = ( + JSON_FILES_DIR + / "rag_answer_overrides.json" +) + RESULTS_DIR = ( - PROJECT_ROOT - / "evaluation" + EVALUATION_DIR / "results" ) -OPENWEBUI_URL = ( - "https://ui.tukekemt.xyz/api/chat/completions" -) -LOCAL_OPENAPI_URL = ( - "http://localhost:8000/openapi.json" -) - -LOCAL_RAG_URL = ( - "http://localhost:8000/rag" -) - -DEFAULT_MODEL = "model120-fast" - -DEFAULT_TIMEOUT = 180 - -NO_ANSWER_TEXT = ( - "V dostupných dokumentoch ZP Wiki sa túto " - "informáciu nepodarilo spoľahlivo nájsť." -) +ALLOWED_OVERRIDE_FIELDS = { + "question", + "expected_answer_contains", + "expected_source_urls", + "should_answer", + "note", +} def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser( description=( - "Answer-level RAG evaluácia cez " - "OpenWebUI + Model120 + ZP Agent." + "Answer-level RAG evaluácia " + "cez OpenWebUI, Model120 " + "a ZP Agent." ) ) parser.add_argument( "--split", - choices=[ + choices=( "dev", "test", "all", - ], + ), default="dev", - help="Dataset split. Default: dev.", + help=( + "Časť datasetu. " + "Predvolené je dev." + ), ) parser.add_argument( @@ -77,8 +103,8 @@ def parse_args() -> argparse.Namespace: type=int, default=None, help=( - "Maximálny počet otázok. " - "Vhodné na prvý kontrolný beh." + "Maximálny počet otázok " + "na vyhodnotenie." ), ) @@ -87,7 +113,7 @@ def parse_args() -> argparse.Namespace: default=DEFAULT_MODEL, help=( "OpenWebUI model ID. " - f"Default: {DEFAULT_MODEL}." + f"Predvolené: {DEFAULT_MODEL}." ), ) @@ -95,14 +121,28 @@ def parse_args() -> argparse.Namespace: "--questions", type=Path, default=QUESTIONS_PATH, - help="Cesta k questions.json.", + help=( + "Cesta k questions.json." + ), + ) + + parser.add_argument( + "--overrides", + type=Path, + default=RAG_ANSWER_OVERRIDES_PATH, + help=( + "Cesta k answer-level override " + "súboru." + ), ) parser.add_argument( "--results-dir", type=Path, default=RESULTS_DIR, - help="Adresár pre výsledky.", + help=( + "Adresár pre výsledky." + ), ) parser.add_argument( @@ -111,7 +151,7 @@ def parse_args() -> argparse.Namespace: default=DEFAULT_TIMEOUT, help=( "HTTP timeout v sekundách. " - f"Default: {DEFAULT_TIMEOUT}." + f"Predvolené: {DEFAULT_TIMEOUT}." ), ) @@ -131,1105 +171,231 @@ def parse_args() -> argparse.Namespace: default=[], help=( "Vyhodnoť iba konkrétne ID. " - "Možno použiť opakovane." + "Parameter možno použiť " + "opakovane." ), ) parser.add_argument( "--fail-fast", action="store_true", - help="Pri prvej chybe ukonči evaluáciu.", - ) - - return parser.parse_args() - - -def load_env_value( - key: str, - env_path: Path | None = None, -) -> str: - value = os.environ.get(key) - - if value: - return value - - if env_path is None: - env_path = PROJECT_ROOT / ".env" - - if not env_path.exists(): - raise RuntimeError( - f"{key} nie je v environment " - f"a {env_path} neexistuje." - ) - - for raw_line in env_path.read_text( - encoding="utf-8" - ).splitlines(): - line = raw_line.strip() - - if ( - not line - or line.startswith("#") - or "=" not in line - ): - continue - - name, value = line.split( - "=", - 1, - ) - - if name.strip() != key: - continue - - value = value.strip() - - if ( - len(value) >= 2 - and value[0] == value[-1] - and value[0] in { - "'", - '"', - } - ): - value = value[1:-1] - - if value: - return value - - raise RuntimeError( - f"{key} sa nepodarilo nájsť." - ) - - -def request_json( - url: str, - *, - method: str = "GET", - headers: dict[str, str] | None = None, - payload: dict[str, Any] | None = None, - timeout: int = DEFAULT_TIMEOUT, -) -> dict[str, Any]: - data = None - - if payload is not None: - data = json.dumps( - payload, - ensure_ascii=False, - ).encode("utf-8") - - request = urllib.request.Request( - url, - data=data, - method=method, - headers=headers or {}, - ) - - try: - with urllib.request.urlopen( - request, - timeout=timeout, - ) as response: - raw = response.read().decode( - "utf-8" - ) - - except urllib.error.HTTPError as exc: - body = exc.read().decode( - "utf-8", - errors="replace", - ) - - raise RuntimeError( - f"HTTP {exc.code} pre {url}: " - f"{body[:1500]}" - ) from exc - - except urllib.error.URLError as exc: - raise RuntimeError( - f"Sieťová chyba pre {url}: " - f"{exc}" - ) from exc - - if not raw.strip(): - raise RuntimeError( - f"Prázdna odpoveď z {url}." - ) - - try: - parsed = json.loads(raw) - - except json.JSONDecodeError as exc: - raise RuntimeError( - f"Neplatný JSON z {url}: " - f"{raw[:1000]}" - ) from exc - - if not isinstance(parsed, dict): - raise RuntimeError( - f"Očakávaný JSON objekt z {url}, " - f"dostal som {type(parsed).__name__}." - ) - - return parsed - - -def resolve_refs( - value: Any, - document: dict[str, Any], -) -> Any: - if isinstance(value, list): - return [ - resolve_refs( - item, - document, - ) - for item in value - ] - - if not isinstance(value, dict): - return value - - ref = value.get("$ref") - - if ref: - if not ref.startswith("#/"): - raise RuntimeError( - f"Nepodporovaný OpenAPI $ref: {ref}" - ) - - current: Any = document - - for part in ref[2:].split("/"): - current = current[part] - - resolved = resolve_refs( - current, - document, - ) - - extra = { - key: item - for key, item in value.items() - if key != "$ref" - } - - if ( - extra - and isinstance( - resolved, - dict, - ) - ): - resolved = { - **resolved, - **resolve_refs( - extra, - document, - ), - } - - return resolved - - return { - key: resolve_refs( - item, - document, - ) - for key, item in value.items() - } - - -def build_rag_tool( - openapi: dict[str, Any], -) -> tuple[str, dict[str, Any]]: - try: - operation = ( - openapi[ - "paths" - ][ - "/rag" - ][ - "post" - ] - ) - - except KeyError as exc: - raise RuntimeError( - "OpenAPI schéma neobsahuje POST /rag." - ) from exc - - operation_id = operation.get( - "operationId" - ) - - if not operation_id: - raise RuntimeError( - "POST /rag nemá operationId." - ) - - try: - schema = ( - operation[ - "requestBody" - ][ - "content" - ][ - "application/json" - ][ - "schema" - ] - ) - - except KeyError as exc: - raise RuntimeError( - "POST /rag nemá request JSON schema." - ) from exc - - parameters = resolve_refs( - schema, - openapi, - ) - - description = ( - operation.get("description") - or operation.get("summary") - or ( - "Vyhľadá relevantný kontext " - "v dokumentoch ZP Wiki." - ) - ) - - tool = { - "type": "function", - "function": { - "name": operation_id, - "description": description, - "parameters": parameters, - }, - } - - return operation_id, tool - - -def normalize_text( - value: str, -) -> str: - value = unicodedata.normalize( - "NFKC", - value, - ) - - value = value.casefold() - - value = " ".join( - value.split() - ) - - return value - - -MARKDOWN_URL_RE = re.compile( - r"\[[^\]]*\]\((https?://[^)]+)\)" -) - - -def normalize_url( - value: str, -) -> str: - value = value.strip() - - match = MARKDOWN_URL_RE.search( - value - ) - - if match: - value = match.group(1) - - return value.rstrip("/") - - -def extract_urls_from_object( - value: Any, -) -> list[str]: - result: list[str] = [] - - if isinstance(value, dict): - for key, item in value.items(): - if ( - key == "source_url" - and isinstance( - item, - str, - ) - ): - result.append( - normalize_url(item) - ) - - result.extend( - extract_urls_from_object( - item - ) - ) - - elif isinstance(value, list): - for item in value: - result.extend( - extract_urls_from_object( - item - ) - ) - - return list( - dict.fromkeys(result) - ) - - -def get_usage( - response: dict[str, Any], -) -> dict[str, int]: - usage = response.get( - "usage" - ) - - if not isinstance( - usage, - dict, - ): - return { - "prompt_tokens": 0, - "completion_tokens": 0, - "total_tokens": 0, - } - - return { - "prompt_tokens": int( - usage.get( - "prompt_tokens", - 0, - ) - or 0 + help=( + "Pri prvej chybe " + "ukonči evaluáciu." ), - "completion_tokens": int( - usage.get( - "completion_tokens", - 0, - ) - or 0 - ), - "total_tokens": int( - usage.get( - "total_tokens", - 0, - ) - or 0 - ), - } - - -def add_usage( - total: dict[str, int], - current: dict[str, int], -) -> None: - for key in ( - "prompt_tokens", - "completion_tokens", - "total_tokens", - ): - total[key] += current[key] - - -def get_first_message( - response: dict[str, Any], -) -> dict[str, Any]: - choices = response.get( - "choices" ) + args = parser.parse_args() + if ( - not isinstance( - choices, - list, - ) - or not choices + args.limit is not None + and args.limit <= 0 ): - raise RuntimeError( - "Model nevrátil choices." + parser.error( + "--limit musí byť > 0" ) - choice = choices[0] + if args.timeout <= 0: + parser.error( + "--timeout musí byť > 0" + ) + + if args.delay < 0: + parser.error( + "--delay nesmie byť záporné" + ) + + return args + + +def apply_question_overrides( + questions: list[ + dict[str, Any] + ], + path: Path, +) -> tuple[ + list[dict[str, Any]], + list[str], +]: + if not path.exists(): + raise FileNotFoundError( + "RAG answer override súbor " + f"neexistuje: {path}" + ) + + with path.open( + "r", + encoding="utf-8", + ) as file: + overrides = json.load( + file + ) if not isinstance( - choice, + overrides, dict, ): - raise RuntimeError( - "Neplatný choices[0]." + raise ValueError( + "rag_answer_overrides.json " + "musí obsahovať JSON objekt." ) - message = choice.get( - "message" - ) - - if not isinstance( - message, - dict, - ): - raise RuntimeError( - "Model nevrátil message." + known_ids = { + str( + item["id"] ) - - return message - - -def parse_tool_arguments( - raw_arguments: Any, -) -> dict[str, Any]: - if isinstance( - raw_arguments, - dict, - ): - return raw_arguments - - if not isinstance( - raw_arguments, - str, - ): - raise RuntimeError( - "Neplatný formát tool arguments." - ) - - try: - parsed = json.loads( - raw_arguments - ) - - except json.JSONDecodeError as exc: - raise RuntimeError( - "Model vrátil neplatné JSON " - "argumenty toolu: " - f"{raw_arguments[:1000]}" - ) from exc - - if not isinstance( - parsed, - dict, - ): - raise RuntimeError( - "Tool arguments nie sú JSON objekt." - ) - - return parsed - - -def evaluate_answer( - question: dict[str, Any], - answer: str, - *, - tool_called: bool, -) -> dict[str, Any]: - normalized_answer = normalize_text( - answer - ) - - expected_contains = question.get( - "expected_answer_contains", - [], - ) - - if not isinstance( - expected_contains, - list, - ): - expected_contains = [] - - answer_matches: list[bool] = [] - - for expected in expected_contains: - expected_text = normalize_text( - str(expected) - ) - - answer_matches.append( - expected_text - in normalized_answer - ) - - if answer_matches: - answer_contains_score = ( - sum(answer_matches) - / len(answer_matches) - ) - else: - answer_contains_score = 1.0 - - expected_urls = question.get( - "expected_source_urls", - [], - ) - - if not isinstance( - expected_urls, - list, - ): - expected_urls = [] - - normalized_expected_urls = [ - normalize_url( - str(url) - ) - for url in expected_urls - ] - - source_matches: list[bool] = [] - - for expected_url in ( - normalized_expected_urls - ): - source_matches.append( - expected_url in answer - ) - - if source_matches: - source_url_score = ( - sum(source_matches) - / len(source_matches) - ) - else: - source_url_score = 1.0 - - should_answer = bool( - question.get( - "should_answer", - True, - ) - ) - - normalized_no_answer = normalize_text( - NO_ANSWER_TEXT - ) - - returned_no_answer = ( - normalized_no_answer - in normalized_answer - ) - - if should_answer: - should_answer_ok = bool( - answer.strip() - ) and not returned_no_answer - - else: - should_answer_ok = ( - returned_no_answer - ) - - tool_score = ( - 1.0 - if tool_called - else 0.0 - ) - - should_answer_score = ( - 1.0 - if should_answer_ok - else 0.0 - ) - - overall_score = statistics.mean( - [ - answer_contains_score, - source_url_score, - should_answer_score, - tool_score, - ] - ) - - strict_pass = ( - bool(answer.strip()) - and all(answer_matches) - and all(source_matches) - and should_answer_ok - and tool_called - ) - - return { - "answer_matches": ( - answer_matches - ), - "answer_contains_score": ( - answer_contains_score - ), - "source_matches": ( - source_matches - ), - "source_url_score": ( - source_url_score - ), - "should_answer_ok": ( - should_answer_ok - ), - "should_answer_score": ( - should_answer_score - ), - "returned_no_answer": ( - returned_no_answer - ), - "tool_score": tool_score, - "overall_score": overall_score, - "strict_pass": strict_pass, + for item in questions } + unknown_ids = ( + set( + overrides.keys() + ) + - known_ids + ) -def run_question( - question: dict[str, Any], - *, - model: str, - operation_id: str, - rag_tool: dict[str, Any], - openwebui_api_key: str, - search_api_key: str, - timeout: int, -) -> dict[str, Any]: - started = time.perf_counter() + if unknown_ids: + raise ValueError( + "Override súbor obsahuje " + "neznáme question ID: " + + ", ".join( + sorted( + unknown_ids + ) + ) + ) - usage_total = { - "prompt_tokens": 0, - "completion_tokens": 0, - "total_tokens": 0, - } - - tool_calls_record: list[ + result: list[ dict[str, Any] ] = [] - rag_source_urls: list[str] = [] + applied_ids: list[ + str + ] = [] - question_text = str( - question.get( - "question", - "", - ) - ).strip() - - if not question_text: - raise RuntimeError( - "Otázka je prázdna." + for question in questions: + item = dict( + question ) - messages: list[ - dict[str, Any] - ] = [ - { - "role": "user", - "content": question_text, - } - ] - - first_started = time.perf_counter() - - first_response = request_json( - OPENWEBUI_URL, - method="POST", - headers={ - "Authorization": ( - f"Bearer " - f"{openwebui_api_key}" - ), - "Content-Type": ( - "application/json" - ), - }, - payload={ - "model": model, - "messages": messages, - "tools": [ - rag_tool, - ], - "tool_choice": "auto", - "stream": False, - }, - timeout=timeout, - ) - - first_latency = ( - time.perf_counter() - - first_started - ) - - add_usage( - usage_total, - get_usage( - first_response - ), - ) - - first_message = get_first_message( - first_response - ) - - tool_calls = ( - first_message.get( - "tool_calls" - ) - or [] - ) - - if not isinstance( - tool_calls, - list, - ): - tool_calls = [] - - tool_called = bool( - tool_calls - ) - - tool_latency_total = 0.0 - - if not tool_calls: - answer = str( - first_message.get( - "content" - ) - or "" + question_id = str( + item["id"] ) - scores = evaluate_answer( - question, - answer, - tool_called=False, + override = overrides.get( + question_id ) - total_latency = ( - time.perf_counter() - - started - ) - - return { - "answer": answer, - "tool_called": False, - "tool_call_count": 0, - "tool_calls": [], - "rag_source_urls": [], - "first_model_latency_seconds": ( - round( - first_latency, - 6, - ) - ), - "tool_latency_seconds": 0.0, - "final_model_latency_seconds": 0.0, - "total_latency_seconds": ( - round( - total_latency, - 6, - ) - ), - "usage": usage_total, - "response_model": ( - first_response.get( - "model" - ) - ), - **scores, - } - - assistant_message = { - "role": "assistant", - "content": ( - first_message.get( - "content" - ) - or "" - ), - "tool_calls": tool_calls, - } - - messages.append( - assistant_message - ) - - for tool_call in tool_calls: - if not isinstance( - tool_call, - dict, - ): - raise RuntimeError( - "Neplatný tool_call objekt." + if override is None: + result.append( + item ) - function = tool_call.get( - "function" - ) + continue if not isinstance( - function, + override, dict, ): - raise RuntimeError( - "tool_call nemá function." + raise ValueError( + f"{question_id}: " + "override musí byť " + "JSON objekt." ) - name = function.get( - "name" - ) - - if name != operation_id: - raise RuntimeError( - "Model zavolal neočakávaný " - f"tool: {name!r}" - ) - - arguments = parse_tool_arguments( - function.get( - "arguments" + unsupported_fields = ( + set( + override.keys() ) + - ALLOWED_OVERRIDE_FIELDS ) - tool_started = ( - time.perf_counter() - ) - - rag_result = request_json( - LOCAL_RAG_URL, - method="POST", - headers={ - "X-API-Key": ( - search_api_key - ), - "Content-Type": ( - "application/json" - ), - }, - payload=arguments, - timeout=timeout, - ) - - tool_latency = ( - time.perf_counter() - - tool_started - ) - - tool_latency_total += ( - tool_latency - ) - - current_urls = ( - extract_urls_from_object( - rag_result - ) - ) - - for url in current_urls: - if url not in rag_source_urls: - rag_source_urls.append( - url - ) - - tool_calls_record.append( - { - "name": name, - "arguments": arguments, - "latency_seconds": ( - round( - tool_latency, - 6, + if unsupported_fields: + raise ValueError( + f"{question_id}: " + "nepovolené override polia: " + + ", ".join( + sorted( + unsupported_fields ) - ), - "source_urls": ( - current_urls - ), - } + ) + ) + + if "question" in override: + overridden_question = ( + override[ + "question" + ] + ) + + if ( + not isinstance( + overridden_question, + str, + ) + or not overridden_question.strip() + ): + raise ValueError( + f"{question_id}: " + "override question musí byť " + "neprázdny reťazec." + ) + + item.update( + override ) - tool_call_id = tool_call.get( - "id" + result.append( + item ) - if not tool_call_id: - raise RuntimeError( - "tool_call nemá id." - ) - - messages.append( - { - "role": "tool", - "tool_call_id": ( - tool_call_id - ), - "name": name, - "content": json.dumps( - rag_result, - ensure_ascii=False, - ), - } + applied_ids.append( + question_id ) - final_started = ( - time.perf_counter() + return ( + result, + applied_ids, ) - final_response = request_json( - OPENWEBUI_URL, - method="POST", - headers={ - "Authorization": ( - f"Bearer " - f"{openwebui_api_key}" - ), - "Content-Type": ( - "application/json" - ), - }, - payload={ - "model": model, - "messages": messages, - "stream": False, - }, - timeout=timeout, - ) - final_latency = ( - time.perf_counter() - - final_started - ) - - add_usage( - usage_total, - get_usage( - final_response - ), - ) - - final_message = get_first_message( - final_response - ) - - answer = str( - final_message.get( - "content" - ) - or "" - ) - - scores = evaluate_answer( - question, - answer, - tool_called=tool_called, - ) - - total_latency = ( - time.perf_counter() - - started - ) - - return { - "answer": answer, - "tool_called": tool_called, - "tool_call_count": ( - len(tool_calls_record) - ), - "tool_calls": ( - tool_calls_record - ), - "rag_source_urls": ( - rag_source_urls - ), - "first_model_latency_seconds": ( - round( - first_latency, - 6, - ) - ), - "tool_latency_seconds": ( - round( - tool_latency_total, - 6, - ) - ), - "final_model_latency_seconds": ( - round( - final_latency, - 6, - ) - ), - "total_latency_seconds": ( - round( - total_latency, - 6, - ) - ), - "usage": usage_total, - "response_model": ( - final_response.get( - "model" - ) - ), - **scores, - } - - -def load_questions( - path: Path, +def select_questions( + questions: list[ + dict[str, Any] + ], *, split: str, question_ids: list[str], limit: int | None, -) -> list[dict[str, Any]]: - data = json.loads( - path.read_text( - encoding="utf-8" +) -> list[ + dict[str, Any] +]: + selected = ( + filter_questions_by_split( + questions, + split, ) ) - if not isinstance( - data, - list, - ): - raise RuntimeError( - "questions.json musí byť JSON pole." - ) - - questions = [ - item - for item in data - if isinstance( - item, - dict, - ) - ] - - if split != "all": - questions = [ - item - for item in questions - if item.get("split") == split - ] - if question_ids: wanted = set( question_ids ) - questions = [ + selected = [ item - for item in questions - if item.get("id") in wanted + for item in selected + if item.get( + "id" + ) in wanted ] if limit is not None: - if limit <= 0: - raise RuntimeError( - "--limit musí byť > 0." - ) - - questions = questions[ + selected = selected[ :limit ] - return questions + return selected def result_prefix( @@ -1243,307 +409,56 @@ def result_prefix( ) if question_ids: - prefix += "_selected" + return ( + prefix + + "_selected" + ) - elif limit is not None: - prefix += ( - f"_limit{limit}" + if limit is not None: + return ( + prefix + + f"_limit{limit}" ) return prefix -def safe_mean( - values: list[float], -) -> float: - if not values: - return 0.0 - - return float( - statistics.mean(values) - ) - - -def build_group_summary( - results: list[ - dict[str, Any] - ], - key: str, -) -> dict[str, Any]: - groups: dict[ - str, - list[dict[str, Any]], - ] = defaultdict(list) - - for result in results: - group_name = str( - result.get( - key, - "unknown", - ) - ) - - groups[ - group_name - ].append(result) - - summary: dict[ - str, - Any, - ] = {} - - for group_name in sorted( - groups - ): - items = groups[ - group_name - ] - - summary[ - group_name - ] = summarize_results( - items, - include_groups=False, - ) - - return summary - - -def summarize_results( - results: list[ - dict[str, Any] - ], - *, - include_groups: bool = True, -) -> dict[str, Any]: - total = len(results) - - errors = [ - item - for item in results - if item.get("error") - ] - - completed = total - len( - errors - ) - - tool_called_values = [ - 1.0 - if item.get( - "tool_called" - ) - else 0.0 - for item in results - ] - - answer_scores = [ - float( - item.get( - "answer_contains_score", - 0.0, - ) - ) - for item in results - ] - - source_scores = [ - float( - item.get( - "source_url_score", - 0.0, - ) - ) - for item in results - ] - - should_answer_scores = [ - float( - item.get( - "should_answer_score", - 0.0, - ) - ) - for item in results - ] - - overall_scores = [ - float( - item.get( - "overall_score", - 0.0, - ) - ) - for item in results - ] - - latencies = [ - float( - item.get( - "total_latency_seconds", - 0.0, - ) - ) - for item in results - if not item.get( - "error" - ) - ] - - strict_passes = sum( - 1 - for item in results - if item.get( - "strict_pass" - ) - ) - - prompt_tokens = sum( - int( - ( - item.get( - "usage" - ) - or {} - ).get( - "prompt_tokens", - 0, - ) - ) - for item in results - ) - - completion_tokens = sum( - int( - ( - item.get( - "usage" - ) - or {} - ).get( - "completion_tokens", - 0, - ) - ) - for item in results - ) - - total_tokens = sum( - int( - ( - item.get( - "usage" - ) - or {} - ).get( - "total_tokens", - 0, - ) - ) - for item in results - ) - - summary = { - "total": total, - "completed": completed, - "errors": len(errors), - "tool_call_rate": round( - safe_mean( - tool_called_values - ), - 6, - ), - "answer_contains_score": round( - safe_mean( - answer_scores - ), - 6, - ), - "source_url_score": round( - safe_mean( - source_scores - ), - 6, - ), - "should_answer_score": round( - safe_mean( - should_answer_scores - ), - 6, - ), - "overall_score": round( - safe_mean( - overall_scores - ), - 6, - ), - "strict_pass_count": ( - strict_passes - ), - "strict_pass_rate": round( - ( - strict_passes / total - if total - else 0.0 - ), - 6, - ), - "mean_latency_seconds": round( - safe_mean( - latencies - ), - 6, - ), - "prompt_tokens": ( - prompt_tokens - ), - "completion_tokens": ( - completion_tokens - ), - "total_tokens": ( - total_tokens - ), - } - - if include_groups: - summary[ - "by_category" - ] = build_group_summary( - results, - "category", - ) - - summary[ - "by_difficulty" - ] = build_group_summary( - results, - "difficulty", - ) - - return summary - - -def save_json( +def save_json_results( path: Path, payload: dict[str, Any], ) -> None: - path.write_text( - json.dumps( + 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, ) - + "\n", - encoding="utf-8", - ) + + file.write( + "\n" + ) -def save_csv( +def save_csv_results( path: Path, results: list[ dict[str, Any] ], ) -> None: + path.parent.mkdir( + parents=True, + exist_ok=True, + ) + fieldnames = [ "id", "split", @@ -1577,9 +492,9 @@ def save_csv( "w", encoding="utf-8", newline="", - ) as handle: + ) as file: writer = csv.DictWriter( - handle, + file, fieldnames=fieldnames, ) @@ -1593,236 +508,294 @@ def save_csv( or {} ) + row = { + "id": result.get( + "id" + ), + "split": result.get( + "split" + ), + "category": result.get( + "category" + ), + "difficulty": result.get( + "difficulty" + ), + "question": result.get( + "question" + ), + "requested_model": result.get( + "requested_model" + ), + "response_model": result.get( + "response_model" + ), + "tool_called": result.get( + "tool_called" + ), + "tool_call_count": result.get( + "tool_call_count" + ), + "answer_contains_score": result.get( + "answer_contains_score" + ), + "source_url_score": result.get( + "source_url_score" + ), + "should_answer_score": result.get( + "should_answer_score" + ), + "overall_score": result.get( + "overall_score" + ), + "strict_pass": result.get( + "strict_pass" + ), + "expected_answer_contains": ( + json.dumps( + result.get( + "expected_answer_contains", + [], + ), + ensure_ascii=False, + ) + ), + "expected_source_urls": ( + json.dumps( + result.get( + "expected_source_urls", + [], + ), + ensure_ascii=False, + ) + ), + "rag_source_urls": ( + json.dumps( + result.get( + "rag_source_urls", + [], + ), + ensure_ascii=False, + ) + ), + "answer": result.get( + "answer", + "", + ), + "first_model_latency_seconds": ( + result.get( + "first_model_latency_seconds" + ) + ), + "tool_latency_seconds": result.get( + "tool_latency_seconds" + ), + "final_model_latency_seconds": ( + result.get( + "final_model_latency_seconds" + ) + ), + "total_latency_seconds": result.get( + "total_latency_seconds" + ), + "prompt_tokens": usage.get( + "prompt_tokens", + 0, + ), + "completion_tokens": usage.get( + "completion_tokens", + 0, + ), + "total_tokens": usage.get( + "total_tokens", + 0, + ), + "error": result.get( + "error", + "", + ), + } + writer.writerow( - { - "id": result.get( - "id" - ), - "split": result.get( - "split" - ), - "category": result.get( - "category" - ), - "difficulty": result.get( - "difficulty" - ), - "question": result.get( - "question" - ), - "requested_model": ( - result.get( - "requested_model" - ) - ), - "response_model": ( - result.get( - "response_model" - ) - ), - "tool_called": ( - result.get( - "tool_called" - ) - ), - "tool_call_count": ( - result.get( - "tool_call_count" - ) - ), - "answer_contains_score": ( - result.get( - "answer_contains_score" - ) - ), - "source_url_score": ( - result.get( - "source_url_score" - ) - ), - "should_answer_score": ( - result.get( - "should_answer_score" - ) - ), - "overall_score": ( - result.get( - "overall_score" - ) - ), - "strict_pass": ( - result.get( - "strict_pass" - ) - ), - "expected_answer_contains": ( - json.dumps( - result.get( - "expected_answer_contains", - [], - ), - ensure_ascii=False, - ) - ), - "expected_source_urls": ( - json.dumps( - result.get( - "expected_source_urls", - [], - ), - ensure_ascii=False, - ) - ), - "rag_source_urls": ( - json.dumps( - result.get( - "rag_source_urls", - [], - ), - ensure_ascii=False, - ) - ), - "answer": result.get( - "answer", - "", - ), - "first_model_latency_seconds": ( - result.get( - "first_model_latency_seconds" - ) - ), - "tool_latency_seconds": ( - result.get( - "tool_latency_seconds" - ) - ), - "final_model_latency_seconds": ( - result.get( - "final_model_latency_seconds" - ) - ), - "total_latency_seconds": ( - result.get( - "total_latency_seconds" - ) - ), - "prompt_tokens": ( - usage.get( - "prompt_tokens", - 0, - ) - ), - "completion_tokens": ( - usage.get( - "completion_tokens", - 0, - ) - ), - "total_tokens": ( - usage.get( - "total_tokens", - 0, - ) - ), - "error": result.get( - "error", - "", - ), - } + row ) +def build_error_result( + base_result: dict[str, Any], + exc: Exception, +) -> dict[str, Any]: + return { + **base_result, + "answer": "", + "tool_called": False, + "tool_call_count": 0, + "tool_calls": [], + "rag_source_urls": [], + "first_model_latency_seconds": 0.0, + "tool_latency_seconds": 0.0, + "final_model_latency_seconds": 0.0, + "total_latency_seconds": 0.0, + "usage": { + "prompt_tokens": 0, + "completion_tokens": 0, + "total_tokens": 0, + }, + "response_model": None, + "answer_matches": [], + "answer_contains_score": 0.0, + "source_matches": [], + "source_url_score": 0.0, + "should_answer_ok": False, + "should_answer_score": 0.0, + "returned_no_answer": False, + "tool_score": 0.0, + "overall_score": 0.0, + "strict_pass": False, + "error": ( + f"{type(exc).__name__}: " + f"{exc}" + ), + } + + def print_summary( summary: dict[str, Any], ) -> None: print() print( - "==============================" - ) - print( - "RAG ANSWER EVALUATION" - ) - print( - "==============================" + "RAG answer evaluation" ) print( - "Total:", - summary["total"], + "=" * 78 ) print( - "Completed:", - summary["completed"], + f"Total: " + f"{summary['total']}" ) print( - "Errors:", - summary["errors"], + f"Completed: " + f"{summary['completed']}" ) print( - "Tool call rate:", - f"{summary['tool_call_rate']:.3f}", + f"Errors: " + f"{summary['errors']}" ) print( - "Answer contains:", - f"{summary['answer_contains_score']:.3f}", + "Tool call rate: " + f"{summary['tool_call_rate']:.3f}" ) print( - "Source URL:", - f"{summary['source_url_score']:.3f}", + "Answer contains: " + f"{summary['answer_contains_score']:.3f}" ) print( - "Should answer:", - f"{summary['should_answer_score']:.3f}", + "Source URL: " + f"{summary['source_url_score']:.3f}" ) print( - "Overall:", - f"{summary['overall_score']:.3f}", + "Should answer: " + f"{summary['should_answer_score']:.3f}" ) print( - "Strict pass:", - ( - f"{summary['strict_pass_count']}" - f"/{summary['total']} " - f"({summary['strict_pass_rate']:.3f})" - ), + "Overall: " + f"{summary['overall_score']:.3f}" ) print( - "Mean latency:", - ( - f"{summary['mean_latency_seconds']:.3f} s" - ), + "Strict pass: " + f"{summary['strict_pass_count']}" + f"/{summary['total']} " + f"({summary['strict_pass_rate']:.3f})" ) print( - "Total tokens:", - summary["total_tokens"], + "Mean latency: " + f"{summary['mean_latency_seconds']:.3f} s" ) + print( + "Total tokens: " + f"{summary['total_tokens']}" + ) + + print( + "=" * 78 + ) + + print() + def main() -> int: args = parse_args() - questions_path = ( - args.questions.resolve() - ) + try: + all_questions = ( + load_questions( + args.questions + ) + ) - if not questions_path.exists(): + ( + all_questions, + applied_override_ids, + ) = apply_question_overrides( + all_questions, + args.overrides, + ) + + except ( + FileNotFoundError, + ValueError, + json.JSONDecodeError, + ) as exc: print( - "ERROR: questions.json neexistuje:", - questions_path, + f"ERROR: {exc}", file=sys.stderr, ) + return 2 + questions = select_questions( + all_questions, + split=args.split, + question_ids=( + args.question_id + ), + limit=args.limit, + ) + + if not questions: + print( + "ERROR: Žiadne otázky " + "nezodpovedajú filtru.", + file=sys.stderr, + ) + + return 2 + + selected_ids = { + str( + question["id"] + ) + for question in questions + } + + selected_override_ids = [ + question_id + for question_id in applied_override_ids + if question_id in selected_ids + ] + try: openwebui_api_key = ( load_env_value( @@ -1841,66 +814,9 @@ def main() -> int: f"ERROR: {exc}", file=sys.stderr, ) + return 2 - try: - questions = load_questions( - questions_path, - split=args.split, - question_ids=( - args.question_id - ), - limit=args.limit, - ) - - except ( - RuntimeError, - json.JSONDecodeError, - ) as exc: - print( - f"ERROR: {exc}", - file=sys.stderr, - ) - return 2 - - if not questions: - print( - "ERROR: Žiadne otázky " - "nezodpovedajú filtru.", - file=sys.stderr, - ) - return 2 - - args.results_dir.mkdir( - parents=True, - exist_ok=True, - ) - - prefix = result_prefix( - split=args.split, - limit=args.limit, - question_ids=args.question_id, - ) - - json_path = ( - args.results_dir - / f"{prefix}.json" - ) - - csv_path = ( - args.results_dir - / f"{prefix}.csv" - ) - - partial_path = ( - args.results_dir - / f"{prefix}.partial.json" - ) - - print( - "Načítavam /openapi.json..." - ) - try: openapi = request_json( LOCAL_OPENAPI_URL, @@ -1918,26 +834,86 @@ def main() -> int: f"ERROR: {exc}", file=sys.stderr, ) + return 2 + prefix = result_prefix( + split=args.split, + limit=args.limit, + question_ids=( + args.question_id + ), + ) + + json_path = ( + args.results_dir + / f"{prefix}.json" + ) + + csv_path = ( + args.results_dir + / f"{prefix}.csv" + ) + + partial_path = ( + args.results_dir + / f"{prefix}.partial.json" + ) + + print() print( - "Tool:", - operation_id, + "RAG answer evaluation" ) print( - "Model:", - args.model, + "=" * 60 ) print( - "Split:", - args.split, + f"Dataset: " + f"{args.questions}" ) print( - "Otázky:", - len(questions), + f"Overrides: " + f"{args.overrides}" + ) + + print( + "Applied overrides: " + f"{len(selected_override_ids)}" + ) + + if selected_override_ids: + print( + "Override IDs: " + + ", ".join( + selected_override_ids + ) + ) + + print( + f"Model: " + f"{args.model}" + ) + + print( + f"Tool: " + f"{operation_id}" + ) + + print( + f"Split: " + f"{args.split}" + ) + + print( + f"Questions: " + f"{len(questions)}" + ) + + print( + "=" * 60 ) print() @@ -1946,30 +922,30 @@ def main() -> int: dict[str, Any] ] = [] - total_questions = len( + total = len( questions ) + stopped_early = False + for index, question in enumerate( questions, start=1, ): question_id = str( - question.get( - "id", - f"unknown-{index}", - ) + question[ + "id" + ] ) question_text = str( - question.get( - "question", - "", - ) + question[ + "question" + ] ) print( - f"[{index}/{total_questions}] " + f"[{index:04d}/{total:04d}] " f"{question_id}: " f"{question_text}" ) @@ -1983,38 +959,36 @@ def main() -> int: "split" ), "category": question.get( - "category" + "category", + "unknown", ), "difficulty": question.get( - "difficulty" + "difficulty", + "unknown", ), "question": question_text, + "override_applied": ( + question_id + in selected_override_ids + ), "requested_model": ( args.model ), - "expected_documents": ( - question.get( - "expected_documents", - [], - ) + "expected_documents": question.get( + "expected_documents", + [], ), - "expected_source_urls": ( - question.get( - "expected_source_urls", - [], - ) + "expected_source_urls": question.get( + "expected_source_urls", + [], ), - "expected_answer_contains": ( - question.get( - "expected_answer_contains", - [], - ) + "expected_answer_contains": question.get( + "expected_answer_contains", + [], ), - "should_answer": ( - question.get( - "should_answer", - True, - ) + "should_answer": question.get( + "should_answer", + True, ), "note": question.get( "note" @@ -2022,8 +996,8 @@ def main() -> int: } try: - current = run_question( - question, + run_result = run_question( + question_text, model=args.model, operation_id=( operation_id @@ -2038,9 +1012,22 @@ def main() -> int: timeout=args.timeout, ) + scores = evaluate_answer( + question, + run_result[ + "answer" + ], + tool_called=bool( + run_result[ + "tool_called" + ] + ), + ) + result = { **base_result, - **current, + **run_result, + **scores, "error": None, } @@ -2050,60 +1037,24 @@ def main() -> int: "tool_called" ], "| answer:", - ( - f"{result['answer_contains_score']:.2f}" - ), + f"{result['answer_contains_score']:.2f}", "| source:", - ( - f"{result['source_url_score']:.2f}" - ), + f"{result['source_url_score']:.2f}", "| strict:", result[ "strict_pass" ], "| latency:", - ( - f"{result['total_latency_seconds']:.2f}s" - ), + f"{result['total_latency_seconds']:.2f}s", ) except Exception as exc: - result = { - **base_result, - "answer": "", - "tool_called": False, - "tool_call_count": 0, - "tool_calls": [], - "rag_source_urls": [], - "first_model_latency_seconds": ( - 0.0 - ), - "tool_latency_seconds": 0.0, - "final_model_latency_seconds": ( - 0.0 - ), - "total_latency_seconds": 0.0, - "usage": { - "prompt_tokens": 0, - "completion_tokens": 0, - "total_tokens": 0, - }, - "response_model": None, - "answer_matches": [], - "answer_contains_score": 0.0, - "source_matches": [], - "source_url_score": 0.0, - "should_answer_ok": False, - "should_answer_score": 0.0, - "returned_no_answer": False, - "tool_score": 0.0, - "overall_score": 0.0, - "strict_pass": False, - "error": ( - f"{type(exc).__name__}: " - f"{exc}" - ), - } + result = ( + build_error_result( + base_result, + exc, + ) + ) print( " ERROR:", @@ -2113,11 +1064,7 @@ def main() -> int: ) if args.fail_fast: - results.append( - result - ) - - break + stopped_early = True results.append( result @@ -2130,101 +1077,152 @@ def main() -> int: ).isoformat() ), "status": "partial", - "split": args.split, + "split": ( + args.split + ), "requested_model": ( args.model ), - "questions_path": ( - str( - questions_path - ) + "questions_file": str( + args.questions ), - "completed_so_far": ( - len(results) + "overrides_file": str( + args.overrides + ), + "applied_override_ids": ( + selected_override_ids + ), + "completed_so_far": len( + results ), "expected_total": ( - total_questions + total ), "summary": ( summarize_results( results ) ), - "results": results, + "results": ( + results + ), } - save_json( + save_json_results( partial_path, partial_payload, ) + if stopped_early: + break + if ( args.delay > 0 - and index < total_questions + and index < total ): time.sleep( args.delay ) - summary = summarize_results( - results + summary = ( + summarize_results( + results + ) ) - final_payload = { + completed_all = ( + len(results) + == total + ) + + payload = { "generated_at": ( datetime.now( timezone.utc ).isoformat() ), - "status": "complete", - "split": args.split, - "requested_model": ( - args.model + "status": ( + "complete" + if completed_all + else "partial" ), - "questions_path": ( - str( - questions_path - ) + "configuration": { + "questions_file": str( + args.questions + ), + "overrides_file": str( + args.overrides + ), + "applied_override_ids": ( + selected_override_ids + ), + "split": ( + args.split + ), + "selected_question_count": ( + total + ), + "requested_model": ( + args.model + ), + "openwebui_url": ( + OPENWEBUI_URL + ), + "rag_url": ( + LOCAL_RAG_URL + ), + "timeout": ( + args.timeout + ), + "delay": ( + args.delay + ), + }, + "summary": ( + summary ), - "openwebui_url": ( - OPENWEBUI_URL + "results": ( + results ), - "rag_url": ( - LOCAL_RAG_URL - ), - "summary": summary, - "results": results, } - save_json( + save_json_results( json_path, - final_payload, + payload, ) - save_csv( + save_csv_results( csv_path, results, ) - if partial_path.exists(): + if ( + partial_path.exists() + and completed_all + ): partial_path.unlink() print_summary( summary ) - print() print( - "JSON:", - json_path, + "Výsledky:" ) print( - "CSV:", - csv_path, + f" JSON: " + f"{json_path}" ) - if summary["errors"]: + print( + f" CSV: " + f"{csv_path}" + ) + + if summary[ + "errors" + ]: return 1 return 0