""" FILE: app/core/retrieval/decision_engine.py DESCRIPTION: Der Agentic Orchestrator für MindNet (WP-25a Edition). Realisiert Multi-Stream Retrieval, Intent-basiertes Routing und die neue Pre-Synthesis Kompression (Module A). VERSION: 1.2.0 (WP-25a: Mixture of Experts Support) STATUS: Active FIX: - WP-25a: Vollständige Integration der llm_profile-Steuerung für Synthese und Kompression. - WP-25a: Implementierung der _compress_stream_content Logik zur Inhaltsverdichtung. - WP-25: Beibehaltung von Stream-Tracing und Pre-Initialization Robustness. - COMPATIBILITY: Erhalt aller Methoden-Signaturen für den System-Merge. """ import asyncio import logging import yaml import os from typing import List, Dict, Any, Optional # Core & Service Imports from app.models.dto import QueryRequest, QueryResponse from app.core.retrieval.retriever import Retriever from app.services.llm_service import LLMService from app.config import get_settings logger = logging.getLogger(__name__) class DecisionEngine: def __init__(self): """Initialisiert die Engine und lädt die modularen Konfigurationen.""" self.settings = get_settings() self.retriever = Retriever() self.llm_service = LLMService() self.config = self._load_engine_config() def _load_engine_config(self) -> Dict[str, Any]: """Lädt die Multi-Stream Konfiguration (WP-25/25a).""" path = os.getenv("MINDNET_DECISION_CONFIG", "config/decision_engine.yaml") if not os.path.exists(path): logger.error(f"❌ Decision Engine Config not found at {path}") return {"strategies": {}} try: with open(path, "r", encoding="utf-8") as f: return yaml.safe_load(f) or {} except Exception as e: logger.error(f"❌ Failed to load decision_engine.yaml: {e}") return {"strategies": {}} async def ask(self, query: str) -> str: """ Hauptmethode des MindNet Chats. Orchestriert den agentischen Prozess: Routing -> Retrieval -> Kompression -> Synthese. """ # 1. Intent Recognition (Strategy Routing) strategy_key = await self._determine_strategy(query) strategies = self.config.get("strategies", {}) strategy = strategies.get(strategy_key) if not strategy: logger.warning(f"⚠️ Unknown strategy '{strategy_key}'. Fallback to FACT_WHAT.") strategy_key = "FACT_WHAT" strategy = strategies.get("FACT_WHAT") if not strategy and strategies: strategy_key = next(iter(strategies)) strategy = strategies[strategy_key] if not strategy: return "Entschuldigung, meine Wissensbasis ist aktuell nicht konfiguriert." # 2. Multi-Stream Retrieval & Pre-Synthesis (Parallel Tasks) # WP-25a: Diese Methode übernimmt nun auch die Kompression. stream_results = await self._execute_parallel_streams(strategy, query) # 3. Finale Synthese return await self._generate_final_answer(strategy_key, strategy, query, stream_results) async def _determine_strategy(self, query: str) -> str: """Nutzt den LLM-Router zur Wahl der Such-Strategie.""" prompt_key = self.config.get("settings", {}).get("router_prompt_key", "intent_router_v1") router_prompt_template = self.llm_service.get_prompt(prompt_key) if not router_prompt_template: return "FACT_WHAT" full_prompt = router_prompt_template.format(query=query) try: # Der Router nutzt den Standard-Provider (auto) response = await self.llm_service.generate_raw_response( full_prompt, max_retries=1, priority="realtime" ) return str(response).strip().upper() except Exception as e: logger.error(f"Strategy Routing failed: {e}") return "FACT_WHAT" async def _execute_parallel_streams(self, strategy: Dict, query: str) -> Dict[str, str]: """ Führt Such-Streams aus und komprimiert überlange Ergebnisse (Pre-Synthesis). WP-25a: MoE-Profile werden für die Kompression berücksichtigt. """ stream_keys = strategy.get("use_streams", []) library = self.config.get("streams_library", {}) # Phase 1: Retrieval Tasks starten retrieval_tasks = [] active_streams = [] for key in stream_keys: stream_cfg = library.get(key) if stream_cfg: active_streams.append(key) retrieval_tasks.append(self._run_single_stream(key, stream_cfg, query)) # Ergebnisse sammeln (Exceptions werden als Objekte zurückgegeben) retrieval_results = await asyncio.gather(*retrieval_tasks, return_exceptions=True) # Phase 2: Formatierung und optionale Kompression final_stream_tasks = [] for name, res in zip(active_streams, retrieval_results): if isinstance(res, Exception): logger.error(f"Stream '{name}' failed during retrieval: {res}") async def _err(): return "[Fehler beim Abruf dieses Wissens-Streams]" final_stream_tasks.append(_err()) continue # Formatierung der Hits in Text formatted_context = self._format_stream_context(res) # WP-25a: Kompressions-Check stream_cfg = library.get(name, {}) threshold = stream_cfg.get("compression_threshold", 4000) if len(formatted_context) > threshold: logger.info(f"⚙️ [WP-25a] Compressing stream '{name}' ({len(formatted_context)} chars)...") comp_profile = stream_cfg.get("compression_profile") final_stream_tasks.append( self._compress_stream_content(name, formatted_context, query, comp_profile) ) else: # Direkt-Übernahme als Coroutine für gather() async def _direct(c=formatted_context): return c final_stream_tasks.append(_direct()) # Finale Inhalte (evtl. komprimiert) parallel fertigstellen final_contents = await asyncio.gather(*final_stream_tasks) return dict(zip(active_streams, final_contents)) async def _compress_stream_content(self, stream_name: str, content: str, query: str, profile: Optional[str]) -> str: """ WP-25a Module A: Inhaltsverdichtung via Experten-Modell. """ # Falls kein Profil definiert, nutzen wir das Default-Profil der Strategie compression_prompt = ( f"Du bist ein Wissens-Analyst. Reduziere den folgenden Wissens-Stream '{stream_name}' " f"auf die Informationen, die für die Beantwortung der Frage '{query}' absolut notwendig sind.\n\n" f"BEIBEHALTEN: Harte Fakten, Projektnamen, konkrete Werte und Quellenangaben.\n" f"ENTFERNEN: Redundante Einleitungen, Füllwörter und irrelevante Details.\n\n" f"STREAM-INHALT:\n{content}\n\n" f"KOMPRIMIERTE ANALYSE:" ) try: summary = await self.llm_service.generate_raw_response( compression_prompt, profile_name=profile, # WP-25a: MoE Support priority="background", max_retries=1 ) return summary.strip() if (summary and len(summary.strip()) > 10) else content except Exception as e: logger.error(f"❌ Compression of {stream_name} failed: {e}") return content async def _run_single_stream(self, name: str, cfg: Dict, query: str) -> QueryResponse: """Spezialisierte Graph-Suche mit Stream-Tracing (WP-25).""" transformed_query = cfg.get("query_template", "{query}").format(query=query) request = QueryRequest( query=transformed_query, top_k=cfg.get("top_k", 5), filters={"type": cfg.get("filter_types", [])}, expand={"depth": 1}, boost_edges=cfg.get("edge_boosts", {}), explain=True ) response = await self.retriever.search(request) # WP-25: STREAM-TRACING for hit in response.results: hit.stream_origin = name return response def _format_stream_context(self, response: QueryResponse) -> str: """Wandelt QueryHits in einen formatierten Kontext-String um.""" if not response.results: return "Keine spezifischen Informationen in diesem Stream gefunden." lines = [] for i, hit in enumerate(response.results, 1): source = hit.source.get("path", "Unbekannt") content = hit.source.get("text", "").strip() lines.append(f"[{i}] QUELLE: {source}\nINHALT: {content}") return "\n\n".join(lines) async def _generate_final_answer( self, strategy_key: str, strategy: Dict, query: str, stream_results: Dict[str, str] ) -> str: """Führt die finale Synthese basierend auf dem Strategie-Profil durch.""" # WP-25a: Nutzt das llm_profile der Strategie profile = strategy.get("llm_profile") template_key = strategy.get("prompt_template", "rag_template") # Hier nutzen wir noch den Provider-String für get_prompt (Kompatibilität zu prompts.yaml) # Der llm_service löst das Profil erst bei generate_raw_response auf. template = self.llm_service.get_prompt(template_key) system_prompt = self.llm_service.get_prompt("system_prompt") # WP-25 ROBUSTNESS: Pre-Initialization all_possible_streams = ["values_stream", "facts_stream", "biography_stream", "risk_stream", "tech_stream"] template_vars = {s: "" for s in all_possible_streams} template_vars.update(stream_results) template_vars["query"] = query prepend = strategy.get("prepend_instruction", "") try: final_prompt = template.format(**template_vars) if prepend: final_prompt = f"{prepend}\n\n{final_prompt}" # WP-25a: MoE Call response = await self.llm_service.generate_raw_response( final_prompt, system=system_prompt, profile_name=profile, priority="realtime" ) # Fallback bei leerer Antwort auf lokales Modell if not response or len(response.strip()) < 5: return await self.llm_service.generate_raw_response( final_prompt, system=system_prompt, provider="ollama", priority="realtime" ) return response except KeyError as e: logger.error(f"Template Variable mismatch in '{template_key}': Missing {e}") fallback_context = "\n\n".join([v for v in stream_results.values() if v]) return await self.llm_service.generate_raw_response( f"Beantworte: {query}\n\nKontext:\n{fallback_context}", system=system_prompt, priority="realtime" ) except Exception as e: logger.error(f"Final Synthesis failed: {e}") return "Ich konnte keine Antwort generieren."