jp170na/dp-zp-agent
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Compare commits

base: jp170na:855b1b106d64d97000040265fae4a395a9ad9429
Branches Tags
jp170na:master
..
compare: jp170na:297431548f99a4c2dd9ea905f6f2548598b0b7a1
Branches Tags
jp170na:master

No commits in common. "855b1b106d64d97000040265fae4a395a9ad9429" and "297431548f99a4c2dd9ea905f6f2548598b0b7a1" have entirely different histories.
855b1b106d ... 297431548f

6 changed files with 1 additions and 1519 deletions
Show Stats Expand all files Collapse all files

5
evaluation/json_files/rag_answer_overrides.json
View File

@ -1,5 +0,0 @@
{
"q0005": {
"question": "Akej téme sa venuje diplomová práca Jána Holpa?"
}
}

2
evaluation/json_files/questions.json → evaluation/questions.json
View File

@ -73,7 +73,7 @@
"split": "dev", "split": "dev",
"category": "paraphrase", "category": "paraphrase",
"difficulty": "medium", "difficulty": "medium",
"question": "O čom je práca osoby Ján Holp?", "question": "Akej téme sa venuje diplomová práca Jána Holpa?",
"expected_documents": [ "expected_documents": [
"pages/students/2016/jan_holp/README.md" "pages/students/2016/jan_holp/README.md"
], ],

0
evaluation/json_files/questions_before_ambiguity_cleanup.json → evaluation/questions_before_ambiguity_cleanup.json
View File

0
evaluation/json_files/questions_before_validation.json → evaluation/questions_before_validation.json
View File

545
evaluation/rag_metrics.py
View File

@ -1,545 +0,0 @@
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()
)
}

968
evaluation/rag_runner.py
View File

@ -1,968 +0,0 @@
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"
)
),
}