File size: 3,341 Bytes
1db7196
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
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()