From 3c67fd5f9b31a71bead12827d7ae6a76abfcf5d3 Mon Sep 17 00:00:00 2001 From: Lars Date: Tue, 30 Sep 2025 12:26:33 +0200 Subject: [PATCH] app/core/chunk_payload.py aktualisiert --- app/core/chunk_payload.py | 94 ++++++++++++++++++--------------------- 1 file changed, 43 insertions(+), 51 deletions(-) diff --git a/app/core/chunk_payload.py b/app/core/chunk_payload.py index 53f53e8..b0c2a76 100644 --- a/app/core/chunk_payload.py +++ b/app/core/chunk_payload.py @@ -2,13 +2,13 @@ # -*- coding: utf-8 -*- """ Modul: app/core/chunk_payload.py -Version: 2.0.0 +Version: 2.0.1 Datum: 2025-09-30 Zweck ----- Erzeugt Chunk-Payloads für Qdrant. Unterstützt abwärtskompatibel bisherige Felder und -ergänzt neue Felder für **verlustfreie Rekonstruktion** bei überlappenden Fenstern: +ergänzt Felder für **verlustfreie Rekonstruktion** bei überlappenden Fenstern: - text : effektiver, nicht-überlappender Segmenttext (für Rekonstruktion) - window : Fenstertext inkl. Overlap (für Embeddings) @@ -16,25 +16,22 @@ ergänzt neue Felder für **verlustfreie Rekonstruktion** bei überlappenden Fen - overlap_left : Anzahl überlappender Zeichen zum **vorigen** Fenster - overlap_right : Anzahl überlappender Zeichen zum **nächsten** Fenster -Abwärtskompatibel bleiben: - - chunk_id (note_id#), chunk_index, seq, path, note_id, type, title, tags, etc. +Abwärtskompatible Aliasse: + - id : == chunk_id (wird u. a. von build_edges_for_note erwartet) + - content/raw : bleiben leer; Primärfelder sind window/text -Aufruf (typisch aus dem Importer) ---------------------------------- +Typische Nutzung +---------------- from app.core.chunk_payload import make_chunk_payloads payloads = make_chunk_payloads(frontmatter, rel_path, chunks, note_text=full_body) -Wobei `chunks` eine Folge von Objekten oder Dicts ist, die mindestens ein Fenster enthalten: - c.text ODER c.content ODER c.raw (falls als Objekt) +`chunks` ist eine Sequenz von Objekten oder Dicts, die mindestens ein Fenster enthalten: + c.text ODER c.content ODER c.raw (falls Objekt) bzw. c["text"] ODER c["content"] ODER c["raw"] (falls Dict) - -Falls `note_text` nicht übergeben wird, wird die effektive Segmentierung über -eine robuste **Overlap-Deduplikation** zwischen Fenstern ermittelt. """ from __future__ import annotations from typing import Any, Dict, Iterable, List, Optional, Tuple, Union -import re # ------------------------------- Utils ------------------------------- # @@ -43,10 +40,16 @@ def _as_text(window_candidate: Any) -> str: if window_candidate is None: return "" # Objekt mit Attributen - for k in ("text", "content", "raw", "window"): - v = getattr(window_candidate, k, None) if not isinstance(window_candidate, dict) else window_candidate.get(k) - if isinstance(v, str) and v: - return v + if not isinstance(window_candidate, dict): + for k in ("window", "text", "content", "raw"): + v = getattr(window_candidate, k, None) + if isinstance(v, str) and v: + return v + else: + for k in ("window", "text", "content", "raw"): + v = window_candidate.get(k) + if isinstance(v, str) and v: + return v # Fallback: string-repr if isinstance(window_candidate, str): return window_candidate @@ -58,10 +61,6 @@ def _get_int(x: Any, default: int = 0) -> int: except Exception: return default -def _norm_lines(s: str) -> str: - """Nur für defensive Gleichheitstests – NICHT für Persistenz.""" - return "\n".join([ln.rstrip() for ln in s.replace("\r\n", "\n").replace("\r", "\n").split("\n")]).strip() - # ---------------------- Overlap-Dedupe Algorithmus ------------------- # def _dedupe_windows_to_segments(windows: List[str]) -> Tuple[List[str], List[int]]: @@ -76,10 +75,9 @@ def _dedupe_windows_to_segments(windows: List[str]) -> Tuple[List[str], List[int reconstructed = "" for w in windows: w = w or "" - # finde größtes k, sodass reconstructed.endswith(w[:k]) - max_k = min(len(w), max(0, len(reconstructed))) + max_k = min(len(w), len(reconstructed)) k = 0 - # Suche von groß nach klein (einfache O(n^2) – ausreichend bei kurzen Fenstern) + # Suche von groß nach klein (einfach, ausreichend bei kurzen Fenstern) for cand in range(max_k, -1, -1): if reconstructed.endswith(w[:cand]): k = cand @@ -111,21 +109,19 @@ def make_chunk_payloads( Rückgabe -------- Liste von Payload-Dicts. Wichtige Felder: - note_id, chunk_id, chunk_index, seq, path, text, window, start, end, - overlap_left, overlap_right, type, title, tags + note_id, id, chunk_id, chunk_index, seq, path, text, window, + start, end, overlap_left, overlap_right, type, title, tags """ note_id = str(frontmatter.get("id") or "").strip() note_type = frontmatter.get("type", None) note_title = frontmatter.get("title", None) note_tags = frontmatter.get("tags", None) - # 1) Fenstertexte extrahieren + # 1) Fenstertexte + Sequenzen extrahieren windows: List[str] = [] seqs: List[int] = [] for idx, c in enumerate(chunks): windows.append(_as_text(c)) - # Bestmögliche seq ermitteln - s = None if isinstance(c, dict): s = c.get("seq", c.get("chunk_index", idx)) else: @@ -134,41 +130,35 @@ def make_chunk_payloads( # 2) Nicht-überlappende Segmente berechnen segments, overlaps_left = _dedupe_windows_to_segments(windows) - overlaps_right = [0] * len(segments) - # right-overlap ist der left-overlap des nächsten Fensters bezogen auf dessen Fenster, - # lässt sich nur approximieren; wir speichern ihn konsistent als 0 bzw. könnte man - # nachträglich bestimmen, falls benötigt. + overlaps_right = [0] * len(segments) # optional: später präzisieren - # 3) Falls note_text gegeben ist, berechne absolute Offsets präzise + # 3) Offsets bestimmen (ohne/mit note_text gleich: kumulativ) starts: List[int] = [0] * len(segments) ends: List[int] = [0] * len(segments) - if isinstance(note_text, str): - pos = 0 - for i, seg in enumerate(segments): - starts[i] = pos - pos += len(seg) - ends[i] = pos - else: - # Ohne Gesamtkorpus: Offsets anhand der kumulativen Segmentlängen - pos = 0 - for i, seg in enumerate(segments): - starts[i] = pos - pos += len(seg) - ends[i] = pos + pos = 0 + for i, seg in enumerate(segments): + starts[i] = pos + pos += len(seg) + ends[i] = pos - # 4) Payload-Dicts aufbauen + # 4) Payload-Dicts zusammenstellen payloads: List[Dict[str, Any]] = [] for i, (win, seg) in enumerate(zip(windows, segments)): + chunk_id = f"{note_id}#{i+1}" pl: Dict[str, Any] = { + # Identität "note_id": note_id, - "chunk_id": f"{note_id}#{i+1}", + "chunk_id": chunk_id, + "id": chunk_id, # <— WICHTIG: Alias für Abwärtskompatibilität (Edges erwarten 'id') + + # Indexierung "chunk_index": i, "seq": seqs[i], "path": rel_path.replace("\\", "/").lstrip("/"), # Texte - "window": win, # für Embeddings (inkl. Overlap) - "text": seg, # effektiver Anteil (verlustfreie Rekonstruktion) + "window": win, # für Embeddings (inkl. Overlap) + "text": seg, # überlappungsfreier Anteil für exakte Rekonstruktion # Offsets & Overlaps "start": starts[i], @@ -182,11 +172,13 @@ def make_chunk_payloads( pl["title"] = note_title if note_tags is not None: pl["tags"] = note_tags + payloads.append(pl) return payloads -# __main__ (optionaler Mini-Test) + +# __main__ (optional: Mini-Demo) if __name__ == "__main__": # pragma: no cover demo_fm = {"id": "demo", "title": "Demo", "type": "concept"} demo_chunks = [