Lars/mindnet
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 21 Wiki Activity

Verbesserung der atomaren Sektions-Chunking-Strategie durch Einführung strikter Look-Ahead-Logik und präventiven Flush zur Gewährleistung von Sektionsgrenzen. Anpassungen an der Token-Schätzung und Umbenennung von Funktionen zur besseren Lesbarkeit.

Browse Source
This commit is contained in:
Lars 2025-12-29 21:05:42 +01:00
parent 1b40e29f40
commit f9ac4e4dbf
2 changed files with 22 additions and 23 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
app/core/chunking/chunking_parser.py
View File

@ -55,7 +55,7 @@ def parse_blocks(md_text: str) -> Tuple[List[RawBlock], str]:
current_section_title = title current_section_title = title
section_path = f"/{current_section_title}" section_path = f"/{current_section_title}"
# Die Überschrift selbst als Block hinzufügen (Fix: H1 wird nicht mehr gefiltert) # Die Überschrift selbst als Block hinzufügen
blocks.append(RawBlock("heading", stripped, level, section_path, current_section_title)) blocks.append(RawBlock("heading", stripped, level, section_path, current_section_title))
continue continue

43
app/core/chunking/chunking_strategies.py
View File

@ -1,7 +1,7 @@
""" """
FILE: app/core/chunking/chunking_strategies.py FILE: app/core/chunking/chunking_strategies.py
DESCRIPTION: Strategien für atomares Sektions-Chunking (WP-15b konform). DESCRIPTION: Strategien für atomares Sektions-Chunking (WP-15b konform).
Fix: Vorausschauende Trennung zur Wahrung von Sektionsgrenzen. v3.3.5: Garantiert atomare Sektionsgrenzen durch präventiven Flush.
""" """
from typing import List, Dict, Any, Optional from typing import List, Dict, Any, Optional
from .chunking_models import RawBlock, Chunk from .chunking_models import RawBlock, Chunk
@ -19,19 +19,20 @@ def _create_context_win(doc_title: str, sec_title: Optional[str], text: str) ->
def strategy_by_heading(blocks: List[RawBlock], config: Dict[str, Any], note_id: str, doc_title: str = "") -> List[Chunk]: def strategy_by_heading(blocks: List[RawBlock], config: Dict[str, Any], note_id: str, doc_title: str = "") -> List[Chunk]:
""" """
Gruppiert Blöcke zu Sektionen und hält diese atomar zusammen. Gruppiert Blöcke zu Sektionen und hält diese atomar zusammen.
Nutzt Look-Ahead, um Sektions-Überhänge zu vermeiden. Nutzt striktes Look-Ahead, um das Zerschneiden von Sektionsübergängen zu verhindern.
""" """
strict = config.get("strict_heading_split", False) strict = config.get("strict_heading_split", False)
target = config.get("target", 400) target = config.get("target", 400)
max_tokens = config.get("max", 600) max_tokens = config.get("max", 600)
split_level = config.get("split_level", 2) split_level = config.get("split_level", 2)
overlap = sum(config.get("overlap", (50, 80))) // 2 overlap_config = config.get("overlap", (50, 80))
overlap = sum(overlap_config) // 2 if isinstance(overlap_config, (list, tuple)) else overlap_config
chunks: List[Chunk] = [] chunks: List[Chunk] = []
buf: List[RawBlock] = [] buf: List[RawBlock] = []
cur_tokens = 0 cur_tokens = 0
def _add_to_chunks(txt, title, path): def _add_chunk(txt, title, path):
idx = len(chunks) idx = len(chunks)
win = _create_context_win(doc_title, title, txt) win = _create_context_win(doc_title, title, txt)
chunks.append(Chunk( chunks.append(Chunk(
@ -48,23 +49,23 @@ def strategy_by_heading(blocks: List[RawBlock], config: Dict[str, Any], note_id:
main_title = buf[0].section_title main_title = buf[0].section_title
main_path = buf[0].section_path main_path = buf[0].section_path
full_text = "\n\n".join([b.text for b in buf]) full_text = "\n\n".join([b.text for b in buf])
actual_tokens = estimate_tokens(full_text)
# Falls die gruppierten Sektionen in das Limit passen # Falls die gruppierten Sektionen in das Limit passen
if estimate_tokens(full_text) <= max_tokens: if actual_tokens <= max_tokens:
_add_to_chunks(full_text, main_title, main_path) _add_chunk(full_text, main_title, main_path)
else: else:
# Nur wenn eine Sektion ALLEINE zu groß ist, wird intern gesplittet # Nur wenn eine Sektion ALLEINE zu groß ist, wird intern gesplittet
sents = split_sentences(full_text) sents = split_sentences(full_text)
cur_sents = []; sub_len = 0 cur_sents = []; sub_len = 0
# Kontext-Sicherung: Heading für Teil-Chunks merken
header_text = buf[0].text if buf[0].kind == "heading" else "" header_text = buf[0].text if buf[0].kind == "heading" else ""
for s in sents: for s in sents:
slen = estimate_tokens(s) slen = estimate_tokens(s)
if sub_len + slen > target and cur_sents: if sub_len + slen > target and cur_sents:
_add_to_chunks(" ".join(cur_sents), main_title, main_path) _add_chunk(" ".join(cur_sents), main_title, main_path)
# Overlap-Erzeugung und Header-Injektion für Folgeschritte # Overlap-Erzeugung und Header-Injektion
ov_s = [header_text] if header_text else [] ov_s = [header_text] if header_text else []
ov_l = estimate_tokens(header_text) if header_text else 0 ov_l = estimate_tokens(header_text) if header_text else 0
for os in reversed(cur_sents): for os in reversed(cur_sents):
@ -80,7 +81,7 @@ def strategy_by_heading(blocks: List[RawBlock], config: Dict[str, Any], note_id:
cur_sents.append(s); sub_len += slen cur_sents.append(s); sub_len += slen
if cur_sents: if cur_sents:
_add_to_chunks(" ".join(cur_sents), main_title, main_path) _add_chunk(" ".join(cur_sents), main_title, main_path)
buf = []; cur_tokens = 0 buf = []; cur_tokens = 0
@ -89,7 +90,6 @@ def strategy_by_heading(blocks: List[RawBlock], config: Dict[str, Any], note_id:
curr_sec: List[RawBlock] = [] curr_sec: List[RawBlock] = []
for b in blocks: for b in blocks:
# Ein Split-Trigger (H1 oder H2) startet eine neue atomare Sektion
if b.kind == "heading" and b.level <= split_level: if b.kind == "heading" and b.level <= split_level:
if curr_sec: sections.append(curr_sec) if curr_sec: sections.append(curr_sec)
curr_sec = [b] curr_sec = [b]
@ -97,26 +97,25 @@ def strategy_by_heading(blocks: List[RawBlock], config: Dict[str, Any], note_id:
curr_sec.append(b) curr_sec.append(b)
if curr_sec: sections.append(curr_sec) if curr_sec: sections.append(curr_sec)
# SCHRITT 2: Verarbeitung der Sektionen mit Vorausschau (Look-Ahead) # SCHRITT 2: Verarbeitung der Sektionen mit strenger Vorausschau
for sec in sections: for sec in sections:
sec_text = "\n\n".join([b.text for b in sec]) sec_text = "\n\n".join([b.text for b in sec])
sec_tokens = estimate_tokens(sec_text) sec_tokens = estimate_tokens(sec_text)
if buf: if buf:
# VORAUSSCHAU: Würde die neue Sektion das Limit sprengen? # PRÜFUNG 1: Passt die gesamte neue Sektion noch in den Chunk (bis max)?
if cur_tokens + sec_tokens > max_tokens: # PRÜFUNG 2: Wenn wir über target sind, fangen wir auf jeden Fall neu an.
_flush_buffer() # Beende den aktuellen Chunk sauber VOR der neuen Sektion if (cur_tokens + sec_tokens > max_tokens) or (cur_tokens >= target):
# Wenn strict: Jede neue Sektion auf split_level erzwingt neuen Chunk
elif strict and sec[0].kind == "heading" and sec[0].level == split_level:
_flush_buffer() _flush_buffer()
# Wenn target erreicht: Neue Sektion startet neuen Chunk # PRÜFUNG 3: Wenn strict-mode aktiv ist und ein split_level erreicht wurde
elif cur_tokens >= target: elif strict and sec[0].kind == "heading" and sec[0].level == split_level:
_flush_buffer() _flush_buffer()
buf.extend(sec) buf.extend(sec)
cur_tokens += sec_tokens # Token-Zähler basierend auf dem tatsächlichen Puffer-Text aktualisieren
cur_tokens = estimate_tokens("\n\n".join([b.text for b in buf]))
# Falls eine einzelne Sektion nach dem Flush (oder als erste) schon zu groß ist # Falls eine einzelne Sektion (selbst nach flush) schon zu groß ist
if cur_tokens >= max_tokens: if cur_tokens >= max_tokens:
_flush_buffer() _flush_buffer()
@ -124,7 +123,7 @@ def strategy_by_heading(blocks: List[RawBlock], config: Dict[str, Any], note_id:
return chunks return chunks
def strategy_sliding_window(blocks: List[RawBlock], config: Dict[str, Any], note_id: str, context_prefix: str = "") -> List[Chunk]: 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.""" """Standard-Sliding-Window für flache Texte."""
target = config.get("target", 400) target = config.get("target", 400)
max_tokens = config.get("max", 600) max_tokens = config.get("max", 600)