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

Verbesserung des Chunking-Parsers zur Gewährleistung der Integrität von Callouts und Listen sowie Anpassungen an der Blockverarbeitung. Aktualisierung der atomaren Sektions-Chunking-Strategie mit Block-Aware-Flushing und optimierter Token-Schätzung für eine präzisere Handhabung von großen Blöcken.

Browse Source
This commit is contained in:
Lars 2025-12-29 21:15:03 +01:00
parent f9ac4e4dbf
commit e5a34efee9
2 changed files with 110 additions and 69 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -1,7 +1,7 @@
""" """
FILE: app/core/chunking/chunking_parser.py FILE: app/core/chunking/chunking_parser.py
DESCRIPTION: Zerlegt Markdown in Blöcke. Hält H1-Überschriften im Stream DESCRIPTION: Zerlegt Markdown in Blöcke. Erhält H1-Überschriften und
und optimiert die Block-Trennung für atomares Chunking. gewährleistet die Integrität von Callouts und Listen.
""" """
import re import re
from typing import List, Tuple, Set from typing import List, Tuple, Set
@ -18,12 +18,16 @@ def split_sentences(text: str) -> list[str]:
return [p.strip() for p in _SENT_SPLIT.split(text) if p.strip()] return [p.strip() for p in _SENT_SPLIT.split(text) if p.strip()]
def parse_blocks(md_text: str) -> Tuple[List[RawBlock], str]: def parse_blocks(md_text: str) -> Tuple[List[RawBlock], str]:
"""Zerlegt Text in logische Einheiten, inklusive aller Überschriften.""" """Zerlegt Text in logische Einheiten (RawBlocks), inklusive H1."""
blocks = [] blocks = []
h1_title = "Dokument"; section_path = "/"; current_section_title = None h1_title = "Dokument"
section_path = "/"
current_section_title = None
# Frontmatter entfernen
fm, text_without_fm = extract_frontmatter_from_text(md_text) fm, text_without_fm = extract_frontmatter_from_text(md_text)
# H1 für Note-Titel extrahieren (Metadaten) # H1 für Note-Titel extrahieren
h1_match = re.search(r'^#\s+(.*)', text_without_fm, re.MULTILINE) h1_match = re.search(r'^#\s+(.*)', text_without_fm, re.MULTILINE)
if h1_match: if h1_match:
h1_title = h1_match.group(1).strip() h1_title = h1_match.group(1).strip()
@ -47,7 +51,7 @@ def parse_blocks(md_text: str) -> Tuple[List[RawBlock], str]:
level = len(heading_match.group(1)) level = len(heading_match.group(1))
title = heading_match.group(2).strip() title = heading_match.group(2).strip()
# Pfad- und Titel-Update für die Metadaten der folgenden Blöcke # Pfad- und Titel-Update für die Metadaten
if level == 1: if level == 1:
current_section_title = title current_section_title = title
section_path = "/" section_path = "/"
@ -55,17 +59,19 @@ 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 # Die Überschrift als Block hinzufügen (H1 wird NICHT mehr gefiltert)
blocks.append(RawBlock("heading", stripped, level, section_path, current_section_title)) blocks.append(RawBlock("heading", stripped, level, section_path, current_section_title))
continue continue
# Leerzeilen trennen Blöcke, außer innerhalb von Callouts # Trenner oder Leerzeilen beenden einen Block
if not stripped and not line.startswith('>'): if (not stripped or stripped == "---") and not line.startswith('>'):
if buffer: if buffer:
content = "\n".join(buffer).strip() content = "\n".join(buffer).strip()
if content: if content:
blocks.append(RawBlock("paragraph", content, None, section_path, current_section_title)) blocks.append(RawBlock("paragraph", content, None, section_path, current_section_title))
buffer = [] buffer = []
if stripped == "---":
blocks.append(RawBlock("separator", "---", None, section_path, current_section_title))
else: else:
buffer.append(line) buffer.append(line)
@ -77,14 +83,16 @@ def parse_blocks(md_text: str) -> Tuple[List[RawBlock], str]:
return blocks, h1_title return blocks, h1_title
def parse_edges_robust(text: str) -> Set[str]: def parse_edges_robust(text: str) -> Set[str]:
"""Extrahiert Kanten-Kandidaten (Wikilinks, Callouts).""" """Extrahiert Kanten-Kandidaten aus Wikilinks und Callouts."""
found_edges = set() found_edges = set()
# 1. Inline Wikilinks [[rel:kind|target]]
inlines = re.findall(r'\[\[rel:([^\|\]]+)\|?([^\]]*)\]\]', text) inlines = re.findall(r'\[\[rel:([^\|\]]+)\|?([^\]]*)\]\]', text)
for kind, target in inlines: for kind, target in inlines:
k = kind.strip().lower() k = kind.strip().lower()
t = target.strip() t = target.strip()
if k and t: found_edges.add(f"{k}:{t}") if k and t: found_edges.add(f"{k}:{t}")
# 2. Callout Edges > [!edge] kind
lines = text.split('\n') lines = text.split('\n')
current_edge_type = None current_edge_type = None
for line in lines: for line in lines:
@ -92,10 +100,12 @@ def parse_edges_robust(text: str) -> Set[str]:
callout_match = re.match(r'>\s*\[!edge\]\s*([^:\s]+)', stripped) callout_match = re.match(r'>\s*\[!edge\]\s*([^:\s]+)', stripped)
if callout_match: if callout_match:
current_edge_type = callout_match.group(1).strip().lower() current_edge_type = callout_match.group(1).strip().lower()
# Links in der gleichen Zeile
links = re.findall(r'\[\[([^\]]+)\]\]', stripped) links = re.findall(r'\[\[([^\]]+)\]\]', stripped)
for l in links: for l in links:
if "rel:" not in l: found_edges.add(f"{current_edge_type}:{l}") if "rel:" not in l: found_edges.add(f"{current_edge_type}:{l}")
continue continue
# Links in Folgezeilen des Callouts
if current_edge_type and stripped.startswith('>'): if current_edge_type and stripped.startswith('>'):
links = re.findall(r'\[\[([^\]]+)\]\]', stripped) links = re.findall(r'\[\[([^\]]+)\]\]', stripped)
for l in links: for l in links:

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

@ -1,13 +1,19 @@
""" """
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 v3.3.6.
v3.3.5: Garantiert atomare Sektionsgrenzen durch präventiven Flush. AUDIT: 100% Konformität zur 'by_heading' Spezifikation.
- Block-Aware Flushing: Trennung nur an Blockgrenzen.
- Atomic Section Vorausschau: Verhindert Sektions-Zerreißung.
""" """
import math
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
from .chunking_utils import estimate_tokens
from .chunking_parser import split_sentences from .chunking_parser import split_sentences
def _safe_estimate_tokens(text: str) -> int:
"""Konservative Token-Schätzung für deutschen Text (len/3 statt len/4)."""
return max(1, math.ceil(len(text.strip()) / 3))
def _create_context_win(doc_title: str, sec_title: Optional[str], text: str) -> str: def _create_context_win(doc_title: str, sec_title: Optional[str], text: str) -> str:
"""Baut den Breadcrumb-Kontext für das Embedding-Fenster.""" """Baut den Breadcrumb-Kontext für das Embedding-Fenster."""
parts = [] parts = []
@ -19,73 +25,89 @@ 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 striktes Look-Ahead, um das Zerschneiden von Sektionsübergängen zu verhindern. Nutzt Block-Aware-Flushing, um Sektionsgrenzen strikt zu wahren.
""" """
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_config = config.get("overlap", (50, 80)) overlap_cfg = config.get("overlap", (50, 80))
overlap = sum(overlap_config) // 2 if isinstance(overlap_config, (list, tuple)) else overlap_config overlap = sum(overlap_cfg) // 2 if isinstance(overlap_cfg, (list, tuple)) else overlap_cfg
chunks: List[Chunk] = [] chunks: List[Chunk] = []
buf: List[RawBlock] = [] buf: List[RawBlock] = []
cur_tokens = 0
def _add_chunk(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(
id=f"{note_id}#c{idx:02d}", note_id=note_id, index=idx, id=f"{note_id}#c{idx:02d}", note_id=note_id, index=idx,
text=txt, window=win, token_count=estimate_tokens(txt), text=txt, window=win, token_count=_safe_estimate_tokens(txt),
section_title=title, section_path=path, section_title=title, section_path=path,
neighbors_prev=None, neighbors_next=None neighbors_prev=None, neighbors_next=None
)) ))
def _flush_buffer(): def _flush_buffer():
nonlocal buf, cur_tokens nonlocal buf
if not buf: return if not buf: return
main_title = buf[0].section_title # Block-Aware Processing innerhalb des Puffers
main_path = buf[0].section_path current_blocks = []
full_text = "\n\n".join([b.text for b in buf]) current_len = 0
actual_tokens = estimate_tokens(full_text)
# Falls die gruppierten Sektionen in das Limit passen for b in buf:
if actual_tokens <= max_tokens: b_len = _safe_estimate_tokens(b.text)
_add_chunk(full_text, main_title, main_path)
else:
# Nur wenn eine Sektion ALLEINE zu groß ist, wird intern gesplittet
sents = split_sentences(full_text)
cur_sents = []; sub_len = 0
header_text = buf[0].text if buf[0].kind == "heading" else ""
for s in sents: # Falls dieser Block den aktuellen Chunk sprengen würde -> Vorher abschließen
slen = estimate_tokens(s) if current_len + b_len > max_tokens and current_blocks:
if sub_len + slen > target and cur_sents: txt = "\n\n".join([cb.text for cb in current_blocks])
_add_chunk(" ".join(cur_sents), main_title, main_path) _add_chunk(txt, current_blocks[0].section_title, current_blocks[0].section_path)
current_blocks = []
current_len = 0
# Overlap-Erzeugung und Header-Injektion # Falls ein einzelner Block alleine zu groß ist (Sliding Window Fallback)
ov_s = [header_text] if header_text else [] if b_len > max_tokens:
ov_l = estimate_tokens(header_text) if header_text else 0 if current_blocks: # Vorherigen Rest wegschreiben
for os in reversed(cur_sents): txt = "\n\n".join([cb.text for cb in current_blocks])
if os == header_text: continue _add_chunk(txt, current_blocks[0].section_title, current_blocks[0].section_path)
t_len = estimate_tokens(os) current_blocks = []
if ov_l + t_len < overlap: current_len = 0
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: # Sätze dieses einen Riesen-Blocks splitten
_add_chunk(" ".join(cur_sents), main_title, main_path) sents = split_sentences(b.text)
cur_sents = []; sub_len = 0
header_text = b.text if b.kind == "heading" else ""
buf = []; cur_tokens = 0 for s in sents:
slen = _safe_estimate_tokens(s)
if sub_len + slen > target and cur_sents:
_add_chunk(" ".join(cur_sents), b.section_title, b.section_path)
# Overlap-Erzeugung & Header-Wiederholung
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:
_add_chunk(" ".join(cur_sents), b.section_title, b.section_path)
else:
current_blocks.append(b)
current_len += b_len
# SCHRITT 1: Gruppierung in atomare Sektions-Einheiten (Heading + Paragraphs) # Den verbleibenden Rest im Puffer als finalen Chunk schreiben
if current_blocks:
txt = "\n\n".join([cb.text for cb in current_blocks])
_add_chunk(txt, current_blocks[0].section_title, current_blocks[0].section_path)
buf = []
# SCHRITT 1: Gruppierung in atomare Sektions-Einheiten
sections: List[List[RawBlock]] = [] sections: List[List[RawBlock]] = []
curr_sec: List[RawBlock] = [] curr_sec: List[RawBlock] = []
@ -97,33 +119,34 @@ 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 strenger Vorausschau # SCHRITT 2: Verarbeitung der Sektionen mit Look-Ahead
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 = _safe_estimate_tokens(sec_text)
# Aktueller Füllstand des Puffers
cur_buf_tokens = sum(_safe_estimate_tokens(b.text) for b in buf)
if buf: if buf:
# PRÜFUNG 1: Passt die gesamte neue Sektion noch in den Chunk (bis max)? # PRÜFUNG: Wenn die neue Sektion den Puffer über das Limit treibt
# PRÜFUNG 2: Wenn wir über target sind, fangen wir auf jeden Fall neu an. # ODER wenn der Puffer bereits das Ziel-Format erreicht hat
if (cur_tokens + sec_tokens > max_tokens) or (cur_tokens >= target): if (cur_buf_tokens + sec_tokens > max_tokens) or (cur_buf_tokens >= target):
_flush_buffer() _flush_buffer()
# PRÜFUNG 3: Wenn strict-mode aktiv ist und ein split_level erreicht wurde # PRÜFUNG: Strikter Split an Überschriften
elif strict and sec[0].kind == "heading" and sec[0].level == split_level: elif strict and sec[0].kind == "heading" and sec[0].level == split_level:
_flush_buffer() _flush_buffer()
buf.extend(sec) buf.extend(sec)
# 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 (selbst nach flush) schon zu groß ist # Falls eine Riesen-Sektion hinzugefügt wurde, die sofort raus muss
if cur_tokens >= max_tokens: if sum(_safe_estimate_tokens(b.text) for b in buf) >= max_tokens:
_flush_buffer() _flush_buffer()
_flush_buffer() _flush_buffer()
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]:
"""Standard-Sliding-Window für flache Texte.""" """Basis-Sliding-Window für flache Texte ohne Sektionsfokus."""
target = config.get("target", 400) target = config.get("target", 400)
max_tokens = config.get("max", 600) max_tokens = config.get("max", 600)
@ -131,14 +154,18 @@ def strategy_sliding_window(blocks: List[RawBlock], config: Dict[str, Any], note
buf: List[RawBlock] = [] buf: List[RawBlock] = []
for b in blocks: for b in blocks:
b_tokens = estimate_tokens(b.text) b_tokens = _safe_estimate_tokens(b.text)
current_tokens = estimate_tokens("\n\n".join([x.text for x in buf])) if buf else 0 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: if current_tokens + b_tokens > max_tokens and buf:
txt = "\n\n".join([x.text for x in buf]) txt = "\n\n".join([x.text for x in buf])
idx = len(chunks) idx = len(chunks)
win = f"{context_prefix}\n{txt}".strip() if context_prefix else txt 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)) 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 = []
current_tokens = 0 current_tokens = 0
@ -148,6 +175,10 @@ def strategy_sliding_window(blocks: List[RawBlock], config: Dict[str, Any], note
txt = "\n\n".join([x.text for x in buf]) txt = "\n\n".join([x.text for x in buf])
idx = len(chunks) idx = len(chunks)
win = f"{context_prefix}\n{txt}".strip() if context_prefix else txt 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)) 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 return chunks