""" FILE: app/core/chunking/chunking_strategies.py DESCRIPTION: Strategien für atomares Sektions-Chunking v3.8.5. Implementiert das 'Pack-and-Carry-Over' Verfahren: 1. Packt ganze Abschnitte basierend auf Schätzung. 2. Kein physischer Overflow-Check während des Packens. 3. Smart-Zerlegung von Übergrößen mit Carry-Over in die Queue. """ 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 _create_win(doc_title: str, sec_title: Optional[str], text: str) -> str: parts = [doc_title] if doc_title else [] 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]: """ Universelle Heading-Strategie mit Fallunterscheidung für Smart-Edge-Allocation. """ smart_edge = config.get("enable_smart_edge_allocation", True) 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(txt, title, path): idx = len(chunks) win = _create_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), section_title=title, section_path=path, neighbors_prev=None, neighbors_next=None )) # --- FALL A: HARD SPLIT (enable_smart_edge_allocation: false) --- if not smart_edge: buf = [] for b in blocks: # Trenne bei jeder Überschrift <= split_level if b.kind == "heading" and b.level <= split_level: # Nur flashen, wenn der Puffer nicht nur aus der aktuellen Überschrift besteht if buf and not (len(buf) == 1 and buf[0].kind == "heading"): _emit("\n\n".join([x.text for x in buf]), buf[0].section_title, buf[0].section_path) buf = [] buf.append(b) if buf: _emit("\n\n".join([x.text for x in buf]), buf[0].section_title, buf[0].section_path) return chunks # --- FALL B: SMART EDGE ALLOCATION (Pack-and-Carry-Over) --- # 1. Gruppierung in atomare Sektions-Einheiten (Sektions-Isolation) sections: List[List[RawBlock]] = [] curr = [] for b in blocks: if b.kind == "heading" and b.level <= split_level: if curr: sections.append(curr) curr = [b] else: curr.append(b) if curr: sections.append(curr) # 2. Queue-Management für Carry-Over processing_queue = [{"blocks": s, "text": "\n\n".join([b.text for b in s])} for s in sections] current_chunk_text = "" current_meta = {"title": None, "path": "/"} while processing_queue: item = processing_queue.pop(0) item_text = item["text"] item_tokens = estimate_tokens(item_text) # Metadaten-Initialisierung falls Chunk leer if not current_chunk_text and "blocks" in item: current_meta["title"] = item["blocks"][0].section_title current_meta["path"] = item["blocks"][0].section_path combined_est = estimate_tokens(current_chunk_text + "\n\n" + item_text) if current_chunk_text else item_tokens # Regel 1: Passt die vollständige Sektion nach Schätzung rein? (Kein harter Overflow-Check) if combined_est <= max_tokens: current_chunk_text = (current_chunk_text + "\n\n" + item_text).strip() else: # Regel 2: Wenn Puffer voll -> Emittieren und Sektion zurücklegen if current_chunk_text: _emit(current_chunk_text, current_meta["title"], current_meta["path"]) current_chunk_text = "" processing_queue.insert(0, item) else: # Regel 3: Einzelne Sektion zu groß -> Smart Zerlegung sents = split_sentences(item_text) header_text = "" if "blocks" in item and item["blocks"][0].kind == "heading": header_text = item["blocks"][0].text take_sents = []; take_len = 0 while sents: s = sents.pop(0) slen = estimate_tokens(s) if take_len + slen > target and take_sents: sents.insert(0, s); break take_sents.append(s); take_len += slen _emit(" ".join(take_sents), current_meta["title"], current_meta["path"]) # Carry-Over: Der Rest wird an den Anfang der Queue geschoben if sents: remainder_text = " ".join(sents) # Falls wir einen Header haben, fügen wir ihn dem Rest für den Kontext hinzu if header_text and not remainder_text.startswith(header_text): remainder_text = header_text + "\n\n" + remainder_text processing_queue.insert(0, {"text": remainder_text, "is_split": True}) if current_chunk_text: _emit(current_chunk_text, current_meta["title"], current_meta["path"]) return chunks def strategy_sliding_window(blocks: List[RawBlock], config: Dict[str, Any], note_id: str, context_prefix: str = "") -> List[Chunk]: """Sliding Window Strategie: Erhalten für alternative Anwendungsfälle.""" target = config.get("target", 400); max_tokens = config.get("max", 600) chunks: List[Chunk] = []; buf: List[RawBlock] = [] for b in blocks: 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) 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=estimate_tokens(txt), section_title=buf[0].section_title, section_path=buf[0].section_path, neighbors_prev=None, neighbors_next=None)) return chunks