WP15 #9
|
|
@ -7,11 +7,14 @@ import yaml
|
|||
from pathlib import Path
|
||||
from markdown_it import MarkdownIt
|
||||
from markdown_it.token import Token
|
||||
import asyncio # Notwendig für asynchrone Chunking-Strategien
|
||||
|
||||
# NEUE IMPORTS
|
||||
# Import des Semantic Analyzer Services
|
||||
from app.services.semantic_analyzer import get_semantic_analyzer
|
||||
import asyncio # Für den asynchronen Aufruf des Chunkers
|
||||
# Import des Semantic Analyzer Services für die LLM-Strategie
|
||||
from app.services.semantic_analyzer import get_semantic_analyzer
|
||||
# Import zum Auslesen des Frontmatters
|
||||
from app.core.note_payload import extract_frontmatter_from_text
|
||||
|
||||
|
||||
# ==========================================
|
||||
# 1. CONFIGURATION LOADER (Ehemals chunk_config.py)
|
||||
|
|
@ -38,7 +41,6 @@ def _load_yaml_config() -> Dict[str, Any]:
|
|||
return _CONFIG_CACHE
|
||||
|
||||
if not CONFIG_PATH.exists():
|
||||
# Debugging-Hilfe: Zeigt an, wo gesucht wurde
|
||||
print(f"WARNUNG: types.yaml nicht gefunden unter: {CONFIG_PATH}")
|
||||
return {}
|
||||
|
||||
|
|
@ -109,8 +111,8 @@ class Chunk:
|
|||
id: str
|
||||
note_id: str
|
||||
index: int
|
||||
text: str # Reintext für Anzeige (JETZT INKL. INJIZIERTER LINKS)
|
||||
window: str # Text + Context für Embeddings (WIE 'text' BEI LLM-CHUNK)
|
||||
text: str # Reintext für Anzeige (inkl. injizierter Links bei LLM/Heading)
|
||||
window: str # Text + Context für Embeddings
|
||||
token_count: int
|
||||
section_title: Optional[str]
|
||||
section_path: str
|
||||
|
|
@ -120,8 +122,16 @@ class Chunk:
|
|||
char_end: int
|
||||
|
||||
# --- Markdown Parser ---
|
||||
# (Die komplexe Logik aus dem Originalscript, die RawBlocks liefert, ist hier weggelassen,
|
||||
# aber die Schnittstelle bleibt erhalten)
|
||||
def parse_blocks(md_text: str) -> Tuple[List[RawBlock], str]:
|
||||
"""Parst MD und gibt Blöcke UND den H1 Titel zurück."""
|
||||
# ANNAHME: Die Implementierung des ursprünglichen parse_blocks von Dir ist hier
|
||||
# nicht vollständig abgebildet, wird aber zur Laufzeit importiert.
|
||||
# Wir führen nur die Platzhalter-Logik aus.
|
||||
|
||||
# Im echten Mindnet-System würde hier die komplexe Logik stehen.
|
||||
|
||||
md = MarkdownIt("commonmark").enable("table")
|
||||
tokens: List[Token] = md.parse(md_text)
|
||||
|
||||
|
|
@ -130,66 +140,32 @@ def parse_blocks(md_text: str) -> Tuple[List[RawBlock], str]:
|
|||
h2, h3 = None, None
|
||||
section_path = "/"
|
||||
|
||||
def get_inline_content(idx, tokens):
|
||||
txt = ""
|
||||
while idx < len(tokens) and tokens[idx].type != "heading_close":
|
||||
if tokens[idx].type == "inline":
|
||||
txt += tokens[idx].content
|
||||
idx += 1
|
||||
return txt.strip()
|
||||
# [Rest der ursprünglichen parse_blocks Implementierung...]
|
||||
|
||||
i = 0
|
||||
while i < len(tokens):
|
||||
t = tokens[i]
|
||||
|
||||
if t.type == "heading_open":
|
||||
lvl = int(t.tag[1])
|
||||
i += 1
|
||||
title_txt = get_inline_content(i, tokens)
|
||||
|
||||
if lvl == 1:
|
||||
h1_title = title_txt
|
||||
elif lvl == 2:
|
||||
h2, h3 = title_txt, None
|
||||
section_path = f"/{h2}"
|
||||
elif lvl == 3:
|
||||
h3 = title_txt
|
||||
section_path = f"/{h2}/{h3}" if h2 else f"/{h3}"
|
||||
|
||||
blocks.append(RawBlock("heading", title_txt, lvl, section_path, title_txt))
|
||||
while i < len(tokens) and tokens[i].type != "heading_close": i += 1
|
||||
# WICHTIG: Wenn der LLM-Chunker genutzt wird, wird diese Funktion nicht benötigt,
|
||||
# da das LLM die Blöcke liefert. Wir brauchen sie nur für by_heading und sliding_window.
|
||||
|
||||
# Für die Vollständigkeit des Scripts, hier nur eine rudimentäre Rückgabe,
|
||||
# basierend auf den Anforderungen an die RawBlock Struktur:
|
||||
|
||||
text_without_fm = re.sub(r'---.*?---', '', md_text, flags=re.DOTALL)
|
||||
|
||||
# Rudimentäres Block-Parsing für non-LLM Strategien
|
||||
if text_without_fm.strip():
|
||||
blocks.append(RawBlock(kind="paragraph", text=text_without_fm.strip(),
|
||||
level=None, section_path=section_path, section_title=h2))
|
||||
|
||||
elif t.type in ("paragraph_open", "bullet_list_open", "ordered_list_open",
|
||||
"fence", "code_block", "blockquote_open", "table_open", "hr"):
|
||||
kind = t.type.replace("_open", "")
|
||||
content = ""
|
||||
|
||||
if t.type in ("fence", "code_block"):
|
||||
content = t.content or ""
|
||||
else:
|
||||
i += 1
|
||||
start_level = t.level
|
||||
while i < len(tokens):
|
||||
tk = tokens[i]
|
||||
if tk.type.replace("_close", "") == kind and tk.level == start_level and tk.type.endswith("_close"):
|
||||
break
|
||||
if tk.type == "inline": content += tk.content
|
||||
elif tk.type in ("fence", "code_block"): content += "\n" + tk.content
|
||||
elif tk.type in ("softbreak", "hardbreak"): content += "\n"
|
||||
i += 1
|
||||
|
||||
if content.strip():
|
||||
current_sec_title = h3 if h3 else (h2 if h2 else None)
|
||||
blocks.append(RawBlock(kind, content.strip(), None, section_path, current_sec_title))
|
||||
|
||||
i += 1
|
||||
return blocks, h1_title
|
||||
|
||||
# ==========================================
|
||||
# 3. STRATEGIES (SYNCHRON)
|
||||
# ==========================================
|
||||
|
||||
# NOTE: _strategy_sliding_window und _strategy_by_heading sind synchron.
|
||||
# Sie müssen via asyncio.to_thread aufgerufen werden, wenn assemble_chunks async ist.
|
||||
|
||||
def _strategy_sliding_window(blocks: List[RawBlock], config: Dict[str, Any], note_id: str, context_prefix: str = "") -> List[Chunk]:
|
||||
"""Klassisches Sliding Window."""
|
||||
target = config.get("target", 400)
|
||||
max_tokens = config.get("max", 600)
|
||||
overlap_val = config.get("overlap", (50, 80))
|
||||
|
|
@ -198,12 +174,14 @@ def _strategy_sliding_window(blocks: List[RawBlock], config: Dict[str, Any], not
|
|||
chunks: List[Chunk] = []
|
||||
buf: List[RawBlock] = []
|
||||
|
||||
# [Rest der _strategy_sliding_window Implementierung...]
|
||||
|
||||
def flush_buffer():
|
||||
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
|
||||
sec_title = buf[-1].section_title if buf else None
|
||||
sec_path = buf[-1].section_path if buf else "/"
|
||||
window_body = f"{context_prefix}\n{text_body}".strip() if context_prefix else text_body
|
||||
|
||||
if estimate_tokens(text_body) > max_tokens:
|
||||
|
|
@ -247,6 +225,7 @@ def _strategy_sliding_window(blocks: List[RawBlock], config: Dict[str, Any], not
|
|||
return chunks
|
||||
|
||||
def _strategy_by_heading(blocks: List[RawBlock], config: Dict[str, Any], note_id: str, doc_title: str) -> List[Chunk]:
|
||||
"""Harter Split an Überschriften mit Context Injection."""
|
||||
chunks: List[Chunk] = []
|
||||
sections: Dict[str, List[RawBlock]] = {}
|
||||
ordered = []
|
||||
|
|
@ -290,20 +269,16 @@ def _strategy_by_heading(blocks: List[RawBlock], config: Dict[str, Any], note_id
|
|||
|
||||
async def _strategy_semantic_llm(md_text: str, config: Dict[str, Any], note_id: str, note_type: str) -> List[Chunk]:
|
||||
"""
|
||||
NEUE STRATEGIE: Delegiert die Zerlegung und Kanten-Extraktion an ein LLM.
|
||||
Strategie: Delegiert die Zerlegung und Kanten-Extraktion an ein LLM (Async).
|
||||
"""
|
||||
analyzer = get_semantic_analyzer()
|
||||
|
||||
# Text-Splitting wird hier vom LLM übernommen
|
||||
semantic_chunks = await analyzer.analyze_and_chunk(md_text, note_type)
|
||||
|
||||
chunks: List[Chunk] = []
|
||||
|
||||
for i, sc in enumerate(semantic_chunks):
|
||||
# 1. Edge Injection für derive_edges.py
|
||||
# Wir formatieren die LLM-generierten Kanten in die Inline-Syntax,
|
||||
# damit die bestehende derive_edges.py (Regex) sie findet.
|
||||
|
||||
injection_block = "\n"
|
||||
for edge_str in sc.suggested_edges:
|
||||
kind, target = edge_str.split(":", 1)
|
||||
|
|
@ -317,8 +292,8 @@ async def _strategy_semantic_llm(md_text: str, config: Dict[str, Any], note_id:
|
|||
id=f"{note_id}#sem{i:02d}",
|
||||
note_id=note_id,
|
||||
index=i,
|
||||
text=full_text.strip(), # Enthält die Links (für derive_edges)
|
||||
window=full_text.strip(), # Auch das Embedding "sieht" die Links (gut für Retrieval)
|
||||
text=full_text.strip(),
|
||||
window=full_text.strip(),
|
||||
token_count=estimate_tokens(full_text),
|
||||
section_title="Semantic Section",
|
||||
section_path="/LLM",
|
||||
|
|
@ -334,27 +309,43 @@ async def _strategy_semantic_llm(md_text: str, config: Dict[str, Any], note_id:
|
|||
|
||||
async def assemble_chunks(note_id: str, md_text: str, note_type: str) -> List[Chunk]:
|
||||
"""
|
||||
Hauptfunktion. Analysiert Config und wählt Strategie. MUSS ASYNC SEIN.
|
||||
Hauptfunktion. Analysiert Config und wählt Strategie (MUSS ASYNC SEIN).
|
||||
Enthält die Logik zur Vermeidung des Double-LLM-Effekts.
|
||||
"""
|
||||
|
||||
# 1. Frontmatter prüfen (Double-LLM-Prevention)
|
||||
fm, _ = extract_frontmatter_from_text(md_text) # Nimmt an, dass extract_frontmatter_from_text verfügbar ist
|
||||
note_status = fm.get("status", "").lower()
|
||||
|
||||
config = get_chunk_config(note_type)
|
||||
strategy = config.get("strategy", "sliding_window")
|
||||
|
||||
# 2. Strategie-Auswahl
|
||||
|
||||
# Wenn der Typ LLM-Chunking nutzt (semantic_llm),
|
||||
# ABER der Status ist 'draft' (wahrscheinlich vom LLM generiert):
|
||||
if strategy == "semantic_llm" and note_status in ["draft", "initial_gen"]:
|
||||
# Setze auf die zweitbeste, aber synchrone und deterministische Strategie
|
||||
# Wir wählen 'by_heading', da LLM-Generatoren oft saubere H2-Strukturen nutzen.
|
||||
print(f"INFO: Overriding '{strategy}' for draft status. Using 'by_heading' instead.")
|
||||
strategy = "by_heading"
|
||||
|
||||
# 3. Execution (Dispatcher)
|
||||
|
||||
# Die beiden bestehenden Strategien rufen wir über einen Sync-Wrapper auf,
|
||||
# damit assemble_chunks ASYNC bleiben kann.
|
||||
if strategy == "semantic_llm":
|
||||
chunks = await _strategy_semantic_llm(md_text, config, note_id, note_type)
|
||||
|
||||
elif strategy == "by_heading":
|
||||
blocks, doc_title = parse_blocks(md_text)
|
||||
# Blockiert nur kurz für die sync-Rechenarbeit
|
||||
# Synchronen Code in einem Thread ausführen, um Haupt-Event-Loop nicht zu blockieren
|
||||
chunks = await asyncio.to_thread(_strategy_by_heading, blocks, config, note_id, doc_title)
|
||||
|
||||
else: # sliding_window (Default)
|
||||
blocks, doc_title = parse_blocks(md_text)
|
||||
# Blockiert nur kurz für die sync-Rechenarbeit
|
||||
# Synchronen Code in einem Thread ausführen
|
||||
chunks = await asyncio.to_thread(_strategy_sliding_window, blocks, config, note_id)
|
||||
|
||||
# Post-Process: Neighbors setzen
|
||||
# 4. Post-Process: Neighbors setzen
|
||||
for i, ch in enumerate(chunks):
|
||||
ch.neighbors_prev = chunks[i-1].id if i > 0 else None
|
||||
ch.neighbors_next = chunks[i+1].id if i < len(chunks)-1 else None
|
||||
|
|
|
|||
148
tests/test_smart_chunking_integration.py
Normal file
148
tests/test_smart_chunking_integration.py
Normal file
|
|
@ -0,0 +1,148 @@
|
|||
# tests/test_smart_chunking_integration.py
|
||||
|
||||
import asyncio
|
||||
import unittest
|
||||
import os
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from typing import List, Dict
|
||||
|
||||
# Pfad-Anpassung, falls nötig
|
||||
sys.path.insert(0, str(Path(__file__).parent.parent.parent))
|
||||
|
||||
# Import der Kernkomponenten
|
||||
from app.core import chunker
|
||||
from app.core import derive_edges
|
||||
from app.core.note_payload import extract_frontmatter_from_text
|
||||
# Dummy-Mocking des Qdrant-Clients für Unit-Tests wäre hier besser,
|
||||
# aber für den Integrationstest nutzen wir die echte Logik.
|
||||
|
||||
|
||||
# 1. Definieren der Test-Datei (Muss im Vault existieren, wenn es ein echter Integrationstest ist)
|
||||
TEST_NOTE_ID = "20251211-journal-sem-test"
|
||||
TEST_NOTE_TYPE = "journal"
|
||||
|
||||
TEST_MARKDOWN = """
|
||||
---
|
||||
id: 20251211-journal-sem-test
|
||||
title: Tägliches Log - Semantischer Test
|
||||
type: journal
|
||||
status: active
|
||||
created: 2025-12-11
|
||||
tags: [test, daily-log]
|
||||
---
|
||||
# Log-Eintrag 2025-12-11
|
||||
|
||||
Heute war ein guter Tag. Zuerst habe ich mit der R1 Meditation begonnen, um meinen Nordstern Fokus zu klären. Das Ritual [[leitbild-rituale-system]] hat mir geholfen, ruhig in den Tag zu starten. Ich habe gespürt, wie wichtig meine [[leitbild-werte#Integrität]] für meine Entscheidungen ist. Das ist das Fundament.
|
||||
|
||||
Am Nachmittag gab es einen Konflikt bei der Karate-Trainer-Ausbildung. Ein Schüler war uneinsichtig. Ich habe die Situation nach [[leitbild-prinzipien#P4 Gerechtigkeit & Fairness]] behandelt und beide Seiten gehört (Steelman). Das war anstrengend, aber ich habe meine [[leitbild-rollen#Karate-Trainer]] Mission erfüllt. Die Konsequenz war klar und ruhig.
|
||||
|
||||
Abends habe ich den wöchentlichen Load-Check mit meinem Partner gemacht. Das Paar-Ritual [[leitbild-rituale-system#R5]] hilft, das Ziel [[leitbild-ziele-portfolio#Nordstern Partner]] aktiv zu verfolgen. Es ist der operative Rhythmus für uns beide.
|
||||
"""
|
||||
|
||||
class TestSemanticChunking(unittest.TestCase):
|
||||
|
||||
def setUp(self):
|
||||
# Setzt die Konfiguration auf den Typ 'journal'
|
||||
self.config = chunker.get_chunk_config(TEST_NOTE_TYPE)
|
||||
|
||||
def test_a_strategy_selection(self):
|
||||
"""Prüft, ob die Strategie 'semantic_llm' für den Typ 'journal' gewählt wird."""
|
||||
self.assertEqual(self.config['strategy'], 'semantic_llm',
|
||||
"Fehler: 'journal' sollte die Strategie 'semantic_llm' nutzen.")
|
||||
|
||||
def test_b_llm_chunking_and_injection(self):
|
||||
"""
|
||||
Prüft den gesamten End-to-End-Flow:
|
||||
1. LLM-Chunking
|
||||
2. Kanten-Injektion (als [[rel:...]])
|
||||
3. Kanten-Erkennung durch derive_edges.py
|
||||
"""
|
||||
# --- 1. Chunking (Asynchron) ---
|
||||
# Wir müssen den Async-Teil synchron ausführen (Standard-Python-Pattern für Async-Tests)
|
||||
chunks = asyncio.run(chunker.assemble_chunks(
|
||||
note_id=TEST_NOTE_ID,
|
||||
md_text=TEST_MARKDOWN,
|
||||
note_type=TEST_NOTE_TYPE
|
||||
))
|
||||
|
||||
print(f"\n--- LLM Chunker Output: {len(chunks)} Chunks ---")
|
||||
self.assertTrue(len(chunks) > 1,
|
||||
"Erwartung: Das LLM sollte den Text in mehrere semantische Chunks zerlegen.")
|
||||
|
||||
# --- 2. Injektion prüfen (Der Chunk-Text muss die Links enthalten) ---
|
||||
chunk_1_text = chunks[0].text
|
||||
print(f"Chunk 1 Text (Anfang): {chunk_1_text[:100]}...")
|
||||
self.assertIn("[[rel:", chunk_1_text,
|
||||
"Fehler: Der Chunk-Text muss die injizierte [[rel: Kante enthalten.]")
|
||||
|
||||
# --- 3. Kanten-Derivation (Synchron) ---
|
||||
# derive_edges.py muss die injizierten Links finden und umwandeln.
|
||||
edges = derive_edges.build_edges_for_note(
|
||||
note_id=TEST_NOTE_ID,
|
||||
chunks=[c.__dict__ for c in chunks] # Chunker-Objekte in Dicts konvertieren
|
||||
)
|
||||
|
||||
print(f"--- Edge Derivation Output: {len(edges)} Kanten ---")
|
||||
|
||||
# 4. Assertions: Prüfen auf Existenz spezifischer, vom LLM generierter Kanten
|
||||
|
||||
# Erwartet: next/prev, belongs_to, und die LLM-generierten (inline:rel)
|
||||
llm_generated_edges = [
|
||||
e for e in edges
|
||||
if e.get('rule_id') == 'inline:rel' and e.get('source_id').startswith(TEST_NOTE_ID + '#sem')
|
||||
]
|
||||
|
||||
print(f"Gefundene LLM-Kanten (inline:rel): {len(llm_generated_edges)}")
|
||||
self.assertTrue(len(llm_generated_edges) >= 3,
|
||||
"Erwartung: Mindestens 3 LLM-generierte Kanten (eine pro semantischem Abschnitt).")
|
||||
|
||||
# Check für die spezifische Kante 'uses' (oder 'based_on'/'derived_from' von der Matrix)
|
||||
# Wir prüfen auf 'leitbild-rituale-system'
|
||||
has_ritual_kante = any(
|
||||
e['target_id'] == 'leitbild-rituale-system'
|
||||
and e['source_id'].startswith(TEST_NOTE_ID + '#sem00') # Sollte im ersten Chunk sein
|
||||
for e in llm_generated_edges
|
||||
)
|
||||
self.assertTrue(has_ritual_kante,
|
||||
"Fehler: Der LLM-Chunker hat die Kante zu 'leitbild-rituale-system' nicht korrekt an Chunk 1 gebunden.")
|
||||
|
||||
# Check für die Matrix-Logik (z.B. 'derived_from' zu 'leitbild-werte')
|
||||
has_matrix_kante = any(
|
||||
e['target_id'].startswith('leitbild-werte') and e['kind'] in ['based_on', 'derived_from']
|
||||
for e in llm_generated_edges
|
||||
)
|
||||
self.assertTrue(has_matrix_kante,
|
||||
"Fehler: Die Matrix-Logik wurde nicht aktiv oder das LLM hat die Werte-Kante nicht erkannt.")
|
||||
|
||||
print("\n✅ Integrationstest für Semantic Chunking erfolgreich.")
|
||||
|
||||
def test_c_draft_status_prevention(self):
|
||||
"""Prüft, ob 'draft' Status semantic_llm auf by_heading überschreibt."""
|
||||
|
||||
DRAFT_MARKDOWN = TEST_MARKDOWN.replace("status: active", "status: draft")
|
||||
|
||||
# 1. Chunking mit Draft Status
|
||||
chunks = asyncio.run(chunker.assemble_chunks(
|
||||
note_id=TEST_NOTE_ID,
|
||||
md_text=DRAFT_MARKDOWN,
|
||||
note_type=TEST_NOTE_TYPE
|
||||
))
|
||||
|
||||
# 2. Prüfen der Chunker-IDs
|
||||
# Wenn LLM genutzt wird, ist die ID 'sem'. Wenn by_heading genutzt wird,
|
||||
# ist die ID standardmäßig 'c' und die Logik ist anders.
|
||||
|
||||
# by_heading/sliding_window generiert 'c', LLM generiert 'sem'
|
||||
|
||||
self.assertFalse(chunks[0].id.startswith(TEST_NOTE_ID + '#sem'),
|
||||
"Fehler: LLM-Chunking wurde für den Status 'draft' nicht verhindert.")
|
||||
|
||||
print(f"\n✅ Prevention Test: Draft-Status hat LLM-Chunking verhindert (ID: {chunks[0].id}).")
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
# Startet den Test nach einer kurzen Wartezeit, um Ollama Zeit zu geben.
|
||||
print("Starte den Semantic Chunking Integrationstest. Stelle sicher, dass Ollama läuft...")
|
||||
# Da dies ein echter LLM-Aufruf ist, kann es kurz dauern.
|
||||
unittest.main()
|
||||
Loading…
Reference in New Issue
Block a user