106 lines
3.8 KiB
Python
106 lines
3.8 KiB
Python
"""
|
|
app/services/semantic_analyzer.py — Edge Validation & Filtering
|
|
Version: 1.1 (Robust JSON Parsing)
|
|
"""
|
|
|
|
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")
|
|
|
|
# Fallback, falls Prompt nicht in YAML definiert ist (für Tests ohne volle Config)
|
|
if not prompt_template:
|
|
prompt_template = (
|
|
"TASK: Wähle aus den Kandidaten die relevanten Kanten für den Text.\n"
|
|
"TEXT: {chunk_text}\n"
|
|
"KANDIDATEN: {edge_list}\n"
|
|
"OUTPUT: JSON Liste von Strings [\"kind:target\"]."
|
|
)
|
|
|
|
# 2. Kandidaten-Liste formatieren
|
|
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],
|
|
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 & Cleaning
|
|
clean_json = response_json.replace("```json", "").replace("```", "").strip()
|
|
if not clean_json: return []
|
|
|
|
data = json.loads(clean_json)
|
|
valid_edges = []
|
|
|
|
# 6. Robuste Validierung (List vs Dict)
|
|
if isinstance(data, list):
|
|
# Standardfall: ["kind:target", ...]
|
|
valid_edges = [str(e) for e in data if isinstance(e, str) and ":" in e]
|
|
|
|
elif isinstance(data, dict):
|
|
# Abweichende Formate behandeln
|
|
for key, val in data.items():
|
|
# Fall A: {"edges": ["kind:target"]}
|
|
if key.lower() in ["edges", "results", "kanten"] and isinstance(val, list):
|
|
valid_edges.extend([str(e) for e in val if isinstance(e, str) and ":" in e])
|
|
|
|
# Fall B: {"kind": "target"} (Das beobachtete Format im Log)
|
|
elif isinstance(val, str):
|
|
# Wir rekonstruieren "kind:target"
|
|
valid_edges.append(f"{key}:{val}")
|
|
|
|
# Fall C: {"kind": ["target1", "target2"]}
|
|
elif isinstance(val, list):
|
|
for target in val:
|
|
if isinstance(target, str):
|
|
valid_edges.append(f"{key}:{target}")
|
|
|
|
# Safety: Filtere nur Kanten, die halbwegs valide aussehen
|
|
return [e for e in valid_edges if ":" in e]
|
|
|
|
except json.JSONDecodeError:
|
|
logger.warning("SemanticAnalyzer: LLM lieferte kein valides JSON. Ignoriere Zuweisung.")
|
|
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 |