From 96b4f65cd1dbf74908642d7ddd6b6040614f57a7 Mon Sep 17 00:00:00 2001 From: Lars Date: Mon, 29 Dec 2025 21:37:11 +0100 Subject: [PATCH] =?UTF-8?q?Aktualisierung=20des=20Chunking-Parsers=20zur?= =?UTF-8?q?=20Verbesserung=20der=20Blockverarbeitung=20und=20Beschreibung.?= =?UTF-8?q?=20Anpassungen=20an=20der=20atomaren=20Sektions-Chunking-Strate?= =?UTF-8?q?gie=20mit=20optimierter=20Token-Sch=C3=A4tzung=20und=20neuen=20?= =?UTF-8?q?Hilfsfunktionen=20zur=20besseren=20Handhabung=20von=20gro=C3=9F?= =?UTF-8?q?en=20Sektionen.?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/core/chunking/chunking_parser.py | 18 ++-- app/core/chunking/chunking_strategies.py | 105 +++++++++++------------ 2 files changed, 55 insertions(+), 68 deletions(-) diff --git a/app/core/chunking/chunking_parser.py b/app/core/chunking/chunking_parser.py index ca67598..bf3801a 100644 --- a/app/core/chunking/chunking_parser.py +++ b/app/core/chunking/chunking_parser.py @@ -1,7 +1,7 @@ """ FILE: app/core/chunking/chunking_parser.py -DESCRIPTION: Zerlegt Markdown in logische Blöcke. Hält H1-Überschriften im Stream - und optimiert die Block-Trennung für atomares Sektions-Chunking. +DESCRIPTION: Zerlegt Markdown in logische Einheiten (RawBlocks). + Hält H1-Überschriften im Inhalts-Stream. """ import re from typing import List, Tuple, Set @@ -15,7 +15,6 @@ def split_sentences(text: str) -> list[str]: """Teilt Text in Sätze auf unter Berücksichtigung deutscher Interpunktion.""" text = _WS.sub(' ', text.strip()) if not text: return [] - # Splittet bei Satzzeichen, gefolgt von Leerzeichen und Großbuchstaben return [p.strip() for p in _SENT_SPLIT.split(text) if p.strip()] def parse_blocks(md_text: str) -> Tuple[List[RawBlock], str]: @@ -28,7 +27,7 @@ def parse_blocks(md_text: str) -> Tuple[List[RawBlock], str]: # Frontmatter entfernen fm, text_without_fm = extract_frontmatter_from_text(md_text) - # H1 für Note-Metadaten extrahieren + # H1 für Note-Titel extrahieren h1_match = re.search(r'^#\s+(.*)', text_without_fm, re.MULTILINE) if h1_match: h1_title = h1_match.group(1).strip() @@ -42,7 +41,7 @@ def parse_blocks(md_text: str) -> Tuple[List[RawBlock], str]: # Heading-Erkennung (H1 bis H6) heading_match = re.match(r'^(#{1,6})\s+(.*)', stripped) if heading_match: - # Vorherigen Block abschließen + # Vorherigen Text-Block abschließen if buffer: content = "\n".join(buffer).strip() if content: @@ -52,7 +51,7 @@ def parse_blocks(md_text: str) -> Tuple[List[RawBlock], str]: level = len(heading_match.group(1)) title = heading_match.group(2).strip() - # Update der Pfad-Metadaten für die folgenden Blöcke + # Pfad- und Titel-Update für die Metadaten der folgenden Blöcke if level == 1: current_section_title = title section_path = "/" @@ -60,11 +59,11 @@ def parse_blocks(md_text: str) -> Tuple[List[RawBlock], str]: current_section_title = title section_path = f"/{current_section_title}" - # Die Überschrift als regulären Block hinzufügen (Fix: H1 bleibt im Text) + # Die Überschrift selbst als Block hinzufügen blocks.append(RawBlock("heading", stripped, level, section_path, current_section_title)) continue - # Trenner (---) oder Leerzeilen beenden Blöcke, außer in Callouts + # Trenner oder Leerzeilen beenden Blöcke, außer innerhalb von Callouts if (not stripped or stripped == "---") and not line.startswith('>'): if buffer: content = "\n".join(buffer).strip() @@ -76,7 +75,6 @@ def parse_blocks(md_text: str) -> Tuple[List[RawBlock], str]: else: buffer.append(line) - # Letzten Puffer leeren if buffer: content = "\n".join(buffer).strip() if content: @@ -87,14 +85,12 @@ def parse_blocks(md_text: str) -> Tuple[List[RawBlock], str]: def parse_edges_robust(text: str) -> Set[str]: """Extrahiert Kanten-Kandidaten aus Wikilinks und Callouts.""" found_edges = set() - # 1. Wikilinks [[rel:kind|target]] inlines = re.findall(r'\[\[rel:([^\|\]]+)\|?([^\]]*)\]\]', text) for kind, target in inlines: k = kind.strip().lower() t = target.strip() if k and t: found_edges.add(f"{k}:{t}") - # 2. Callout Edges > [!edge] kind lines = text.split('\n') current_edge_type = None for line in lines: diff --git a/app/core/chunking/chunking_strategies.py b/app/core/chunking/chunking_strategies.py index 9e2943a..fa22382 100644 --- a/app/core/chunking/chunking_strategies.py +++ b/app/core/chunking/chunking_strategies.py @@ -1,20 +1,15 @@ """ FILE: app/core/chunking/chunking_strategies.py -DESCRIPTION: Strategien für atomares Sektions-Chunking v3.4.1. - Garantiert Sektions-Integrität (Atomic Units) durch Look-Ahead. +DESCRIPTION: Universelle Strategie für atomares Sektions-Chunking v3.5.0. + Garantiert Sektions-Integrität durch präventives Chunk-Management. """ -import math from typing import List, Dict, Any, Optional from .chunking_models import RawBlock, Chunk from .chunking_utils import estimate_tokens from .chunking_parser import split_sentences -def _safe_estimate_tokens(text: str) -> int: - """Konservative Schätzung für MD und deutsche Texte (len/2.8).""" - return max(1, math.ceil(len(text.strip()) / 2.8)) - def _create_context_win(doc_title: str, sec_title: Optional[str], text: str) -> str: - """Baut den Breadcrumb-Kontext für das Embedding-Fenster (H1 > H2).""" + """Baut den Breadcrumb-Kontext für das Embedding-Fenster.""" parts = [] if doc_title: parts.append(doc_title) if sec_title and sec_title != doc_title: parts.append(sec_title) @@ -23,10 +18,9 @@ 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]: """ - Sektions-Chunking: Behandelt Abschnitte als unteilbare Einheiten. - Schiebt ganze Abschnitte in den nächsten Chunk, falls das Limit erreicht ist. + Universelles Sektions-Chunking: Packt Sektionen in Chunks. + Bei Überlauf wird die komplette Sektion in den nächsten Chunk geschoben. """ - 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) @@ -35,24 +29,23 @@ def strategy_by_heading(blocks: List[RawBlock], config: Dict[str, Any], note_id: chunks: List[Chunk] = [] - def _emit_chunk(block_list: List[RawBlock]): - """Erzeugt ein finales Chunk-Objekt aus einer Liste von Blöcken.""" + # --- HILFSFUNKTION: Erzeugt einen Chunk aus einer Blockliste --- + def _create_chunk_from_blocks(block_list: List[RawBlock]): if not block_list: return txt = "\n\n".join([b.text for b in block_list]) idx = len(chunks) - # Metadaten vom ersten Block der Gruppe (Header) title = block_list[0].section_title path = block_list[0].section_path win = _create_context_win(doc_title, title, txt) chunks.append(Chunk( id=f"{note_id}#c{idx:02d}", note_id=note_id, index=idx, - text=txt, window=win, token_count=_safe_estimate_tokens(txt), + text=txt, window=win, token_count=estimate_tokens(txt), section_title=title, section_path=path, neighbors_prev=None, neighbors_next=None )) - def _emit_split_section(sec_blocks: List[RawBlock]): - """Splittet eine einzelne Sektion, die für sich allein zu groß ist.""" + # --- HILFSFUNKTION: Splittet eine einzelne Sektion, die > max ist --- + def _split_giant_section(sec_blocks: List[RawBlock]): full_text = "\n\n".join([b.text for b in sec_blocks]) main_title = sec_blocks[0].section_title main_path = sec_blocks[0].section_path @@ -62,15 +55,17 @@ def strategy_by_heading(blocks: List[RawBlock], config: Dict[str, Any], note_id: cur_sents = []; sub_len = 0 for s in sents: - slen = _safe_estimate_tokens(s) + slen = estimate_tokens(s) if sub_len + slen > target and cur_sents: - _emit_chunk([RawBlock("paragraph", " ".join(cur_sents), None, main_path, main_title)]) - # Header Injection für den Kontext im nächsten Teil-Chunk + combined_text = " ".join(cur_sents) + _create_chunk_from_blocks([RawBlock("paragraph", combined_text, None, main_path, main_title)]) + + # Context Injection: Überschrift für den nächsten Teil-Chunk ov_s = [header_text] if header_text else [] - ov_l = _safe_estimate_tokens(header_text) if header_text else 0 + ov_l = estimate_tokens(header_text) if header_text else 0 for os in reversed(cur_sents): if os == header_text: continue - t_len = _safe_estimate_tokens(os) + t_len = estimate_tokens(os) if ov_l + t_len < overlap: ov_s.insert(len(ov_s)-1 if header_text else 0, os) ov_l += t_len @@ -78,13 +73,15 @@ def strategy_by_heading(blocks: List[RawBlock], config: Dict[str, Any], note_id: cur_sents = list(ov_s); cur_sents.append(s); sub_len = ov_l + slen else: cur_sents.append(s); sub_len += slen + if cur_sents: - _emit_chunk([RawBlock("paragraph", " ".join(cur_sents), None, main_path, main_title)]) + _create_chunk_from_blocks([RawBlock("paragraph", " ".join(cur_sents), None, main_path, main_title)]) - # SCHRITT 1: Gruppierung in atomare Einheiten (Sektionen) + # 1. SCHRITT: Gruppierung in atomare Sektions-Einheiten sections: List[List[RawBlock]] = [] curr_sec: List[RawBlock] = [] for b in blocks: + # Eine neue Überschrift auf oder unter dem split_level startet eine neue Sektion if b.kind == "heading" and b.level <= split_level: if curr_sec: sections.append(curr_sec) curr_sec = [b] @@ -92,54 +89,48 @@ 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 strengem Look-Ahead - current_chunk_buf = [] - current_tokens = 0 + # 2. SCHRITT: Sektionen in Chunks packen (Das universelle Pack-Verfahren) + candidate_chunk: List[RawBlock] = [] + candidate_tokens = 0 for sec in sections: sec_text = "\n\n".join([b.text for b in sec]) - sec_tokens = _safe_estimate_tokens(sec_text) + sec_tokens = estimate_tokens(sec_text) - if current_chunk_buf: - # PRÜFUNG: Würde die neue Sektion den aktuellen Chunk sprengen? - # ODER: Haben wir das Target bereits erreicht und fangen lieber neu an? - if (current_tokens + sec_tokens > max_tokens) or (current_tokens >= target): - _emit_chunk(current_chunk_buf) - current_chunk_buf = [] - current_tokens = 0 - # PRÜFUNG: Harter Split gefordert an Überschriften - elif strict and sec[0].kind == "heading" and sec[0].level == split_level: - _emit_chunk(current_chunk_buf) - current_chunk_buf = [] - current_tokens = 0 - - # Wenn eine EINZELNE Sektion alleine schon das Limit sprengt - if sec_tokens > max_tokens: - if current_chunk_buf: - _emit_chunk(current_chunk_buf) - current_chunk_buf = [] - current_tokens = 0 - _emit_split_section(sec) + # Passt diese gesamte Sektion noch in den laufenden Chunk? + if candidate_tokens + sec_tokens <= max_tokens: + candidate_chunk.extend(sec) + candidate_tokens = estimate_tokens("\n\n".join([b.text for b in candidate_chunk])) else: - current_chunk_buf.extend(sec) - current_tokens += sec_tokens + 2 # +2 für Newline Join + # Falls der aktuelle Chunk nicht leer ist: Raus damit, bevor die neue Sektion kommt + if candidate_chunk: + _create_chunk_from_blocks(candidate_chunk) + candidate_chunk = [] + candidate_tokens = 0 + + # Die neue Sektion ist nun allein. Ist sie selbst zu groß? + if sec_tokens > max_tokens: + _split_giant_section(sec) + else: + candidate_chunk = list(sec) + candidate_tokens = sec_tokens - # Letzten Puffer schreiben - if current_chunk_buf: - _emit_chunk(current_chunk_buf) + # Letzten Rest wegschreiben + if candidate_chunk: + _create_chunk_from_blocks(candidate_chunk) 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 ohne Sektionsfokus.""" target = config.get("target", 400) max_tokens = config.get("max", 600) chunks: List[Chunk] = [] buf: List[RawBlock] = [] for b in blocks: - b_tokens = _safe_estimate_tokens(b.text) - current_tokens = sum(_safe_estimate_tokens(x.text) for x in buf) if buf else 0 + b_tokens = estimate_tokens(b.text) + current_tokens = sum(estimate_tokens(x.text) for x in buf) if buf else 0 if current_tokens + b_tokens > max_tokens and buf: txt = "\n\n".join([x.text for x in buf]) idx = len(chunks) @@ -150,5 +141,5 @@ def strategy_sliding_window(blocks: List[RawBlock], config: Dict[str, Any], note if buf: txt = "\n\n".join([x.text for x in buf]); idx = len(chunks) win = f"{context_prefix}\n{txt}".strip() if context_prefix else txt - chunks.append(Chunk(id=f"{note_id}#c{idx:02d}", note_id=note_id, index=idx, text=txt, window=win, token_count=_safe_estimate_tokens(txt), section_title=buf[0].section_title, section_path=buf[0].section_path, neighbors_prev=None, neighbors_next=None)) + chunks.append(Chunk(id=f"{note_id}#c{idx:02d}", note_id=note_id, index=idx, text=txt, window=win, token_count=estimate_tokens(txt), section_title=buf[0].section_title, section_path=buf[0].section_path, neighbors_prev=None, neighbors_next=None)) return chunks \ No newline at end of file