""" 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. """ 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).""" parts = [] if doc_title: parts.append(doc_title) if sec_title and sec_title != doc_title: parts.append(sec_title) prefix = " > ".join(parts) return f"{prefix}\n{text}".strip() if prefix else text 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. """ 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_cfg = config.get("overlap", (50, 80)) overlap = sum(overlap_cfg) // 2 if isinstance(overlap_cfg, (list, tuple)) else overlap_cfg chunks: List[Chunk] = [] def _emit_chunk(block_list: List[RawBlock]): """Erzeugt ein finales Chunk-Objekt aus einer Liste von Blöcken.""" 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), 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.""" 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 header_text = sec_blocks[0].text if sec_blocks[0].kind == "heading" else "" sents = split_sentences(full_text) cur_sents = []; sub_len = 0 for s in sents: slen = _safe_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 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: _emit_chunk([RawBlock("paragraph", " ".join(cur_sents), None, main_path, main_title)]) # SCHRITT 1: Gruppierung in atomare Einheiten (Sektionen) sections: List[List[RawBlock]] = [] curr_sec: List[RawBlock] = [] for b in blocks: if b.kind == "heading" and b.level <= split_level: if curr_sec: sections.append(curr_sec) curr_sec = [b] else: 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 for sec in sections: sec_text = "\n\n".join([b.text for b in sec]) sec_tokens = _safe_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) else: current_chunk_buf.extend(sec) current_tokens += sec_tokens + 2 # +2 für Newline Join # Letzten Puffer schreiben if current_chunk_buf: _emit_chunk(current_chunk_buf) 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.""" 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 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)) buf = [] buf.append(b) 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)) return chunks