Code / scripts/ml_training/essay_scoring/features/spacy_pipeline.py
scripts/ml_training/essay_scoring/features/spacy_pipeline.py
1
"""spaCy pipeline utilities for the essay scoring research build.
2
3
This module centralizes spaCy model loading for both:
4
5
- The local research pipeline (`scripts.ml_training.essay_scoring.features.*`)
6
- The Hemma offload server (`scripts.ml_training.essay_scoring.offload.spacy_runtime`)
7
8
Keeping this in one place makes it easier to tune performance (e.g., disabling
9
unused pipeline components) while ensuring the same NLP semantics are used
10
across local and offloaded feature extraction.
11
"""
12
13
from __future__ import annotations
14
15
from collections.abc import Sequence
16
from dataclasses import dataclass
17
18
import spacy
19
import textdescriptives
20
from spacy.language import Language
21
22
23
@dataclass(frozen=True)
24
class CoherenceComponentState:
25
"""TextDescriptives coherence component state for a loaded spaCy model."""
26
27
enabled: bool
28
status: str
29
vectors_length: int
30
pipe_name: str | None = None
31
message: str | None = None
32
33
34
def ensure_textdescriptives_readability(nlp: Language) -> None:
35
"""Ensure the TextDescriptives readability component is attached."""
36
37
_ = textdescriptives.__version__
38
if "textdescriptives/readability" not in nlp.pipe_names:
39
nlp.add_pipe("textdescriptives/readability", last=True)
40
41
42
def ensure_textdescriptives_coherence(nlp: Language) -> CoherenceComponentState:
43
"""Attach TextDescriptives coherence only for vector-backed spaCy models."""
44
45
_ = textdescriptives.__version__
46
vectors_length = int(nlp.vocab.vectors_length)
47
pipe_name = "textdescriptives/coherence"
48
if vectors_length <= 0:
49
return CoherenceComponentState(
50
enabled=False,
51
status="degraded_no_vectors",
52
vectors_length=vectors_length,
53
message="spaCy model has no static vectors; coherence cannot be trusted.",
54
)
55
56
if pipe_name not in nlp.pipe_names:
57
nlp.add_pipe(pipe_name, last=True)
58
return CoherenceComponentState(
59
enabled=True,
60
status="enabled",
61
vectors_length=vectors_length,
62
pipe_name=pipe_name,
63
)
64
65
66
def load_spacy_model(
67
model_name: str,
68
*,
69
enable_readability: bool = True,
70
enable_coherence: bool = False,
71
disable_components: Sequence[str] | None = None,
72
) -> Language:
73
"""Load a spaCy model for the essay scoring pipeline.
74
75
Tier 1 requires TextDescriptives readability metrics; Tier 2 does not.
76
"""
77
78
nlp = spacy.load(model_name, disable=list(disable_components or ()))
79
if enable_readability:
80
ensure_textdescriptives_readability(nlp)
81
if enable_coherence:
82
ensure_textdescriptives_coherence(nlp)
83
return nlp