diff --git a/app/core/chunking/chunking_strategies.py b/app/core/chunking/chunking_strategies.py index 8945fee..5e09512 100644 --- a/app/core/chunking/chunking_strategies.py +++ b/app/core/chunking/chunking_strategies.py @@ -1,10 +1,9 @@ """ FILE: app/core/chunking/chunking_strategies.py -DESCRIPTION: Mathematische Splitting-Strategien. - AUDIT v3.3.2: 100% Konformität zur 'by_heading' Spezifikation. - - Implementiert Hybrid-Safety-Net (Sliding Window für Übergrößen). - - Breadcrumb-Kontext im Window (H1 > H2). - - Sliding Window mit H1-Kontext (Gold-Standard v3.1.0). +DESCRIPTION: Korrigierte Splitting-Strategien für Mindnet v3.3.3. + - Fix: Erhalt von Überschriften im Chunk-Text. + - Fix: Atomares Buffering (Blöcke fallen als Ganzes in den nächsten Chunk). + - Fix: Korrekte Zuordnung von Sektions-Metadaten. """ from typing import List, Dict, Any, Optional from .chunking_models import RawBlock, Chunk @@ -19,71 +18,9 @@ def _create_context_win(doc_title: str, sec_title: Optional[str], text: str) -> prefix = " > ".join(parts) return f"{prefix}\n{text}".strip() if prefix else text -def strategy_sliding_window(blocks: List[RawBlock], - config: Dict[str, Any], - note_id: str, - context_prefix: str = "") -> List[Chunk]: - """ - Fasst Blöcke zusammen und schneidet bei 'target' Tokens. - Ignoriert H2-Überschriften beim Splitting, um Kontext zu wahren. - """ - target = config.get("target", 400) - max_tokens = config.get("max", 600) - overlap_val = config.get("overlap", (50, 80)) - overlap = sum(overlap_val) // 2 if isinstance(overlap_val, tuple) else overlap_val - - chunks: List[Chunk] = [] - buf: List[RawBlock] = [] - - def _add(txt, sec, path): - idx = len(chunks) - # H1-Kontext Präfix für das Window-Feld - 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=sec, section_path=path, - neighbors_prev=None, neighbors_next=None - )) - - def flush(): - nonlocal buf - if not buf: return - text_body = "\n\n".join([b.text for b in buf]) - sec_title = buf[-1].section_title; sec_path = buf[-1].section_path - - if estimate_tokens(text_body) <= max_tokens: - _add(text_body, sec_title, sec_path) - else: - sents = split_sentences(text_body); cur_sents = []; cur_len = 0 - for s in sents: - slen = estimate_tokens(s) - if cur_len + slen > target and cur_sents: - _add(" ".join(cur_sents), sec_title, sec_path) - ov_s = []; ov_l = 0 - for os in reversed(cur_sents): - if ov_l + estimate_tokens(os) < overlap: - ov_s.insert(0, os); ov_l += estimate_tokens(os) - else: break - cur_sents = list(ov_s); cur_sents.append(s); cur_len = ov_l + slen - else: - cur_sents.append(s); cur_len += slen - if cur_sents: - _add(" ".join(cur_sents), sec_title, sec_path) - buf = [] - - for b in blocks: - # H2-Überschriften werden ignoriert, um den Zusammenhang zu wahren - if b.kind == "heading": continue - if estimate_tokens("\n\n".join([x.text for x in buf])) + estimate_tokens(b.text) >= target: - flush() - buf.append(b) - flush() - return chunks - def strategy_by_heading(blocks: List[RawBlock], config: Dict[str, Any], note_id: str, doc_title: str = "") -> List[Chunk]: """ - Splittet Text basierend auf Markdown-Überschriften mit Hybrid-Safety-Net. + Splittet Text basierend auf Markdown-Überschriften mit atomarem Block-Erhalt. """ strict = config.get("strict_heading_split", False) target = config.get("target", 400) @@ -92,7 +29,7 @@ def strategy_by_heading(blocks: List[RawBlock], config: Dict[str, Any], note_id: overlap = sum(config.get("overlap", (50, 80))) // 2 chunks: List[Chunk] = [] - buf: List[str] = [] + buf: List[RawBlock] = [] cur_tokens = 0 def _add_to_chunks(txt, title, path): @@ -105,18 +42,26 @@ def strategy_by_heading(blocks: List[RawBlock], config: Dict[str, Any], note_id: neighbors_prev=None, neighbors_next=None )) - def _flush(title, path): + def _flush(): nonlocal buf, cur_tokens if not buf: return - full_text = "\n\n".join(buf) + + # Metadaten stammen immer vom ersten Block im Puffer (meist die Überschrift) + main_title = buf[0].section_title + main_path = buf[0].section_path + full_text = "\n\n".join([b.text for b in buf]) + + # Falls der gesamte Puffer in einen Chunk passt if estimate_tokens(full_text) <= max_tokens: - _add_to_chunks(full_text, title, path) + _add_to_chunks(full_text, main_title, main_path) else: - sents = split_sentences(full_text); cur_sents = []; sub_len = 0 + # Nur wenn ein einzelner Abschnitt größer als 'max' ist, wird intern gesplittet + sents = split_sentences(full_text) + cur_sents = []; sub_len = 0 for s in sents: slen = estimate_tokens(s) if sub_len + slen > target and cur_sents: - _add_to_chunks(" ".join(cur_sents), title, path) + _add_to_chunks(" ".join(cur_sents), main_title, main_path) ov_s = []; ov_l = 0 for os in reversed(cur_sents): if ov_l + estimate_tokens(os) < overlap: @@ -124,19 +69,74 @@ def strategy_by_heading(blocks: List[RawBlock], config: Dict[str, Any], note_id: 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_to_chunks(" ".join(cur_sents), title, path) + if cur_sents: _add_to_chunks(" ".join(cur_sents), main_title, main_path) + buf = []; cur_tokens = 0 for b in blocks: + b_tokens = estimate_tokens(b.text) + + # Prüfung auf Split-Trigger (Überschriften) + is_split_trigger = False if b.kind == "heading": - if b.level < split_level: _flush(b.section_title, b.section_path) + if b.level < split_level: + is_split_trigger = True elif b.level == split_level: - if strict or cur_tokens >= target: _flush(b.section_title, b.section_path) - continue - bt = estimate_tokens(b.text) - if cur_tokens + bt > max_tokens and buf: _flush(b.section_title, b.section_path) - buf.append(b.text); cur_tokens += bt - if buf: - last_b = blocks[-1] if blocks else None - _flush(last_b.section_title if last_b else None, last_b.section_path if last_b else "/") + if strict or cur_tokens >= target: + is_split_trigger = True + + if is_split_trigger: + _flush() # Vorherigen Puffer leeren + buf.append(b) # Neue Überschrift in den neuen Puffer aufnehmen + cur_tokens = b_tokens + else: + # Atomarer Check: Wenn der neue Block den aktuellen Chunk sprengen würde + if cur_tokens + b_tokens > max_tokens and buf: + _flush() # Puffer leeren, Block 'b' wird Teil des nächsten Chunks + + buf.append(b) + cur_tokens += b_tokens + + _flush() # Letzten Puffer leeren + return chunks + +def strategy_sliding_window(blocks: List[RawBlock], + config: Dict[str, Any], + note_id: str, + context_prefix: str = "") -> List[Chunk]: + """ + Standard Sliding Window mit Korrektur für Heading-Retention. + """ + target = config.get("target", 400) + max_tokens = config.get("max", 600) + overlap_val = config.get("overlap", (50, 80)) + overlap = sum(overlap_val) // 2 if isinstance(overlap_val, tuple) else overlap_val + + chunks: List[Chunk] = [] + buf: List[RawBlock] = [] + + def _flush_window(): + nonlocal buf + if not buf: return + txt = "\n\n".join([b.text for b 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 + )) + buf = [] + + for b in blocks: + # Auch hier: Überschriften mitnehmen + b_tokens = estimate_tokens(b.text) + current_buf_tokens = estimate_tokens("\n\n".join([x.text for x in buf])) if buf else 0 + + if current_buf_tokens + b_tokens >= target and buf: + _flush_window() + buf.append(b) + + _flush_window() return chunks \ No newline at end of file