Spaces:
Running on Zero
Running on Zero
File size: 7,948 Bytes
531d9de 5d05dd0 19d779e 531d9de 5d05dd0 19d779e 5d05dd0 531d9de 5d05dd0 19d779e | 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 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 | """Candidate construction for constrained classification prompts."""
from __future__ import annotations
from pydantic import BaseModel, ConfigDict, Field
from gcmd_classifier.llm.prompts import PromptCandidate
from gcmd_classifier.models import CanonicalConceptRecord
from gcmd_classifier.vocabulary.index import VocabularyIndex
class TopicCandidate(BaseModel):
"""Application candidate for one UUID-bearing GCMD Topic."""
model_config = ConfigDict(extra="forbid", frozen=True)
candidate_id: str = Field(min_length=1)
topic_uuid: str = Field(min_length=1)
prompt_candidate: PromptCandidate
@property
def record_uuid(self) -> str:
"""Authoritative UUID for the candidate's Topic record."""
return self.topic_uuid
class TermCandidate(BaseModel):
"""Application candidate for one direct Term beneath a selected Topic."""
model_config = ConfigDict(extra="forbid", frozen=True)
candidate_id: str = Field(min_length=1)
topic_uuid: str = Field(min_length=1)
term_uuid: str = Field(min_length=1)
prompt_candidate: PromptCandidate
@property
def record_uuid(self) -> str:
"""Authoritative UUID for the candidate's Term record."""
return self.term_uuid
class VariableCandidate(BaseModel):
"""Application candidate for one direct Variable child beneath a selected parent."""
model_config = ConfigDict(extra="forbid", frozen=True)
candidate_id: str = Field(min_length=1)
parent_uuid: str = Field(min_length=1)
variable_uuid: str = Field(min_length=1)
prompt_candidate: PromptCandidate
@property
def record_uuid(self) -> str:
"""Authoritative UUID for the candidate's Variable record."""
return self.variable_uuid
def build_topic_candidates(index: VocabularyIndex) -> tuple[TopicCandidate, ...]:
"""Build deterministic prompt candidates from UUID-bearing Topic records only."""
candidates: list[TopicCandidate] = []
for position, topic in enumerate(index.topics(), start=1):
_validate_topic_record(topic)
candidate_id = f"topic_{position:04d}"
candidates.append(
TopicCandidate(
candidate_id=candidate_id,
topic_uuid=topic.UUID,
prompt_candidate=PromptCandidate(
candidate_id=candidate_id,
name=topic.name,
level=topic.level,
definition=topic.definition,
canonical_path=topic.canonical_path,
),
)
)
return tuple(candidates)
def build_term_candidates(
index: VocabularyIndex,
*,
topic_uuid: str,
) -> tuple[TermCandidate, ...]:
"""Build deterministic prompt candidates from direct Term children of one Topic."""
topic = index.get(topic_uuid)
_validate_topic_record(topic)
candidates: list[TermCandidate] = []
for position, term in enumerate(index.terms_for_topic(topic_uuid), start=1):
_validate_term_record(term, topic_uuid=topic_uuid, index=index)
candidate_id = f"term_{position:04d}"
candidates.append(
TermCandidate(
candidate_id=candidate_id,
topic_uuid=topic_uuid,
term_uuid=term.UUID,
prompt_candidate=PromptCandidate(
candidate_id=candidate_id,
name=term.name,
level=term.level,
definition=term.definition,
canonical_path=term.canonical_path,
parent_context=topic.canonical_path,
),
)
)
return tuple(candidates)
def build_variable_candidates(
index: VocabularyIndex,
*,
parent_uuid: str,
) -> tuple[VariableCandidate, ...]:
"""Build deterministic prompt candidates from direct Variable children of one parent."""
parent = index.get(parent_uuid)
_validate_variable_parent_record(parent)
candidates: list[VariableCandidate] = []
for position, variable in enumerate(index.variables_for_parent(parent_uuid), start=1):
_validate_variable_child_record(variable, parent_uuid=parent_uuid, index=index)
candidate_id = f"variable_{position:04d}"
candidates.append(
VariableCandidate(
candidate_id=candidate_id,
parent_uuid=parent_uuid,
variable_uuid=variable.UUID,
prompt_candidate=PromptCandidate(
candidate_id=candidate_id,
name=variable.name,
level=variable.level,
definition=variable.definition,
canonical_path=variable.canonical_path,
parent_context=parent.canonical_path,
),
)
)
return tuple(candidates)
def validate_variable_candidate_relationship(
candidate: VariableCandidate,
*,
selected_parent_uuid: str,
index: VocabularyIndex,
) -> None:
"""Validate that a Variable candidate is a direct child of the selected parent."""
if candidate.parent_uuid != selected_parent_uuid:
raise ValueError(
f"Variable candidate {candidate.candidate_id!r} was built for parent "
f"{candidate.parent_uuid!r}, not selected parent {selected_parent_uuid!r}."
)
variable = index.get(candidate.variable_uuid)
_validate_variable_child_record(variable, parent_uuid=selected_parent_uuid, index=index)
def validate_term_candidate_relationship(
candidate: TermCandidate,
*,
selected_topic_uuid: str,
index: VocabularyIndex,
) -> None:
"""Validate that a Term candidate is a direct child of the selected Topic."""
if candidate.topic_uuid != selected_topic_uuid:
raise ValueError(
f"Term candidate {candidate.candidate_id!r} was built for Topic "
f"{candidate.topic_uuid!r}, not selected Topic {selected_topic_uuid!r}."
)
term = index.get(candidate.term_uuid)
_validate_term_record(term, topic_uuid=selected_topic_uuid, index=index)
def _validate_topic_record(record: CanonicalConceptRecord) -> None:
if record.level != "Topic":
raise ValueError(f"Expected Topic record, got {record.level!r} for {record.UUID!r}.")
if not record.UUID:
raise ValueError(f"Topic record {record.name!r} does not have a UUID.")
def _validate_term_record(
record: CanonicalConceptRecord,
*,
topic_uuid: str,
index: VocabularyIndex,
) -> None:
if record.level != "Term":
raise ValueError(f"Expected Term record, got {record.level!r} for {record.UUID!r}.")
if not record.UUID:
raise ValueError(f"Term record {record.name!r} does not have a UUID.")
if index.parent_of(record.UUID) != topic_uuid:
raise ValueError(f"Term {record.UUID!r} is not a direct child of Topic {topic_uuid!r}.")
def _validate_variable_parent_record(record: CanonicalConceptRecord) -> None:
if record.level not in {"Term", "Variable_Level_1", "Variable_Level_2"}:
raise ValueError(
f"Expected Term or Variable parent, got {record.level!r} for {record.UUID!r}."
)
if not record.UUID:
raise ValueError(f"Variable parent record {record.name!r} does not have a UUID.")
def _validate_variable_child_record(
record: CanonicalConceptRecord,
*,
parent_uuid: str,
index: VocabularyIndex,
) -> None:
if record.level not in {"Variable_Level_1", "Variable_Level_2", "Variable_Level_3"}:
raise ValueError(f"Expected Variable record, got {record.level!r} for {record.UUID!r}.")
if not record.UUID:
raise ValueError(f"Variable record {record.name!r} does not have a UUID.")
if index.parent_of(record.UUID) != parent_uuid:
raise ValueError(
f"Variable {record.UUID!r} is not a direct child of parent {parent_uuid!r}."
)
|