""" FILE: app/core/chunking/chunking_strategies.py DESCRIPTION: Universelle Strategie für atomares Sektions-Chunking v3.5.0. Garantiert Sektions-Integrität durch präventives Chunk-Management. """ 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_context_win(doc_title: str, sec_title: Optional[str], text: str) -> str: """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) 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]: """ Universelles Sektions-Chunking: Packt Sektionen in Chunks. Bei Überlauf wird die komplette Sektion in den nächsten Chunk geschoben. """ 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] = [] # --- 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) 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=estimate_tokens(txt), section_title=title, section_path=path, neighbors_prev=None, neighbors_next=None )) # --- 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 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 = estimate_tokens(s) if sub_len + slen > target and cur_sents: 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 = 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 if cur_sents: _create_chunk_from_blocks([RawBlock("paragraph", " ".join(cur_sents), None, main_path, main_title)]) # 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] else: curr_sec.append(b) if curr_sec: sections.append(curr_sec) # 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 = estimate_tokens(sec_text) # 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: # 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 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]: """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 = 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