Code / scripts/ml_training/essay_scoring/features/schema.py
scripts/ml_training/essay_scoring/features/schema.py
1
"""Pydantic feature-schema adapter for the essay scoring pipeline.
2
3
Purpose:
4
Preserve the ML training schema API while delegating canonical scorer
5
feature names, order, and snapshot identity to
6
`huleedu_nlp_shared.feature_contracts`.
7
8
Relationships:
9
- Consumed by feature combination, reusable stores, offload metadata, and
10
runner artifacts.
11
- The shared registry owns feature extraction schema; this module remains a
12
compatibility adapter for ML training callers.
13
"""
14
15
from __future__ import annotations
16
17
from huleedu_nlp_shared.feature_contracts import (
18
TIER1_FEATURE_NAMES,
19
TIER2_FEATURE_NAMES,
20
TIER3_CONSTRUCT_COMPLEX_NOMINAL_FEATURE_NAMES,
21
TIER3_RST_GUM_PRIMARY_TREE_PURE_FEATURE_NAMES,
22
TIER3_RST_GUM_PRIMARY_TREE_SHARED_FEATURE_NAMES,
23
FeatureSchemaSnapshotV1,
24
)
25
from pydantic import BaseModel, ConfigDict, Field, model_validator
26
27
from ..features.feature_schema_identity import (
28
SCHEMA_ADAPTER_ARTIFACT_DIGEST,
29
SCHEMA_ADAPTER_RUNTIME_FINGERPRINT,
30
CurrentScorerInputId,
31
build_shared_schema_snapshot,
32
shared_feature_names_for,
33
)
34
35
36
class Tier1Features(BaseModel):
37
"""Tier 1 complete measurement (ledger issue density, readability, core length).
38
39
This models the complete measured Tier 1 record, which retains
40
``flesch_kincaid``. The narrower accepted scorer-vector projection is an
41
explicit, separate operation; see :meth:`accepted_projection`.
42
"""
43
44
model_config = ConfigDict(extra="forbid")
45
46
grammar_errors_per_100_words: float
47
language_tool_spelling_errors_per_100_words: float
48
l2_dictionary_corrections_per_100_words: float
49
final_spelling_issues_per_100_words: float
50
punctuation_errors_per_100_words: float
51
flesch_kincaid: float
52
smog: float
53
coleman_liau: float
54
ari: float
55
avg_sentence_length: float
56
ttr: float
57
word_count: float
58
avg_word_length: float
59
60
def to_list(self) -> list[float]:
61
"""Return the complete Tier 1 measurement as a list in canonical order."""
62
63
return [
64
self.grammar_errors_per_100_words,
65
self.language_tool_spelling_errors_per_100_words,
66
self.l2_dictionary_corrections_per_100_words,
67
self.final_spelling_issues_per_100_words,
68
self.punctuation_errors_per_100_words,
69
self.flesch_kincaid,
70
self.smog,
71
self.coleman_liau,
72
self.ari,
73
self.avg_sentence_length,
74
self.ttr,
75
self.word_count,
76
self.avg_word_length,
77
]
78
79
def accepted_projection(self) -> list[float]:
80
"""Return the accepted scorer-vector projection (exactly the canonical 12 Tier 1 values)."""
81
82
return [
83
self.grammar_errors_per_100_words,
84
self.language_tool_spelling_errors_per_100_words,
85
self.l2_dictionary_corrections_per_100_words,
86
self.final_spelling_issues_per_100_words,
87
self.punctuation_errors_per_100_words,
88
self.smog,
89
self.coleman_liau,
90
self.ari,
91
self.avg_sentence_length,
92
self.ttr,
93
self.word_count,
94
self.avg_word_length,
95
]
96
97
98
class AppliedSpellingFeatureCounts(BaseModel):
99
"""Mutually exclusive spelling corrections applied to one parser text."""
100
101
model_config = ConfigDict(extra="forbid", frozen=True)
102
103
total: int = Field(ge=0)
104
non_core: int = Field(ge=0)
105
core_l2: int = Field(ge=0)
106
107
@model_validator(mode="after")
108
def _validate_partition(self) -> "AppliedSpellingFeatureCounts":
109
if self.total != self.non_core + self.core_l2:
110
raise ValueError("applied spelling total must equal non-core plus core-L2")
111
return self
112
113
114
class Tier2Features(BaseModel):
115
"""Tier 2 complete measurement set (syntactic complexity, cohesion, prompt relevance).
116
117
This is the complete Tier 2 record, which retains ``clause_count``. The
118
narrower accepted scorer-vector projection is an explicit, separate
119
operation; see :meth:`accepted_projection`.
120
"""
121
122
model_config = ConfigDict(extra="forbid")
123
124
parse_tree_depth: float
125
clause_count: float
126
passive_ratio: float
127
dep_distance: float
128
sent_similarity_variance: float
129
prompt_similarity: float
130
intro_prompt_sim: float
131
min_para_relevance: float
132
133
def to_list(self) -> list[float]:
134
"""Return the complete Tier 2 entry as a list in canonical order."""
135
136
return [
137
self.parse_tree_depth,
138
self.clause_count,
139
self.passive_ratio,
140
self.dep_distance,
141
self.sent_similarity_variance,
142
self.prompt_similarity,
143
self.intro_prompt_sim,
144
self.min_para_relevance,
145
]
146
147
def accepted_projection(self) -> list[float]:
148
"""Return the accepted projection (canonical seven Tier 2 values)."""
149
150
return [
151
self.parse_tree_depth,
152
self.passive_ratio,
153
self.dep_distance,
154
self.sent_similarity_variance,
155
self.prompt_similarity,
156
self.intro_prompt_sim,
157
self.min_para_relevance,
158
]
159
160
161
class ConstructComplexNominalFeatures(BaseModel):
162
"""Accepted construct-aligned complex nominal scorer features."""
163
164
model_config = ConfigDict(extra="forbid")
165
166
procedural_or_event_head_complex_np_density_log1p: float
167
168
def to_list(self) -> list[float]:
169
"""Return features as a list in canonical order."""
170
171
return [self.procedural_or_event_head_complex_np_density_log1p]
172
173
174
class FeatureSchema(BaseModel):
175
"""Schema describing feature ordering for training artifacts."""
176
177
model_config = ConfigDict(extra="forbid")
178
179
embedding_dim: int
180
tier1: list[str]
181
tier2: list[str]
182
tier3_construct: list[str] = Field(default_factory=list)
183
tier3_rst_gum_primary_tree_pure: list[str] = Field(default_factory=list)
184
tier3_rst_gum_primary_tree_shared: list[str] = Field(default_factory=list)
185
combined: list[str]
186
schema_snapshot: FeatureSchemaSnapshotV1
187
188
189
def build_feature_schema(embedding_dim: int) -> FeatureSchema:
190
"""Build a feature schema aligned with EPIC-010 naming.
191
192
Args:
193
embedding_dim: Number of embedding dimensions.
194
195
Returns:
196
FeatureSchema for combined features.
197
"""
198
199
snapshot = build_shared_schema_snapshot(
200
scorer_input_id="combined.current.v1",
201
embedding_dim=embedding_dim,
202
runtime_fingerprint=SCHEMA_ADAPTER_RUNTIME_FINGERPRINT,
203
artifact_digest=SCHEMA_ADAPTER_ARTIFACT_DIGEST,
204
)
205
return FeatureSchema(
206
embedding_dim=embedding_dim,
207
tier1=list(TIER1_FEATURE_NAMES),
208
tier2=list(TIER2_FEATURE_NAMES),
209
tier3_construct=list(TIER3_CONSTRUCT_COMPLEX_NOMINAL_FEATURE_NAMES),
210
tier3_rst_gum_primary_tree_pure=list(TIER3_RST_GUM_PRIMARY_TREE_PURE_FEATURE_NAMES),
211
tier3_rst_gum_primary_tree_shared=list(TIER3_RST_GUM_PRIMARY_TREE_SHARED_FEATURE_NAMES),
212
combined=list(snapshot.ordered_feature_names),
213
schema_snapshot=snapshot,
214
)
215
216
217
def feature_names_for(scorer_input_id: CurrentScorerInputId, embedding_dim: int) -> list[str]:
218
"""Return ordered feature names for one exact scorer input."""
219
220
return shared_feature_names_for(scorer_input_id, embedding_dim)