diff --git a/evaluation/rag_metrics.py b/evaluation/rag_metrics.py new file mode 100644 index 0000000..4db6c7d --- /dev/null +++ b/evaluation/rag_metrics.py @@ -0,0 +1,545 @@ +from __future__ import annotations + +import re +import statistics +import unicodedata +from collections import defaultdict +from typing import Any + + +NO_ANSWER_TEXT = ( + "V dostupných dokumentoch ZP Wiki sa túto " + "informáciu nepodarilo spoľahlivo nájsť." +) + + +MARKDOWN_URL_RE = re.compile( + r"\[[^\]]*\]\((https?://[^)]+)\)" +) + + +def normalize_text( + value: str, +) -> str: + value = unicodedata.normalize( + "NFKC", + value, + ) + + value = value.casefold() + + return " ".join( + value.split() + ) + + +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 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 + ), + } + + +def safe_mean( + values: list[float], +) -> float: + if not values: + return 0.0 + + return float( + statistics.mean( + values + ) + ) + + +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: dict[ + str, + Any, + ] = { + "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" + ] = aggregate_by_field( + results, + "category", + ) + + summary[ + "by_difficulty" + ] = aggregate_by_field( + results, + "difficulty", + ) + + return summary + + +def aggregate_by_field( + results: list[ + dict[str, Any] + ], + field: str, +) -> dict[ + str, + dict[str, Any], +]: + grouped: dict[ + str, + list[dict[str, Any]], + ] = defaultdict( + list + ) + + for result in results: + group_name = str( + result.get( + field, + "unknown", + ) + ) + + grouped[ + group_name + ].append( + result + ) + + return { + group_name: summarize_results( + group_rows, + include_groups=False, + ) + for ( + group_name, + group_rows, + ) in sorted( + grouped.items() + ) + } diff --git a/evaluation/rag_runner.py b/evaluation/rag_runner.py new file mode 100644 index 0000000..29c389d --- /dev/null +++ b/evaluation/rag_runner.py @@ -0,0 +1,968 @@ +from __future__ import annotations + +import json +import os +import re +import time +import urllib.error +import urllib.request +from pathlib import Path +from typing import Any + + +PROJECT_ROOT = Path( + __file__ +).resolve().parents[1] + + +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 + + +MARKDOWN_URL_RE = re.compile( + r"\[[^\]]*\]\((https?://[^)]+)\)" +) + + +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( + "Očakávaný JSON objekt " + f"z {url}, dostal som " + f"{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( + "Nepodporovaný " + f"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_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 + ), + "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" + ) + + if ( + not isinstance( + choices, + list, + ) + or not choices + ): + raise RuntimeError( + "Model nevrátil choices." + ) + + choice = choices[ + 0 + ] + + if not isinstance( + choice, + dict, + ): + raise RuntimeError( + "Neplatný choices[0]." + ) + + message = choice.get( + "message" + ) + + if not isinstance( + message, + dict, + ): + raise RuntimeError( + "Model nevrátil message." + ) + + 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 run_question( + question: str, + *, + model: str, + operation_id: str, + rag_tool: dict[str, Any], + openwebui_api_key: str, + search_api_key: str, + timeout: int, +) -> dict[str, Any]: + question = question.strip() + + if not question: + raise RuntimeError( + "Otázka je prázdna." + ) + + started = ( + time.perf_counter() + ) + + usage_total = { + "prompt_tokens": 0, + "completion_tokens": 0, + "total_tokens": 0, + } + + tool_calls_record: list[ + dict[str, Any] + ] = [] + + rag_source_urls: list[ + str + ] = [] + + messages: list[ + dict[str, Any] + ] = [ + { + "role": "user", + "content": question, + } + ] + + first_started = ( + time.perf_counter() + ) + + first_response = request_json( + OPENWEBUI_URL, + method="POST", + headers={ + "Authorization": ( + "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 = [] + + if not tool_calls: + answer = str( + first_message.get( + "content" + ) + or "" + ) + + 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" + ) + ), + } + + messages.append( + { + "role": "assistant", + "content": ( + first_message.get( + "content" + ) + or "" + ), + "tool_calls": ( + tool_calls + ), + } + ) + + tool_latency_total = 0.0 + + for tool_call in tool_calls: + if not isinstance( + tool_call, + dict, + ): + raise RuntimeError( + "Neplatný tool_call objekt." + ) + + function = tool_call.get( + "function" + ) + + if not isinstance( + function, + dict, + ): + raise RuntimeError( + "tool_call nemá function." + ) + + name = function.get( + "name" + ) + + if name != operation_id: + raise RuntimeError( + "Model zavolal " + "neočakávaný tool: " + f"{name!r}" + ) + + arguments = ( + parse_tool_arguments( + function.get( + "arguments" + ) + ) + ) + + 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, + ) + ), + "source_urls": ( + current_urls + ), + } + ) + + tool_call_id = ( + tool_call.get( + "id" + ) + ) + + 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, + ), + } + ) + + final_started = ( + time.perf_counter() + ) + + final_response = request_json( + OPENWEBUI_URL, + method="POST", + headers={ + "Authorization": ( + "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 "" + ) + + total_latency = ( + time.perf_counter() + - started + ) + + return { + "answer": answer, + "tool_called": True, + "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" + ) + ), + }