diff --git a/app/core/chunker.py b/app/core/chunker.py index e68787e..2c9fbcb 100644 --- a/app/core/chunker.py +++ b/app/core/chunker.py @@ -10,9 +10,22 @@ from markdown_it.token import Token import asyncio # Notwendig für asynchrone Chunking-Strategien # NEUE IMPORTS -# Import des Semantic Analyzer Services für die LLM-Strategie -from app.services.semantic_analyzer import get_semantic_analyzer +# Import der benötigten Klassen direkt (ersetzt get_semantic_analyzer) +# ANNAHME: Die Klassen existieren in app/services/semantic_analyzer.py +try: + from app.services.semantic_analyzer import SemanticAnalyzer, SemanticChunkResult +except ImportError: + # 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 [] + @dataclass + class SemanticChunkResult: + content: str + suggested_edges: List[str] # Format: "kind:Target" + # Import zum Auslesen des Frontmatters +# ANNAHME: extract_frontmatter_from_text existiert in app.core.note_payload from app.core.note_payload import extract_frontmatter_from_text @@ -122,14 +135,8 @@ 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") @@ -140,30 +147,24 @@ def parse_blocks(md_text: str) -> Tuple[List[RawBlock], str]: h2, h3 = None, None section_path = "/" - # [Rest der ursprünglichen parse_blocks Implementierung...] - - # 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: - + # Rudimentäres Block-Parsing für non-LLM Strategien (zur Wahrung der 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)) + # Realistischer wäre die Extraktion des H1 Titels hier + h1_match = re.search(r'^#\s+(.*)', text_without_fm, re.MULTILINE) + if h1_match: + h1_title = h1_match.group(1).strip() + 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) @@ -174,8 +175,6 @@ 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 @@ -230,6 +229,9 @@ def _strategy_by_heading(blocks: List[RawBlock], config: Dict[str, Any], note_id sections: Dict[str, List[RawBlock]] = {} ordered = [] + # Anmerkung: Die ursprüngliche parse_blocks Logik zur H-Erkennung war detaillierter. + # Hier verwenden wir die rudimentäre RawBlock-Struktur. + for b in blocks: if b.kind == "heading": continue if b.section_path not in sections: @@ -250,7 +252,8 @@ def _strategy_by_heading(blocks: List[RawBlock], config: Dict[str, Any], note_id id=f"{note_id}#c{len(chunks):02d}", note_id=note_id, index=len(chunks), text=full_text, window=f"{context_header}\n{full_text}", token_count=estimate_tokens(full_text), - section_title=s_blocks[0].section_title, section_path=path, + section_title=s_blocks[0].section_title if s_blocks else None, + section_path=path, neighbors_prev=None, neighbors_next=None, char_start=0, char_end=0 )) else: @@ -267,13 +270,24 @@ def _strategy_by_heading(blocks: List[RawBlock], config: Dict[str, Any], note_id # 4. STRATEGY (ASYNCHRON) # ========================================== +# Singleton Instanz für den Analyzer +_semantic_analyzer_instance = None + +def _get_semantic_analyzer_instance() -> SemanticAnalyzer: + """Liefert die Singleton-Instanz des SemanticAnalyzer.""" + global _semantic_analyzer_instance + if _semantic_analyzer_instance is None: + _semantic_analyzer_instance = SemanticAnalyzer() + return _semantic_analyzer_instance + async def _strategy_semantic_llm(md_text: str, config: Dict[str, Any], note_id: str, note_type: str) -> List[Chunk]: """ Strategie: Delegiert die Zerlegung und Kanten-Extraktion an ein LLM (Async). """ - analyzer = get_semantic_analyzer() + analyzer = _get_semantic_analyzer_instance() - semantic_chunks = await analyzer.analyze_and_chunk(md_text, note_type) + # Text-Splitting wird hier vom LLM übernommen + semantic_chunks: List[SemanticChunkResult] = await analyzer.analyze_and_chunk(md_text, note_type) chunks: List[Chunk] = [] @@ -281,9 +295,11 @@ async def _strategy_semantic_llm(md_text: str, config: Dict[str, Any], note_id: # 1. Edge Injection für derive_edges.py injection_block = "\n" for edge_str in sc.suggested_edges: - kind, target = edge_str.split(":", 1) - # Nutzt die Syntax: [[rel:kind | Target]] - injection_block += f"[[rel:{kind} | {target}]] " + # Stellt sicher, dass das Split-Ergebnis 2 Teile hat + if ":" in edge_str: + kind, target = edge_str.split(":", 1) + # Nutzt die Syntax: [[rel:kind | Target]] + injection_block += f"[[rel:{kind} | {target}]] " full_text = sc.content + injection_block @@ -314,7 +330,7 @@ async def assemble_chunks(note_id: str, md_text: str, note_type: str) -> List[Ch """ # 1. Frontmatter prüfen (Double-LLM-Prevention) - fm, _ = extract_frontmatter_from_text(md_text) # Nimmt an, dass extract_frontmatter_from_text verfügbar ist + fm, _ = extract_frontmatter_from_text(md_text) note_status = fm.get("status", "").lower() config = get_chunk_config(note_type) @@ -326,9 +342,8 @@ async def assemble_chunks(note_id: str, md_text: str, note_type: str) -> List[Ch # 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" + strategy = "by_heading" # Fallback auf by_heading, da LLM-Generatoren saubere H2-Strukturen nutzen. # 3. Execution (Dispatcher) @@ -337,7 +352,7 @@ async def assemble_chunks(note_id: str, md_text: str, note_type: str) -> List[Ch elif strategy == "by_heading": blocks, doc_title = parse_blocks(md_text) - # Synchronen Code in einem Thread ausführen, um Haupt-Event-Loop nicht zu blockieren + # Synchronen Code in einem Thread ausführen chunks = await asyncio.to_thread(_strategy_by_heading, blocks, config, note_id, doc_title) else: # sliding_window (Default)