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

Aktualisierung der atomaren Sektions-Chunking-Strategie auf Version 3.9.8 mit verbesserten Implementierungen des 'Pack-and-Carry-Over' Verfahrens. Einführung von Look-Ahead zur strikten Einhaltung von Sektionsgrenzen und Vermeidung redundanter Kanten-Injektionen. Anpassungen an der Chunk-Erstellung und Optimierung der Handhabung von leeren Überschriften.

Browse Source
This commit is contained in:
Lars 2025-12-30 07:54:54 +01:00
parent 06fc42ed37
commit 65d697b7be
2 changed files with 32 additions and 20 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -1,7 +1,8 @@
""" """
FILE: app/core/chunking/chunking_parser.py 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 Einheiten (RawBlocks).
und extrahiert Kanten-Kandidaten (parse_edges_robust). Hält alle Überschriftenebenen (H1-H6) im Stream.
Stellt die Funktion parse_edges_robust zur Verfügung.
""" """
import re import re
from typing import List, Tuple, Set from typing import List, Tuple, Set
@ -21,10 +22,14 @@ def split_sentences(text: str) -> list[str]:
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 (RawBlocks), inklusive H1-H6.""" """Zerlegt Text in logische Einheiten (RawBlocks), inklusive H1-H6."""
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 Metadaten extrahieren # H1 für Note-Titel extrahieren (Metadaten-Zweck)
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()
@ -38,6 +43,7 @@ def parse_blocks(md_text: str) -> Tuple[List[RawBlock], str]:
# Heading-Erkennung (H1 bis H6) # Heading-Erkennung (H1 bis H6)
heading_match = re.match(r'^(#{1,6})\s+(.*)', stripped) heading_match = re.match(r'^(#{1,6})\s+(.*)', stripped)
if heading_match: if heading_match:
# Vorherigen Text-Block abschließen
if buffer: if buffer:
content = "\n".join(buffer).strip() content = "\n".join(buffer).strip()
if content: if content:
@ -47,16 +53,17 @@ 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 # Pfad- und Titel-Update für die Metadaten der folgenden Blöcke
if level == 1: if level == 1:
current_section_title = title; section_path = "/" current_section_title = title; section_path = "/"
elif level == 2: elif level == 2:
current_section_title = title; section_path = f"/{current_section_title}" current_section_title = title; section_path = f"/{current_section_title}"
# Die Überschrift selbst als regulären 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
# Trenner oder Leerzeilen beenden Blöcke, außer innerhalb von Callouts # Trenner (---) oder Leerzeilen beenden Blöcke, außer innerhalb von Callouts
if (not stripped or 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()
@ -93,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 des Callouts
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:

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

@ -1,11 +1,9 @@
""" """
FILE: app/core/chunking/chunking_strategies.py FILE: app/core/chunking/chunking_strategies.py
DESCRIPTION: Strategien für atomares Sektions-Chunking v3.9.6. DESCRIPTION: Strategien für atomares Sektions-Chunking v3.9.8.
Implementiert das 'Pack-and-Carry-Over' Verfahren: Implementiert das 'Pack-and-Carry-Over' Verfahren nach Regel 1-3.
1. Packt ganze Abschnitte basierend auf Schätzung. - Keine redundante Kanten-Injektion.
2. Kein physischer Overflow-Check während des Packens. - Strikte Einhaltung von Sektionsgrenzen via Look-Ahead.
3. Smart-Zerlegung von Übergrößen mit Carry-Over in die Queue.
- Hard-Split-Logik für strict_heading_split integriert.
""" """
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
@ -13,6 +11,7 @@ from .chunking_utils import estimate_tokens
from .chunking_parser import split_sentences from .chunking_parser import split_sentences
def _create_win(doc_title: str, sec_title: Optional[str], text: str) -> str: def _create_win(doc_title: str, sec_title: Optional[str], text: str) -> str:
"""Baut den Breadcrumb-Kontext für das Embedding-Fenster."""
parts = [doc_title] if doc_title else [] parts = [doc_title] if doc_title else []
if sec_title and sec_title != doc_title: parts.append(sec_title) if sec_title and sec_title != doc_title: parts.append(sec_title)
prefix = " > ".join(parts) prefix = " > ".join(parts)
@ -33,6 +32,7 @@ def strategy_by_heading(blocks: List[RawBlock], config: Dict[str, Any], note_id:
chunks: List[Chunk] = [] chunks: List[Chunk] = []
def _emit(txt, title, path): def _emit(txt, title, path):
"""Schreibt den finalen Chunk ohne Text-Modifikationen."""
idx = len(chunks) idx = len(chunks)
win = _create_win(doc_title, title, txt) win = _create_win(doc_title, title, txt)
chunks.append(Chunk( chunks.append(Chunk(
@ -54,7 +54,6 @@ def strategy_by_heading(blocks: List[RawBlock], config: Dict[str, Any], note_id:
}) })
curr_blocks = [b] curr_blocks = [b]
else: else:
curr_sec_has_content = True
curr_blocks.append(b) curr_blocks.append(b)
if curr_blocks: if curr_blocks:
sections.append({ sections.append({
@ -75,26 +74,27 @@ def strategy_by_heading(blocks: List[RawBlock], config: Dict[str, Any], note_id:
item = queue.pop(0) item = queue.pop(0)
item_text = item["text"] item_text = item["text"]
# Initialisierung für neuen Chunk
if not current_chunk_text: if not current_chunk_text:
current_meta["title"] = item["meta"].section_title current_meta["title"] = item["meta"].section_title
current_meta["path"] = item["meta"].section_path current_meta["path"] = item["meta"].section_path
# FALL A: HARD SPLIT MODUS # FALL A: HARD SPLIT MODUS
if is_hard_split_mode: if is_hard_split_mode:
# Leere Überschriften (H1 vor H2) werden mit dem nächsten Item verschmolzen # Leere Überschriften (z.B. H1 direkt vor H2) verbleiben am nächsten Chunk
if item.get("is_empty", False) and queue: if item.get("is_empty", False) and queue:
current_chunk_text = (current_chunk_text + "\n\n" + item_text).strip() current_chunk_text = (current_chunk_text + "\n\n" + item_text).strip()
continue continue
combined = (current_chunk_text + "\n\n" + item_text).strip() combined = (current_chunk_text + "\n\n" + item_text).strip()
# Wenn durch das Verschmelzen das Limit gesprengt würde, flashen wir vorher # Wenn durch Verschmelzung das Limit gesprengt würde, vorher flashen
if estimate_tokens(combined) > max_tokens and current_chunk_text: if estimate_tokens(combined) > max_tokens and current_chunk_text:
_emit(current_chunk_text, current_meta["title"], current_meta["path"]) _emit(current_chunk_text, current_meta["title"], current_meta["path"])
current_chunk_text = item_text current_chunk_text = item_text
else: else:
current_chunk_text = combined current_chunk_text = combined
# Im Hard-Split wird nach jeder nicht-leeren Sektion geflasht # Im Hard-Split wird nach jeder Sektion geflasht
_emit(current_chunk_text, current_meta["title"], current_meta["path"]) _emit(current_chunk_text, current_meta["title"], current_meta["path"])
current_chunk_text = "" current_chunk_text = ""
continue continue
@ -104,7 +104,7 @@ def strategy_by_heading(blocks: List[RawBlock], config: Dict[str, Any], note_id:
combined_est = estimate_tokens(combined_text) combined_est = estimate_tokens(combined_text)
if combined_est <= max_tokens: if combined_est <= max_tokens:
# Regel 1 & 2: Passt rein -> Aufnehmen # Regel 1 & 2: Passt rein laut Schätzung -> Aufnehmen
current_chunk_text = combined_text current_chunk_text = combined_text
else: else:
if current_chunk_text: if current_chunk_text:
@ -119,8 +119,7 @@ def strategy_by_heading(blocks: List[RawBlock], config: Dict[str, Any], note_id:
take_sents = []; take_len = 0 take_sents = []; take_len = 0
while sents: while sents:
s = sents.pop(0) s = sents.pop(0); slen = estimate_tokens(s)
slen = estimate_tokens(s)
if take_len + slen > target and take_sents: if take_len + slen > target and take_sents:
sents.insert(0, s); break sents.insert(0, s); break
take_sents.append(s); take_len += slen take_sents.append(s); take_len += slen
@ -129,6 +128,7 @@ def strategy_by_heading(blocks: List[RawBlock], config: Dict[str, Any], note_id:
if sents: if sents:
remainder = " ".join(sents) remainder = " ".join(sents)
# Kontext-Erhalt: Überschrift für den Rest wiederholen
if header_prefix and not remainder.startswith(header_prefix): if header_prefix and not remainder.startswith(header_prefix):
remainder = header_prefix + "\n\n" + remainder remainder = header_prefix + "\n\n" + remainder
# Carry-Over: Rest wird vorne in die Queue geschoben # Carry-Over: Rest wird vorne in die Queue geschoben
@ -140,9 +140,10 @@ 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, doc_title: str = "") -> List[Chunk]: def strategy_sliding_window(blocks: List[RawBlock], config: Dict[str, Any], note_id: str, doc_title: str = "") -> List[Chunk]:
"""Basis-Sliding-Window für flache Texte.""" """Standard-Sliding-Window für flache Texte ohne Sektionsfokus."""
target = config.get("target", 400); max_tokens = config.get("max", 600) target = config.get("target", 400); max_tokens = config.get("max", 600)
chunks: List[Chunk] = []; buf: List[RawBlock] = [] chunks: List[Chunk] = []; buf: List[RawBlock] = []
for b in blocks: for b in blocks:
b_tokens = estimate_tokens(b.text) b_tokens = estimate_tokens(b.text)
curr_tokens = sum(estimate_tokens(x.text) for x in buf) if buf else 0 curr_tokens = sum(estimate_tokens(x.text) for x in buf) if buf else 0
@ -152,8 +153,10 @@ def strategy_sliding_window(blocks: List[RawBlock], config: Dict[str, Any], note
chunks.append(Chunk(id=f"{note_id}#c{idx:02d}", note_id=note_id, index=idx, text=txt, window=win, token_count=curr_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=curr_tokens, section_title=buf[0].section_title, section_path=buf[0].section_path, neighbors_prev=None, neighbors_next=None))
buf = [] buf = []
buf.append(b) buf.append(b)
if buf: if buf:
txt = "\n\n".join([x.text for x in buf]); idx = len(chunks) txt = "\n\n".join([x.text for x in buf]); idx = len(chunks)
win = _create_win(doc_title, buf[0].section_title, txt) win = _create_win(doc_title, buf[0].section_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=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 return chunks