diff --git a/app/core/type_registry.py b/app/core/type_registry.py index 93b17a0..83c4703 100644 --- a/app/core/type_registry.py +++ b/app/core/type_registry.py @@ -1,111 +1,117 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- """ -type_registry.py v1.0.0 +Modul: app/core/type_registry.py +Version: 1.0.0 +Datum: 2025-11-08 -Zweck: - - Optionale, konfigurierbare Type-Registry laden (YAML/JSON), um pro "type" - (aus Frontmatter) Chunk-Profile, Default-Edges und optionale - Retriever-Gewichte bereitzustellen – ohne bestehende Funktionen zu brechen. +Zweck +----- +Lädt eine optionale Typ-Registry (config/types.yaml) und stellt +komfortable Zugriffsfunktionen bereit. Die Registry ist *optional*: +- Fehlt die Datei oder ist das YAML defekt, wird ein konservativer + Default (Typ "concept") verwendet und es wird eine Warnung ausgegeben. +- Änderungen an der Datei greifen nach einem Neustart des Prozesses. -Kompatibilität: - - Keine Abhängigkeiten von anderen Modulen. - - Keine harten Fehler, wenn Registry fehlt oder unvollständig ist. +Öffentliche API +--------------- +- load_type_registry(path: str = "config/types.yaml") -> dict +- get_type_config(note_type: str, reg: dict) -> dict +- resolve_note_type(fm_type: str | None, reg: dict) -> str +- effective_chunk_profile(note_type: str, reg: dict) -> str | None +- profile_overlap(profile: str | None) -> tuple[int,int] # nur Overlap-Empfehlung -Nutzung: - from app.core.type_registry import ( - load_type_registry, resolve_chunk_profile, get_edge_defaults_for_type, - get_retriever_weight_for_type - ) - -Umgebungsvariablen: - TYPE_REGISTRY_PATH (default: "config/types.yaml") - -Format (Beispiel): - version: 1.0 - types: - concept: - chunk_profile: long - edge_defaults: [references, related_to] - retriever_weight: 1.0 - task: - chunk_profile: short - edge_defaults: [depends_on, belongs_to] - retriever_weight: 0.8 - experience: - chunk_profile: medium - edge_defaults: [derived_from, inspired_by] - retriever_weight: 0.9 +Hinweis +------- +Die Registry steuert KEINE Breaking Changes. Ohne Datei/Typ bleibt das +Verhalten exakt wie im Release-Stand 20251105. """ - from __future__ import annotations + +from functools import lru_cache +from typing import Dict, Any, Optional, Tuple import os -import json -from typing import Any, Dict, List, Optional try: - import yaml # type: ignore -except Exception: # yaml ist optional; JSON wird ebenfalls unterstützt - yaml = None # type: ignore + import yaml # PyYAML +except Exception: + yaml = None # wird erst benötigt, wenn eine Datei gelesen werden soll -_CACHE: Dict[str, Dict[str, Any]] = {} +# Konservativer Default – bewusst minimal +_DEFAULT_REGISTRY: Dict[str, Any] = { + "version": "1.0", + "types": { + "concept": { + "chunk_profile": "medium", + "edge_defaults": ["references", "related_to"], + "retriever_weight": 1.0, + } + }, + "_using_defaults": True, + "_warning": "types.yaml missing or invalid – using built-in defaults (type=concept).", +} + +# Chunk-Profile → Overlap-Empfehlungen (nur für synthetische Fensterbildung) +# Die absoluten Chunk-Längen bleiben Aufgabe des Chunkers (assemble_chunks). +_PROFILE_TO_OVERLAP: Dict[str, Tuple[int, int]] = { + "short": (20, 30), + "medium": (40, 60), + "long": (60, 80), +} + + +@lru_cache(maxsize=1) +def load_type_registry(path: str = "config/types.yaml") -> Dict[str, Any]: + """ + Lädt die Registry aus 'path'. Bei Fehlern wird ein konserviver Default geliefert. + Die Rückgabe ist *prozessweit* gecached. + """ + if not path: + return dict(_DEFAULT_REGISTRY) + + if not os.path.isfile(path): + return dict(_DEFAULT_REGISTRY) + + if yaml is None: + # PyYAML fehlt → auf Default zurückfallen + return dict(_DEFAULT_REGISTRY) -def _safe_load_yaml_or_json(path: str) -> Dict[str, Any]: - if not os.path.exists(path): - return {} try: with open(path, "r", encoding="utf-8") as f: - data = f.read() + data = yaml.safe_load(f) or {} + # Minimal validieren + if not isinstance(data, dict) or "types" not in data or not isinstance(data["types"], dict): + return dict(_DEFAULT_REGISTRY) + data.setdefault("version", "1.0") + data.setdefault("_using_defaults", False) + return data except Exception: - return {} - # YAML bevorzugen, wenn verfügbar und Datei nach YAML aussieht - if path.lower().endswith((".yaml", ".yml")) and yaml is not None: - try: - return yaml.safe_load(data) or {} - except Exception: - return {} - # JSON fallback - try: - return json.loads(data) - except Exception: - return {} + return dict(_DEFAULT_REGISTRY) -def load_type_registry(path: Optional[str] = None) -> Dict[str, Any]: - """ - Lädt einmalig die Registry und cached sie. - Fehlt sie, wird ein leeres Dict geliefert (keine Fehler). - """ - key = path or os.getenv("TYPE_REGISTRY_PATH", "config/types.yaml") - if key in _CACHE: - return _CACHE[key] - obj = _safe_load_yaml_or_json(key) - if not isinstance(obj, dict): - obj = {} - _CACHE[key] = obj - return obj -def _types_map(reg: Dict[str, Any]) -> Dict[str, Any]: - return reg.get("types", {}) if isinstance(reg, dict) else {} +def get_type_config(note_type: Optional[str], reg: Dict[str, Any]) -> Dict[str, Any]: + t = (note_type or "concept").strip().lower() + types = (reg or {}).get("types", {}) if isinstance(reg, dict) else {} + return types.get(t) or types.get("concept") or _DEFAULT_REGISTRY["types"]["concept"] -def resolve_chunk_profile(note_type: str, default_profile: str = "default") -> str: - reg = load_type_registry() - tmap = _types_map(reg) - entry = tmap.get(note_type, {}) - return str(entry.get("chunk_profile", default_profile)) -def get_edge_defaults_for_type(note_type: str) -> List[str]: - reg = load_type_registry() - tmap = _types_map(reg) - entry = tmap.get(note_type, {}) - v = entry.get("edge_defaults", []) - if not isinstance(v, list): - return [] - return [str(x) for x in v] +def resolve_note_type(fm_type: Optional[str], reg: Dict[str, Any]) -> str: + """Liefert einen gültigen Typ (unbekannt → 'concept').""" + t = (fm_type or "concept").strip().lower() + types = (reg or {}).get("types", {}) if isinstance(reg, dict) else {} + return t if t in types else "concept" -def get_retriever_weight_for_type(note_type: str) -> Optional[float]: - reg = load_type_registry() - tmap = _types_map(reg) - entry = tmap.get(note_type, {}) - v = entry.get("retriever_weight", None) - try: - return float(v) if v is not None else None - except Exception: - return None + +def effective_chunk_profile(note_type: Optional[str], reg: Dict[str, Any]) -> Optional[str]: + cfg = get_type_config(note_type, reg) + prof = cfg.get("chunk_profile") + if isinstance(prof, str) and prof.strip(): + return prof.strip().lower() + return None + + +def profile_overlap(profile: Optional[str]) -> Tuple[int, int]: + """Gibt eine Overlap-Empfehlung (low, high) für das Profil zurück.""" + if not profile: + return _PROFILE_TO_OVERLAP["medium"] + return _PROFILE_TO_OVERLAP.get(profile.strip().lower(), _PROFILE_TO_OVERLAP["medium"])