| import json |
| import os |
|
|
| import dspy |
| from dspy.evaluate import Evaluate |
|
|
|
|
| LLM_CPP_API_BASE = os.environ.get("LLM_CPP_API_BASE", "http://172.16.34.21:8034/v1") |
| MODEL_PATH = ( |
| "/home/mshahidul/readctrl/code/text_classifier/dspy_model/vllm-Meta-Llama-3.1-8B-Instruct_teacher-gpt5_v1/model.json" |
| ) |
| TEST_PATH = "/home/mshahidul/readctrl/code/text_classifier/test.jsonl" |
|
|
|
|
| llama_cpp_lm = dspy.LM( |
| model="openai/dspy", |
| api_base=LLM_CPP_API_BASE, |
| api_key="EMPTY", |
| temperature=0.0, |
| ) |
| dspy.configure(lm=llama_cpp_lm) |
|
|
|
|
| class HealthLiteracySignature(dspy.Signature): |
| """ |
| Analyze the linguistic complexity, use of medical jargon, and sentence |
| structure of 'generated_text' to determine the health literacy level. |
| """ |
|
|
| 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) |
|
|
|
|
| def load_testset(path): |
| examples = [] |
| with open(path, "r") as f: |
| for line in f: |
| if not line.strip(): |
| continue |
| record = json.loads(line) |
| example = dspy.Example( |
| generated_text=record["generated_text"], |
| literacy_label=record["literacy_label"], |
| ).with_inputs("generated_text") |
| examples.append(example) |
| return examples |
|
|
|
|
| def health_literacy_metric(gold, pred, trace=None): |
| if not pred or not hasattr(pred, "literacy_label"): |
| return False |
|
|
| gold_label = str(gold.literacy_label).strip().lower() |
| pred_label = str(pred.literacy_label).strip().lower() |
| return gold_label in pred_label |
|
|
|
|
| def load_compiled_classifier(path): |
| 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 main(): |
| if not os.path.exists(MODEL_PATH): |
| raise FileNotFoundError(f"Model file not found: {MODEL_PATH}") |
| if not os.path.exists(TEST_PATH): |
| raise FileNotFoundError(f"Test file not found: {TEST_PATH}") |
|
|
| testset = load_testset(TEST_PATH) |
| compiled_classifier = load_compiled_classifier(MODEL_PATH) |
|
|
| evaluator = Evaluate( |
| devset=testset, |
| metric=health_literacy_metric, |
| num_threads=1, |
| display_progress=True, |
| ) |
| evaluation_result = evaluator(compiled_classifier) |
| accuracy_score = ( |
| float(evaluation_result.score) |
| if hasattr(evaluation_result, "score") |
| else float(evaluation_result) |
| ) |
| print(evaluation_result) |
| print(f"accuracy_score: {accuracy_score}") |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
