93 lines
3.2 KiB
Python
93 lines
3.2 KiB
Python
"""
|
|
app/services/semantic_analyzer.py
|
|
Zweck: Asynchroner Service zur Zuweisung von Kanten zu Text-Chunks mittels LLM.
|
|
Nutzt Templates aus prompts.yaml.
|
|
"""
|
|
|
|
import json
|
|
import logging
|
|
from typing import List, Optional
|
|
from dataclasses import dataclass
|
|
|
|
# Importe
|
|
from app.services.llm_service import LLMService
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
class SemanticAnalyzer:
|
|
def __init__(self):
|
|
self.llm = LLMService()
|
|
|
|
async def assign_edges_to_chunk(self, chunk_text: str, all_edges: List[str], note_type: str) -> List[str]:
|
|
"""
|
|
Sendet einen Chunk und eine Liste potenzieller Kanten an das LLM.
|
|
Das LLM filtert heraus, welche Kanten für diesen Chunk relevant sind.
|
|
"""
|
|
if not all_edges:
|
|
return []
|
|
|
|
# 1. Prompt laden
|
|
prompt_template = self.llm.prompts.get("edge_allocation_template")
|
|
if not prompt_template:
|
|
logger.error("Prompt 'edge_allocation_template' in prompts.yaml nicht gefunden.")
|
|
return []
|
|
|
|
# 2. Kandidaten-Liste formatieren
|
|
# Wir übergeben die Kanten als einfache Liste, damit das LLM sie auswählen kann.
|
|
edges_str = "\n".join([f"- {e}" for e in all_edges])
|
|
|
|
# 3. Prompt füllen
|
|
final_prompt = prompt_template.format(
|
|
chunk_text=chunk_text[:3000], # Truncate safety
|
|
edge_list=edges_str
|
|
)
|
|
|
|
try:
|
|
# 4. LLM Call mit JSON Erzwingung
|
|
response_json = await self.llm.generate_raw_response(
|
|
prompt=final_prompt,
|
|
force_json=True
|
|
)
|
|
|
|
# 5. Parsing
|
|
clean_json = response_json.replace("```json", "").replace("```", "").strip()
|
|
|
|
# Fallback für leere Antworten
|
|
if not clean_json:
|
|
return []
|
|
|
|
data = json.loads(clean_json)
|
|
|
|
# 6. Validierung: Wir erwarten eine Liste von Strings
|
|
if isinstance(data, list):
|
|
# Filtern: Nur Strings zurückgeben, die auch in der Input-Liste waren (Sicherheit)
|
|
# oder zumindest das korrekte Format haben.
|
|
valid_edges = [str(e) for e in data if isinstance(e, str) and ":" in e]
|
|
return valid_edges
|
|
elif isinstance(data, dict):
|
|
# Manchmal packt das LLM es in {"edges": [...]}
|
|
for key, val in data.items():
|
|
if isinstance(val, list):
|
|
return [str(e) for e in val if isinstance(e, str)]
|
|
|
|
logger.warning(f"SemanticAnalyzer: Unerwartetes JSON Format: {str(data)[:100]}")
|
|
return []
|
|
|
|
except json.JSONDecodeError:
|
|
logger.warning("SemanticAnalyzer: LLM lieferte kein valides JSON. Keine Kanten zugewiesen.")
|
|
return []
|
|
except Exception as e:
|
|
logger.error(f"SemanticAnalyzer Error: {e}")
|
|
return []
|
|
|
|
async def close(self):
|
|
if self.llm:
|
|
await self.llm.close()
|
|
|
|
# Singleton Helper
|
|
_analyzer_instance = None
|
|
def get_semantic_analyzer():
|
|
global _analyzer_instance
|
|
if _analyzer_instance is None:
|
|
_analyzer_instance = SemanticAnalyzer()
|
|
return _analyzer_instance |