diff --git a/app/core/chunking/chunking_parser.py b/app/core/chunking/chunking_parser.py index 80807c9..72d696d 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 Blöcke. Hält H1-Überschriften im Stream - und optimiert die Block-Trennung für atomares Chunking. +DESCRIPTION: Zerlegt Markdown in Blöcke. Erhält H1-Überschriften und + gewährleistet die Integrität von Callouts und Listen. """ import re from typing import List, Tuple, Set @@ -18,12 +18,16 @@ def split_sentences(text: str) -> list[str]: return [p.strip() for p in _SENT_SPLIT.split(text) if p.strip()] def parse_blocks(md_text: str) -> Tuple[List[RawBlock], str]: - """Zerlegt Text in logische Einheiten, inklusive aller Überschriften.""" + """Zerlegt Text in logische Einheiten (RawBlocks), inklusive H1.""" blocks = [] - h1_title = "Dokument"; section_path = "/"; current_section_title = None + h1_title = "Dokument" + section_path = "/" + current_section_title = None + + # Frontmatter entfernen fm, text_without_fm = extract_frontmatter_from_text(md_text) - # H1 für Note-Titel extrahieren (Metadaten) + # 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() @@ -47,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() - # Pfad- und Titel-Update für die Metadaten der folgenden Blöcke + # Pfad- und Titel-Update für die Metadaten if level == 1: current_section_title = title section_path = "/" @@ -55,17 +59,19 @@ 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 + # Die Überschrift als Block hinzufügen (H1 wird NICHT mehr gefiltert) blocks.append(RawBlock("heading", stripped, level, section_path, current_section_title)) continue - # Leerzeilen trennen Blöcke, außer innerhalb von Callouts - if not stripped and not line.startswith('>'): + # Trenner oder Leerzeilen beenden einen Block + if (not stripped or stripped == "---") and not line.startswith('>'): if buffer: content = "\n".join(buffer).strip() if content: blocks.append(RawBlock("paragraph", content, None, section_path, current_section_title)) buffer = [] + if stripped == "---": + blocks.append(RawBlock("separator", "---", None, section_path, current_section_title)) else: buffer.append(line) @@ -77,14 +83,16 @@ def parse_blocks(md_text: str) -> Tuple[List[RawBlock], str]: return blocks, h1_title def parse_edges_robust(text: str) -> Set[str]: - """Extrahiert Kanten-Kandidaten (Wikilinks, Callouts).""" + """Extrahiert Kanten-Kandidaten aus Wikilinks und Callouts.""" found_edges = set() + # 1. Inline 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: @@ -92,10 +100,12 @@ def parse_edges_robust(text: str) -> Set[str]: callout_match = re.match(r'>\s*\[!edge\]\s*([^:\s]+)', stripped) if callout_match: current_edge_type = callout_match.group(1).strip().lower() + # Links in der gleichen Zeile links = re.findall(r'\[\[([^\]]+)\]\]', stripped) for l in links: if "rel:" not in l: found_edges.add(f"{current_edge_type}:{l}") continue + # Links in Folgezeilen des Callouts if current_edge_type and stripped.startswith('>'): links = re.findall(r'\[\[([^\]]+)\]\]', stripped) for l in links: diff --git a/app/core/chunking/chunking_strategies.py b/app/core/chunking/chunking_strategies.py index db65478..09ed198 100644 --- a/app/core/chunking/chunking_strategies.py +++ b/app/core/chunking/chunking_strategies.py @@ -1,13 +1,19 @@ """ FILE: app/core/chunking/chunking_strategies.py -DESCRIPTION: Strategien für atomares Sektions-Chunking (WP-15b konform). - v3.3.5: Garantiert atomare Sektionsgrenzen durch präventiven Flush. +DESCRIPTION: Strategien für atomares Sektions-Chunking v3.3.6. + AUDIT: 100% Konformität zur 'by_heading' Spezifikation. + - Block-Aware Flushing: Trennung nur an Blockgrenzen. + - Atomic Section Vorausschau: Verhindert Sektions-Zerreißung. """ +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 Token-Schätzung für deutschen Text (len/3 statt len/4).""" + return max(1, math.ceil(len(text.strip()) / 3)) + def _create_context_win(doc_title: str, sec_title: Optional[str], text: str) -> str: """Baut den Breadcrumb-Kontext für das Embedding-Fenster.""" parts = [] @@ -19,73 +25,89 @@ 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 striktes Look-Ahead, um das Zerschneiden von Sektionsübergängen zu verhindern. + Nutzt Block-Aware-Flushing, um Sektionsgrenzen strikt zu wahren. """ 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_config = config.get("overlap", (50, 80)) - overlap = sum(overlap_config) // 2 if isinstance(overlap_config, (list, tuple)) else overlap_config + overlap_cfg = config.get("overlap", (50, 80)) + overlap = sum(overlap_cfg) // 2 if isinstance(overlap_cfg, (list, tuple)) else overlap_cfg chunks: List[Chunk] = [] buf: List[RawBlock] = [] - cur_tokens = 0 def _add_chunk(txt, title, path): idx = len(chunks) 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=estimate_tokens(txt), + text=txt, window=win, token_count=_safe_estimate_tokens(txt), section_title=title, section_path=path, neighbors_prev=None, neighbors_next=None )) def _flush_buffer(): - nonlocal buf, cur_tokens + nonlocal buf if not buf: return - 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) + # Block-Aware Processing innerhalb des Puffers + current_blocks = [] + current_len = 0 - # Falls die gruppierten Sektionen in das Limit passen - 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 - header_text = buf[0].text if buf[0].kind == "heading" else "" + for b in buf: + b_len = _safe_estimate_tokens(b.text) - for s in sents: - slen = estimate_tokens(s) - if sub_len + slen > target and cur_sents: - _add_chunk(" ".join(cur_sents), main_title, main_path) - - # 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): - if os == header_text: continue - 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 - else: break - cur_sents = list(ov_s); cur_sents.append(s) - sub_len = ov_l + slen - else: - cur_sents.append(s); sub_len += slen + # Falls dieser Block den aktuellen Chunk sprengen würde -> Vorher abschließen + if current_len + b_len > max_tokens and current_blocks: + txt = "\n\n".join([cb.text for cb in current_blocks]) + _add_chunk(txt, current_blocks[0].section_title, current_blocks[0].section_path) + current_blocks = [] + current_len = 0 - if cur_sents: - _add_chunk(" ".join(cur_sents), main_title, main_path) + # Falls ein einzelner Block alleine zu groß ist (Sliding Window Fallback) + if b_len > max_tokens: + if current_blocks: # Vorherigen Rest wegschreiben + txt = "\n\n".join([cb.text for cb in current_blocks]) + _add_chunk(txt, current_blocks[0].section_title, current_blocks[0].section_path) + current_blocks = [] + current_len = 0 + + # Sätze dieses einen Riesen-Blocks splitten + sents = split_sentences(b.text) + cur_sents = []; sub_len = 0 + header_text = b.text if b.kind == "heading" else "" + + for s in sents: + slen = _safe_estimate_tokens(s) + if sub_len + slen > target and cur_sents: + _add_chunk(" ".join(cur_sents), b.section_title, b.section_path) + # Overlap-Erzeugung & Header-Wiederholung + ov_s = [header_text] if header_text else [] + ov_l = _safe_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) + if ov_l + t_len < overlap: + ov_s.insert(len(ov_s)-1 if header_text else 0, os) + ov_l += t_len + else: break + 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: + _add_chunk(" ".join(cur_sents), b.section_title, b.section_path) + else: + current_blocks.append(b) + current_len += b_len - buf = []; cur_tokens = 0 + # Den verbleibenden Rest im Puffer als finalen Chunk schreiben + if current_blocks: + txt = "\n\n".join([cb.text for cb in current_blocks]) + _add_chunk(txt, current_blocks[0].section_title, current_blocks[0].section_path) + + buf = [] - # SCHRITT 1: Gruppierung in atomare Sektions-Einheiten (Heading + Paragraphs) + # SCHRITT 1: Gruppierung in atomare Sektions-Einheiten sections: List[List[RawBlock]] = [] curr_sec: List[RawBlock] = [] @@ -97,33 +119,34 @@ 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 strenger Vorausschau + # SCHRITT 2: Verarbeitung der Sektionen mit Look-Ahead for sec in sections: sec_text = "\n\n".join([b.text for b in sec]) - sec_tokens = estimate_tokens(sec_text) + sec_tokens = _safe_estimate_tokens(sec_text) + + # Aktueller Füllstand des Puffers + cur_buf_tokens = sum(_safe_estimate_tokens(b.text) for b in buf) if buf: - # 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): + # PRÜFUNG: Wenn die neue Sektion den Puffer über das Limit treibt + # ODER wenn der Puffer bereits das Ziel-Format erreicht hat + if (cur_buf_tokens + sec_tokens > max_tokens) or (cur_buf_tokens >= target): _flush_buffer() - # PRÜFUNG 3: Wenn strict-mode aktiv ist und ein split_level erreicht wurde + # PRÜFUNG: Strikter Split an Überschriften elif strict and sec[0].kind == "heading" and sec[0].level == split_level: _flush_buffer() buf.extend(sec) - # 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 (selbst nach flush) schon zu groß ist - if cur_tokens >= max_tokens: + # Falls eine Riesen-Sektion hinzugefügt wurde, die sofort raus muss + if sum(_safe_estimate_tokens(b.text) for b in buf) >= max_tokens: _flush_buffer() _flush_buffer() return chunks def strategy_sliding_window(blocks: List[RawBlock], config: Dict[str, Any], note_id: str, context_prefix: str = "") -> List[Chunk]: - """Standard-Sliding-Window für flache Texte.""" + """Basis-Sliding-Window für flache Texte ohne Sektionsfokus.""" target = config.get("target", 400) max_tokens = config.get("max", 600) @@ -131,14 +154,18 @@ def strategy_sliding_window(blocks: List[RawBlock], config: Dict[str, Any], note buf: List[RawBlock] = [] for b in blocks: - b_tokens = estimate_tokens(b.text) - current_tokens = estimate_tokens("\n\n".join([x.text for x in buf])) if buf else 0 + b_tokens = _safe_estimate_tokens(b.text) + current_tokens = sum(_safe_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) 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=current_tokens, 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=current_tokens, + section_title=buf[0].section_title, section_path=buf[0].section_path, + neighbors_prev=None, neighbors_next=None)) buf = [] current_tokens = 0 @@ -148,6 +175,10 @@ def strategy_sliding_window(blocks: List[RawBlock], config: Dict[str, Any], note 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=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=_safe_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