diff --git a/app/core/chunker.py b/app/core/chunker.py index e75b835..7634bcd 100644 --- a/app/core/chunker.py +++ b/app/core/chunker.py @@ -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": diff --git a/tests/test_smart_chunking_integration.py b/tests/test_smart_chunking_integration.py index 8a2ca33..fcb167f 100644 --- a/tests/test_smart_chunking_integration.py +++ b/tests/test_smart_chunking_integration.py @@ -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 --- \ No newline at end of file +if __name__ == '__main__': + print("Starte den Semantic Chunking Integrationstest. Stelle sicher, dass Ollama und die Konfiguration korrekt sind.") + unittest.main() \ No newline at end of file