WP15-Chunker und Test
This commit is contained in:
parent
9a38daafc0
commit
f6d8751f23
|
|
@ -10,17 +10,19 @@ from markdown_it.token import Token
|
|||
import asyncio
|
||||
|
||||
# NEUE IMPORTS
|
||||
# Import der benötigten Klassen direkt (ersetzt get_semantic_analyzer)
|
||||
try:
|
||||
# ANNAHME: Die Klassen SemanticAnalyzer und SemanticChunkResult existieren in app.services.semantic_analyzer.py
|
||||
from app.services.semantic_analyzer import SemanticAnalyzer, SemanticChunkResult
|
||||
except ImportError:
|
||||
# Fallback für Tests
|
||||
# Fallback für Tests, wenn der Service noch nicht auf dem Pfad ist
|
||||
print("WARNUNG: SemanticAnalyzer Service nicht gefunden. Semantic Chunking wird fehlschlagen.")
|
||||
class SemanticAnalyzer:
|
||||
async def analyze_and_chunk(self, text, type): return [SemanticChunkResult(content=text, suggested_edges=[])]
|
||||
@dataclass
|
||||
class SemanticChunkResult:
|
||||
content: str
|
||||
suggested_edges: List[str]
|
||||
suggested_edges: List[str] # Format: "kind:Target"
|
||||
|
||||
|
||||
# ==========================================
|
||||
|
|
@ -30,8 +32,10 @@ except ImportError:
|
|||
def extract_frontmatter_from_text(md_text: str) -> Tuple[Dict[str, Any], str]:
|
||||
"""
|
||||
Extrakte das YAML Frontmatter aus dem Markdown-Text und gibt den Body zurück.
|
||||
(Lokalisiert im Chunker zur Vermeidung von Import-Fehlern)
|
||||
"""
|
||||
fm_match = re.match(r'^---\s*\n(.*?)\n---', md_text, re.DOTALL)
|
||||
# Regex toleriert Whitespace/Newline vor dem ersten ---
|
||||
fm_match = re.match(r'^\s*---\s*\n(.*?)\n---', md_text, re.DOTALL)
|
||||
|
||||
if not fm_match:
|
||||
return {}, md_text
|
||||
|
|
@ -47,7 +51,7 @@ def extract_frontmatter_from_text(md_text: str) -> Tuple[Dict[str, Any], str]:
|
|||
frontmatter = {}
|
||||
|
||||
# Entferne den Frontmatter Block aus dem Text
|
||||
text_without_fm = re.sub(r'^---\s*\n(.*?)\n---', '', md_text, flags=re.DOTALL)
|
||||
text_without_fm = re.sub(r'^\s*---\s*\n(.*?)\n---', '', md_text, flags=re.DOTALL)
|
||||
|
||||
return frontmatter, text_without_fm.strip()
|
||||
|
||||
|
|
@ -368,8 +372,6 @@ async def assemble_chunks(note_id: str, md_text: str, note_type: str) -> List[Ch
|
|||
# 3. Execution (Dispatcher)
|
||||
|
||||
# Der Text, der an die Chunker-Strategie geht.
|
||||
# Da extract_frontmatter_from_text den Frontmatter entfernt hat,
|
||||
# ist der Body der saubere Text. md_text enthält ihn noch für non-Frontmatter-Logik.
|
||||
md_to_chunk = md_text
|
||||
|
||||
if strategy == "semantic_llm":
|
||||
|
|
|
|||
|
|
@ -8,7 +8,7 @@ from pathlib import Path
|
|||
from typing import List, Dict
|
||||
|
||||
# --- PFAD-KORREKTUR ---
|
||||
# Fügt das Root-Verzeichnis zum Python-Pfad hinzu (wie zuvor besprochen)
|
||||
# Fügt das Root-Verzeichnis zum Python-Pfad hinzu
|
||||
ROOT_DIR = Path(__file__).resolve().parent.parent
|
||||
sys.path.insert(0, str(ROOT_DIR))
|
||||
# ----------------------
|
||||
|
|
@ -16,14 +16,10 @@ sys.path.insert(0, str(ROOT_DIR))
|
|||
# Import der Kernkomponenten
|
||||
from app.core import chunker
|
||||
from app.core import derive_edges
|
||||
from app.services.semantic_analyzer import SemanticAnalyzer # Import der Klasse für die Instanziierung
|
||||
|
||||
# WICHTIG: Wir importieren extract_frontmatter_from_text NICHT mehr aus
|
||||
# note_payload.py, sondern entfernen den Import, da er für den Test nicht direkt nötig ist.
|
||||
|
||||
# ANNAHME: Der Test kann die Logik des Parsings und der Edge-Derivation nutzen,
|
||||
# ohne note_payload direkt zu importieren.
|
||||
|
||||
# 1. Definieren der Test-Datei (Muss im Vault existieren, wenn es ein echter Integrationstest ist)
|
||||
# 1. Definieren der Test-Note (Simuliert eine journal.md Datei)
|
||||
TEST_NOTE_ID = "20251211-journal-sem-test"
|
||||
TEST_NOTE_TYPE = "journal"
|
||||
|
||||
|
|
@ -47,8 +43,26 @@ Abends habe ich den wöchentlichen Load-Check mit meinem Partner gemacht. Das Pa
|
|||
|
||||
class TestSemanticChunking(unittest.TestCase):
|
||||
|
||||
# 2. Ressourcen-Management (Schließt den httpx.AsyncClient sauber)
|
||||
_analyzer_instance = None
|
||||
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
"""Initialisiert den SemanticAnalyzer einmalig und asynchron."""
|
||||
# Da LLMService async ist, nutzen wir die Singleton-Instanz der Klasse
|
||||
cls._analyzer_instance = SemanticAnalyzer()
|
||||
# Stellen Sie sicher, dass der Chunker diese Instanz verwenden kann.
|
||||
# Dies ist im chunker.py Code über _get_semantic_analyzer_instance() abgedeckt.
|
||||
chunker._semantic_analyzer_instance = cls._analyzer_instance
|
||||
|
||||
@classmethod
|
||||
def tearDownClass(cls):
|
||||
"""Schließt den httpx.AsyncClient nach allen Tests."""
|
||||
if cls._analyzer_instance:
|
||||
asyncio.run(cls._analyzer_instance.close())
|
||||
|
||||
def setUp(self):
|
||||
# Setzt die Konfiguration auf den Typ 'journal'
|
||||
# Lädt die Konfiguration, um die Strategie zu prüfen
|
||||
self.config = chunker.get_chunk_config(TEST_NOTE_TYPE)
|
||||
|
||||
def test_a_strategy_selection(self):
|
||||
|
|
@ -59,10 +73,11 @@ class TestSemanticChunking(unittest.TestCase):
|
|||
def test_b_llm_chunking_and_injection(self):
|
||||
"""
|
||||
Prüft den gesamten End-to-End-Flow:
|
||||
1. LLM-Chunking
|
||||
1. LLM-Chunking (muss > 1 Chunk liefern)
|
||||
2. Kanten-Injektion (als [[rel:...]])
|
||||
3. Kanten-Erkennung durch derive_edges.py
|
||||
"""
|
||||
|
||||
# --- 1. Chunking (Asynchron) ---
|
||||
chunks = asyncio.run(chunker.assemble_chunks(
|
||||
note_id=TEST_NOTE_ID,
|
||||
|
|
@ -71,50 +86,42 @@ class TestSemanticChunking(unittest.TestCase):
|
|||
))
|
||||
|
||||
print(f"\n--- LLM Chunker Output: {len(chunks)} Chunks ---")
|
||||
|
||||
# Assertion B1: Zerlegung (Die Fallback-Logik des LLM liefert bei Fehler 1 Chunk)
|
||||
self.assertTrue(len(chunks) > 1,
|
||||
"Erwartung: Das LLM sollte den Text in mehrere semantische Chunks zerlegen.")
|
||||
"Assertion B1 Fehler: 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.]")
|
||||
"Assertion B2 Fehler: Der Chunk-Text muss die injizierte [[rel: Kante enthalten.")
|
||||
|
||||
# --- 3. Kanten-Derivation (Synchron) ---
|
||||
edges = derive_edges.build_edges_for_note(
|
||||
note_id=TEST_NOTE_ID,
|
||||
chunks=[c.__dict__ for c in chunks] # Chunker-Objekte in Dicts konvertieren
|
||||
chunks=[c.__dict__ for c in chunks]
|
||||
)
|
||||
|
||||
print(f"--- Edge Derivation Output: {len(edges)} Kanten ---")
|
||||
|
||||
# 4. Assertions: Prüfen auf Existenz spezifischer, vom LLM generierter Kanten
|
||||
|
||||
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)}")
|
||||
# Assertion B3: Mindestens 3 LLM-Kanten (eine pro semantischem Abschnitt)
|
||||
self.assertTrue(len(llm_generated_edges) >= 3,
|
||||
"Erwartung: Mindestens 3 LLM-generierte Kanten (eine pro semantischem Abschnitt).")
|
||||
"Assertion B3 Fehler: Mindestens 3 LLM-generierte Kanten (eine pro semantischem Abschnitt).")
|
||||
|
||||
# Check für die spezifische Kante 'uses' (oder 'based_on'/'derived_from' von der Matrix)
|
||||
has_ritual_kante = any(
|
||||
e['target_id'] == 'leitbild-rituale-system'
|
||||
and e['source_id'].startswith(TEST_NOTE_ID + '#sem00')
|
||||
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')
|
||||
# Assertion B4: Check für die Matrix-Logik / Werte-Kante (Chunk 1)
|
||||
# Erwartet: derived_from oder based_on zu 'leitbild-werte'
|
||||
has_matrix_kante = any(
|
||||
e['target_id'].startswith('leitbild-werte') and e['kind'] in ['based_on', 'derived_from', 'references']
|
||||
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.")
|
||||
"Assertion B4 Fehler: Die Matrix-Logik / Werte-Kante wurde nicht erkannt.")
|
||||
|
||||
print("\n✅ Integrationstest für Semantic Chunking erfolgreich.")
|
||||
|
||||
|
|
@ -131,16 +138,16 @@ class TestSemanticChunking(unittest.TestCase):
|
|||
))
|
||||
|
||||
# 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.
|
||||
|
||||
# Assertion C1: LLM-Chunking muss verhindert werden (darf NICHT mit '#sem' starten)
|
||||
self.assertFalse(chunks[0].id.startswith(TEST_NOTE_ID + '#sem'),
|
||||
"Fehler: LLM-Chunking wurde für den Status 'draft' nicht verhindert (ID startet mit #sem).")
|
||||
"Assertion C1 Fehler: LLM-Chunking wurde für den Status 'draft' nicht verhindert.")
|
||||
|
||||
# Da 'by_heading' der Fallback ist, sollte die ID mit '#c' starten
|
||||
# Assertion C2: Fallback-Strategie sollte by_heading sein (ID muss mit '#c' starten)
|
||||
self.assertTrue(chunks[0].id.startswith(TEST_NOTE_ID + '#c'),
|
||||
"Fehler: Fallback-Strategie 'by_heading' wurde nicht korrekt ausgeführt.")
|
||||
"Assertion C2 Fehler: Fallback-Strategie 'by_heading' wurde nicht korrekt ausgeführt.")
|
||||
|
||||
print(f"\n✅ Prevention Test: Draft-Status hat LLM-Chunking verhindert (Fallback ID: {chunks[0].id}).")
|
||||
|
||||
# --- Ende des Test-Skripts ---
|
||||
if __name__ == '__main__':
|
||||
print("Starte den Semantic Chunking Integrationstest. Stelle sicher, dass Ollama und die Konfiguration korrekt sind.")
|
||||
unittest.main()
|
||||
Loading…
Reference in New Issue
Block a user