188 lines
10 KiB
Python
188 lines
10 KiB
Python
"""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()
|