134 lines
5.7 KiB
Python
134 lines
5.7 KiB
Python
"""
|
|
app/services/semantic_analyzer.py — Edge Validation & Filtering
|
|
Version: Final (Nutzt Prompts.yaml Template)
|
|
"""
|
|
|
|
import json
|
|
import logging
|
|
from typing import List, Dict, Any, Optional
|
|
from dataclasses import dataclass
|
|
|
|
# Import der benötigten Services (Annahme: llm_service und discovery sind verfügbar.)
|
|
from app.services.llm_service import LLMService
|
|
from app.services.discovery import DiscoveryService
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
@dataclass
|
|
class SemanticChunkResult:
|
|
content: str
|
|
suggested_edges: List[str] # Format: "kind:Target"
|
|
|
|
class SemanticAnalyzer:
|
|
def __init__(self):
|
|
self.llm = LLMService()
|
|
self.discovery = DiscoveryService()
|
|
self.MAX_CONTEXT_TOKENS = 3000
|
|
# NEU: Prompts aus dem LLMService laden
|
|
self.edge_template = self.llm.prompts.get("edge_allocation_template", "")
|
|
|
|
async def analyze_and_chunk(
|
|
self,
|
|
text: str,
|
|
source_type: str,
|
|
# NEU: all_note_edges ist jetzt ein zwingendes Argument für diesen Workflow
|
|
all_note_edges: List[str],
|
|
target_type_resolver: Optional[callable] = None
|
|
) -> List[SemanticChunkResult]:
|
|
"""
|
|
[WP-15] Führt die semantische Analyse durch (Zerlegung ODER Kantenfilterung).
|
|
Da wir nur den 5-Schritte-Workflow nutzen, wird dies primär als Kantenfilter genutzt.
|
|
"""
|
|
|
|
if not self.edge_template:
|
|
logger.error("SemanticAnalyzer: 'edge_allocation_template' fehlt in prompts.yaml!")
|
|
return [SemanticChunkResult(content=text, suggested_edges=[])]
|
|
|
|
# Standard-Resolver verwenden, wenn keiner übergeben wird
|
|
if target_type_resolver is None:
|
|
target_type_resolver = self._default_target_type_resolver
|
|
|
|
edge_list_str = "\n".join([f"- {e}" for e in all_note_edges])
|
|
|
|
# 1. Prompt mit Template füllen
|
|
# Wir nutzen den ersten Teil des Templates als System/Rolle und den Rest als User-Prompt
|
|
|
|
# NOTE: Da wir das Template direkt aus prompts.yaml laden, enthält es die SYSTEM/ROLLE direkt
|
|
final_prompt = self.edge_template.format(
|
|
note_type=source_type,
|
|
chunk_text=text,
|
|
edge_list_str=edge_list_str
|
|
)
|
|
|
|
# Wir trennen den System-Teil (bis zur ANWEISUNG) nicht mehr manuell,
|
|
# sondern übergeben den gesamten Prompt und lassen das LLM die Rolle verstehen.
|
|
|
|
try:
|
|
# 2. LLM Call (Async)
|
|
response_json = await self.llm.generate_raw_response(
|
|
final_prompt,
|
|
system=None, # System-Rolle ist im Template enthalten
|
|
force_json=True
|
|
)
|
|
|
|
clean_json = response_json.replace("```json", "").replace("```", "").strip()
|
|
data = json.loads(clean_json)
|
|
|
|
# --- Robuste Parsing-Logik (wie in den Korrekturen etabliert) ---
|
|
if isinstance(data, dict):
|
|
data = [data]
|
|
elif not isinstance(data, list):
|
|
logger.error("SemanticAnalyzer: JSON root ist weder Array noch Objekt. Fehlerhafte LLM-Antwort.")
|
|
raise ValueError("Root element is not a list or dictionary.")
|
|
|
|
# Da wir im 5-Schritte-Workflow nur ein Array von Kanten-Strings erwarten:
|
|
# Wir behandeln das Resultat (data) als die gefilterte Kantenliste
|
|
|
|
if not data:
|
|
return []
|
|
|
|
filtered_edges = []
|
|
for item in data:
|
|
# WENN data ein Array von Strings ist (wie im edge_allocation_template):
|
|
if isinstance(item, str) and ":" in item:
|
|
# Um die Matrix-Logik zu aktivieren, muss jedes Item einmal durch die Matrix.
|
|
# Dies ist komplex, da wir den Typ der ZIEL-Entität benötigen.
|
|
|
|
target = item.split(":", 1)[1].strip()
|
|
target_entity_type = target_type_resolver(target)
|
|
|
|
# Hier MUSS der Edge-String manuell korrigiert werden, da der LLM-Output
|
|
# die Matrix-Logik ignoriert. Wir simulieren die Korrektur hier nicht mehr,
|
|
# sondern vertrauen dem LLM-Output (item) und überlassen die Matrix-Anwendung
|
|
# dem derive_edges.py (wo sie hingehört).
|
|
filtered_edges.append(item) # Füge den LLM-generierten String hinzu
|
|
|
|
# Wenn das LLM fälschlicherweise das alte Format {content:..., relations:[...]} liefert,
|
|
# ignorieren wir dies, da das edge_allocation_template ein Array von Strings erwartet.
|
|
|
|
# Das LLM hat nun die Kanten für den Chunk gefiltert.
|
|
return [SemanticChunkResult(content=text, suggested_edges=filtered_edges)]
|
|
|
|
except json.JSONDecodeError:
|
|
logger.error("SemanticAnalyzer: LLM lieferte KEIN valides JSON. Fallback auf Raw Text.")
|
|
return [SemanticChunkResult(content=text, suggested_edges=[])]
|
|
except Exception as e:
|
|
logger.error(f"SemanticAnalyzer Unbehandelter Fehler: {e}")
|
|
return [SemanticChunkResult(content=text, suggested_edges=[])]
|
|
|
|
# NEU: Abstrakter Fallback-Resolver
|
|
def _default_target_type_resolver(self, title: str) -> str:
|
|
"""Standard-Fallback, wenn kein Resolver übergeben wird (immer 'concept')."""
|
|
return "concept"
|
|
|
|
async def close(self):
|
|
if self.llm:
|
|
await self.llm.close()
|
|
|
|
# Export des Singleton-Helpers
|
|
_analyzer_instance = None
|
|
def get_semantic_analyzer():
|
|
global _analyzer_instance
|
|
if _analyzer_instance is None:
|
|
_analyzer_instance = SemanticAnalyzer()
|
|
return _analyzer_instance |