From f18a40d76c5c7325c4ea69cfd7322573b5b4a9f2 Mon Sep 17 00:00:00 2001 From: Lars Date: Sun, 16 Nov 2025 18:51:06 +0100 Subject: [PATCH] Dateien nach "app/core" hochladen --- app/core/chunk_payload.py | 289 ++++++++++++++++++-------------------- 1 file changed, 134 insertions(+), 155 deletions(-) diff --git a/app/core/chunk_payload.py b/app/core/chunk_payload.py index d7ed9c2..7ce39fe 100644 --- a/app/core/chunk_payload.py +++ b/app/core/chunk_payload.py @@ -1,182 +1,161 @@ #!/usr/bin/env python3 # -*- coding: utf-8 -*- """ -app/core/chunk_payload.py — Mindnet V2 (compat) +app/core/chunk_payload.py (Mindnet V2 — robust) -Ziele (unveränderte v1-Basis, weniger Duplikate): -- **Kanonicum:** `index` -- **Standard‑Alias (v2):** `ord` (abschaltbar über ENV MINDNET_CHUNK_INCLUDE_ORD=0) -- **Optionale Aliase:** gesteuert über ENV MINDNET_CHUNK_INDEX_ALIASES - (z. B. "chunk_num,Chunk_Nummer" oder "Chunk_Number"). Standard: kein zusätzlicher Alias. -- Verarbeitet Chunks als Dict **oder** Objekt (Dataclass) und setzt immer `id` (= `chunk_id`) -- Berechnet `neighbors.prev/next`, falls nicht vorhanden -- Denormalisiert Note‑`tags` auf Chunks -- Akzeptiert `file_path=` als Alias zu `path_arg` +Aufgabe +------- +Erzeugt Chunk-Payloads aus den vom Chunker gelieferten "Chunk"-Objekten. +- Spiegelt `retriever_weight` und `chunk_profile` in **jedem** Chunk-Payload. +- Werteauflösung: Frontmatter > types.yaml > Defaults. +- Lädt `config/types.yaml` selbst, wenn `types_cfg` nicht übergeben wurde. -ENV: -- MINDNET_CHUNK_INCLUDE_ORD: "1" (Default) | "0" -- MINDNET_CHUNK_INDEX_ALIASES: CSV‑Liste zulässiger Namen: chunk_num,Chunk_Nummer,Chunk_Number +Eingang +------- +- note: Dict mit mind. { frontmatter: {...}, id, type, title, path } +- note_path: Pfad der Note (für Payload-Feld `path`) +- chunks_from_chunker: Liste von Objekten mit Attributen/Feldern: + id, note_id, index, text, window, neighbors_prev, neighbors_next +- note_text: voller Text der Note (optional, kann leer sein) +- types_cfg: optional; wenn None → config wird intern geladen +- file_path: optional, für Debug/Tracing im Payload + +Ausgang (pro Chunk) +------------------- +- Pflichtfelder: note_id, chunk_id, index (0-basiert), ord (1-basiert), type, tags +- Texte: text, window +- Nachbarn: neighbors_prev, neighbors_next +- Spiegelungen: retriever_weight, chunk_profile +- Meta: source_path, path, section (leer), created/updated opt. aus Frontmatter -Hinweis: `edge_defaults` sind Note‑Regeln (nicht pro Chunk). """ from __future__ import annotations - -import json -import os -import pathlib -import hashlib from typing import Any, Dict, List, Optional +import os, yaml -from app.core.chunker import assemble_chunks +def _env(n: str, d: Optional[str]=None) -> str: + v = os.getenv(n) + return v if v is not None else (d or "") -# ---------- Helpers ---------- +def _deep_get(root: Any, path: str) -> Any: + cur = root + for key in path.split("."): + if not isinstance(cur, dict) or key not in cur: + return None + cur = cur[key] + return cur -def _as_dict(obj): - if isinstance(obj, dict): - return obj - d = {} - for k in ("index","ord","chunk_index","text","window","id","chunk_id","neighbors","note_id","type","title"): - if hasattr(obj, k): - d[k] = getattr(obj, k) - return d +def _as_float(x: Any): + try: + return float(x) + except Exception: + return None -def _coalesce(*vals): - for v in vals: - if v is not None: - return v +def _load_types_local() -> dict: + p = _env("MINDNET_TYPES_FILE", "./config/types.yaml") + try: + with open(p, "r", encoding="utf-8") as f: + return yaml.safe_load(f) or {} + except Exception: + return {} + +def _effective_chunk_profile(note_type: str, fm: Dict[str, Any], reg: dict) -> Optional[str]: + # Frontmatter zuerst + if isinstance(fm.get("chunk_profile"), str): + return fm.get("chunk_profile") + # Registry + types = reg.get("types") if isinstance(reg.get("types"), dict) else reg + if isinstance(types, dict): + v = types.get(note_type, {}) + if isinstance(v, dict): + cp = v.get("chunk_profile") + if isinstance(cp, str): + return cp return None -def _env_float(name: str, default: float) -> float: - try: - return float(os.environ.get(name, default)) - except Exception: - return default +def _effective_retriever_weight(note_type: str, fm: Dict[str, Any], reg: dict) -> float: + # Frontmatter zuerst + if fm.get("retriever_weight") is not None: + v = _as_float(fm.get("retriever_weight")) + if v is not None: + return float(v) + # Registry-Pfade + types = reg.get("types") if isinstance(reg.get("types"), dict) else reg + candidates = [ + f"{note_type}.retriever_weight", + f"{note_type}.retriever.weight", + f"{note_type}.retrieval.weight", + "defaults.retriever_weight", + "defaults.retriever.weight", + "global.retriever_weight", + "global.retriever.weight", + ] + for path in candidates: + # Wenn types == reg-root (flatten), erlauben sowohl "types.X" als auch "X" + val = _deep_get(types, path) if "." in path else (types.get(path) if isinstance(types, dict) else None) + if val is None and isinstance(reg, dict): + # versuche absolute Pfade + val = _deep_get(reg, f"types.{path}") + v = _as_float(val) + if v is not None: + return float(v) + return 1.0 -def _ensure_list(x) -> list: - if x is None: return [] - if isinstance(x, list): return [str(i) for i in x] - if isinstance(x, (set, tuple)): return [str(i) for i in x] - return [str(x)] +def make_chunk_payloads(note: Dict[str, Any], + note_path: str, + chunks_from_chunker: List[Any], + *, + note_text: str = "", + types_cfg: Optional[dict] = None, + file_path: Optional[str] = None) -> List[Dict[str, Any]]: + fm = (note or {}).get("frontmatter", {}) + note_type = fm.get("type") or note.get("type") or "concept" + reg = types_cfg if isinstance(types_cfg, dict) else _load_types_local() -def _text_from_note(note: Dict[str, Any]) -> str: - return note.get("body") or note.get("text") or "" + # Effektive Werte bestimmen + cp = _effective_chunk_profile(note_type, fm, reg) + rw = _effective_retriever_weight(note_type, fm, reg) + + tags = fm.get("tags") or [] + if isinstance(tags, str): + tags = [tags] -def _iter_chunks(note: Dict[str, Any], chunk_profile: str, fulltext: str) -> List[Dict[str, Any]]: - """Nutze bestehenden assemble_chunks(note_id, body, type). Rückgabe kann Objektliste sein.""" - note_id = note.get("id") or (note.get("frontmatter") or {}).get("id") - ntype = (note.get("frontmatter") or {}).get("type") or note.get("type") or "note" - raw_list = assemble_chunks(note_id, fulltext, ntype) out: List[Dict[str, Any]] = [] - for c in raw_list: - out.append(_as_dict(c) if not isinstance(c, dict) else c) - return out -# ---------- Main ---------- + for idx, ch in enumerate(chunks_from_chunker): + # Chunk-Grunddaten (Attribute oder Keys) + cid = getattr(ch, "id", None) or (ch.get("id") if isinstance(ch, dict) else None) + nid = getattr(ch, "note_id", None) or (ch.get("note_id") if isinstance(ch, dict) else fm.get("id")) + index = getattr(ch, "index", None) or (ch.get("index") if isinstance(ch, dict) else idx) + text = getattr(ch, "text", None) or (ch.get("text") if isinstance(ch, dict) else "") + window = getattr(ch, "window", None) or (ch.get("window") if isinstance(ch, dict) else text) + prev_id = getattr(ch, "neighbors_prev", None) or (ch.get("neighbors_prev") if isinstance(ch, dict) else None) + next_id = getattr(ch, "neighbors_next", None) or (ch.get("neighbors_next") if isinstance(ch, dict) else None) -def make_chunk_payloads( - note: Any, - path_arg: Optional[str] = None, - chunks_from_chunker: Optional[List[Dict[str, Any]]] = None, - *, - file_path: Optional[str] = None, - note_text: Optional[str] = None, - types_cfg: Optional[dict] = None, -) -> List[Dict[str, Any]]: - n = note if isinstance(note, dict) else {"frontmatter": {}} - fm = n.get("frontmatter") or {} - note_type = str(fm.get("type") or n.get("type") or "note") - types_cfg = types_cfg or {} - cfg_for_type = types_cfg.get(note_type, {}) if isinstance(types_cfg, dict) else {} - - default_rw = _env_float("MINDNET_DEFAULT_RETRIEVER_WEIGHT", 1.0) - retriever_weight = _coalesce(fm.get("retriever_weight"), cfg_for_type.get("retriever_weight"), default_rw) - try: - retriever_weight = float(retriever_weight) - except Exception: - retriever_weight = default_rw - - chunk_profile = _coalesce(fm.get("chunk_profile"), cfg_for_type.get("chunk_profile"), os.environ.get("MINDNET_DEFAULT_CHUNK_PROFILE","medium")) - chunk_profile = chunk_profile if isinstance(chunk_profile, str) else "medium" - - note_id = n.get("note_id") or n.get("id") or fm.get("id") - title = n.get("title") or fm.get("title") or "" - - # Pfad (file_path > note['path'] > path_arg) - path = file_path or n.get("path") or path_arg - if isinstance(path, pathlib.Path): - path = str(path) - path = path or "" - - # Tags denormalisieren (optional) - tags = fm.get("tags") or fm.get("keywords") or n.get("tags") - tags_list = _ensure_list(tags) if tags else [] - - # Chunks holen - fulltext = note_text if isinstance(note_text, str) else _text_from_note(n) - raw_chunks = chunks_from_chunker if isinstance(chunks_from_chunker, list) else _iter_chunks(n, chunk_profile, fulltext) - - include_ord = (os.environ.get("MINDNET_CHUNK_INCLUDE_ORD", "1") != "0") - alias_csv = os.environ.get("MINDNET_CHUNK_INDEX_ALIASES", "").strip() - extra_aliases = [a.strip() for a in alias_csv.split(",") if a.strip()] if alias_csv else [] - - payloads: List[Dict[str, Any]] = [] - for c in raw_chunks: - cdict = c if isinstance(c, dict) else _as_dict(c) - idx = _coalesce(cdict.get("index"), cdict.get("ord"), cdict.get("chunk_index"), len(payloads)) - try: - idx = int(idx) - except Exception: - idx = len(payloads) - - text = _coalesce(cdict.get("window"), cdict.get("text"), "") - if not isinstance(text, str): - text = str(text or "") - - # deterministische ID - key = f"{note_id}|{idx}" - h = hashlib.sha1(key.encode("utf-8")).hexdigest()[:12] if note_id else hashlib.sha1(f"{path}|{idx}".encode("utf-8")).hexdigest()[:12] - chunk_id = cdict.get("chunk_id") or cdict.get("id") or (f"{note_id}-{idx:03d}-{h}" if note_id else h) - - payload = { - "id": chunk_id, # v1 erwartet 'id' - "chunk_id": chunk_id, - "index": idx, # Kanonisch - "note_id": note_id, + pl: Dict[str, Any] = { + "note_id": nid, + "chunk_id": cid, + "index": int(index), + "ord": int(index) + 1, "type": note_type, - "title": title, - "path": path, + "tags": tags, "text": text, - "window": text, - "retriever_weight": retriever_weight, - "chunk_profile": chunk_profile, + "window": window, + "neighbors_prev": prev_id, + "neighbors_next": next_id, + "section": getattr(ch, "section", None) or (ch.get("section") if isinstance(ch, dict) else ""), + "path": note_path, + "source_path": file_path or note_path, + "retriever_weight": float(rw), } - if include_ord: - payload["ord"] = idx # v2‑Standard, abschaltbar - for alias in extra_aliases: - # nur whitelisted Namen zulassen - if alias in ("chunk_num","Chunk_Nummer","Chunk_Number"): - payload[alias] = idx + if cp is not None: + pl["chunk_profile"] = cp - nb = cdict.get("neighbors") - if isinstance(nb, dict): - prev_id = nb.get("prev"); next_id = nb.get("next") - payload["neighbors"] = {"prev": prev_id, "next": next_id} + # Aufräumen: keine historischen Aliasfelder + for alias in ("chunk_num", "Chunk_Number"): + if alias in pl: + pl.pop(alias, None) - if tags_list: - payload["tags"] = tags_list + out.append(pl) - json.loads(json.dumps(payload, ensure_ascii=False)) - payloads.append(payload) - - # neighbors berechnen, falls fehlend - for i, p in enumerate(payloads): - nb = p.get("neighbors") or {} - prev_id = nb.get("prev") - next_id = nb.get("next") - if prev_id is None and i > 0: - prev_id = payloads[i-1]["id"] - if next_id is None and i+1 < len(payloads): - next_id = payloads[i+1]["id"] - p["neighbors"] = {"prev": prev_id, "next": next_id} - - return payloads + return out