From f9ac4e4dbff345910c99c3b131366a942a64ebb3 Mon Sep 17 00:00:00 2001 From: Lars Date: Mon, 29 Dec 2025 21:05:42 +0100 Subject: [PATCH] =?UTF-8?q?Verbesserung=20der=20atomaren=20Sektions-Chunki?= =?UTF-8?q?ng-Strategie=20durch=20Einf=C3=BChrung=20strikter=20Look-Ahead-?= =?UTF-8?q?Logik=20und=20pr=C3=A4ventiven=20Flush=20zur=20Gew=C3=A4hrleist?= =?UTF-8?q?ung=20von=20Sektionsgrenzen.=20Anpassungen=20an=20der=20Token-S?= =?UTF-8?q?ch=C3=A4tzung=20und=20Umbenennung=20von=20Funktionen=20zur=20be?= =?UTF-8?q?sseren=20Lesbarkeit.?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/core/chunking/chunking_parser.py | 2 +- app/core/chunking/chunking_strategies.py | 43 ++++++++++++------------ 2 files changed, 22 insertions(+), 23 deletions(-) diff --git a/app/core/chunking/chunking_parser.py b/app/core/chunking/chunking_parser.py index 6bc866d..80807c9 100644 --- a/app/core/chunking/chunking_parser.py +++ b/app/core/chunking/chunking_parser.py @@ -55,7 +55,7 @@ def parse_blocks(md_text: str) -> Tuple[List[RawBlock], str]: current_section_title = title section_path = f"/{current_section_title}" - # Die Überschrift selbst als Block hinzufügen (Fix: H1 wird nicht mehr gefiltert) + # Die Überschrift selbst als Block hinzufügen blocks.append(RawBlock("heading", stripped, level, section_path, current_section_title)) continue diff --git a/app/core/chunking/chunking_strategies.py b/app/core/chunking/chunking_strategies.py index af19d4d..db65478 100644 --- a/app/core/chunking/chunking_strategies.py +++ b/app/core/chunking/chunking_strategies.py @@ -1,7 +1,7 @@ """ FILE: app/core/chunking/chunking_strategies.py DESCRIPTION: Strategien für atomares Sektions-Chunking (WP-15b konform). - Fix: Vorausschauende Trennung zur Wahrung von Sektionsgrenzen. + v3.3.5: Garantiert atomare Sektionsgrenzen durch präventiven Flush. """ from typing import List, Dict, Any, Optional from .chunking_models import RawBlock, Chunk @@ -19,19 +19,20 @@ def _create_context_win(doc_title: str, sec_title: Optional[str], text: str) -> def strategy_by_heading(blocks: List[RawBlock], config: Dict[str, Any], note_id: str, doc_title: str = "") -> List[Chunk]: """ Gruppiert Blöcke zu Sektionen und hält diese atomar zusammen. - Nutzt Look-Ahead, um Sektions-Überhänge zu vermeiden. + Nutzt striktes Look-Ahead, um das Zerschneiden von Sektionsübergängen zu verhindern. """ strict = config.get("strict_heading_split", False) target = config.get("target", 400) max_tokens = config.get("max", 600) split_level = config.get("split_level", 2) - overlap = sum(config.get("overlap", (50, 80))) // 2 + overlap_config = config.get("overlap", (50, 80)) + overlap = sum(overlap_config) // 2 if isinstance(overlap_config, (list, tuple)) else overlap_config chunks: List[Chunk] = [] buf: List[RawBlock] = [] cur_tokens = 0 - def _add_to_chunks(txt, title, path): + def _add_chunk(txt, title, path): idx = len(chunks) win = _create_context_win(doc_title, title, txt) chunks.append(Chunk( @@ -48,23 +49,23 @@ def strategy_by_heading(blocks: List[RawBlock], config: Dict[str, Any], note_id: main_title = buf[0].section_title main_path = buf[0].section_path full_text = "\n\n".join([b.text for b in buf]) + actual_tokens = estimate_tokens(full_text) # Falls die gruppierten Sektionen in das Limit passen - if estimate_tokens(full_text) <= max_tokens: - _add_to_chunks(full_text, main_title, main_path) + if actual_tokens <= max_tokens: + _add_chunk(full_text, main_title, main_path) else: # Nur wenn eine Sektion ALLEINE zu groß ist, wird intern gesplittet sents = split_sentences(full_text) cur_sents = []; sub_len = 0 - # Kontext-Sicherung: Heading für Teil-Chunks merken header_text = buf[0].text if buf[0].kind == "heading" else "" for s in sents: slen = estimate_tokens(s) if sub_len + slen > target and cur_sents: - _add_to_chunks(" ".join(cur_sents), main_title, main_path) + _add_chunk(" ".join(cur_sents), main_title, main_path) - # Overlap-Erzeugung und Header-Injektion für Folgeschritte + # Overlap-Erzeugung und Header-Injektion ov_s = [header_text] if header_text else [] ov_l = estimate_tokens(header_text) if header_text else 0 for os in reversed(cur_sents): @@ -80,7 +81,7 @@ def strategy_by_heading(blocks: List[RawBlock], config: Dict[str, Any], note_id: cur_sents.append(s); sub_len += slen if cur_sents: - _add_to_chunks(" ".join(cur_sents), main_title, main_path) + _add_chunk(" ".join(cur_sents), main_title, main_path) buf = []; cur_tokens = 0 @@ -89,7 +90,6 @@ def strategy_by_heading(blocks: List[RawBlock], config: Dict[str, Any], note_id: curr_sec: List[RawBlock] = [] for b in blocks: - # Ein Split-Trigger (H1 oder H2) startet eine neue atomare Sektion if b.kind == "heading" and b.level <= split_level: if curr_sec: sections.append(curr_sec) curr_sec = [b] @@ -97,26 +97,25 @@ def strategy_by_heading(blocks: List[RawBlock], config: Dict[str, Any], note_id: curr_sec.append(b) if curr_sec: sections.append(curr_sec) - # SCHRITT 2: Verarbeitung der Sektionen mit Vorausschau (Look-Ahead) + # SCHRITT 2: Verarbeitung der Sektionen mit strenger Vorausschau for sec in sections: sec_text = "\n\n".join([b.text for b in sec]) sec_tokens = estimate_tokens(sec_text) if buf: - # VORAUSSCHAU: Würde die neue Sektion das Limit sprengen? - if cur_tokens + sec_tokens > max_tokens: - _flush_buffer() # Beende den aktuellen Chunk sauber VOR der neuen Sektion - # Wenn strict: Jede neue Sektion auf split_level erzwingt neuen Chunk - elif strict and sec[0].kind == "heading" and sec[0].level == split_level: + # PRÜFUNG 1: Passt die gesamte neue Sektion noch in den Chunk (bis max)? + # PRÜFUNG 2: Wenn wir über target sind, fangen wir auf jeden Fall neu an. + if (cur_tokens + sec_tokens > max_tokens) or (cur_tokens >= target): _flush_buffer() - # Wenn target erreicht: Neue Sektion startet neuen Chunk - elif cur_tokens >= target: + # PRÜFUNG 3: Wenn strict-mode aktiv ist und ein split_level erreicht wurde + elif strict and sec[0].kind == "heading" and sec[0].level == split_level: _flush_buffer() buf.extend(sec) - cur_tokens += sec_tokens + # Token-Zähler basierend auf dem tatsächlichen Puffer-Text aktualisieren + cur_tokens = estimate_tokens("\n\n".join([b.text for b in buf])) - # Falls eine einzelne Sektion nach dem Flush (oder als erste) schon zu groß ist + # Falls eine einzelne Sektion (selbst nach flush) schon zu groß ist if cur_tokens >= max_tokens: _flush_buffer() @@ -124,7 +123,7 @@ def strategy_by_heading(blocks: List[RawBlock], config: Dict[str, Any], note_id: return chunks def strategy_sliding_window(blocks: List[RawBlock], config: Dict[str, Any], note_id: str, context_prefix: str = "") -> List[Chunk]: - """Basis-Sliding-Window für flache Texte ohne Sektionsfokus.""" + """Standard-Sliding-Window für flache Texte.""" target = config.get("target", 400) max_tokens = config.get("max", 600)