231 lines
7.8 KiB
Python
Executable File
231 lines
7.8 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import csv
|
|
import json
|
|
import re
|
|
import time
|
|
from concurrent.futures import ThreadPoolExecutor, as_completed
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import requests
|
|
|
|
from src.rag.project_config import env
|
|
|
|
|
|
def atomic_json(path: Path, payload: Any) -> None:
|
|
tmp = path.with_suffix(path.suffix + ".tmp")
|
|
tmp.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
tmp.replace(path)
|
|
|
|
|
|
def extract_json(text: str) -> dict[str, Any]:
|
|
text = (text or "").strip()
|
|
try:
|
|
return json.loads(text)
|
|
except Exception:
|
|
pass
|
|
match = re.search(r"\{.*\}", text, re.S)
|
|
if match:
|
|
try:
|
|
return json.loads(match.group(0))
|
|
except Exception:
|
|
pass
|
|
return {}
|
|
|
|
|
|
def judge_one(
|
|
gold: dict[str, Any],
|
|
answer: dict[str, Any],
|
|
output_dir: Path,
|
|
api_url: str,
|
|
api_key: str,
|
|
model: str,
|
|
) -> dict[str, Any]:
|
|
qid = str(gold["id"])
|
|
output = output_dir / f"{qid}.json"
|
|
if output.exists():
|
|
try:
|
|
existing = json.loads(output.read_text(encoding="utf-8"))
|
|
if existing.get("answer_verdict"):
|
|
return {"question_id": qid, "status": "ok", "resumed": True}
|
|
except Exception:
|
|
pass
|
|
candidate = str(answer.get("deepseek_answer") or "")
|
|
prompt = f"""
|
|
Evaluate one direct-LLM answer to a plant-biology question.
|
|
The GOLD SOURCE TEXT is authoritative. The short GOLD ANSWER may be incomplete.
|
|
Do not reward verbosity. Judge only factual correctness and completeness.
|
|
|
|
Question:
|
|
{gold['question']}
|
|
|
|
Gold answer:
|
|
{gold['gold_answer']}
|
|
|
|
Gold source text:
|
|
{gold['source_text']}
|
|
|
|
Candidate answer:
|
|
{candidate}
|
|
|
|
Return one compact JSON object only:
|
|
{{"answer_verdict":"correct|partial|wrong|insufficient","usable":true,"reason":"maximum 12 words"}}
|
|
|
|
Definitions:
|
|
- correct: all facts needed by the question are correct.
|
|
- partial: some required facts are correct, but important requested facts are missing.
|
|
- wrong: substantive answer conflicts with the gold source text.
|
|
- insufficient: no substantive answer or explicitly cannot answer.
|
|
- usable is true only for correct or partial.
|
|
""".strip()
|
|
last_error = ""
|
|
for attempt in range(1, 4):
|
|
try:
|
|
response = requests.post(
|
|
api_url,
|
|
headers={
|
|
"Authorization": f"Bearer {api_key}",
|
|
"Content-Type": "application/json",
|
|
},
|
|
json={
|
|
"model": model,
|
|
"messages": [
|
|
{"role": "system", "content": "Return valid JSON only."},
|
|
{"role": "user", "content": prompt},
|
|
],
|
|
"temperature": 0,
|
|
"max_tokens": 1200,
|
|
},
|
|
timeout=180,
|
|
)
|
|
response.raise_for_status()
|
|
content = (response.json().get("choices") or [{}])[0].get(
|
|
"message", {}
|
|
).get("content", "")
|
|
parsed = extract_json(content)
|
|
verdict = parsed.get("answer_verdict")
|
|
if verdict in {"correct", "partial", "wrong", "insufficient"}:
|
|
payload = {
|
|
"question_id": qid,
|
|
"question": gold["question"],
|
|
"gold_answer": gold["gold_answer"],
|
|
"gold_source_text": gold["source_text"],
|
|
"deepseek_answer": candidate,
|
|
"answer_verdict": verdict,
|
|
"usable": bool(parsed.get("usable")),
|
|
"reason": parsed.get("reason", ""),
|
|
"attempt": attempt,
|
|
"model": model,
|
|
}
|
|
atomic_json(output, payload)
|
|
return {"question_id": qid, "status": "ok"}
|
|
last_error = "invalid verdict"
|
|
except Exception as exc:
|
|
last_error = f"{type(exc).__name__}: {exc}"
|
|
time.sleep(3 * attempt)
|
|
atomic_json(output, {"question_id": qid, "error": last_error})
|
|
return {"question_id": qid, "status": "error", "error": last_error}
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--answers", required=True)
|
|
parser.add_argument("--questions", required=True)
|
|
parser.add_argument("--output-dir", required=True)
|
|
parser.add_argument("--workers", type=int, default=4)
|
|
args = parser.parse_args()
|
|
|
|
api_key = env("DEEPSEEK_API_KEY")
|
|
if not api_key:
|
|
raise RuntimeError("No DEEPSEEK_API_KEY configured")
|
|
base_url = env("DEEPSEEK_BASE_URL", "https://api.deepseek.com").rstrip("/")
|
|
model = env("V15_RELATION_LLM_MODEL", "deepseek-v4-flash")
|
|
|
|
answers = json.loads(Path(args.answers).read_text(encoding="utf-8-sig"))
|
|
questions = json.loads(Path(args.questions).read_text(encoding="utf-8-sig"))
|
|
answer_by_id = {str(item["id"]): item for item in answers}
|
|
output_dir = Path(args.output_dir)
|
|
per_question = output_dir / "per_question"
|
|
per_question.mkdir(parents=True, exist_ok=True)
|
|
|
|
results: list[dict[str, Any]] = []
|
|
with ThreadPoolExecutor(max_workers=max(1, args.workers)) as pool:
|
|
futures = [
|
|
pool.submit(
|
|
judge_one,
|
|
gold,
|
|
answer_by_id[str(gold["id"])],
|
|
per_question,
|
|
f"{base_url}/chat/completions",
|
|
api_key,
|
|
model,
|
|
)
|
|
for gold in questions
|
|
]
|
|
for index, future in enumerate(as_completed(futures), 1):
|
|
result = future.result()
|
|
results.append(result)
|
|
print(
|
|
f"[{index}/{len(questions)}] {result['question_id']} "
|
|
f"{result['status']}",
|
|
flush=True,
|
|
)
|
|
errors = [item for item in results if item["status"] == "error"]
|
|
if errors:
|
|
raise RuntimeError(errors)
|
|
|
|
rows: list[dict[str, Any]] = []
|
|
for gold in questions:
|
|
qid = str(gold["id"])
|
|
judged = json.loads(
|
|
(per_question / f"{qid}.json").read_text(encoding="utf-8")
|
|
)
|
|
verdict = judged["answer_verdict"]
|
|
rows.append(
|
|
{
|
|
"id": qid,
|
|
"question": gold["question"],
|
|
"gold_answer": gold["gold_answer"],
|
|
"deepseek_answer": judged["deepseek_answer"],
|
|
"score": {"correct": 1.0, "partial": 0.5}.get(verdict, 0.0),
|
|
"label": verdict,
|
|
"reason": judged["reason"],
|
|
"model": "DeepSeek V4Flash direct, unified judge",
|
|
"status": "answered",
|
|
}
|
|
)
|
|
with (output_dir / "direct_same_scale_correctness.csv").open(
|
|
"w", encoding="utf-8-sig", newline=""
|
|
) as handle:
|
|
writer = csv.DictWriter(handle, fieldnames=list(rows[0]))
|
|
writer.writeheader()
|
|
writer.writerows(rows)
|
|
(output_dir / "direct_same_scale_correctness.json").write_text(
|
|
json.dumps(rows, ensure_ascii=False, indent=2),
|
|
encoding="utf-8",
|
|
)
|
|
counts: dict[str, int] = {}
|
|
for row in rows:
|
|
counts[row["label"]] = counts.get(row["label"], 0) + 1
|
|
summary = {
|
|
"model": "DeepSeek V4Flash direct, unified evidence-aware judge",
|
|
"n": len(rows),
|
|
"total_score": sum(float(row["score"]) for row in rows),
|
|
"answer_correctness": sum(float(row["score"]) for row in rows)
|
|
/ len(rows),
|
|
"label_counts": counts,
|
|
"scoring_note": (
|
|
"Same correct=1, partial=0.5, wrong/insufficient=0 rubric "
|
|
"used for the RAG ablation configurations."
|
|
),
|
|
}
|
|
atomic_json(output_dir / "direct_same_scale_correctness_summary.json", summary)
|
|
print(json.dumps(summary, ensure_ascii=False, indent=2))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|