"""Evaluate frozen ADC answers using the actual RAGAS collections API. The reference is never added to retrieved_contexts. These contexts are the source chunks returned with each stored answer, not a reconstruction of its full prompt. """ from __future__ import annotations import argparse import asyncio import contextvars import hashlib import importlib.metadata import inspect import json import math import os import platform import sys import time from datetime import datetime, timezone from pathlib import Path os.environ.setdefault("RAGAS_DO_NOT_TRACK", "true") ROOT = Path(__file__).resolve().parents[2] sys.path.insert(0, str(ROOT)) from scripts.evaluation.ragas_data import METRICS, aggregate, checked_score, sample_from_answer, skip_reason def lines(path): return [json.loads(s) for s in path.read_text(encoding="utf-8").splitlines() if s.strip()] if path.exists() else [] def sha(path): return hashlib.sha256(path.read_bytes()).hexdigest() async def run(args): import httpx from dotenv import dotenv_values from openai import AsyncOpenAI from ragas.embeddings.base import embedding_factory from ragas.llms import llm_factory from ragas.metrics.collections import AnswerCorrectness, AnswerRelevancy, ContextRecall, Faithfulness cfg = dotenv_values(args.root/"lightrag/.env") host = os.getenv("RAGAS_LLM_BASE_URL") or cfg.get("LLM_BINDING_HOST") key = os.getenv("RAGAS_LLM_API_KEY") or cfg.get("LLM_BINDING_API_KEY") or cfg.get("OPENAI_API_KEY") model = os.getenv("RAGAS_LLM_MODEL") or cfg.get("LLM_MODEL") if not host or not key or not model: raise SystemExit("LLM endpoint, model and API key are required; configure .env or RAGAS_LLM_* variables") args.out.mkdir(parents=True, exist_ok=True) trace_path = args.out/"api_trace.jsonl" current_cell = contextvars.ContextVar("ragas_cell",default=None) async def trace(response): await response.aread() packet = {"path":response.request.url.path,"status_code":response.status_code,"metric_cell":current_cell.get()} if response.status_code < 400: try: packet["request"] = json.loads(response.request.content) packet["response"] = response.json() except (ValueError, UnicodeDecodeError): packet["non_json"] = True # Do not persist headers, credentials, or authentication error bodies. with trace_path.open("a",encoding="utf-8") as f: f.write(json.dumps(packet,ensure_ascii=False)+"\n") http_client = httpx.AsyncClient(timeout=args.timeout,event_hooks={"response":[trace]}, limits=httpx.Limits(max_connections=args.workers)) client = AsyncOpenAI(base_url=host,api_key=key,http_client=http_client,max_retries=1) embedding_client = AsyncOpenAI(base_url=args.embedding_url,api_key="local-only",timeout=args.timeout,max_retries=1) effective_max_tokens = args.retry_max_tokens or 4096 llm = llm_factory(model,client=client,temperature=0,max_tokens=effective_max_tokens, extra_body={"chat_template_kwargs":{"enable_thinking":False}}) embeddings = embedding_factory("openai",model="local-embed",client=embedding_client) scorers = {"faithfulness":Faithfulness(llm=llm), "answer_relevancy":AnswerRelevancy(llm=llm,embeddings=embeddings), "context_recall":ContextRecall(llm=llm), "answer_correctness":AnswerCorrectness(llm=llm,embeddings=embeddings)} cases_path = args.root/"evaluation/scenarios.json" answers_path = args.root/"outputs/evaluation_pilot_fixed/answers.jsonl" cases = {c["id"]:c for c in json.loads(cases_path.read_text(encoding="utf-8"))} answers = list({(r["case_id"],r["mode"]):r for r in lines(answers_path)}.values()) if args.limit: selected_ids = list(dict.fromkeys(r["case_id"] for r in answers))[:args.limit] answers = [r for r in answers if r["case_id"] in selected_ids] answers = [r for r in answers if r["mode"] in args.modes] if not answers: raise SystemExit("No matching stored answers") identity = {"ragas_version":importlib.metadata.version("ragas"),"openai_version":importlib.metadata.version("openai"), "answers_sha256":sha(answers_path),"cases_sha256":sha(cases_path), "scorer_source_sha256":{name:hashlib.sha256(inspect.getsource(type(s)).encode()).hexdigest() for name,s in scorers.items()}, "case_mode_pairs":[[r["case_id"],r["mode"]] for r in answers],"llm_model":model,"llm_endpoint":host, "temperature":0,"max_tokens":4096,"enable_thinking":False,"embedding_model":"paraphrase-multilingual-MiniLM-L12-v2", "embedding_endpoint":args.embedding_url,"metric_api":"ragas.metrics.collections","answer_correctness_weights":[0.75,0.25], "context_definition":"Exact source fragments returned with stored answer; graph descriptions and full generator prompt were not archived.", "prompt_language":"Default RAGAS prompts (English); questions, references and answers are Slovak.", "negative_cases":"Standard answerable metrics skipped; abstention retained in separate existing rubric."} manifest_path = args.out/"manifest.json" ledger = args.out/"scores.jsonl" if manifest_path.exists(): previous = json.loads(manifest_path.read_text(encoding="utf-8")) if not args.resume or previous["identity"] != identity: raise SystemExit("Output exists or evaluation identity changed; use a new output directory") previous.setdefault("resume_events",[]).append({"utc":datetime.now(timezone.utc).isoformat(), "workers":args.workers,"timeout":args.timeout, "retry_max_tokens":args.retry_max_tokens}) manifest_path.write_text(json.dumps(previous,ensure_ascii=False,indent=2),encoding="utf-8") else: manifest_path.write_text(json.dumps({"identity":identity,"started_utc":datetime.now(timezone.utc).isoformat(), "python":platform.python_version(),"workers":args.workers, "limitation":"Evaluator is the same school model service as the generator; references are source-derived and not independently clinically reviewed."}, ensure_ascii=False,indent=2),encoding="utf-8") dataset = [{"case_id":r["case_id"],"mode":r["mode"],"sample":sample_from_answer(r,cases[r["case_id"]])} for r in answers] (args.out/"samples.json").write_text(json.dumps(dataset,ensure_ascii=False,indent=2),encoding="utf-8") saved = lines(ledger) latest = {(r["case_id"],r["mode"],r["metric"]):r for r in saved} if args.retry_max_tokens and len(latest) != len(answers)*len(METRICS): raise SystemExit("A larger retry budget is only allowed after a complete first pass") completed = {k for k,r in latest.items() if r["status"] in {"ok","skipped","undefined"}} gate = asyncio.Semaphore(args.workers) async def score(row, metric): pair = (row["case_id"],row["mode"],metric) if pair in completed: return sample = sample_from_answer(row,cases[row["case_id"]]) record = {"case_id":pair[0],"mode":pair[1],"metric":metric,"llm_max_tokens":effective_max_tokens} reason = skip_reason(row,cases[row["case_id"]],metric,sample) if reason: record.update(status="skipped",reason=reason,value=None) else: async with gate: start = time.perf_counter() cell_token = current_cell.set(list(pair)) try: if metric == "faithfulness": inputs = {k:sample[k] for k in ("user_input","response","retrieved_contexts")} elif metric == "answer_relevancy": inputs = {k:sample[k] for k in ("user_input","response")} elif metric == "context_recall": inputs = {k:sample[k] for k in ("user_input","reference","retrieved_contexts")} else: inputs = {k:sample[k] for k in ("user_input","response","reference")} result = await asyncio.wait_for(scorers[metric].ascore(**inputs),timeout=args.timeout) if not math.isfinite(float(result.value)): record.update(status="undefined",value=None,reason="RAGAS returned a nonfinite value") else: record.update(status="ok",value=checked_score(metric,result.value), reason=getattr(result,"reason",None),ragas_traces=getattr(result,"traces",None)) except Exception as exc: record.update(status="error",value=None,error_type=type(exc).__name__) record["seconds"] = round(time.perf_counter()-start,3) current_cell.reset(cell_token) saved.append(record) with ledger.open("a",encoding="utf-8") as f: f.write(json.dumps(record,ensure_ascii=False,default=str)+"\n") print(f"{pair[0]} {pair[1]} {metric}: {record['status']} {record.get('value')}",flush=True) try: await asyncio.gather(*(score(row,metric) for row in answers for metric in METRICS)) finally: (args.out/"summary.json").write_text(json.dumps(aggregate(saved),indent=2),encoding="utf-8") await client.close() await embedding_client.close() def main(): p = argparse.ArgumentParser(description=__doc__) p.add_argument("--root",type=Path,default=ROOT) p.add_argument("--out",type=Path,required=True) p.add_argument("--embedding-url",default="http://localhost:8010/v1") p.add_argument("--modes",nargs="+",default=["bypass","naive","local","hybrid","mix"]) p.add_argument("--limit",type=int) p.add_argument("--workers",type=int,default=3) p.add_argument("--timeout",type=int,default=180) p.add_argument("--resume",action="store_true") p.add_argument("--retry-max-tokens",type=int,help="Larger output budget for failed cells after a complete first pass") args=p.parse_args() if args.workers<1 or args.timeout<=0 or (args.limit is not None and args.limit<1): p.error("workers, timeout and limit must be positive") if args.retry_max_tokens is not None and (not args.resume or args.retry_max_tokens<4096): p.error("retry-max-tokens requires --resume and must be at least 4096") asyncio.run(run(args)) if __name__ == "__main__": main()