Spaces:
Running
Running
File size: 6,612 Bytes
fb1e53a | 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 | from __future__ import annotations
import re
import unicodedata
from typing import Any
from src.common.cohort import normalize_cohort
CANONICAL_TYPES_BY_LOOKUP = {
"foreign_language": {"foreign_language", "foreign_language_equivalency"},
"study_duration": {"study_duration"},
"scholarship_classification": {"scholarship", "scholarship_classification"},
"scoring": {
"scoring",
"grade_scale",
"grade_10_to_letter",
"pass_fail_ungraded",
"letter_to_grade4",
"academic_classification",
"conduct",
"conduct_classification",
},
}
SUBTYPES_BY_OPERATION = {
"grade_10_to_letter": {
"grade_scale",
"grade_10_to_letter",
},
"pass_fail_ungraded": {
"grade_scale",
"grade_10_to_letter",
"pass_fail_ungraded",
},
"pass_threshold": {
"grade_scale",
"grade_10_to_letter",
"pass_fail_ungraded",
},
"letter_to_grade_4": {"letter_to_grade4", "letter_to_grade_4"},
"academic_classification": {"academic_classification"},
"conduct_classification": {"conduct", "conduct_classification"},
}
def _normalize(value: Any) -> str:
text = str(value or "").lower().replace("đ", "d")
text = unicodedata.normalize("NFD", text)
text = "".join(char for char in text if unicodedata.category(char) != "Mn")
text = re.sub(r"[^a-z0-9+.,-]+", " ", text)
return re.sub(r"\s+", " ", text).strip()
def _mentioned_cohorts(query: str, selected_cohort: str | None) -> set[str]:
selected = normalize_cohort(selected_cohort)
if selected:
return {selected}
normalized = _normalize(query)
cohorts: set[str] = set()
if re.search(r"\bk\s*48\s*[-/]?\s*k?\s*49\b|\bk\s*48\b|\bk\s*49\b", normalized):
cohorts.add("K48-K49")
if re.search(r"\bk\s*50\b", normalized):
cohorts.add("K50")
if re.search(r"\bk\s*51\b", normalized):
cohorts.add("K51")
return cohorts
def _select_rows(table: dict[str, Any]) -> tuple[list[dict[str, Any]], str]:
rows = [row for row in table.get("rows") or [] if isinstance(row, dict)]
return rows, "full_table"
def _table_text(table: dict[str, Any]) -> str:
return " ".join(
str(table.get(key) or "")
for key in (
"table_id",
"table_name",
"table_type",
"table_subtype",
"applicability",
)
)
def _matches_study_duration_slots(table: dict[str, Any], slots: dict[str, Any]) -> bool:
training_mode = str(slots.get("training_mode") or "")
if not training_mode:
return True
table_norm = _normalize(_table_text(table))
mode_norm = _normalize(training_mode)
if mode_norm == "chinh quy":
return "chinh quy" in table_norm
if mode_norm == "vua lam vua hoc":
return "vua lam vua hoc" in table_norm
return mode_norm in table_norm
def build_structured_context(
decision: dict[str, Any],
tables: list[dict[str, Any]],
*,
query: str,
cohort: str | None,
) -> dict[str, Any] | None:
"""Select authoritative table data for reasoning without vector indexing."""
lookup_type = str(decision.get("lookup_type") or "")
allowed_types = CANONICAL_TYPES_BY_LOOKUP.get(lookup_type)
if not allowed_types:
return None
slots = decision.get("slots") if isinstance(decision.get("slots"), dict) else {}
operation = str(slots.get("operation") or "")
allowed_subtypes = SUBTYPES_BY_OPERATION.get(operation)
wanted_cohorts = _mentioned_cohorts(query, cohort or decision.get("cohort"))
eligible_tables: list[dict[str, Any]] = []
for table in tables:
if not isinstance(table, dict) or not table.get("used_by_runtime", True):
continue
table_type = str(table.get("table_type") or "")
table_subtype = str(table.get("table_subtype") or table_type)
if table_type not in allowed_types and table_subtype not in allowed_types:
continue
if allowed_subtypes and table_subtype not in allowed_subtypes:
continue
if lookup_type == "study_duration" and not _matches_study_duration_slots(
table, slots
):
continue
eligible_tables.append(table)
available_cohorts = {
normalized
for table in eligible_tables
if (normalized := normalize_cohort(table.get("cohort")))
}
if not wanted_cohorts:
if len(available_cohorts) != 1:
return None
wanted_cohorts = set(available_cohorts)
selected_tables: list[dict[str, Any]] = []
for table in eligible_tables:
table_type = str(table.get("table_type") or "")
table_subtype = str(table.get("table_subtype") or table_type)
table_cohort = normalize_cohort(table.get("cohort"))
if wanted_cohorts and table_cohort not in wanted_cohorts:
continue
rows, selection_method = _select_rows(table)
if not rows:
continue
selected_tables.append(
{
"table_id": table.get("table_id"),
"table_type": table_type,
"table_subtype": table_subtype,
"table_name": table.get("table_name"),
"applicability": table.get("applicability"),
"cohort": table_cohort,
"document_id": table.get("document_id"),
"source_parent_id": table.get("source_parent_id")
or table.get("source_section_id"),
"source_pages": table.get("source_pages") or [],
"columns": table.get("columns") or [],
"rows": rows,
"total_row_count": len(table.get("rows") or []),
"selection_method": selection_method,
"derived_from": table.get("derived_from"),
}
)
if not selected_tables:
return None
selected_tables.sort(
key=lambda item: (str(item.get("cohort") or ""), str(item.get("table_id") or ""))
)
return {
"lookup_type": "structured_context",
"source_lookup_type": lookup_type,
"execution_mode": decision.get("execution_mode"),
"cohort": (
next(iter(wanted_cohorts)) if len(wanted_cohorts) == 1 else None
),
"items": selected_tables,
"source_parent_ids": list(
dict.fromkeys(
str(item["source_parent_id"])
for item in selected_tables
if item.get("source_parent_id")
)
),
}
|