Spaces:
Sleeping
Sleeping
| """Per-document structured concept profile assembly. | |
| Builds the embedding-ready profile blob: kind, topic, source path, concepts | |
| present (with definitions), connected concepts, excerpt. Per-kind variations | |
| keep profile shape compatible. Pure function for testability. | |
| PROFILE_SCHEMA_VERSION is bumped when the profile shape changes (forces | |
| re-embed of all docs since the embedding text changes). | |
| """ | |
| from __future__ import annotations | |
| import re | |
| from dataclasses import dataclass, field | |
| from scripts.lib.term_extraction import find_terms, expand_one_hop | |
| PROFILE_SCHEMA_VERSION = 1 | |
| class ConceptProfile: | |
| text: str | |
| concepts_present: list[str] | |
| connected_concepts: list[str] | |
| token_count: int | |
| packing_hint: str # "whole" | "prefer_section_excerpts" | "prefer_topic_excerpts" | |
| sections: list[dict] = field(default_factory=list) | |
| _PACKING_BY_KIND = { | |
| "lecture_note": "prefer_section_excerpts", | |
| "slide": "whole", | |
| "slide_answer": "whole", | |
| "syntopicon_entry": "whole", | |
| "course_meta": "whole", | |
| "lecture_problem": "prefer_topic_excerpts", | |
| } | |
| def _estimate_tokens(text: str) -> int: | |
| """Rough token estimate: ~4 chars per token.""" | |
| return max(1, len(text) // 4) | |
| def _truncate_to_budget(text: str, budget_tokens: int) -> str: | |
| char_budget = budget_tokens * 4 | |
| if len(text) <= char_budget: | |
| return text | |
| truncated = text[:char_budget] | |
| last_space = truncated.rfind(" ") | |
| if last_space > char_budget // 2: | |
| truncated = truncated[:last_space] | |
| return truncated + "..." | |
| def _format_concept_line(slug: str, registry: dict) -> str: | |
| term = registry.get("terms", {}).get(slug, {}) | |
| definition = term.get("definition", "") | |
| return f"- {slug}: {definition}" | |
| def assemble( | |
| doc: dict, | |
| source_text: str, | |
| term_regex: re.Pattern, | |
| registry: dict, | |
| excerpt_token_budget: int = 300, | |
| ) -> ConceptProfile: | |
| """Build a structured concept profile for a document.""" | |
| kind = doc.get("kind", "unknown") | |
| if kind == "lecture_turn": | |
| raise ValueError( | |
| "concept_profile.assemble() called with kind='lecture_turn'; " | |
| "lecture_turn records are citation-only and must skip profile generation." | |
| ) | |
| topic = doc.get("topic", "") | |
| source_path = doc.get("source_path", "") | |
| linked = doc.get("linked_syntopicon_entry") | |
| body_text = source_text | |
| concepts_present = find_terms(body_text, term_regex, registry) | |
| if linked and linked not in concepts_present: | |
| concepts_present = [linked] + concepts_present | |
| connected = [c for c in expand_one_hop(concepts_present, registry) if c not in concepts_present] | |
| parts: list[str] = [] | |
| parts.append(f"DOCUMENT KIND: {kind}") | |
| parts.append(f"TOPIC: {topic}") | |
| parts.append(f"SOURCE: {source_path}") | |
| parts.append("") | |
| if concepts_present: | |
| parts.append("CONCEPTS PRESENT (matched against syntopicon term_registry.words[]):") | |
| for slug in concepts_present: | |
| parts.append(_format_concept_line(slug, registry)) | |
| parts.append("") | |
| parts.append("CONNECTED CONCEPTS (1-hop syntopicon graph from terms above):") | |
| parts.append(", ".join(connected) if connected else "(none beyond concepts already present)") | |
| parts.append("") | |
| if kind == "syntopicon_entry": | |
| parts.append(_truncate_to_budget(source_text, excerpt_token_budget)) | |
| else: | |
| parts.append(f"EXCERPT (first ~{excerpt_token_budget} tokens):") | |
| parts.append(_truncate_to_budget(source_text, excerpt_token_budget)) | |
| text = "\n".join(parts).rstrip() + "\n" | |
| packing_hint = _PACKING_BY_KIND.get(kind, "whole") | |
| token_count = _estimate_tokens(source_text) | |
| return ConceptProfile( | |
| text=text, | |
| concepts_present=concepts_present, | |
| connected_concepts=connected, | |
| token_count=token_count, | |
| packing_hint=packing_hint, | |
| sections=[], | |
| ) | |