87 lines
3.7 KiB
Python
87 lines
3.7 KiB
Python
"""
|
|
app/services/semantic_analyzer.py — Edge Validation & Filtering
|
|
Der Service ist nun primär dafür zuständig, Kanten aus einer Liste dem gegebenen Chunk zuzuordnen.
|
|
"""
|
|
import json
|
|
import logging
|
|
from typing import List, Dict, Any, Optional
|
|
|
|
# Import der benötigten Services (Annahme: llm_service ist verfügbar.)
|
|
from app.services.llm_service import LLMService
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Ein Singleton-Muster für den Analyzer (wie zuvor)
|
|
_analyzer_instance: Optional['SemanticAnalyzer'] = None
|
|
|
|
def get_semantic_analyzer():
|
|
global _analyzer_instance
|
|
if _analyzer_instance is None:
|
|
_analyzer_instance = SemanticAnalyzer()
|
|
return _analyzer_instance
|
|
|
|
class SemanticAnalyzer:
|
|
def __init__(self):
|
|
# Der DiscoveryService wird hier nicht mehr direkt benötigt.
|
|
self.llm = LLMService()
|
|
|
|
async def filter_edges_for_chunk(self, chunk_text: str, all_note_edges: List[str], note_type: str) -> List[str]:
|
|
"""
|
|
[Schritt 4 des Workflows] Sendet Chunk und alle Kanten an LLM, um die relevanten Kanten für diesen Chunk zu filtern.
|
|
:param chunk_text: Der Text des Chunks zur Analyse.
|
|
:param all_note_edges: Alle für die gesamte Notiz gefundenen Kanten (Format: "kind:Target").
|
|
:param note_type: Der Typ der Notiz.
|
|
:return: Liste der relevanten Kanten für diesen Chunk.
|
|
"""
|
|
if not all_note_edges:
|
|
return []
|
|
|
|
edge_list_str = "\n".join([f"- {e}" for e in all_note_edges])
|
|
|
|
system_prompt = (
|
|
"Du bist ein Edge Filter Agent. Deine Aufgabe ist es, aus einer gegebenen Liste von potentiellen "
|
|
"Knowledge Graph Kanten (Edges) jene auszuwählen, die *semantisch relevant* für den vorliegenden "
|
|
"Textausschnitt sind. Alle Kanten beziehen sich auf die Hauptnotiz.\n"
|
|
"Antworte AUSSCHLIESSLICH mit einer validen JSON-Liste von Kanten-Strings, die im Text direkt erwähnt oder "
|
|
"klar impliziert werden. Es ist KEIN Array von Objekten, sondern ein Array von Strings.\n"
|
|
"Format: [\"kind:Target\", \"kind:Target\", ...]\n"
|
|
"Wähle nur Kanten, die der Chunk *aktiv* benötigt oder referenziert."
|
|
)
|
|
|
|
user_prompt = (
|
|
f"Notiz-Typ: {note_type}\n"
|
|
f"Textausschnitt:\n---\n{chunk_text}\n---\n\n"
|
|
f"Gesamte Kanten der Notiz (AUSWAHL):\n{edge_list_str}\n\n"
|
|
"Welche der oben genannten Kanten sind für diesen Textabschnitt relevant? Liste sie im JSON-Array auf."
|
|
)
|
|
|
|
try:
|
|
# 1. LLM Call
|
|
response_json = await self.llm.generate_raw_response(
|
|
user_prompt,
|
|
system=system_prompt,
|
|
force_json=True
|
|
)
|
|
|
|
# 2. Robustes JSON Parsing
|
|
clean_json = response_json.replace("```json", "").replace("```", "").strip()
|
|
data = json.loads(clean_json)
|
|
|
|
if isinstance(data, list):
|
|
# Filtere nach Strings, die den Doppelpunkt enthalten, um das Format "kind:Target" zu garantieren.
|
|
return [s for s in data if isinstance(s, str) and ":" in s]
|
|
|
|
logger.warning(f"SemanticAnalyzer: LLM lieferte non-list beim Edge-Filtern: {data}")
|
|
return []
|
|
|
|
except json.JSONDecodeError as e:
|
|
logger.error(f"SemanticAnalyzer: LLM lieferte KEIN valides JSON beim Edge-Filtern: {e}")
|
|
return []
|
|
except Exception as e:
|
|
logger.error(f"SemanticAnalyzer Unbehandelter Fehler beim Edge-Filtern: {e}")
|
|
return []
|
|
|
|
async def close(self):
|
|
# Stellt sicher, dass der AsyncClient geschlossen wird (gute Praxis)
|
|
if self.llm:
|
|
await self.llm.close() |