From 6b83879741ec4267bff06dfc3dc80200c05fc8b7 Mon Sep 17 00:00:00 2001 From: Lars Date: Mon, 29 Dec 2025 22:04:23 +0100 Subject: [PATCH] =?UTF-8?q?Aktualisierung=20des=20Chunking-Parsers=20zur?= =?UTF-8?q?=20Verbesserung=20der=20Satzverarbeitung=20und=20Blocktrennung.?= =?UTF-8?q?=20Einf=C3=BChrung=20des=20'Pack-and-Carry-Over'=20Verfahrens?= =?UTF-8?q?=20in=20der=20Sektions-Chunking-Strategie=20zur=20Optimierung?= =?UTF-8?q?=20der=20Handhabung=20von=20gro=C3=9Fen=20Sektionen=20und=20Gew?= =?UTF-8?q?=C3=A4hrleistung=20der=20Sektionsintegrit=C3=A4t.=20Anpassungen?= =?UTF-8?q?=20an=20der=20Token-Sch=C3=A4tzung=20und=20Verbesserung=20der?= =?UTF-8?q?=20Metadatenverarbeitung.?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/core/chunking/chunking_parser.py | 36 ++--- app/core/chunking/chunking_strategies.py | 178 ++++++++++++----------- 2 files changed, 108 insertions(+), 106 deletions(-) diff --git a/app/core/chunking/chunking_parser.py b/app/core/chunking/chunking_parser.py index fb3b65c..696650a 100644 --- a/app/core/chunking/chunking_parser.py +++ b/app/core/chunking/chunking_parser.py @@ -1,6 +1,6 @@ """ FILE: app/core/chunking/chunking_parser.py -DESCRIPTION: Zerlegt Markdown in atomare Blöcke. Hält H1-Überschriften im Stream +DESCRIPTION: Zerlegt Markdown in logische Blöcke. Hält H1-Überschriften im Stream und gewährleistet die strukturelle Integrität von Callouts. """ import re @@ -8,28 +8,25 @@ from typing import List, Tuple, Set from .chunking_models import RawBlock from .chunking_utils import extract_frontmatter_from_text +_WS = re.compile(r'\s+') +_SENT_SPLIT = re.compile(r'(?<=[.!?])\s+(?=[A-ZÄÖÜ0-9„(])') + def split_sentences(text: str) -> list[str]: """Teilt Text in Sätze auf unter Berücksichtigung deutscher Interpunktion.""" - text = re.sub(r'\s+', ' ', text.strip()) + text = _WS.sub(' ', text.strip()) if not text: return [] - # Splittet bei Satzzeichen, gefolgt von Leerzeichen und Großbuchstaben - sentences = re.split(r'(?<=[.!?])\s+(?=[A-ZÄÖÜ0-9„(])', text) - return [s.strip() for s in sentences if s.strip()] + # Splittet bei Punkt, Ausrufezeichen oder Fragezeichen, gefolgt von Großbuchstabe + 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 (RawBlocks), inklusive H1-H6.""" blocks = [] - h1_title = "Dokument" - section_path = "/" - current_section_title = None - - # Frontmatter entfernen + h1_title = "Dokument"; section_path = "/"; current_section_title = None fm, text_without_fm = extract_frontmatter_from_text(md_text) # 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() + if h1_match: h1_title = h1_match.group(1).strip() lines = text_without_fm.split('\n') buffer = [] @@ -57,18 +54,15 @@ def parse_blocks(md_text: str) -> Tuple[List[RawBlock], str]: blocks.append(RawBlock("heading", stripped, level, section_path, current_section_title)) continue - # Trenner (---) beenden Blöcke, Leerzeilen nur wenn nicht in Callout - if stripped == "---" and not line.startswith('>'): + # Trenner oder Leerzeilen beenden Blöcke, außer innerhalb von Callouts + 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 = [] - blocks.append(RawBlock("separator", "---", None, section_path, current_section_title)) - elif not 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)) + 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) diff --git a/app/core/chunking/chunking_strategies.py b/app/core/chunking/chunking_strategies.py index 7936e88..d3e8cb6 100644 --- a/app/core/chunking/chunking_strategies.py +++ b/app/core/chunking/chunking_strategies.py @@ -1,135 +1,143 @@ """ FILE: app/core/chunking/chunking_strategies.py -DESCRIPTION: Strategie für atomares Sektions-Chunking v3.7.0. - Garantiert Sektions-Integrität durch ein flexibles Toleranz-Limit. - Kein Splitting von Sektionen, solange sie 'ungefähr' passen. +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. """ -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 -# Toleranz-Faktor: Erlaubt Chunks, bis zu 15% über 'max' zu wachsen, -# um eine Sektion vollständig zu erhalten. -FLEX_FACTOR = 1.15 - -def _safe_estimate(text: str) -> int: - """Sicherere Token-Schätzung für MD/Deutsch (Faktor 3.0 statt 4.0).""" - return max(1, math.ceil(len(text.strip()) / 3.0)) - -def _create_context_win(doc_title: str, sec_title: Optional[str], text: str) -> str: - parts = [] - if doc_title: parts.append(doc_title) +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 + 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 - # Das flexible Maximum, das Sektionen unzertrennt lässt - soft_max = int(max_tokens * FLEX_FACTOR) - chunks: List[Chunk] = [] - def _emit_chunk(block_list: List[RawBlock]): - """Schreibt eine Liste von Blöcken als einen einzigen Chunk ohne internes Splitting.""" - if not block_list: return - txt = "\n\n".join([b.text for b in block_list]) + def _emit(txt, title, path): idx = len(chunks) - title = block_list[0].section_title - path = block_list[0].section_path - win = _create_context_win(doc_title, title, txt) + 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=_safe_estimate(txt), - section_title=title, section_path=path, - neighbors_prev=None, neighbors_next=None + text=txt, window=win, token_count=estimate_tokens(txt), + section_title=title, section_path=path, neighbors_prev=None, neighbors_next=None )) - def _split_giant_section(sec_blocks: List[RawBlock]): - """Notfall-Split: Nur wenn eine EINZELNE Sektion bereits > soft_max 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(s) - if sub_len + slen > target and cur_sents: - _emit_chunk([RawBlock("paragraph", " ".join(cur_sents), None, main_path, main_title)]) - ov_s = [header_text] if header_text else [] - ov_l = _safe_estimate(header_text) if header_text else 0 - for os in reversed(cur_sents): - if os == header_text: continue - t_len = _safe_estimate(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)]) + # --- 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 - # 1. Gruppierung in atomare Sektions-Einheiten + # --- FALL B: SMART EDGE ALLOCATION (Pack-and-Carry-Over) --- + # 1. Gruppierung in atomare Sektions-Einheiten (Sektions-Isolation) sections: List[List[RawBlock]] = [] - curr_sec: List[RawBlock] = [] + curr = [] 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) + if curr: sections.append(curr) + curr = [b] + else: curr.append(b) + if curr: sections.append(curr) - # 2. Das flexible Pack-Verfahren - current_chunk_buf: List[RawBlock] = [] - current_tokens = 0 + # 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": "/"} - for sec in sections: - sec_text = "\n\n".join([b.text for b in sec]) - sec_tokens = _safe_estimate(sec_text) + while processing_queue: + item = processing_queue.pop(0) + item_text = item["text"] + item_tokens = estimate_tokens(item_text) - if current_chunk_buf: - # PRÜFUNG: Würde die neue Sektion das FLEXIBLE Limit sprengen? - if (current_tokens + sec_tokens > soft_max): - _emit_chunk(current_chunk_buf) - current_chunk_buf = [] - current_tokens = 0 - # Haben wir das Ziel-Maß erreicht und es kommt eine neue Sektion? - elif (current_tokens >= target): - _emit_chunk(current_chunk_buf) - current_chunk_buf = [] - current_tokens = 0 + # 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 - # Wenn eine EINZELNE Sektion alleine schon das weiche Limit sprengt - if not current_chunk_buf and sec_tokens > soft_max: - _split_giant_section(sec) + 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: - current_chunk_buf.extend(sec) - current_tokens = _safe_estimate("\n\n".join([b.text for b in current_chunk_buf])) + # 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 - if current_chunk_buf: - _emit_chunk(current_chunk_buf) + 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 = _safe_estimate(b.text) - current_tokens = sum(_safe_estimate(x.text) for x in buf) if buf else 0 + 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=_safe_estimate(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=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