neuer Chunking parser, der Headings mitführt und nicht mitten im Abschnitt schneidet
This commit is contained in:
parent
b7d1bcce3d
commit
8f5eb36b5f
|
|
@ -1,10 +1,9 @@
|
|||
"""
|
||||
FILE: app/core/chunking/chunking_strategies.py
|
||||
DESCRIPTION: Mathematische Splitting-Strategien.
|
||||
AUDIT v3.3.2: 100% Konformität zur 'by_heading' Spezifikation.
|
||||
- Implementiert Hybrid-Safety-Net (Sliding Window für Übergrößen).
|
||||
- Breadcrumb-Kontext im Window (H1 > H2).
|
||||
- Sliding Window mit H1-Kontext (Gold-Standard v3.1.0).
|
||||
DESCRIPTION: Korrigierte Splitting-Strategien für Mindnet v3.3.3.
|
||||
- Fix: Erhalt von Überschriften im Chunk-Text.
|
||||
- Fix: Atomares Buffering (Blöcke fallen als Ganzes in den nächsten Chunk).
|
||||
- Fix: Korrekte Zuordnung von Sektions-Metadaten.
|
||||
"""
|
||||
from typing import List, Dict, Any, Optional
|
||||
from .chunking_models import RawBlock, Chunk
|
||||
|
|
@ -19,71 +18,9 @@ def _create_context_win(doc_title: str, sec_title: Optional[str], text: str) ->
|
|||
prefix = " > ".join(parts)
|
||||
return f"{prefix}\n{text}".strip() if prefix else text
|
||||
|
||||
def strategy_sliding_window(blocks: List[RawBlock],
|
||||
config: Dict[str, Any],
|
||||
note_id: str,
|
||||
context_prefix: str = "") -> List[Chunk]:
|
||||
"""
|
||||
Fasst Blöcke zusammen und schneidet bei 'target' Tokens.
|
||||
Ignoriert H2-Überschriften beim Splitting, um Kontext zu wahren.
|
||||
"""
|
||||
target = config.get("target", 400)
|
||||
max_tokens = config.get("max", 600)
|
||||
overlap_val = config.get("overlap", (50, 80))
|
||||
overlap = sum(overlap_val) // 2 if isinstance(overlap_val, tuple) else overlap_val
|
||||
|
||||
chunks: List[Chunk] = []
|
||||
buf: List[RawBlock] = []
|
||||
|
||||
def _add(txt, sec, path):
|
||||
idx = len(chunks)
|
||||
# H1-Kontext Präfix für das Window-Feld
|
||||
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=sec, section_path=path,
|
||||
neighbors_prev=None, neighbors_next=None
|
||||
))
|
||||
|
||||
def flush():
|
||||
nonlocal buf
|
||||
if not buf: return
|
||||
text_body = "\n\n".join([b.text for b in buf])
|
||||
sec_title = buf[-1].section_title; sec_path = buf[-1].section_path
|
||||
|
||||
if estimate_tokens(text_body) <= max_tokens:
|
||||
_add(text_body, sec_title, sec_path)
|
||||
else:
|
||||
sents = split_sentences(text_body); cur_sents = []; cur_len = 0
|
||||
for s in sents:
|
||||
slen = estimate_tokens(s)
|
||||
if cur_len + slen > target and cur_sents:
|
||||
_add(" ".join(cur_sents), sec_title, sec_path)
|
||||
ov_s = []; ov_l = 0
|
||||
for os in reversed(cur_sents):
|
||||
if ov_l + estimate_tokens(os) < overlap:
|
||||
ov_s.insert(0, os); ov_l += estimate_tokens(os)
|
||||
else: break
|
||||
cur_sents = list(ov_s); cur_sents.append(s); cur_len = ov_l + slen
|
||||
else:
|
||||
cur_sents.append(s); cur_len += slen
|
||||
if cur_sents:
|
||||
_add(" ".join(cur_sents), sec_title, sec_path)
|
||||
buf = []
|
||||
|
||||
for b in blocks:
|
||||
# H2-Überschriften werden ignoriert, um den Zusammenhang zu wahren
|
||||
if b.kind == "heading": continue
|
||||
if estimate_tokens("\n\n".join([x.text for x in buf])) + estimate_tokens(b.text) >= target:
|
||||
flush()
|
||||
buf.append(b)
|
||||
flush()
|
||||
return chunks
|
||||
|
||||
def strategy_by_heading(blocks: List[RawBlock], config: Dict[str, Any], note_id: str, doc_title: str = "") -> List[Chunk]:
|
||||
"""
|
||||
Splittet Text basierend auf Markdown-Überschriften mit Hybrid-Safety-Net.
|
||||
Splittet Text basierend auf Markdown-Überschriften mit atomarem Block-Erhalt.
|
||||
"""
|
||||
strict = config.get("strict_heading_split", False)
|
||||
target = config.get("target", 400)
|
||||
|
|
@ -92,7 +29,7 @@ def strategy_by_heading(blocks: List[RawBlock], config: Dict[str, Any], note_id:
|
|||
overlap = sum(config.get("overlap", (50, 80))) // 2
|
||||
|
||||
chunks: List[Chunk] = []
|
||||
buf: List[str] = []
|
||||
buf: List[RawBlock] = []
|
||||
cur_tokens = 0
|
||||
|
||||
def _add_to_chunks(txt, title, path):
|
||||
|
|
@ -105,18 +42,26 @@ def strategy_by_heading(blocks: List[RawBlock], config: Dict[str, Any], note_id:
|
|||
neighbors_prev=None, neighbors_next=None
|
||||
))
|
||||
|
||||
def _flush(title, path):
|
||||
def _flush():
|
||||
nonlocal buf, cur_tokens
|
||||
if not buf: return
|
||||
full_text = "\n\n".join(buf)
|
||||
|
||||
# Metadaten stammen immer vom ersten Block im Puffer (meist die Überschrift)
|
||||
main_title = buf[0].section_title
|
||||
main_path = buf[0].section_path
|
||||
full_text = "\n\n".join([b.text for b in buf])
|
||||
|
||||
# Falls der gesamte Puffer in einen Chunk passt
|
||||
if estimate_tokens(full_text) <= max_tokens:
|
||||
_add_to_chunks(full_text, title, path)
|
||||
_add_to_chunks(full_text, main_title, main_path)
|
||||
else:
|
||||
sents = split_sentences(full_text); cur_sents = []; sub_len = 0
|
||||
# Nur wenn ein einzelner Abschnitt größer als 'max' ist, wird intern gesplittet
|
||||
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:
|
||||
_add_to_chunks(" ".join(cur_sents), title, path)
|
||||
_add_to_chunks(" ".join(cur_sents), main_title, main_path)
|
||||
ov_s = []; ov_l = 0
|
||||
for os in reversed(cur_sents):
|
||||
if ov_l + estimate_tokens(os) < overlap:
|
||||
|
|
@ -124,19 +69,74 @@ def strategy_by_heading(blocks: List[RawBlock], config: Dict[str, Any], note_id:
|
|||
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_to_chunks(" ".join(cur_sents), title, path)
|
||||
if cur_sents: _add_to_chunks(" ".join(cur_sents), main_title, main_path)
|
||||
|
||||
buf = []; cur_tokens = 0
|
||||
|
||||
for b in blocks:
|
||||
b_tokens = estimate_tokens(b.text)
|
||||
|
||||
# Prüfung auf Split-Trigger (Überschriften)
|
||||
is_split_trigger = False
|
||||
if b.kind == "heading":
|
||||
if b.level < split_level: _flush(b.section_title, b.section_path)
|
||||
if b.level < split_level:
|
||||
is_split_trigger = True
|
||||
elif b.level == split_level:
|
||||
if strict or cur_tokens >= target: _flush(b.section_title, b.section_path)
|
||||
continue
|
||||
bt = estimate_tokens(b.text)
|
||||
if cur_tokens + bt > max_tokens and buf: _flush(b.section_title, b.section_path)
|
||||
buf.append(b.text); cur_tokens += bt
|
||||
if buf:
|
||||
last_b = blocks[-1] if blocks else None
|
||||
_flush(last_b.section_title if last_b else None, last_b.section_path if last_b else "/")
|
||||
if strict or cur_tokens >= target:
|
||||
is_split_trigger = True
|
||||
|
||||
if is_split_trigger:
|
||||
_flush() # Vorherigen Puffer leeren
|
||||
buf.append(b) # Neue Überschrift in den neuen Puffer aufnehmen
|
||||
cur_tokens = b_tokens
|
||||
else:
|
||||
# Atomarer Check: Wenn der neue Block den aktuellen Chunk sprengen würde
|
||||
if cur_tokens + b_tokens > max_tokens and buf:
|
||||
_flush() # Puffer leeren, Block 'b' wird Teil des nächsten Chunks
|
||||
|
||||
buf.append(b)
|
||||
cur_tokens += b_tokens
|
||||
|
||||
_flush() # Letzten Puffer leeren
|
||||
return chunks
|
||||
|
||||
def strategy_sliding_window(blocks: List[RawBlock],
|
||||
config: Dict[str, Any],
|
||||
note_id: str,
|
||||
context_prefix: str = "") -> List[Chunk]:
|
||||
"""
|
||||
Standard Sliding Window mit Korrektur für Heading-Retention.
|
||||
"""
|
||||
target = config.get("target", 400)
|
||||
max_tokens = config.get("max", 600)
|
||||
overlap_val = config.get("overlap", (50, 80))
|
||||
overlap = sum(overlap_val) // 2 if isinstance(overlap_val, tuple) else overlap_val
|
||||
|
||||
chunks: List[Chunk] = []
|
||||
buf: List[RawBlock] = []
|
||||
|
||||
def _flush_window():
|
||||
nonlocal buf
|
||||
if not buf: return
|
||||
txt = "\n\n".join([b.text for b 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
|
||||
))
|
||||
buf = []
|
||||
|
||||
for b in blocks:
|
||||
# Auch hier: Überschriften mitnehmen
|
||||
b_tokens = estimate_tokens(b.text)
|
||||
current_buf_tokens = estimate_tokens("\n\n".join([x.text for x in buf])) if buf else 0
|
||||
|
||||
if current_buf_tokens + b_tokens >= target and buf:
|
||||
_flush_window()
|
||||
buf.append(b)
|
||||
|
||||
_flush_window()
|
||||
return chunks
|
||||
Loading…
Reference in New Issue
Block a user