| import argparse |
| import json |
| import os |
| import re |
| import traceback |
| import urllib.error |
| import urllib.request |
| from datetime import datetime |
| from typing import Any, Dict, List, Tuple |
|
|
| import dspy |
| from openai import OpenAI |
| from tqdm import tqdm |
|
|
|
|
| DEFAULT_CLASSIFIER_API_BASE = "http://172.16.34.22:8040/v1" |
| DEFAULT_SUPPORT_API_BASE = "http://172.16.34.22:3090/v1" |
| DEFAULT_MODEL_PATH = ( |
| "/home/mshahidul/readctrl/code/text_classifier/" |
| "dspy_model/vllm-Meta-Llama-3.1-8B-Instruct_teacher-gpt5_v1/model.json" |
| ) |
| DEFAULT_INPUT_FILE = ( |
| "/home/mshahidul/readctrl/code/RL_model/inference_data/" |
| "RL_model_inference_v1.jsonl" |
| ) |
| DEFAULT_REFERENCE_SUBCLAIMS_FILE = ( |
| "/home/mshahidul/readctrl/code/text_classifier/data/" |
| "verified_combined_0-80_clean200_with_subclaims.json" |
| ) |
| DEFAULT_OUTPUT_DIR = "/home/mshahidul/readctrl/code/rl_inference/test_result_v2" |
|
|
| CHAT_TEMPLATE = ( |
| "<|begin_of_text|><|start_header_id|>system<|end_header_id|>\n\n" |
| "Cutting Knowledge Date: December 2023\n" |
| "Today Date: 26 July 2024\n\n" |
| "<|eot_id|><|start_header_id|>user<|end_header_id|>\n\n" |
| "{user_prompt}" |
| "<|eot_id|><|start_header_id|>assistant<|end_header_id|>\n\n" |
| ) |
|
|
| VALID_LABELS = { |
| "low_health_literacy", |
| "intermediate_health_literacy", |
| "proficient_health_literacy", |
| } |
|
|
|
|
| class HealthLiteracySignature(dspy.Signature): |
| generated_text = dspy.InputField( |
| desc="A version of the source text rewritten for a specific audience." |
| ) |
| literacy_label = dspy.OutputField( |
| desc=( |
| "Classification: low_health_literacy (simple words, no jargon), " |
| "intermediate_health_literacy (moderate technicality), or " |
| "proficient_health_literacy (highly technical/original level)." |
| ) |
| ) |
|
|
|
|
| class HealthLiteracyClassifier(dspy.Module): |
| def __init__(self): |
| super().__init__() |
| self.classifier = dspy.ChainOfThought(HealthLiteracySignature) |
|
|
| def forward(self, generated_text): |
| return self.classifier(generated_text=generated_text) |
|
|
|
|
| class MedicalClaimVerifier: |
| def __init__(self, base_url: str, model_name: str): |
| self.model_name = model_name |
| self.base_url = base_url |
| self.client = OpenAI(api_key="EMPTY", base_url=self.base_url) |
| self.cov_iqr_ranges = { |
| "low": (0.1765, 0.3226), |
| "intermediate": (0.1818, 0.4091), |
| "proficient": (0.7725, 0.9347), |
| } |
|
|
| def build_user_prompt(self, text: str, subclaims: List[str]) -> str: |
| numbered_subclaims = "\n".join( |
| f"{idx + 1}. {subclaim}" for idx, subclaim in enumerate(subclaims) |
| ) |
| return ( |
| "You are a medical evidence checker.\n" |
| "Given a medical passage and a list of subclaims, return labels for each " |
| "subclaim in the same order.\n\n" |
| "Allowed labels: supported, not_supported.\n" |
| "Output format: a JSON array of strings only.\n\n" |
| f"Medical text:\n{text}\n\n" |
| f"Subclaims:\n{numbered_subclaims}" |
| ) |
|
|
| def render_chat_prompt(self, user_prompt: str) -> str: |
| return CHAT_TEMPLATE.format(user_prompt=user_prompt) |
|
|
| def extract_label_list(self, text: str) -> List[str]: |
| cleaned = text.strip() |
| try: |
| parsed = json.loads(cleaned) |
| if isinstance(parsed, list): |
| return parsed |
| except json.JSONDecodeError: |
| pass |
|
|
| match = re.search(r"\[[\s\S]*\]", cleaned) |
| if match: |
| try: |
| parsed = json.loads(match.group(0)) |
| if isinstance(parsed, list): |
| return parsed |
| except json.JSONDecodeError: |
| return [] |
| return [] |
|
|
| def check_support_api(self, context: str, subclaims: List[str]) -> List[str]: |
| if not context or not subclaims: |
| return [] |
|
|
| user_prompt = self.build_user_prompt(context, subclaims) |
| prompt = self.render_chat_prompt(user_prompt) |
| try: |
| response = self.client.completions.create( |
| model=self.model_name, |
| prompt=prompt, |
| max_tokens=256, |
| temperature=0, |
| ) |
| pred_text = response.choices[0].text.strip() |
| labels = self.extract_label_list(pred_text) |
| return [str(x).strip().lower() for x in labels] |
| except Exception: |
| return [] |
|
|
| @staticmethod |
| def average_supported(labels: List[str], expected_len: int) -> float: |
| if expected_len <= 0: |
| return 0.0 |
| normalized = [str(x).strip().lower() for x in labels] |
| if len(normalized) < expected_len: |
| normalized.extend(["invalid"] * (expected_len - len(normalized))) |
| elif len(normalized) > expected_len: |
| normalized = normalized[:expected_len] |
| supported_count = sum(1 for item in normalized if item == "supported") |
| return supported_count / expected_len |
|
|
| def evaluate_level( |
| self, gen_text: str, gold_subs: List[str], full_subs: List[str] |
| ) -> Tuple[float, float]: |
| if not gen_text or not gold_subs or not full_subs: |
| return 0.0, 0.0 |
| comp_labels = self.check_support_api(gen_text, gold_subs) |
| cov_labels = self.check_support_api(gen_text, full_subs) |
| comp_score = self.average_supported(comp_labels, len(gold_subs)) |
| cov_score = self.average_supported(cov_labels, len(full_subs)) |
| return comp_score, cov_score |
|
|
|
|
| def parse_args() -> argparse.Namespace: |
| parser = argparse.ArgumentParser( |
| description=( |
| "Evaluate classifier accuracy plus subclaim support thresholds " |
| "(completeness + coverage)." |
| ) |
| ) |
| parser.add_argument("--model-path", default=DEFAULT_MODEL_PATH) |
| parser.add_argument( |
| "--input-file", |
| default=DEFAULT_INPUT_FILE, |
| help="Path to RL inference JSONL (e.g. RL_model_inference_v1.jsonl).", |
| ) |
| parser.add_argument( |
| "--reference-subclaims-file", |
| default=DEFAULT_REFERENCE_SUBCLAIMS_FILE, |
| help=( |
| "JSON list file that contains summary_subclaims/fulltext_subclaims " |
| "(used for lookup by doc_id + label)." |
| ), |
| ) |
| parser.add_argument( |
| "--classifier-api-base", |
| default=os.environ.get("VLLM_API_BASE", DEFAULT_CLASSIFIER_API_BASE), |
| ) |
| parser.add_argument( |
| "--support-api-base", |
| default=os.environ.get("SUPPORT_API_BASE", DEFAULT_SUPPORT_API_BASE), |
| ) |
| parser.add_argument( |
| "--support-model", |
| default=os.environ.get("VLLM_MODEL", "sc"), |
| ) |
| parser.add_argument("--output-dir", default=DEFAULT_OUTPUT_DIR) |
| parser.add_argument( |
| "--generated-text-key", |
| default="generated_text", |
| help="Field name to evaluate text from input JSONL.", |
| ) |
| parser.add_argument( |
| "--comp-min-threshold", |
| type=float, |
| default=0.9, |
| help="Completeness pass lower bound (inclusive).", |
| ) |
| parser.add_argument( |
| "--comp-max-threshold", |
| type=float, |
| default=1.0, |
| help="Completeness pass upper bound (inclusive).", |
| ) |
| parser.add_argument( |
| "--max-samples", |
| type=int, |
| default=-1, |
| help="Use -1 for all rows.", |
| ) |
| parser.add_argument( |
| "--provide-traceback", |
| action="store_true", |
| help="Print full traceback if runtime error happens.", |
| ) |
| return parser.parse_args() |
|
|
|
|
| def check_api_base(api_base: str) -> None: |
| models_url = api_base.rstrip("/") + "/models" |
| req = urllib.request.Request(models_url, method="GET") |
| try: |
| with urllib.request.urlopen(req, timeout=5) as resp: |
| if resp.status >= 400: |
| raise RuntimeError( |
| f"Endpoint reachable but unhealthy: {models_url} (status={resp.status})" |
| ) |
| except urllib.error.URLError as exc: |
| raise ConnectionError( |
| "Cannot reach OpenAI-compatible endpoint. " |
| f"api_base={api_base}. " |
| "Start your vLLM server or pass correct api base." |
| ) from exc |
|
|
|
|
| def load_compiled_classifier(path: str): |
| if hasattr(dspy, "load"): |
| try: |
| return dspy.load(path) |
| except Exception: |
| pass |
| classifier = HealthLiteracyClassifier() |
| try: |
| classifier.load(path) |
| except Exception as exc: |
| raise RuntimeError(f"Failed to load compiled model from {path}") from exc |
| return classifier |
|
|
|
|
| def normalize_pred_label(pred_obj: Any) -> str: |
| if not pred_obj or not hasattr(pred_obj, "literacy_label"): |
| return "" |
| return str(pred_obj.literacy_label).strip().lower() |
|
|
|
|
| def load_items(path: str) -> List[Dict[str, Any]]: |
| items: List[Dict[str, Any]] = [] |
| with open(path, "r", encoding="utf-8") as f: |
| for line_no, line in enumerate(f, start=1): |
| if not line.strip(): |
| continue |
| row = json.loads(line) |
| items.append( |
| { |
| "line_no": line_no, |
| "row_index": row.get("row_index"), |
| "doc_id": row.get("doc_id"), |
| "gold_label": str(row.get("gold_label", "")).strip(), |
| "generated_text": str(row.get("generated_text", "")).strip(), |
| } |
| ) |
| return items |
|
|
|
|
| def load_subclaim_lookup( |
| reference_path: str, |
| ) -> Dict[Tuple[Any, str], Tuple[List[str], List[str]]]: |
| with open(reference_path, "r", encoding="utf-8") as f: |
| rows = json.load(f) |
| if not isinstance(rows, list): |
| raise ValueError("Reference subclaims file must be a JSON list.") |
|
|
| lookup: Dict[Tuple[Any, str], Tuple[List[str], List[str]]] = {} |
| for row in rows: |
| doc_id = row.get("doc_id") |
| label = str(row.get("label", "")).strip() |
| gold_subs = row.get("summary_subclaims", []) |
| full_subs = row.get("fulltext_subclaims", []) |
| if label not in VALID_LABELS: |
| continue |
| if not isinstance(gold_subs, list) or not isinstance(full_subs, list): |
| continue |
| if not gold_subs or not full_subs: |
| continue |
| key = (doc_id, label) |
| if key not in lookup: |
| lookup[key] = (gold_subs, full_subs) |
| return lookup |
|
|
|
|
| def to_level_key(label: str) -> str: |
| mapping = { |
| "low_health_literacy": "low", |
| "intermediate_health_literacy": "intermediate", |
| "proficient_health_literacy": "proficient", |
| } |
| return mapping.get(label, "") |
|
|
|
|
| def in_range(value: float, lower: float, upper: float) -> bool: |
| return lower <= value <= upper |
|
|
|
|
| def main() -> None: |
| args = parse_args() |
| if not os.path.exists(args.model_path): |
| raise FileNotFoundError(f"Model file not found: {args.model_path}") |
| if not os.path.exists(args.input_file): |
| raise FileNotFoundError(f"Input file not found: {args.input_file}") |
| if not os.path.exists(args.reference_subclaims_file): |
| raise FileNotFoundError( |
| f"Reference subclaims file not found: {args.reference_subclaims_file}" |
| ) |
|
|
| try: |
| check_api_base(args.classifier_api_base) |
| check_api_base(args.support_api_base) |
|
|
| lm = dspy.LM( |
| model="openai/dspy", |
| api_base=args.classifier_api_base, |
| api_key="EMPTY", |
| temperature=0.0, |
| ) |
| dspy.configure(lm=lm) |
| classifier = load_compiled_classifier(args.model_path) |
| verifier = MedicalClaimVerifier( |
| base_url=args.support_api_base, |
| model_name=args.support_model, |
| ) |
| subclaim_lookup = load_subclaim_lookup(args.reference_subclaims_file) |
|
|
| rows = load_items(args.input_file) |
| if args.max_samples > 0: |
| rows = rows[: args.max_samples] |
|
|
| unmatched_rows = 0 |
| total = 0 |
| classifier_correct = 0 |
| comp_pass_count = 0 |
| cov_pass_count = 0 |
| cls_and_comp_pass_count = 0 |
| cls_comp_cov_pass_count = 0 |
| details: List[Dict[str, Any]] = [] |
|
|
| for idx, row in enumerate(tqdm(rows, desc="Evaluating"), start=1): |
| gold_label = str(row.get("gold_label", "")).strip() |
| if gold_label not in VALID_LABELS: |
| continue |
|
|
| generated_text = str(row.get(args.generated_text_key, "")).strip() |
| subclaims = subclaim_lookup.get((row.get("doc_id"), gold_label)) |
| if not generated_text or not subclaims: |
| if not subclaims: |
| unmatched_rows += 1 |
| continue |
| gold_subs, full_subs = subclaims |
|
|
| total += 1 |
| pred = classifier(generated_text=generated_text) |
| pred_label = normalize_pred_label(pred) |
| is_cls_correct = gold_label in pred_label |
| classifier_correct += int(is_cls_correct) |
|
|
| comp_score, cov_score = verifier.evaluate_level( |
| gen_text=generated_text, |
| gold_subs=gold_subs, |
| full_subs=full_subs, |
| ) |
|
|
| comp_pass = in_range( |
| comp_score, args.comp_min_threshold, args.comp_max_threshold |
| ) |
| comp_pass_count += int(comp_pass) |
|
|
| level_key = to_level_key(gold_label) |
| cov_low, cov_high = verifier.cov_iqr_ranges[level_key] |
| cov_pass = in_range(cov_score, cov_low, cov_high) |
| cov_pass_count += int(cov_pass) |
|
|
| cls_and_comp_pass = is_cls_correct and comp_pass |
| cls_comp_cov_pass = cls_and_comp_pass and cov_pass |
| cls_and_comp_pass_count += int(cls_and_comp_pass) |
| cls_comp_cov_pass_count += int(cls_comp_cov_pass) |
|
|
| details.append( |
| { |
| "idx": idx, |
| "line_no": row.get("line_no"), |
| "row_index": row.get("row_index"), |
| "doc_id": row.get("doc_id"), |
| "gold_label": gold_label, |
| "pred_label": pred_label, |
| "classifier_correct": is_cls_correct, |
| "completeness_score": comp_score, |
| "coverage_score": cov_score, |
| "completeness_threshold": [ |
| args.comp_min_threshold, |
| args.comp_max_threshold, |
| ], |
| "completeness_pass": comp_pass, |
| "coverage_iqr_threshold": [cov_low, cov_high], |
| "coverage_pass": cov_pass, |
| "pass_cls_and_completeness": cls_and_comp_pass, |
| "pass_cls_comp_cov": cls_comp_cov_pass, |
| } |
| ) |
|
|
| if total == 0: |
| raise RuntimeError("No valid rows were found for evaluation.") |
|
|
| def safe_rate(n: int) -> float: |
| return n / total if total else 0.0 |
|
|
| os.makedirs(args.output_dir, exist_ok=True) |
| ts = datetime.now().strftime("%Y%m%d_%H%M%S") |
| summary_path = os.path.join( |
| args.output_dir, f"classifier_subclaim_threshold_eval_{ts}.json" |
| ) |
| details_path = os.path.join( |
| args.output_dir, f"classifier_subclaim_threshold_eval_{ts}.jsonl" |
| ) |
|
|
| summary = { |
| "model_path": args.model_path, |
| "input_file": args.input_file, |
| "reference_subclaims_file": args.reference_subclaims_file, |
| "generated_text_key": args.generated_text_key, |
| "classifier_api_base": args.classifier_api_base, |
| "support_api_base": args.support_api_base, |
| "support_model": args.support_model, |
| "total_samples": total, |
| "unmatched_rows": unmatched_rows, |
| "classifier_only_accuracy": safe_rate(classifier_correct), |
| "completeness_pass_rate": safe_rate(comp_pass_count), |
| "coverage_pass_rate": safe_rate(cov_pass_count), |
| "accuracy_cls_and_completeness_threshold": safe_rate( |
| cls_and_comp_pass_count |
| ), |
| "accuracy_cls_completeness_coverage_threshold": safe_rate( |
| cls_comp_cov_pass_count |
| ), |
| "completeness_threshold": [args.comp_min_threshold, args.comp_max_threshold], |
| "coverage_thresholds": verifier.cov_iqr_ranges, |
| "details_path": details_path, |
| } |
|
|
| with open(summary_path, "w", encoding="utf-8") as f: |
| json.dump(summary, f, indent=2) |
|
|
| with open(details_path, "w", encoding="utf-8") as f: |
| for item in details: |
| f.write(json.dumps(item, ensure_ascii=False) + "\n") |
|
|
| print(json.dumps(summary, indent=2)) |
| print(f"[DONE] Summary saved: {summary_path}") |
| print(f"[DONE] Details saved: {details_path}") |
|
|
| except Exception as exc: |
| print(f"[error] {type(exc).__name__}: {exc}") |
| if args.provide_traceback: |
| traceback.print_exc() |
| raise |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
