Lars/mindnet
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 21 Wiki Activity

chunker angepasst

Browse Source
This commit is contained in:
Lars 2025-12-12 08:53:30 +01:00
parent 8436f4192d
commit 2d58220a3c
1 changed files with 47 additions and 32 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

79
app/core/chunker.py
View File

@ -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)