07_run_weknora_qa.py
2.9 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
from __future__ import annotations
import sys
import _bootstrap # noqa: F401
from weknora_eval.api import client_from_config
from weknora_eval.config import load_config
from weknora_eval.loaders import append_jsonl, read_jsonl, setup_logging, write_jsonl
def main() -> int:
setup_logging()
config = load_config()
client = client_from_config(config)
qa_config = config.get("qa", {})
rows = [row for row in read_jsonl("data/testsets/testset.reviewed.jsonl") if row.get("review_status") == "approved"]
answers = []
for index, row in enumerate(rows, start=1):
sample_id = row["sample_id"]
try:
session = client.create_session(title=f"ragas-eval-{sample_id}")
session_id = session.get("id")
if not session_id:
raise RuntimeError(f"create_session returned no id for {sample_id}")
result = client.knowledge_chat_sse(
session_id=session_id,
query=row["user_input"],
disable_title=bool(qa_config.get("disable_title", True)),
enable_memory=bool(qa_config.get("enable_memory", False)),
channel=str(qa_config.get("channel", "api")),
)
answer = {
"sample_id": sample_id,
"user_input": row["user_input"],
"session_id": session_id,
"request_id": result.get("request_id"),
"response": result.get("response") or "",
"retrieved_contexts": result.get("retrieved_contexts") or [],
"weknora_references": result.get("weknora_references") or [],
"error": None,
}
if not answer["response"]:
answer["error"] = "empty_response"
append_jsonl("data/runs/failed_requests.jsonl", answer)
elif not answer["retrieved_contexts"]:
append_jsonl("data/runs/failed_requests.jsonl", {**answer, "error": "empty_retrieval"})
answers.append(answer)
print(f"[{index}/{len(rows)}] {sample_id} response_chars={len(answer['response'])}")
except Exception as exc: # noqa: BLE001
failed = {
"sample_id": sample_id,
"user_input": row.get("user_input"),
"response": "",
"retrieved_contexts": [],
"weknora_references": [],
"session_id": None,
"request_id": None,
"error": str(exc),
}
answers.append(failed)
append_jsonl("data/runs/failed_requests.jsonl", failed)
print(f"[{index}/{len(rows)}] {sample_id} failed: {exc}")
write_jsonl("data/runs/weknora_answers.jsonl", answers)
failures = [row for row in answers if row.get("error") and row.get("error") != "empty_retrieval"]
return 1 if failures else 0
if __name__ == "__main__":
sys.exit(main())