from fastapi import FastAPI, HTTPException from fastapi.responses import Response from pydantic import BaseModel from typing import Literal, Optional, Dict, Any, List import asyncio import json import os import re import yaml import httpx import logging import hashlib import time # For latency metrics # CrewAI Integration try: from crewai_client import should_use_crewai, call_crewai, get_crewai_health CREWAI_CLIENT_AVAILABLE = True except ImportError: CREWAI_CLIENT_AVAILABLE = False should_use_crewai = None call_crewai = None from neo4j import AsyncGraphDatabase # Memory Retrieval Pipeline v3.0 try: from memory_retrieval import memory_retrieval, MemoryBrief MEMORY_RETRIEVAL_AVAILABLE = True except ImportError: MEMORY_RETRIEVAL_AVAILABLE = False memory_retrieval = None # Tool Manager for Function Calling try: from tool_manager import ToolManager, ToolResult, format_tool_calls_for_response TOOL_MANAGER_AVAILABLE = True except ImportError: TOOL_MANAGER_AVAILABLE = False ToolManager = None # Runtime Guard (Envelope/Artifact validation for CLAN orchestration) try: from runtime_guard import RuntimeGuard RUNTIME_GUARD_AVAILABLE = True except ImportError: RUNTIME_GUARD_AVAILABLE = False RuntimeGuard = None logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) TRUSTED_DOMAINS_CONFIG_PATH = os.getenv("TRUSTED_DOMAINS_CONFIG_PATH", "./trusted_domains.yml") _trusted_domains_cache: Dict[str, Any] = {"mtime": None, "data": {}} def _load_trusted_domains_overrides() -> Dict[str, Any]: """Load optional trusted domains overrides editable by mentors.""" global _trusted_domains_cache try: if not os.path.exists(TRUSTED_DOMAINS_CONFIG_PATH): return {} mtime = os.path.getmtime(TRUSTED_DOMAINS_CONFIG_PATH) if _trusted_domains_cache.get("mtime") == mtime: return _trusted_domains_cache.get("data") or {} with open(TRUSTED_DOMAINS_CONFIG_PATH, "r", encoding="utf-8") as f: raw = yaml.safe_load(f) or {} if not isinstance(raw, dict): raw = {} _trusted_domains_cache = {"mtime": mtime, "data": raw} return raw except Exception as e: logger.warning(f"⚠️ Failed to load trusted domains overrides: {e}") return {} def _strip_dsml_keep_text_before(text: str) -> str: """If response contains DSML, return only the part before the first DSML-like tag. Otherwise return empty (caller will use fallback).""" if not text or len(text.strip()) < 10: return "" # Find first occurrence of DSML-like patterns (tag or keyword that starts markup) dsml_start_patterns = [ r"", # DSML variants (ASCII and Unicode separators, e.g. <｜DSML｜invoke ...>) r"<\s*(?:\||｜)?\s*DSML", r"DSML\s*(?:\||｜)", r"DSML\s*>\s*", ] earliest = len(text) for pat in dsml_start_patterns: m = re.search(pat, text, re.IGNORECASE | re.DOTALL) if m: earliest = min(earliest, m.start()) if earliest == 0: return "" prefix = text[:earliest].strip() # Remove trailing incomplete tags prefix = re.sub(r"<[^>]*$", "", prefix).strip() return prefix if len(prefix) > 30 else "" def _vision_prompt_wants_web(prompt: str) -> bool: if not prompt: return False p = prompt.lower() markers = [ "знайди", "пошукай", "пошук", "в інтернет", "в інтернеті", "у відкритих джерелах", "що це", "що на фото", "який це", "яка це", "identify", "find online", "search web", "назва", "бренд", "виробник", "інструкція", "дозування", "регламент", "де купити", "ціна", ] return any(m in p for m in markers) def _vision_answer_uncertain(answer: str) -> bool: if not answer: return True a = answer.lower() uncertain_markers = [ "ймовірно", "можливо", "схоже", "не впевнений", "не можу визначити", "важко сказати", "probably", "maybe", "looks like", "not sure", "cannot identify" ] return any(m in a for m in uncertain_markers) EMPTY_ANSWER_GUARD_AGENTS = {"devtools", "monitor"} def _normalize_text_response(text: str) -> str: return re.sub(r"\s+", " ", str(text or "")).strip() def _needs_empty_answer_recovery(text: str) -> bool: normalized = _normalize_text_response(text) if not normalized: return True low = normalized.lower() if len(normalized) < 8: return True meta_markers = ( "the user", "user asked", "i need", "let me", "analysis", "thinking", "користувач", "потрібно", "спочатку", "сначала" ) if any(m in low for m in meta_markers) and len(normalized) < 80: return True if normalized in {"...", "ok", "done"}: return True return False def _image_response_needs_retry(text: str) -> bool: normalized = _normalize_text_response(text) if _needs_empty_answer_recovery(normalized): return True low = normalized.lower() blocked_markers = ( "не можу бачити", "не можу аналізувати зображення", "опишіть фото словами", "cannot view images", "cannot analyze image", "as a text model" ) if any(m in low for m in blocked_markers): return True return len(normalized) < 24 def _vision_response_is_blurry(text: str) -> bool: low = _normalize_text_response(text).lower() if not low: return False blurry_markers = ( "розмит", "нечітк", "не дуже чітк", "blur", "blurry", "out of focus", "low quality" ) return any(m in low for m in blurry_markers) def _build_image_fallback_response(agent_id: str, prompt: str = "") -> str: if (agent_id or "").lower() == "agromatrix": return ( "Фото поки занадто нечітке, тому діагноз неточний. " "Надішли, будь ласка, 2-3 чіткі фото: загальний вигляд рослини, крупний план проблемної ділянки " "і (для листка) нижній бік. Якщо можеш, додай культуру та стадію росту." ) return "Я поки не бачу достатньо деталей на фото. Надішли, будь ласка, чіткіше фото або крупний план об'єкта." def _sanitize_vision_text_for_user(text: str) -> str: if not text: return "" normalized = re.sub(r"\s+", " ", str(text)).strip() if not normalized: return "" sentences = [seg.strip() for seg in re.split(r"(?<=[.!?])\s+", normalized) if seg.strip()] meta_markers = ( "okay", "the user", "user sent", "they want", "i need", "let me", "i will", "first, look at the image", "look at the image", "first, analyze", "first, looking at the image", "looking at the image", "хорошо", "користувач", "пользователь", "потрібно", "нужно", "спочатку", "сначала" ) cleaned = [sent for sent in sentences if not any(m in sent.lower() for m in meta_markers)] if cleaned: out = " ".join(cleaned[:3]).strip() else: # If text is only meta-reasoning, prefer empty over leaking service text to user. if any(m in normalized.lower() for m in meta_markers): return "" out = " ".join(sentences[:3]).strip() if len(out) > 700: out = out[:700].rsplit(" ", 1)[0] + "..." return out def _extract_vision_search_facts(text: str, max_chars: int = 220) -> str: fact = _sanitize_vision_text_for_user(text) # If sanitizer dropped everything (meta-only), try to recover object phrase. if not fact and text: raw = re.sub(r"\s+", " ", str(text)).strip() raw = re.sub(r"(?i)^okay,?\s*", "", raw) raw = re.sub(r"(?i)^let\'s\s+see\.?\s*", "", raw) raw = re.sub(r"(?i)^the user sent (an image|a photo|a picture) of\s+", "", raw) raw = re.sub(r"(?i)^user sent (an image|a photo|a picture) of\s+", "", raw) raw = re.sub(r"(?i)^an image of\s+", "", raw) raw = re.sub(r"(?i)they want.*$", "", raw).strip(" .") fact = raw if not fact: return "" fact = re.sub(r"(?i)джерела\s*:\s*.*$", "", fact).strip() fact = re.sub(r"[*_`#\[\]()]", "", fact) fact = re.sub(r"\s{2,}", " ", fact).strip(" .,") if len(fact) > max_chars: fact = fact[:max_chars].rsplit(" ", 1)[0] return fact def _build_vision_web_query(prompt: str, vision_text: str) -> str: # Keep query compact and deterministic for web_search tool. source_intent = any(k in (prompt or "").lower() for k in ("джерел", "підтвердж", "source", "reference")) prompt_part = (prompt or "").strip() # Remove generic question wrappers that pollute search quality. prompt_part = re.sub(r"(?i)що\s*це\s*на\s*фото\??", "", prompt_part).strip(" .") prompt_part = re.sub(r"(?i)дай\s*2-?3\s*джерела", "", prompt_part).strip(" .") prompt_part = re.sub(r"(?i)дай\s*\d+\s*джерел[а-я]*\s*для", "", prompt_part).strip(" .") prompt_part = re.sub(r"(?i)дай\s*\d+\s*джерел[а-я]*", "", prompt_part).strip(" .") prompt_part = re.sub(r"(?i)знайди\s*в\s*інтернеті", "", prompt_part).strip(" .") prompt_part = re.sub(r"(?i)знайди\s*в\s*інтернеті\s*схожі\s*джерела", "", prompt_part).strip(" .") prompt_part = re.sub(r"(?i)підтвердження", "", prompt_part).strip(" .") prompt_part = re.sub(r"(?i)якщо\s*не\s*впевнений.*$", "", prompt_part).strip(" .") prompt_part = re.sub(r"(?i)пошукай.*$", "", prompt_part).strip(" .") prompt_part = re.sub(r"(?iu)\bі\b\.?$", "", prompt_part).strip(" .") vision_part = _extract_vision_search_facts(vision_text) if vision_part: tokens = re.findall(r"[a-zA-Zа-яА-ЯіїєІЇЄ0-9]{3,}", vision_part.lower()) generic_tokens = { "first", "look", "image", "photo", "picture", "context", "the", "and", "спочатку", "подивись", "зображення", "фото", "картинка", "контекст", } if len(tokens) < 3 or len(vision_part) < 18 or all(t in generic_tokens for t in tokens): # Too vague entity extraction (e.g., single word "rex") -> skip web enrichment. return "" if vision_part: if prompt_part: q = f"{vision_part}. контекст: {prompt_part}".strip(" .") else: q = vision_part if source_intent: q = f"{q} wikipedia encyclopedia" return q.strip() return "" def _compact_web_search_result(raw: str, query: str = "", agent_id: str = "", max_chars: int = 900) -> str: if not raw: return "" text = str(raw).strip() if not text: return "" def _extract_domain(url: str) -> str: if not url: return "" d = url.lower().strip() d = d.replace("https://", "").replace("http://", "") d = d.split("/")[0] if d.startswith("www."): d = d[4:] return d low_signal_tokens = ( "grid maker", "converter", "convert", "download", "wallpaper", "stock photo", "instagram", "pinterest", "tiktok", "youtube", "facebook", "generator", "meme", ) low_signal_domains = ( "pinterest.com", "instagram.com", "tiktok.com", "youtube.com", "facebook.com", "vk.com", "yandex.", "stackexchange.com", "zhihu.com", "baidu.com", ) trusted_common_domains = ( "wikipedia.org", "wikidata.org", "britannica.com", "who.int", "fao.org", "oecd.org", "worldbank.org", "un.org", "europa.eu", "nature.com", "science.org", "sciencedirect.com", "springer.com", ) trusted_agro_domains = ( "fao.org", "europa.eu", "ec.europa.eu", "usda.gov", "nass.usda.gov", "ukragroconsult.com", "minagro.gov.ua", "rada.gov.ua", "kmu.gov.ua", "agroportal.ua", "latifundist.com", "kurkul.com", ) trusted_by_agent = { "agromatrix": trusted_agro_domains, "alateya": ( "europa.eu", "un.org", "worldbank.org", "oecd.org", ), "clan": ( "europa.eu", "un.org", "wikipedia.org", ), "daarwizz": ( "openai.com", "anthropic.com", "mistral.ai", "huggingface.co", "python.org", "github.com", ), "devtools": ( "github.com", "docs.python.org", "pypi.org", "docker.com", "kubernetes.io", "fastapi.tiangolo.com", "postgresql.org", ), "druid": ( "who.int", "nih.gov", "ncbi.nlm.nih.gov", "wikipedia.org", ), "eonarch": ( "iea.org", "irena.org", "entsoe.eu", "europa.eu", "worldbank.org", ), "greenfood": ( "fao.org", "who.int", "efsa.europa.eu", "usda.gov", "ec.europa.eu", ), "senpai": ( "binance.com", "bybit.com", "coinbase.com", "kraken.com", "coindesk.com", "cointelegraph.com", "tradingview.com", "cftc.gov", "sec.gov", "esma.europa.eu", ), "sofiia": ( "who.int", "nih.gov", "ncbi.nlm.nih.gov", "ema.europa.eu", "fda.gov", "mayoclinic.org", "nhs.uk", ), "helion": ( "iea.org", "irena.org", "entsoe.eu", "europa.eu", "worldbank.org", ), "nutra": ( "fao.org", "who.int", "efsa.europa.eu", "fda.gov", ), "microdao_orchestrator": ( "openai.com", "anthropic.com", "mistral.ai", "github.com", "europa.eu", "un.org", "worldbank.org", ), "monitor": ( "grafana.com", "prometheus.io", "elastic.co", "datadoghq.com", "opentelemetry.io", ), "soul": ( "who.int", "nih.gov", "ncbi.nlm.nih.gov", "wikipedia.org", ), "yaromir": ( "europa.eu", "un.org", "worldbank.org", "wikipedia.org", ), } def _norm_domain_entry(value: Any) -> str: if isinstance(value, dict): value = value.get("url") or value.get("domain") or "" value = str(value or "").strip().lower() if not value: return "" value = value.replace("https://", "").replace("http://", "") value = value.split("/")[0] if value.startswith("www."): value = value[4:] return value def _norm_domain_list(values: Any) -> List[str]: out: List[str] = [] if not isinstance(values, list): return out for v in values: d = _norm_domain_entry(v) if d: out.append(d) return out overrides = _load_trusted_domains_overrides() extra_low_signal = _norm_domain_list(overrides.get("low_signal_domains")) if extra_low_signal: low_signal_domains = tuple(dict.fromkeys([*low_signal_domains, *extra_low_signal])) extra_common = _norm_domain_list(overrides.get("common_domains")) if extra_common: trusted_common_domains = tuple(dict.fromkeys([*trusted_common_domains, *extra_common])) agents_overrides = overrides.get("agents") if isinstance(overrides.get("agents"), dict) else {} for a, cfg in agents_overrides.items(): if not isinstance(cfg, dict): continue doms = _norm_domain_list(cfg.get("domains")) if doms: base = trusted_by_agent.get(str(a).lower(), ()) merged = tuple(dict.fromkeys([*base, *doms])) trusted_by_agent[str(a).lower()] = merged agro_query_terms = { "агро", "agro", "crop", "crops", "fertilizer", "fertilizers", "field", "soil", "harvest", "yield", "pesticide", "herbicide", "farm", "farming", "tractor", "зерно", "пшениц", "кукурудз", "соняшник", "ріпак", "врожай", "ґрунт", "поле", "добрив", "насіння", "ззр", "фермер", } query_terms = {t for t in re.findall(r"[a-zA-Zа-яА-ЯіїєІЇЄ0-9]{3,}", (query or "").lower())} agro_mode = any(any(k in term for k in agro_query_terms) for term in query_terms) agent_trusted_domains = trusted_by_agent.get((agent_id or "").lower(), ()) # Parse bullet blocks from tool output. chunks = [] current = [] for line in text.splitlines(): ln = line.rstrip() if ln.startswith("- ") and current: chunks.append("\n".join(current)) current = [ln] else: current.append(ln) if current: chunks.append("\n".join(current)) scored = [] for chunk in chunks: lines = [ln.strip() for ln in chunk.splitlines() if ln.strip()] title = lines[0][2:].strip() if lines and lines[0].startswith("- ") else (lines[0] if lines else "") url_line = next((ln for ln in lines if ln.lower().startswith("url:")), "") url = url_line.split(":", 1)[1].strip() if ":" in url_line else "" domain = _extract_domain(url) text_blob = " ".join(lines).lower() if any(x in domain for x in low_signal_domains): continue score = 0 for t in query_terms: if t in text_blob: score += 2 if any(tok in text_blob for tok in low_signal_tokens): score -= 3 if domain.endswith(".gov") or domain.endswith(".gov.ua") or domain.endswith(".edu"): score += 2 if any(domain == d or domain.endswith("." + d) for d in trusted_common_domains): score += 2 if any(domain == d or domain.endswith("." + d) for d in agent_trusted_domains): score += 2 if any(domain.endswith(d) for d in ("wikipedia.org", "wikidata.org", "fao.org", "europa.eu")): score += 2 if agro_mode: if any(domain == d or domain.endswith("." + d) for d in trusted_agro_domains): score += 3 else: score -= 1 if not url: score -= 1 if len(title) < 6: score -= 1 scored.append((score, domain, chunk)) def _is_trusted_agro(domain: str) -> bool: if not domain: return False if any(domain == d or domain.endswith("." + d) for d in trusted_common_domains): return True return any(domain == d or domain.endswith("." + d) for d in trusted_agro_domains) scored.sort(key=lambda x: x[0], reverse=True) kept = [] seen_domains = set() if agro_mode: for s, domain, chunk in scored: if s < 1 or not _is_trusted_agro(domain): continue if domain and domain in seen_domains: continue if domain: seen_domains.add(domain) kept.append(chunk) if len(kept) >= 3: break if kept: compact = "\n\n".join(kept).strip() if len(compact) > max_chars: compact = compact[:max_chars].rstrip() + "..." return compact for s, domain, chunk in scored: if s < 2: continue if domain and domain in seen_domains: continue if domain: seen_domains.add(domain) kept.append(chunk) if len(kept) >= 3: break if not kept: return "" compact = "\n\n".join(kept).strip() if len(compact) > max_chars: compact = compact[:max_chars].rstrip() + "..." return compact def _extract_sources_from_compact(compact: str, max_items: int = 3) -> List[Dict[str, str]]: if not compact: return [] items: List[Dict[str, str]] = [] chunks = [c for c in compact.split("\n\n") if c.strip()] for chunk in chunks: lines = [ln.strip() for ln in chunk.splitlines() if ln.strip()] if not lines: continue title = lines[0][2:].strip() if lines[0].startswith("- ") else lines[0] url_line = next((ln for ln in lines if ln.lower().startswith("url:")), "") url = url_line.split(":", 1)[1].strip() if ":" in url_line else "" if not url: continue items.append({"title": title[:180], "url": url[:500]}) if len(items) >= max_items: break return items def _condition_matches(cond: Dict[str, Any], agent_id: str, metadata: Dict[str, Any]) -> bool: """Minimal matcher for router-config `when` conditions.""" if not isinstance(cond, dict): return True meta = metadata or {} if "agent" in cond and cond.get("agent") != agent_id: return False if "mode" in cond and meta.get("mode") != cond.get("mode"): return False if "metadata_has" in cond: key = cond.get("metadata_has") if key not in meta: return False if "metadata_equals" in cond: eq = cond.get("metadata_equals") or {} for k, v in eq.items(): if meta.get(k) != v: return False if "task_type" in cond: expected = cond.get("task_type") actual = meta.get("task_type") if isinstance(expected, list): if actual not in expected: return False elif actual != expected: return False if "api_key_available" in cond: env_name = cond.get("api_key_available") if not (isinstance(env_name, str) and os.getenv(env_name)): return False if "and" in cond: clauses = cond.get("and") or [] if not isinstance(clauses, list): return False for clause in clauses: if not _condition_matches(clause, agent_id, meta): return False return True def _select_default_llm(agent_id: str, metadata: Dict[str, Any], base_llm: str, routing_rules: List[Dict[str, Any]]) -> str: """Select LLM by first matching routing rule with `use_llm`.""" for rule in routing_rules: when = rule.get("when", {}) if _condition_matches(when, agent_id, metadata): use_llm = rule.get("use_llm") if use_llm: logger.info(f"🎯 Agent {agent_id} routing rule {rule.get('id', '')} -> {use_llm}") return use_llm return base_llm app = FastAPI(title="DAARION Router", version="2.0.0") # Configuration NATS_URL = os.getenv("NATS_URL", "nats://nats:4222") SWAPPER_URL = os.getenv("SWAPPER_URL", "http://swapper-service:8890") # All multimodal services now through Swapper STT_URL = os.getenv("STT_URL", "http://swapper-service:8890") # Swapper /stt endpoint TTS_URL = os.getenv("TTS_URL", "http://swapper-service:8890") # Swapper /tts endpoint VISION_URL = os.getenv("VISION_URL", "http://172.18.0.1:11434") # Host Ollama OCR_URL = os.getenv("OCR_URL", "http://swapper-service:8890") # Swapper /ocr endpoint DOCUMENT_URL = os.getenv("DOCUMENT_URL", "http://swapper-service:8890") # Swapper /document endpoint CITY_SERVICE_URL = os.getenv("CITY_SERVICE_URL", "http://daarion-city-service:7001") # CrewAI Routing Configuration CREWAI_ROUTING_ENABLED = os.getenv("CREWAI_ROUTING_ENABLED", "true").lower() == "true" CREWAI_URL = os.getenv("CREWAI_URL", "http://dagi-staging-crewai-service:9010") CLAN_RUNTIME_GUARD_ENABLED = os.getenv("CLAN_RUNTIME_GUARD_ENABLED", "true").lower() == "true" CLAN_RUNTIME_GUARD_MODE = os.getenv("CLAN_RUNTIME_GUARD_MODE", "soft").lower() CLAN_GUARD_TEST_MODE = os.getenv("CLAN_GUARD_TEST_MODE", "false").lower() == "true" CLAN_RUNTIME_REGISTRY_PATH = os.getenv("CLAN_RUNTIME_REGISTRY_PATH", "/app/config/roles/clan/zhos/agents_registry.yaml") CLAN_RUNTIME_ENVELOPE_SCHEMA_PATH = os.getenv("CLAN_RUNTIME_ENVELOPE_SCHEMA_PATH", "/app/docs/contracts/clan-envelope.schema.json") CLAN_RUNTIME_ARTIFACT_SCHEMA_PATH = os.getenv("CLAN_RUNTIME_ARTIFACT_SCHEMA_PATH", "/app/docs/contracts/clan-artifact.schema.json") CLAN_RUNTIME_CONSENT_EVENT_SCHEMA_PATH = os.getenv( "CLAN_RUNTIME_CONSENT_EVENT_SCHEMA_PATH", "/app/docs/contracts/clan-consent-event.schema.json", ) # Neo4j Configuration NEO4J_URI = os.getenv("NEO4J_BOLT_URL", "bolt://neo4j:7687") NEO4J_USER = os.getenv("NEO4J_USER", "neo4j") NEO4J_PASSWORD = os.getenv("NEO4J_PASSWORD", "DaarionNeo4j2026!") # HTTP client for backend services http_client: Optional[httpx.AsyncClient] = None # Neo4j driver neo4j_driver = None neo4j_available = False # NATS client nc = None nats_available = False # Tool Manager tool_manager = None runtime_guard_engine = None # Models class FilterDecision(BaseModel): channel_id: str message_id: Optional[str] = None matrix_event_id: str microdao_id: str decision: Literal["allow", "deny", "modify"] target_agent_id: Optional[str] = None rewrite_prompt: Optional[str] = None class AgentInvocation(BaseModel): agent_id: str entrypoint: Literal["channel_message", "direct", "cron"] = "channel_message" payload: Dict[str, Any] # Load config def load_config(): config_path = "router_config.yaml" if os.path.exists(config_path): with open(config_path, 'r') as f: return yaml.safe_load(f) return { "messaging_inbound": { "enabled": True, "source_subject": "agent.filter.decision", "target_subject": "router.invoke.agent" } } def load_router_config(): """Load main router-config.yml with agents and LLM profiles""" # Try multiple locations paths = [ "router-config.yml", "/app/router-config.yml", "../router-config.yml", "../../router-config.yml" ] for path in paths: if os.path.exists(path): with open(path, 'r') as f: logger.info(f"✅ Loaded router config from {path}") return yaml.safe_load(f) logger.warning("⚠️ router-config.yml not found, using empty config") return {"agents": {}} config = load_config() router_config = load_router_config() @app.on_event("startup") async def startup_event(): """Initialize NATS connection and subscriptions""" global nc, nats_available, http_client, neo4j_driver, neo4j_available, runtime_guard_engine logger.info("🚀 DAGI Router v2.0.0 starting up...") # Initialize HTTP client http_client = httpx.AsyncClient(timeout=60.0) logger.info("✅ HTTP client initialized") # Initialize Neo4j connection try: neo4j_driver = AsyncGraphDatabase.driver( NEO4J_URI, auth=(NEO4J_USER, NEO4J_PASSWORD) ) # Verify connection async with neo4j_driver.session() as session: result = await session.run("RETURN 1 as test") await result.consume() neo4j_available = True logger.info(f"✅ Connected to Neo4j at {NEO4J_URI}") except Exception as e: logger.warning(f"⚠️ Neo4j not available: {e}") neo4j_available = False # Try to connect to NATS try: import nats nc = await nats.connect(NATS_URL) nats_available = True logger.info(f"✅ Connected to NATS at {NATS_URL}") # Subscribe to filter decisions if enabled if config.get("messaging_inbound", {}).get("enabled", True): asyncio.create_task(subscribe_to_filter_decisions()) else: logger.warning("⚠️ Messaging inbound routing disabled in config") except Exception as e: logger.warning(f"⚠️ NATS not available: {e}") logger.warning("⚠️ Running in test mode (HTTP only)") nats_available = False # Initialize Memory Retrieval Pipeline if MEMORY_RETRIEVAL_AVAILABLE and memory_retrieval: try: await memory_retrieval.initialize() logger.info("✅ Memory Retrieval Pipeline initialized") except Exception as e: logger.warning(f"⚠️ Memory Retrieval init failed: {e}") # Initialize Tool Manager for function calling global tool_manager if TOOL_MANAGER_AVAILABLE and ToolManager: try: tool_manager = ToolManager(router_config) logger.info("✅ Tool Manager initialized with function calling") except Exception as e: logger.warning(f"⚠️ Tool Manager init failed: {e}") tool_manager = None else: tool_manager = None # Initialize CLAN runtime guard if RUNTIME_GUARD_AVAILABLE and RuntimeGuard and CLAN_RUNTIME_GUARD_ENABLED: try: runtime_guard_engine = RuntimeGuard( registry_path=CLAN_RUNTIME_REGISTRY_PATH, envelope_schema_path=CLAN_RUNTIME_ENVELOPE_SCHEMA_PATH, artifact_schema_path=CLAN_RUNTIME_ARTIFACT_SCHEMA_PATH, consent_event_schema_path=CLAN_RUNTIME_CONSENT_EVENT_SCHEMA_PATH, mode=CLAN_RUNTIME_GUARD_MODE, ) logger.info( "✅ CLAN Runtime Guard initialized " f"(mode={CLAN_RUNTIME_GUARD_MODE}, registry={CLAN_RUNTIME_REGISTRY_PATH})" ) except Exception as e: logger.warning(f"⚠️ Runtime Guard init failed: {e}") runtime_guard_engine = None else: runtime_guard_engine = None # Log backend URLs logger.info(f"📡 Swapper URL: {SWAPPER_URL}") logger.info(f"📡 STT URL: {STT_URL}") logger.info(f"📡 Vision URL: {VISION_URL}") logger.info(f"📡 OCR URL: {OCR_URL}") logger.info(f"📡 Neo4j URL: {NEO4J_URI}") async def subscribe_to_filter_decisions(): """Subscribe to agent.filter.decision events""" if not nc: return source_subject = config.get("messaging_inbound", {}).get( "source_subject", "agent.filter.decision" ) try: sub = await nc.subscribe(source_subject) print(f"✅ Subscribed to {source_subject}") async for msg in sub.messages: try: decision_data = json.loads(msg.data.decode()) await handle_filter_decision(decision_data) except Exception as e: print(f"❌ Error processing decision: {e}") import traceback traceback.print_exc() except Exception as e: print(f"❌ Subscription error: {e}") async def handle_filter_decision(decision_data: dict): """ Process agent.filter.decision events Only processes 'allow' decisions Creates AgentInvocation and publishes to router.invoke.agent """ try: print(f"\n🔀 Processing filter decision") decision = FilterDecision(**decision_data) # Only process 'allow' decisions if decision.decision != "allow": print(f"⏭️ Ignoring non-allow decision: {decision.decision}") return if not decision.target_agent_id: print(f"⚠️ No target agent specified, skipping") return print(f"✅ Decision: allow") print(f"📝 Target: {decision.target_agent_id}") print(f"📝 Channel: {decision.channel_id}") # Create AgentInvocation invocation = AgentInvocation( agent_id=decision.target_agent_id, entrypoint="channel_message", payload={ "channel_id": decision.channel_id, "message_id": decision.message_id, "matrix_event_id": decision.matrix_event_id, "microdao_id": decision.microdao_id, "rewrite_prompt": decision.rewrite_prompt } ) print(f"🚀 Created invocation for {invocation.agent_id}") # Publish to NATS await publish_agent_invocation(invocation) except Exception as e: print(f"❌ Error handling decision: {e}") import traceback traceback.print_exc() async def publish_agent_invocation(invocation: AgentInvocation): """Publish AgentInvocation to router.invoke.agent""" if nc and nats_available: target_subject = config.get("messaging_inbound", {}).get( "target_subject", "router.invoke.agent" ) try: await nc.publish(target_subject, invocation.json().encode()) print(f"✅ Published invocation to {target_subject}") except Exception as e: print(f"❌ Error publishing to NATS: {e}") else: print(f"⚠️ NATS not available, invocation not published: {invocation.json()}") # ============================================================== # PROMETHEUS METRICS ENDPOINT # ============================================================== @app.get("/metrics") async def prometheus_metrics(): """Prometheus metrics endpoint.""" try: from agent_metrics import get_metrics, get_content_type return Response(content=get_metrics(), media_type=get_content_type()) except Exception as e: logger.error(f"Metrics error: {e}") return Response(content=b"# Error generating metrics", media_type="text/plain") @app.get("/health") async def health(): """Health check endpoint""" return { "status": "ok", "service": "router", "version": "1.0.0", "nats_connected": nats_available, "messaging_inbound_enabled": config.get("messaging_inbound", {}).get("enabled", True) } @app.post("/internal/router/test-messaging", response_model=AgentInvocation) async def test_messaging_route(decision: FilterDecision): """ Test endpoint for routing logic Tests filter decision → agent invocation mapping without NATS """ print(f"\n🧪 Test routing request") if decision.decision != "allow" or not decision.target_agent_id: raise HTTPException( status_code=400, detail=f"Decision not routable: {decision.decision}, agent: {decision.target_agent_id}" ) invocation = AgentInvocation( agent_id=decision.target_agent_id, entrypoint="channel_message", payload={ "channel_id": decision.channel_id, "message_id": decision.message_id, "matrix_event_id": decision.matrix_event_id, "microdao_id": decision.microdao_id, "rewrite_prompt": decision.rewrite_prompt } ) print(f"✅ Test invocation created for {invocation.agent_id}") return invocation @app.on_event("shutdown") async def shutdown_event(): """Clean shutdown""" global nc, http_client if nc: await nc.close() logger.info("✅ NATS connection closed") if http_client: await http_client.aclose() logger.info("✅ HTTP client closed") # ============================================================================ # Backend Integration Endpoints # ============================================================================ class InferRequest(BaseModel): """Request for agent inference""" prompt: str model: Optional[str] = None max_tokens: Optional[int] = 2048 temperature: Optional[float] = 0.7 system_prompt: Optional[str] = None images: Optional[List[str]] = None # List of base64 data URLs for vision metadata: Optional[Dict[str, Any]] = None # Additional metadata (user_id, chat_id, etc.) class InferResponse(BaseModel): """Response from agent inference""" response: str model: str tokens_used: Optional[int] = None backend: str image_base64: Optional[str] = None # Generated image in base64 format file_base64: Optional[str] = None file_name: Optional[str] = None file_mime: Optional[str] = None # ========================================================================= # INTERNAL LLM API (for CrewAI and internal services) # ========================================================================= class InternalLLMRequest(BaseModel): prompt: str system_prompt: Optional[str] = None llm_profile: Optional[str] = "reasoning" model: Optional[str] = None max_tokens: Optional[int] = 2048 temperature: Optional[float] = 0.2 role_context: Optional[str] = None metadata: Optional[Dict[str, Any]] = None class InternalLLMResponse(BaseModel): text: str model: str provider: str tokens_used: int = 0 latency_ms: int = 0 class BackendStatus(BaseModel): """Status of a backend service""" name: str url: str status: str # online, offline, error active_model: Optional[str] = None error: Optional[str] = None @app.get("/backends/status", response_model=List[BackendStatus]) async def get_backends_status(): """Get status of all backend services""" backends = [] # Check Swapper try: resp = await http_client.get(f"{SWAPPER_URL}/health", timeout=5.0) if resp.status_code == 200: data = resp.json() backends.append(BackendStatus( name="swapper", url=SWAPPER_URL, status="online", active_model=data.get("active_model") )) else: backends.append(BackendStatus( name="swapper", url=SWAPPER_URL, status="error", error=f"HTTP {resp.status_code}" )) except Exception as e: backends.append(BackendStatus( name="swapper", url=SWAPPER_URL, status="offline", error=str(e) )) # Check STT try: resp = await http_client.get(f"{STT_URL}/health", timeout=5.0) backends.append(BackendStatus( name="stt", url=STT_URL, status="online" if resp.status_code == 200 else "error" )) except Exception as e: backends.append(BackendStatus( name="stt", url=STT_URL, status="offline", error=str(e) )) # Check Vision (Ollama) try: resp = await http_client.get(f"{VISION_URL}/api/tags", timeout=5.0) if resp.status_code == 200: data = resp.json() models = [m.get("name") for m in data.get("models", [])] backends.append(BackendStatus( name="vision", url=VISION_URL, status="online", active_model=", ".join(models[:3]) if models else None )) else: backends.append(BackendStatus( name="vision", url=VISION_URL, status="error" )) except Exception as e: backends.append(BackendStatus( name="vision", url=VISION_URL, status="offline", error=str(e) )) # Check OCR try: resp = await http_client.get(f"{OCR_URL}/health", timeout=5.0) backends.append(BackendStatus( name="ocr", url=OCR_URL, status="online" if resp.status_code == 200 else "error" )) except Exception as e: backends.append(BackendStatus( name="ocr", url=OCR_URL, status="offline", error=str(e) )) return backends # ========================================================================= # INTERNAL LLM COMPLETE ENDPOINT (for CrewAI) # ========================================================================= @app.post("/internal/llm/complete", response_model=InternalLLMResponse) async def internal_llm_complete(request: InternalLLMRequest): """ Internal LLM completion endpoint. NO routing, NO CrewAI decision, NO agent selection. Used by CrewAI service for multi-role orchestration. """ import time as time_module t0 = time_module.time() logger.info(f"Internal LLM: profile={request.llm_profile}, role={request.role_context}") # Get LLM profile configuration llm_profiles = router_config.get("llm_profiles", {}) profile_name = request.llm_profile or "reasoning" llm_profile = llm_profiles.get(profile_name, {}) provider = llm_profile.get("provider", "deepseek") model = request.model or llm_profile.get("model", "deepseek-chat") max_tokens = request.max_tokens or llm_profile.get("max_tokens", 2048) temperature = request.temperature or llm_profile.get("temperature", 0.2) # Build messages messages = [] if request.system_prompt: system_content = request.system_prompt if request.role_context: system_content = f"[Role: {request.role_context}]\n\n{system_content}" messages.append({"role": "system", "content": system_content}) elif request.role_context: messages.append({"role": "system", "content": f"You are acting as {request.role_context}. Respond professionally."}) messages.append({"role": "user", "content": request.prompt}) # Cloud providers cloud_providers = [ {"name": "deepseek", "api_key_env": "DEEPSEEK_API_KEY", "base_url": "https://api.deepseek.com", "model": "deepseek-chat", "timeout": 60}, {"name": "mistral", "api_key_env": "MISTRAL_API_KEY", "base_url": "https://api.mistral.ai", "model": "mistral-large-latest", "timeout": 60}, {"name": "grok", "api_key_env": "GROK_API_KEY", "base_url": "https://api.x.ai", "model": "grok-2-1212", "timeout": 60} ] # Respect configured provider: local profiles should stay local. if provider in ["deepseek", "mistral", "grok"]: cloud_providers = sorted(cloud_providers, key=lambda x: 0 if x["name"] == provider else 1) elif provider == "ollama": cloud_providers = [] # Try cloud providers (only when provider is cloud) for cloud in cloud_providers: api_key = os.getenv(cloud["api_key_env"]) if not api_key: continue try: logger.debug(f"Internal LLM trying {cloud['name']}") cloud_resp = await http_client.post( f"{cloud['base_url']}/v1/chat/completions", headers={"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}, json={"model": cloud["model"], "messages": messages, "max_tokens": max_tokens, "temperature": temperature, "stream": False}, timeout=cloud["timeout"] ) if cloud_resp.status_code == 200: data = cloud_resp.json() response_text = data.get("choices", [{}])[0].get("message", {}).get("content", "") tokens = data.get("usage", {}).get("total_tokens", 0) latency = int((time_module.time() - t0) * 1000) logger.info(f"Internal LLM success: {cloud['name']}, {tokens} tokens, {latency}ms") return InternalLLMResponse(text=response_text, model=cloud["model"], provider=cloud["name"], tokens_used=tokens, latency_ms=latency) except Exception as e: logger.warning(f"Internal LLM {cloud['name']} failed: {e}") continue # Fallback/target local provider (Ollama) try: logger.info("Internal LLM to Ollama") ollama_model = model or "qwen3:8b" ollama_resp = await http_client.post( "http://172.18.0.1:11434/api/generate", json={"model": ollama_model, "prompt": request.prompt, "system": request.system_prompt or "", "stream": False, "options": {"num_predict": max_tokens, "temperature": temperature}}, timeout=120.0 ) if ollama_resp.status_code == 200: data = ollama_resp.json() latency = int((time_module.time() - t0) * 1000) return InternalLLMResponse(text=data.get("response", ""), model=ollama_model, provider="ollama", tokens_used=0, latency_ms=latency) except Exception as e: logger.error(f"Internal LLM Ollama failed: {e}") raise HTTPException(status_code=503, detail="All LLM providers unavailable") @app.post("/v1/agents/{agent_id}/infer", response_model=InferResponse) async def agent_infer(agent_id: str, request: InferRequest): """ Route inference request to appropriate backend. Router decides which backend to use based on: - Agent configuration (model, capabilities) - Request type (text, vision, audio) - Backend availability System prompt is fetched from database via city-service API. Memory context is retrieved via Memory Retrieval Pipeline v3.0. """ logger.info(f"🔀 Inference request for agent: {agent_id}") logger.info(f"📝 Prompt: {request.prompt[:100]}...") # ========================================================================= # MEMORY RETRIEVAL (v4.0 - Universal for all agents) # ========================================================================= memory_brief_text = "" # Extract metadata once for both retrieval and storage metadata = request.metadata or {} channel = "telegram" # Default chat_id = str(metadata.get("chat_id", "")) user_id = str(metadata.get("user_id", "")).replace("tg:", "") username = metadata.get("username") # Get agent_id from metadata or URL parameter request_agent_id = metadata.get("agent_id", agent_id).lower() if MEMORY_RETRIEVAL_AVAILABLE and memory_retrieval: try: if chat_id and user_id: brief = await memory_retrieval.retrieve( channel=channel, chat_id=chat_id, user_id=user_id, agent_id=request_agent_id, # Agent-specific collections username=username, message=request.prompt ) memory_brief_text = brief.to_text(max_lines=10) if memory_brief_text: logger.info(f"🧠 Memory brief for {request_agent_id}: {len(memory_brief_text)} chars") except Exception as e: logger.warning(f"⚠️ Memory retrieval failed for {request_agent_id}: {e}") # Get system prompt from database or config system_prompt = request.system_prompt system_prompt_source = "request" if system_prompt: logger.info(f"📝 Received system_prompt from request: {len(system_prompt)} chars") logger.debug(f"System prompt preview: {system_prompt[:200]}...") else: system_prompt_source = "city_service" logger.info(f"ℹ️ No system_prompt in request for agent {agent_id}, loading from configured sources") if not system_prompt: try: from prompt_builder import get_agent_system_prompt system_prompt = await get_agent_system_prompt( agent_id, city_service_url=CITY_SERVICE_URL, router_config=router_config ) logger.info(f"✅ Loaded system prompt from database for {agent_id}") except Exception as e: logger.warning(f"⚠️ Could not load prompt from database: {e}") # Fallback to config system_prompt_source = "router_config" agent_config = router_config.get("agents", {}).get(agent_id, {}) system_prompt = agent_config.get("system_prompt") if not system_prompt: system_prompt_source = "empty" logger.warning(f"⚠️ System prompt unavailable for {agent_id}; continuing with provider defaults") system_prompt_hash = hashlib.sha256((system_prompt or "").encode("utf-8")).hexdigest()[:12] effective_metadata = dict(metadata) effective_metadata["system_prompt_hash"] = system_prompt_hash effective_metadata["system_prompt_source"] = system_prompt_source effective_metadata["system_prompt_version"] = ( metadata.get("system_prompt_version") or f"{agent_id}:{system_prompt_hash}" ) logger.info( f"🧩 Prompt meta for {agent_id}: source={system_prompt_source}, " f"version={effective_metadata['system_prompt_version']}, hash={system_prompt_hash}" ) # Determine which backend to use # Use router config to get default model for agent, fallback to qwen3:8b agent_config = router_config.get("agents", {}).get(agent_id, {}) # ========================================================================= # CREWAI DECISION: Use orchestration or direct LLM? # ========================================================================= if CREWAI_ROUTING_ENABLED and CREWAI_CLIENT_AVAILABLE: try: runtime_envelope = None if runtime_guard_engine and request_agent_id == "clan": runtime_envelope = runtime_guard_engine.build_envelope( agent_id=request_agent_id, prompt=request.prompt, metadata=effective_metadata, ) ok_pre, pre_info = runtime_guard_engine.pre_dispatch_checks(runtime_envelope) if not ok_pre: stop_payload = runtime_guard_engine.stop_payload(runtime_envelope, pre_info) logger.warning( "🛑 Runtime guard pre-dispatch stop: " f"code={stop_payload.get('stop_code')} request_id={stop_payload.get('request_id')} " f"input_hash={stop_payload.get('input_hash')}" ) return InferResponse( response=json.dumps(stop_payload, ensure_ascii=False), model="runtime-guard", backend="runtime-guard", tokens_used=0, ) if ( CLAN_GUARD_TEST_MODE and effective_metadata.get("guard_self_test") is True and isinstance(effective_metadata.get("__inject_fake_agent_result"), dict) ): fake_result = effective_metadata.get("__inject_fake_agent_result") ok_post, post_info = runtime_guard_engine.post_return_checks(runtime_envelope, fake_result) if not ok_post: stop_payload = runtime_guard_engine.stop_payload(runtime_envelope, post_info) logger.warning( "🧪 Runtime guard self-test stop: " f"code={stop_payload.get('stop_code')} request_id={stop_payload.get('request_id')} " f"input_hash={stop_payload.get('input_hash')}" ) return InferResponse( response=json.dumps(stop_payload, ensure_ascii=False), model="runtime-guard", backend="runtime-guard", tokens_used=0, ) logger.info("🧪 Runtime guard self-test passed (fake result accepted)") return InferResponse( response=json.dumps({"ok": True, "self_test": True}, ensure_ascii=False), model="runtime-guard", backend="runtime-guard", tokens_used=0, ) # Get agent CrewAI config from registry (or router_config fallback) crewai_cfg = agent_config.get("crewai", {}) use_crewai, crewai_reason = should_use_crewai( agent_id=agent_id, prompt=request.prompt, agent_config=agent_config, metadata=effective_metadata, force_crewai=effective_metadata.get("force_crewai", False), ) logger.info(f"🎭 CrewAI decision for {agent_id}: {use_crewai} ({crewai_reason})") if use_crewai: t0 = time.time() crew_result = await call_crewai( agent_id=agent_id, task=request.prompt, context={ "memory_brief": memory_brief_text, "system_prompt": system_prompt, "system_prompt_meta": { "source": system_prompt_source, "version": effective_metadata.get("system_prompt_version"), "hash": system_prompt_hash, }, "metadata": effective_metadata, "runtime_envelope": runtime_envelope, }, team=crewai_cfg.get("team"), profile=effective_metadata.get("crewai_profile") ) latency = time.time() - t0 if crew_result.get("success") and crew_result.get("result"): if runtime_guard_engine and request_agent_id == "clan" and runtime_envelope: ok_post, post_info = runtime_guard_engine.post_return_checks(runtime_envelope, crew_result) if not ok_post: stop_payload = runtime_guard_engine.stop_payload(runtime_envelope, post_info) logger.warning( "🛑 Runtime guard post-return stop: " f"code={stop_payload.get('stop_code')} request_id={stop_payload.get('request_id')} " f"input_hash={stop_payload.get('input_hash')}" ) return InferResponse( response=json.dumps(stop_payload, ensure_ascii=False), model="runtime-guard", backend="runtime-guard", tokens_used=0, ) crew_result = runtime_guard_engine.stamp_result_artifacts(runtime_envelope, crew_result) ok_stamp, stamp_info = runtime_guard_engine.ensure_stamped_trails(crew_result) if not ok_stamp: stop_payload = runtime_guard_engine.stop_payload(runtime_envelope, stamp_info) logger.warning( "🛑 Runtime guard stamped-trail stop: " f"code={stop_payload.get('stop_code')} request_id={stop_payload.get('request_id')} " f"input_hash={stop_payload.get('input_hash')}" ) return InferResponse( response=json.dumps(stop_payload, ensure_ascii=False), model="runtime-guard", backend="runtime-guard", tokens_used=0, ) for row in runtime_guard_engine.artifact_runtime_rows(runtime_envelope, crew_result): logger.info(json.dumps(row, ensure_ascii=False)) for row in runtime_guard_engine.consent_runtime_rows(runtime_envelope, crew_result): logger.info(json.dumps(row, ensure_ascii=False)) for row in (crew_result.get("artifact_state_transition_rows") or []): if isinstance(row, dict): logger.info(json.dumps(row, ensure_ascii=False)) logger.info(f"✅ CrewAI success for {agent_id}: {latency:.2f}s") # Store interaction in memory if MEMORY_RETRIEVAL_AVAILABLE and memory_retrieval and chat_id and user_id: try: await memory_retrieval.store_interaction( channel=channel, chat_id=chat_id, user_id=user_id, agent_id=request_agent_id, username=username, user_message=request.prompt, assistant_response=crew_result["result"] ) except Exception as e: logger.warning(f"⚠️ Memory storage failed: {e}") return InferResponse( response=crew_result["result"], model="crewai-" + agent_id, backend="crewai", tokens_used=0 ) else: logger.warning(f"⚠️ CrewAI failed, falling back to direct LLM") except Exception as e: logger.exception(f"❌ CrewAI error: {e}, falling back to direct LLM") default_llm = agent_config.get("default_llm", "qwen3:8b") routing_rules = router_config.get("routing", []) default_llm = _select_default_llm(agent_id, metadata, default_llm, routing_rules) # Get LLM profile configuration llm_profiles = router_config.get("llm_profiles", {}) llm_profile = llm_profiles.get(default_llm, {}) provider = llm_profile.get("provider", "ollama") # Determine model name if provider in ["deepseek", "openai", "anthropic", "mistral"]: model = llm_profile.get("model", "deepseek-chat") else: # For local ollama, use swapper model name format model = request.model or "qwen3:8b" # ========================================================================= # VISION PROCESSING (if images present) # ========================================================================= if request.images and len(request.images) > 0: logger.info(f"🖼️ Vision request: {len(request.images)} image(s)") try: # Use Swapper's /vision endpoint (manages model loading) vision_payload = { "model": "qwen3-vl-8b", "prompt": request.prompt, "images": request.images, # Swapper handles data URL conversion "max_tokens": request.max_tokens or 1024, "temperature": request.temperature or 0.7 } # Add system prompt if available if system_prompt: if memory_brief_text: vision_payload["system"] = f"{system_prompt}\n\n[INTERNAL MEMORY - do NOT repeat to user]\n{memory_brief_text}" else: vision_payload["system"] = system_prompt logger.info(f"🖼️ Sending to Swapper /vision: {SWAPPER_URL}/vision") vision_resp = await http_client.post( f"{SWAPPER_URL}/vision", json=vision_payload, timeout=120.0 ) if vision_resp.status_code == 200: vision_data = vision_resp.json() raw_response = vision_data.get("text", "") full_response = _sanitize_vision_text_for_user(raw_response) vision_web_query = "" vision_sources: List[Dict[str, str]] = [] # Debug: log full response structure logger.info( f"✅ Vision response: raw={len(raw_response)} chars, sanitized={len(full_response)} chars, " f"success={vision_data.get('success')}, keys={list(vision_data.keys())}" ) if raw_response and not full_response: full_response = _extract_vision_search_facts(raw_response, max_chars=280) if not full_response: logger.warning(f"⚠️ Empty vision response! Full data: {str(vision_data)[:500]}") # Optional vision -> web enrichment (soft policy): # if prompt explicitly asks to search online OR vision answer is uncertain. if (full_response or raw_response) and TOOL_MANAGER_AVAILABLE and tool_manager: try: wants_web = _vision_prompt_wants_web(request.prompt) uncertain = _vision_answer_uncertain(full_response or raw_response) if wants_web or uncertain: query = _build_vision_web_query(request.prompt, full_response or raw_response) if not query: logger.info("🔎 Vision web enrich skipped: query not actionable") else: vision_web_query = query search_result = await tool_manager.execute_tool( "web_search", {"query": query, "max_results": 3}, agent_id=request_agent_id, chat_id=chat_id, user_id=user_id, ) if search_result and search_result.success and search_result.result: compact_search = _compact_web_search_result( search_result.result, query=query, agent_id=request_agent_id, ) if compact_search and "Нічого не знайдено" not in compact_search: vision_sources = _extract_sources_from_compact(compact_search) base_text = full_response or "Не вдалося надійно ідентифікувати об'єкт на фото." full_response = ( f"{base_text}\n\n" f"Додатково знайшов у відкритих джерелах:\n{compact_search}" ) logger.info( "🌐 Vision web enrichment applied " f"for agent={request_agent_id}, wants_web={wants_web}, uncertain={uncertain}, " f"sources={len(vision_sources)}" ) except Exception as e: logger.warning(f"⚠️ Vision web enrichment failed: {e}") if vision_web_query: logger.info( f"🗂️ Vision enrichment metadata: agent={request_agent_id}, " f"query='{vision_web_query[:180]}', sources={len(vision_sources)}" ) # Image quality gate: one soft retry if response looks empty/meta. if _image_response_needs_retry(full_response): try: logger.warning(f"⚠️ Vision quality gate triggered for agent={request_agent_id}, retrying once") retry_payload = dict(vision_payload) retry_payload["prompt"] = ( "Опиши зображення по суті: що зображено, ключові деталі, можливий контекст. " "Відповідай українською 2-4 реченнями, без службових фраз. " f"Запит користувача: {request.prompt}" ) retry_resp = await http_client.post( f"{SWAPPER_URL}/vision", json=retry_payload, timeout=120.0 ) if retry_resp.status_code == 200: retry_data = retry_resp.json() retry_raw = retry_data.get("text", "") retry_text = _sanitize_vision_text_for_user(retry_raw) if retry_raw and not retry_text: retry_text = _extract_vision_search_facts(retry_raw, max_chars=280) if retry_text and not _image_response_needs_retry(retry_text): full_response = retry_text logger.info(f"✅ Vision quality retry improved response for agent={request_agent_id}") except Exception as e: logger.warning(f"⚠️ Vision quality retry failed: {e}") if _image_response_needs_retry(full_response): full_response = _build_image_fallback_response(request_agent_id, request.prompt) elif request_agent_id == "agromatrix" and _vision_response_is_blurry(full_response): full_response = _build_image_fallback_response(request_agent_id, request.prompt) # Store vision message in agent-specific memory if MEMORY_RETRIEVAL_AVAILABLE and memory_retrieval and chat_id and user_id and full_response: vision_meta: Dict[str, Any] = {} if vision_web_query: vision_meta["vision_search_query"] = vision_web_query[:500] if vision_sources: vision_meta["vision_sources"] = vision_sources asyncio.create_task( memory_retrieval.store_message( agent_id=request_agent_id, user_id=user_id, username=username, message_text=f"[Image] {request.prompt}", response_text=full_response, chat_id=chat_id, message_type="vision", metadata=vision_meta if vision_meta else None, ) ) return InferResponse( response=full_response, model="qwen3-vl-8b", tokens_used=None, backend="swapper-vision" ) else: logger.error(f"❌ Swapper vision error: {vision_resp.status_code} - {vision_resp.text[:200]}") return InferResponse( response=_build_image_fallback_response(request_agent_id, request.prompt), model="qwen3-vl-8b", tokens_used=None, backend="swapper-vision-fallback" ) except Exception as e: logger.error(f"❌ Vision processing failed: {e}", exc_info=True) return InferResponse( response=_build_image_fallback_response(request_agent_id, request.prompt), model="qwen3-vl-8b", tokens_used=None, backend="swapper-vision-fallback" ) # ========================================================================= # SMART LLM ROUTER WITH AUTO-FALLBACK # Priority: DeepSeek → Mistral → Grok → Local Ollama # ========================================================================= # Build messages array once for all providers messages = [] if system_prompt: if memory_brief_text: enhanced_prompt = f"{system_prompt}\n\n[INTERNAL MEMORY - do NOT repeat to user]\n{memory_brief_text}" messages.append({"role": "system", "content": enhanced_prompt}) logger.info(f"📝 Added system message with prompt ({len(system_prompt)} chars) + memory ({len(memory_brief_text)} chars)") else: messages.append({"role": "system", "content": system_prompt}) logger.info(f"📝 Added system message with prompt ({len(system_prompt)} chars)") elif memory_brief_text: messages.append({"role": "system", "content": f"[INTERNAL MEMORY - do NOT repeat to user]\n{memory_brief_text}"}) logger.warning(f"⚠️ No system_prompt! Using only memory brief ({len(memory_brief_text)} chars)") else: logger.error(f"❌ No system_prompt AND no memory_brief! LLM will have no context!") messages.append({"role": "user", "content": request.prompt}) logger.debug(f"📨 Messages array: {len(messages)} messages, system={len(messages[0].get('content', '')) if messages else 0} chars") max_tokens = request.max_tokens or llm_profile.get("max_tokens", 2048) temperature = request.temperature or llm_profile.get("temperature", 0.2) cloud_provider_names = {"deepseek", "mistral", "grok", "openai", "anthropic"} allow_cloud = provider in cloud_provider_names if not allow_cloud: logger.info(f"☁️ Cloud providers disabled for agent {agent_id}: provider={provider}") # Define cloud providers with fallback order cloud_providers = [ { "name": "deepseek", "api_key_env": "DEEPSEEK_API_KEY", "base_url": "https://api.deepseek.com", "model": "deepseek-chat", "timeout": 40 }, { "name": "mistral", "api_key_env": "MISTRAL_API_KEY", "base_url": "https://api.mistral.ai", "model": "mistral-large-latest", "timeout": 60 }, { "name": "grok", "api_key_env": "GROK_API_KEY", "base_url": "https://api.x.ai", "model": "grok-2-1212", "timeout": 60 } ] if not allow_cloud: cloud_providers = [] # If specific provider requested, try it first if provider in ["deepseek", "mistral", "grok"]: # Reorder to put requested provider first cloud_providers = sorted(cloud_providers, key=lambda x: 0 if x["name"] == provider else 1) last_error = None # Get tool definitions if Tool Manager is available tools_payload = None if TOOL_MANAGER_AVAILABLE and tool_manager: tools_payload = tool_manager.get_tool_definitions(request_agent_id) logger.debug(f"🔧 {len(tools_payload)} tools available for function calling") for cloud in cloud_providers: api_key = os.getenv(cloud["api_key_env"]) if not api_key: logger.debug(f"⏭️ Skipping {cloud['name']}: API key not configured") continue try: logger.info(f"🌐 Trying {cloud['name'].upper()} API with model: {cloud['model']}") # Build request payload request_payload = { "model": cloud["model"], "messages": messages, "max_tokens": max_tokens, "temperature": temperature, "stream": False } # Add tools for function calling (if available and supported) if tools_payload and cloud["name"] in ["deepseek", "mistral", "grok"]: request_payload["tools"] = tools_payload request_payload["tool_choice"] = "auto" cloud_resp = await http_client.post( f"{cloud['base_url']}/v1/chat/completions", headers={ "Authorization": f"Bearer {api_key}", "Content-Type": "application/json" }, json=request_payload, timeout=cloud["timeout"] ) if cloud_resp.status_code == 200: data = cloud_resp.json() choice = data.get("choices", [{}])[0] message = choice.get("message", {}) response_text = message.get("content", "") or "" tokens_used = data.get("usage", {}).get("total_tokens", 0) # Initialize tool_results to avoid UnboundLocalError tool_results = [] # Check for tool calls (standard format) tool_calls = message.get("tool_calls", []) # Also check for DSML format in content (DeepSeek sometimes returns this) # AGGRESSIVE check - any DSML-like pattern should be caught import re has_dsml = False if response_text: # Check for DSML patterns with regex (handles Unicode variations) dsml_patterns_check = [ r'DSML', # Any mention of DSML r'function_calls>', r'invoke\s*name\s*=', r'parameter\s*name\s*=', r'<[^>]*invoke[^>]*>', r']*invoke[^>]*>', ] for pattern in dsml_patterns_check: if re.search(pattern, response_text, re.IGNORECASE): has_dsml = True logger.warning(f"⚠️ DSML detected via pattern: {pattern}") break if has_dsml: logger.warning("⚠️ Detected DSML format in content, parsing...") # Extract tool name and parameters from DSML import re # Try multiple DSML patterns dsml_patterns = [ r'invoke name="(\w+)".*?parameter name="(\w+)"[^>]*>([^<]+)', r'invoke\s+name="(\w+)".*?parameter\s+name="(\w+)"[^>]*>([^<]+)', r'name="web_extract".*?url.*?>([^\s<]+)', ] dsml_match = None for pattern in dsml_patterns: dsml_match = re.search(pattern, response_text, re.DOTALL | re.IGNORECASE) if dsml_match: break if dsml_match and len(dsml_match.groups()) >= 3: tool_name = dsml_match.group(1) param_name = dsml_match.group(2) param_value = dsml_match.group(3).strip() # Create synthetic tool call with Mistral-compatible ID (exactly 9 chars, a-zA-Z0-9) import string import random tool_call_id = ''.join(random.choices(string.ascii_letters + string.digits, k=9)) tool_calls = [{ "id": tool_call_id, "function": { "name": tool_name, "arguments": json.dumps({param_name: param_value}) } }] logger.info(f"🔧 Parsed DSML tool call: {tool_name}({param_name}={param_value[:50]}...) id={tool_call_id}") # ALWAYS clear DSML content - never show raw DSML to user logger.warning(f"🧹 Clearing DSML content from response ({len(response_text)} chars)") response_text = "" if tool_calls and tool_manager: max_tool_rounds = int(os.getenv("ROUTER_TOOL_MAX_ROUNDS", "10")) logger.info(f"🔧 LLM requested tool calls; running iterative mode up to {max_tool_rounds} rounds") all_tool_results = [] current_tool_calls = tool_calls rounds_done = 0 oneok_ctx = { "client_id": None, "site_id": None, "calc_result": None, "quote_id": None, } repeated_failures = {} while current_tool_calls and rounds_done < max_tool_rounds: rounds_done += 1 logger.info(f"🔁 Tool round {rounds_done}/{max_tool_rounds}: {len(current_tool_calls)} call(s)") round_results = [] abort_loop_due_repeats = False for tc in current_tool_calls: func = tc.get("function", {}) tool_name = func.get("name", "") try: tool_args = json.loads(func.get("arguments", "{}")) except Exception: tool_args = {} # Light auto-repair for 1OK multi-step flows when model omits required args. if request_agent_id == "oneok": if tool_name == "calc_window_quote": ip = (tool_args or {}).get("input_payload") if isinstance(ip, dict) and isinstance(ip.get("windows"), list) and "window_units" not in ip: ip2 = dict(ip) ip2["window_units"] = ip.get("windows") tool_args = dict(tool_args or {}) tool_args["input_payload"] = ip2 elif tool_name == "crm_create_quote": quote_payload = (tool_args or {}).get("quote_payload") if not isinstance(quote_payload, dict): calc_res = oneok_ctx.get("calc_result") or {} line_items = calc_res.get("line_items") if isinstance(calc_res, dict) else None totals = calc_res.get("totals") if isinstance(calc_res, dict) else None if isinstance(line_items, list) and isinstance(totals, dict): tool_args = { "quote_payload": { "client_id": oneok_ctx.get("client_id"), "site_id": oneok_ctx.get("site_id"), "currency": calc_res.get("currency", "UAH"), "line_items": line_items, "totals": totals, "assumptions": calc_res.get("assumptions", []), "validity_days": 14, "lead_time_estimate": calc_res.get("lead_time_if_known") or calc_res.get("lead_time_estimate"), } } logger.info("🛠️ oneok: auto-filled crm_create_quote.quote_payload from calc context") elif tool_name == "docs_render_quote_pdf": quote_id = (tool_args or {}).get("quote_id") quote_payload = (tool_args or {}).get("quote_payload") if not quote_id and not isinstance(quote_payload, dict) and oneok_ctx.get("quote_id"): tool_args = {"quote_id": oneok_ctx.get("quote_id")} logger.info("🛠️ oneok: auto-filled docs_render_quote_pdf.quote_id from quote context") elif tool_name == "schedule_propose_slots": params = (tool_args or {}).get("params") if not isinstance(params, dict): tool_args = {"params": {"count": 3, "timezone": "Europe/Kyiv"}} logger.info("🛠️ oneok: auto-filled schedule_propose_slots.params") result = await tool_manager.execute_tool( tool_name, tool_args, agent_id=request_agent_id, chat_id=chat_id, user_id=user_id, ) tool_result_dict = { "tool_call_id": tc.get("id", ""), "name": tool_name, "success": result.success, "result": result.result, "error": result.error, "image_base64": result.image_base64, "file_base64": result.file_base64, "file_name": result.file_name, "file_mime": result.file_mime, } if result.image_base64: logger.info(f"🖼️ Tool {tool_name} generated image: {len(result.image_base64)} chars") round_results.append(tool_result_dict) all_tool_results.append(tool_result_dict) # Track oneok context to help subsequent tool calls in the same request. if request_agent_id == "oneok" and result.success and isinstance(result.result, dict): if tool_name == "crm_upsert_client": oneok_ctx["client_id"] = result.result.get("client_id") or oneok_ctx.get("client_id") elif tool_name == "crm_upsert_site": oneok_ctx["site_id"] = result.result.get("site_id") or oneok_ctx.get("site_id") elif tool_name == "calc_window_quote": oneok_ctx["calc_result"] = result.result elif tool_name == "crm_create_quote": oneok_ctx["quote_id"] = result.result.get("quote_id") or oneok_ctx.get("quote_id") # Guardrail: stop if model repeats same failing tool call too many times. sig = f"{tool_name}:{json.dumps(tool_args, ensure_ascii=False, sort_keys=True, default=str)}" if result.success: repeated_failures.pop(sig, None) else: repeated_failures[sig] = repeated_failures.get(sig, 0) + 1 if repeated_failures[sig] >= 3: logger.warning(f"⚠️ Repeated failing tool call detected ({tool_name}) x{repeated_failures[sig]}; breaking iterative loop") abort_loop_due_repeats = True break if abort_loop_due_repeats: current_tool_calls = [] response_text = response_text or format_tool_calls_for_response(all_tool_results, fallback_mode="empty_response") break messages.append({"role": "assistant", "content": None, "tool_calls": current_tool_calls}) for tr in round_results: messages.append({ "role": "tool", "tool_call_id": tr["tool_call_id"], "content": str(tr["result"]) if tr["success"] else f"Error: {tr['error']}" }) logger.info("🔄 Calling LLM again after tool round") loop_payload = { "model": cloud["model"], "messages": messages, "max_tokens": max_tokens, "temperature": temperature, "stream": False } # Keep tools enabled for multi-step chains. if tools_payload and cloud["name"] in ["deepseek", "mistral", "grok"]: loop_payload["tools"] = tools_payload loop_payload["tool_choice"] = "auto" loop_resp = await http_client.post( f"{cloud['base_url']}/v1/chat/completions", headers={ "Authorization": f"Bearer {api_key}", "Content-Type": "application/json" }, json=loop_payload, timeout=cloud["timeout"] ) if loop_resp.status_code != 200: logger.error(f"❌ {cloud['name'].upper()} loop call failed: {loop_resp.status_code} - {loop_resp.text[:200]}") response_text = format_tool_calls_for_response(all_tool_results, fallback_mode="empty_response") current_tool_calls = [] break loop_data = loop_resp.json() loop_message = loop_data.get("choices", [{}])[0].get("message", {}) response_text = loop_message.get("content", "") or "" tokens_used += loop_data.get("usage", {}).get("total_tokens", 0) current_tool_calls = loop_message.get("tool_calls", []) # DSML fallback parsing for providers that return markup instead of tool_calls. has_dsml_loop = False if response_text: dsml_patterns_check = [ r'DSML', r'function_calls>', r'invoke\s*name\s*=', r'parameter\s*name\s*=', r'<[^>]*invoke[^>]*>', r']*invoke[^>]*>', ] for pattern in dsml_patterns_check: if re.search(pattern, response_text, re.IGNORECASE): has_dsml_loop = True logger.warning(f"⚠️ DSML detected in loop via pattern: {pattern}") break if has_dsml_loop and not current_tool_calls: dsml_patterns = [ r'invoke name="(\w+)".*?parameter name="(\w+)"[^>]*>([^<]+)', r'invoke\s+name="(\w+)".*?parameter\s+name="(\w+)"[^>]*>([^<]+)', ] dsml_match = None for pattern in dsml_patterns: dsml_match = re.search(pattern, response_text, re.DOTALL | re.IGNORECASE) if dsml_match: break if dsml_match and len(dsml_match.groups()) >= 3: import string import random tool_call_id = ''.join(random.choices(string.ascii_letters + string.digits, k=9)) current_tool_calls = [{ "id": tool_call_id, "function": { "name": dsml_match.group(1), "arguments": json.dumps({dsml_match.group(2): dsml_match.group(3).strip()}) } }] response_text = "" tool_results = all_tool_results if current_tool_calls: logger.warning(f"⚠️ Reached max tool rounds ({max_tool_rounds}) for {request_agent_id}, returning summary") response_text = response_text or format_tool_calls_for_response(tool_results, fallback_mode="empty_response") if response_text and ("DSML" in response_text or "invoke name=" in response_text or "function_calls>" in response_text): prefix_before_dsml = _strip_dsml_keep_text_before(response_text) if prefix_before_dsml: logger.warning(f"🧹 DSML in loop final response: keeping text before DSML ({len(prefix_before_dsml)} chars)") response_text = prefix_before_dsml else: response_text = format_tool_calls_for_response(tool_results, fallback_mode="dsml_detected") if not response_text: logger.warning(f"⚠️ {cloud['name'].upper()} returned empty response after iterative tool calls") response_text = format_tool_calls_for_response(tool_results, fallback_mode="empty_response") if response_text: # FINAL DSML check before returning - never show DSML to user if "DSML" in response_text or "invoke name=" in response_text or "function_calls>" in response_text: prefix_before_dsml = _strip_dsml_keep_text_before(response_text) if prefix_before_dsml: logger.warning(f"🧹 DSML in final response: keeping text before DSML ({len(prefix_before_dsml)} chars)") response_text = prefix_before_dsml else: logger.warning(f"🧹 DSML in final response! Replacing with fallback ({len(response_text)} chars)") response_text = format_tool_calls_for_response(tool_results, fallback_mode="dsml_detected") # Check if any tool generated an image generated_image = None generated_file_base64 = None generated_file_name = None generated_file_mime = None logger.debug(f"🔍 Checking {len(tool_results)} tool results for images...") for tr in tool_results: img_b64 = tr.get("image_base64") if img_b64: generated_image = img_b64 logger.info(f"🖼️ Image generated by tool: {tr['name']} ({len(img_b64)} chars)") file_b64 = tr.get("file_base64") if file_b64 and not generated_file_base64: generated_file_base64 = file_b64 generated_file_name = tr.get("file_name") generated_file_mime = tr.get("file_mime") logger.info(f"📄 File generated by tool: {tr['name']} ({len(file_b64)} chars)") if not img_b64: logger.debug(f" Tool {tr['name']}: no image_base64") logger.info(f"✅ {cloud['name'].upper()} response received, {tokens_used} tokens") # Store message in agent-specific memory (async, non-blocking) if MEMORY_RETRIEVAL_AVAILABLE and memory_retrieval and chat_id and user_id: asyncio.create_task( memory_retrieval.store_message( agent_id=request_agent_id, user_id=user_id, username=username, message_text=request.prompt, response_text=response_text, chat_id=chat_id ) ) return InferResponse( response=response_text, model=cloud["model"], tokens_used=tokens_used, backend=f"{cloud['name']}-cloud", image_base64=generated_image, file_base64=generated_file_base64, file_name=generated_file_name, file_mime=generated_file_mime, ) else: logger.warning(f"⚠️ {cloud['name'].upper()} returned empty response, trying next provider") continue else: logger.warning(f"⚠️ {cloud['name'].upper()} returned {cloud_resp.status_code}, trying next...") last_error = f"{cloud['name']}: {cloud_resp.status_code}" continue except Exception as e: import traceback error_details = traceback.format_exc() logger.warning(f"⚠️ {cloud['name'].upper()} failed: {e}") if not str(e).strip(): # Empty error string logger.error(f"❌ {cloud['name'].upper()} failed with empty error! Check traceback:") logger.error(error_details) else: logger.debug(f"Full error traceback: {error_details}") last_error = f"{cloud['name']}: {str(e)}" continue # If all cloud providers failed, log and fall through to local if last_error: logger.warning(f"⚠️ All cloud providers failed ({last_error}), falling back to local Ollama") # ========================================================================= # LOCAL PROVIDERS (Ollama via Swapper) # ========================================================================= # Determine local model from config (not hardcoded) # Strategy: Use agent's default_llm if it's local (ollama), otherwise find first local model local_model = None # Check if default_llm is local if llm_profile.get("provider") == "ollama": # Extract model name and convert format (qwen3:8b → qwen3:8b for Swapper) ollama_model = llm_profile.get("model", "qwen3:8b") local_model = ollama_model.replace(":", "-") # qwen3:8b → qwen3:8b logger.debug(f"✅ Using agent's default local model: {local_model}") else: # Find first local model from config for profile_name, profile in llm_profiles.items(): if profile.get("provider") == "ollama": ollama_model = profile.get("model", "qwen3:8b") local_model = ollama_model.replace(":", "-") logger.info(f"🔄 Found fallback local model: {local_model} from profile {profile_name}") break # Final fallback if no local model found if not local_model: local_model = "qwen3:8b" logger.warning(f"⚠️ No local model in config, using hardcoded fallback: {local_model}") try: # Check if Swapper is available health_resp = await http_client.get(f"{SWAPPER_URL}/health", timeout=5.0) if health_resp.status_code == 200: logger.info(f"📡 Calling Swapper with local model: {local_model}") # Generate response via Swapper (which handles model loading) generate_resp = await http_client.post( f"{SWAPPER_URL}/generate", json={ "model": local_model, "prompt": request.prompt, "system": system_prompt, "max_tokens": request.max_tokens, "temperature": request.temperature, "stream": False }, timeout=300.0 ) if generate_resp.status_code == 200: data = generate_resp.json() local_response = _normalize_text_response(data.get("response", "")) # Empty-answer gate for selected local top-level agents. if request_agent_id in EMPTY_ANSWER_GUARD_AGENTS and _needs_empty_answer_recovery(local_response): logger.warning(f"⚠️ Empty-answer gate triggered for {request_agent_id}, retrying local generate once") retry_prompt = ( f"{request.prompt}\n\n" "Відповідай коротко і конкретно (2-5 речень), без службових або мета-фраз." ) retry_resp = await http_client.post( f"{SWAPPER_URL}/generate", json={ "model": local_model, "prompt": retry_prompt, "system": system_prompt, "max_tokens": request.max_tokens, "temperature": request.temperature, "stream": False }, timeout=300.0 ) if retry_resp.status_code == 200: retry_data = retry_resp.json() retry_text = _normalize_text_response(retry_data.get("response", "")) if retry_text and not _needs_empty_answer_recovery(retry_text): local_response = retry_text if _needs_empty_answer_recovery(local_response): local_response = ( "Я не отримав корисну відповідь з першої спроби. " "Сформулюй запит коротко ще раз, і я відповім конкретно." ) # Store in agent-specific memory if MEMORY_RETRIEVAL_AVAILABLE and memory_retrieval and chat_id and user_id and local_response: asyncio.create_task( memory_retrieval.store_message( agent_id=request_agent_id, user_id=user_id, username=username, message_text=request.prompt, response_text=local_response, chat_id=chat_id ) ) return InferResponse( response=local_response, model=local_model, tokens_used=data.get("eval_count", 0), backend="swapper+ollama" ) else: logger.error(f"❌ Swapper error: {generate_resp.status_code} - {generate_resp.text}") except Exception as e: logger.error(f"❌ Swapper/Ollama error: {e}") # Fallback to direct Ollama if Swapper fails try: logger.info(f"🔄 Falling back to direct Ollama connection") generate_resp = await http_client.post( f"{VISION_URL}/api/generate", json={ "model": "qwen3:8b", # Use actual Ollama model name "prompt": request.prompt, "system": system_prompt, "stream": False, "options": { "num_predict": request.max_tokens, "temperature": request.temperature } }, timeout=120.0 ) if generate_resp.status_code == 200: data = generate_resp.json() return InferResponse( response=data.get("response", ""), model=model, tokens_used=data.get("eval_count", 0), backend="ollama-direct" ) except Exception as e2: logger.error(f"❌ Direct Ollama fallback also failed: {e2}") # Fallback: return error raise HTTPException( status_code=503, detail=f"No backend available for model: {model}" ) @app.get("/v1/models") async def list_available_models(): """List all available models across backends""" models = [] # Get Swapper models try: resp = await http_client.get(f"{SWAPPER_URL}/models", timeout=5.0) if resp.status_code == 200: data = resp.json() for m in data.get("models", []): models.append({ "id": m.get("name"), "backend": "swapper", "size_gb": m.get("size_gb"), "status": m.get("status", "available") }) except Exception as e: logger.warning(f"Cannot get Swapper models: {e}") # Get Ollama models try: resp = await http_client.get(f"{VISION_URL}/api/tags", timeout=5.0) if resp.status_code == 200: data = resp.json() for m in data.get("models", []): # Avoid duplicates model_name = m.get("name") if not any(x.get("id") == model_name for x in models): models.append({ "id": model_name, "backend": "ollama", "size_gb": round(m.get("size", 0) / 1e9, 1), "status": "loaded" }) except Exception as e: logger.warning(f"Cannot get Ollama models: {e}") return {"models": models, "total": len(models)} # ============================================================================= # NEO4J GRAPH API ENDPOINTS # ============================================================================= class GraphNode(BaseModel): """Model for creating/updating a graph node""" label: str # Node type: User, Agent, Topic, Fact, Entity, etc. properties: Dict[str, Any] node_id: Optional[str] = None # If provided, update existing node class GraphRelationship(BaseModel): """Model for creating a relationship between nodes""" from_node_id: str to_node_id: str relationship_type: str # KNOWS, MENTIONED, RELATED_TO, CREATED_BY, etc. properties: Optional[Dict[str, Any]] = None class GraphQuery(BaseModel): """Model for querying the graph""" cypher: Optional[str] = None # Direct Cypher query (advanced) # Or use structured query: node_label: Optional[str] = None node_id: Optional[str] = None relationship_type: Optional[str] = None depth: int = 1 # How many hops to traverse limit: int = 50 class GraphSearchRequest(BaseModel): """Natural language search in graph""" query: str entity_types: Optional[List[str]] = None # Filter by types limit: int = 20 @app.post("/v1/graph/nodes") async def create_graph_node(node: GraphNode): """Create or update a node in the knowledge graph""" if not neo4j_available or not neo4j_driver: raise HTTPException(status_code=503, detail="Neo4j not available") try: async with neo4j_driver.session() as session: # Generate node_id if not provided node_id = node.node_id or f"{node.label.lower()}_{os.urandom(8).hex()}" # Build properties with node_id props = {**node.properties, "node_id": node_id, "updated_at": "datetime()"} # Create or merge node cypher = f""" MERGE (n:{node.label} {{node_id: $node_id}}) SET n += $properties SET n.updated_at = datetime() RETURN n """ result = await session.run(cypher, node_id=node_id, properties=node.properties) record = await result.single() if record: created_node = dict(record["n"]) logger.info(f"📊 Created/updated node: {node.label} - {node_id}") return {"status": "ok", "node_id": node_id, "node": created_node} raise HTTPException(status_code=500, detail="Failed to create node") except Exception as e: logger.error(f"❌ Neo4j error creating node: {e}") raise HTTPException(status_code=500, detail=str(e)) @app.post("/v1/graph/relationships") async def create_graph_relationship(rel: GraphRelationship): """Create a relationship between two nodes""" if not neo4j_available or not neo4j_driver: raise HTTPException(status_code=503, detail="Neo4j not available") try: async with neo4j_driver.session() as session: props_clause = "" if rel.properties: props_clause = " SET r += $properties" cypher = f""" MATCH (a {{node_id: $from_id}}) MATCH (b {{node_id: $to_id}}) MERGE (a)-[r:{rel.relationship_type}]->(b) {props_clause} SET r.created_at = datetime() RETURN a.node_id as from_id, b.node_id as to_id, type(r) as rel_type """ result = await session.run( cypher, from_id=rel.from_node_id, to_id=rel.to_node_id, properties=rel.properties or {} ) record = await result.single() if record: logger.info(f"🔗 Created relationship: {rel.from_node_id} -[{rel.relationship_type}]-> {rel.to_node_id}") return { "status": "ok", "from_id": record["from_id"], "to_id": record["to_id"], "relationship_type": record["rel_type"] } raise HTTPException(status_code=404, detail="One or both nodes not found") except Exception as e: logger.error(f"❌ Neo4j error creating relationship: {e}") raise HTTPException(status_code=500, detail=str(e)) @app.post("/v1/graph/query") async def query_graph(query: GraphQuery): """Query the knowledge graph""" if not neo4j_available or not neo4j_driver: raise HTTPException(status_code=503, detail="Neo4j not available") try: async with neo4j_driver.session() as session: # If direct Cypher provided, use it (with safety check) if query.cypher: # Basic safety: only allow read queries if any(kw in query.cypher.upper() for kw in ["DELETE", "REMOVE", "DROP", "CREATE", "MERGE", "SET"]): raise HTTPException(status_code=400, detail="Only read queries allowed via cypher parameter") result = await session.run(query.cypher) records = await result.data() return {"status": "ok", "results": records, "count": len(records)} # Build structured query if query.node_id: # Get specific node with relationships cypher = f""" MATCH (n {{node_id: $node_id}}) OPTIONAL MATCH (n)-[r]-(related) RETURN n, collect({{rel: type(r), node: related}}) as connections LIMIT 1 """ result = await session.run(cypher, node_id=query.node_id) elif query.node_label: # Get nodes by label cypher = f""" MATCH (n:{query.node_label}) RETURN n ORDER BY n.updated_at DESC LIMIT $limit """ result = await session.run(cypher, limit=query.limit) else: # Get recent nodes cypher = """ MATCH (n) RETURN n, labels(n) as labels ORDER BY n.updated_at DESC LIMIT $limit """ result = await session.run(cypher, limit=query.limit) records = await result.data() return {"status": "ok", "results": records, "count": len(records)} except Exception as e: logger.error(f"❌ Neo4j query error: {e}") raise HTTPException(status_code=500, detail=str(e)) @app.get("/v1/graph/search") async def search_graph(q: str, types: Optional[str] = None, limit: int = 20): """Search nodes by text in properties""" if not neo4j_available or not neo4j_driver: raise HTTPException(status_code=503, detail="Neo4j not available") try: async with neo4j_driver.session() as session: # Build label filter label_filter = "" if types: labels = [t.strip() for t in types.split(",")] label_filter = " AND (" + " OR ".join([f"n:{l}" for l in labels]) + ")" # Search in common text properties cypher = f""" MATCH (n) WHERE ( n.name CONTAINS $query OR n.title CONTAINS $query OR n.text CONTAINS $query OR n.description CONTAINS $query OR n.content CONTAINS $query ){label_filter} RETURN n, labels(n) as labels ORDER BY n.updated_at DESC LIMIT $limit """ result = await session.run(cypher, query=q, limit=limit) records = await result.data() return {"status": "ok", "query": q, "results": records, "count": len(records)} except Exception as e: logger.error(f"❌ Neo4j search error: {e}") raise HTTPException(status_code=500, detail=str(e)) @app.get("/v1/graph/stats") async def get_graph_stats(): """Get knowledge graph statistics""" if not neo4j_available or not neo4j_driver: raise HTTPException(status_code=503, detail="Neo4j not available") try: async with neo4j_driver.session() as session: # Get node counts by label labels_result = await session.run(""" CALL db.labels() YIELD label CALL apoc.cypher.run('MATCH (n:`' + label + '`) RETURN count(n) as count', {}) YIELD value RETURN label, value.count as count """) # If APOC not available, use simpler query try: labels_data = await labels_result.data() except: labels_result = await session.run(""" MATCH (n) RETURN labels(n)[0] as label, count(*) as count ORDER BY count DESC """) labels_data = await labels_result.data() # Get relationship counts rels_result = await session.run(""" MATCH ()-[r]->() RETURN type(r) as type, count(*) as count ORDER BY count DESC """) rels_data = await rels_result.data() # Get total counts total_result = await session.run(""" MATCH (n) RETURN count(n) as nodes """) total_nodes = (await total_result.single())["nodes"] total_rels_result = await session.run(""" MATCH ()-[r]->() RETURN count(r) as relationships """) total_rels = (await total_rels_result.single())["relationships"] return { "status": "ok", "total_nodes": total_nodes, "total_relationships": total_rels, "nodes_by_label": labels_data, "relationships_by_type": rels_data } except Exception as e: logger.error(f"❌ Neo4j stats error: {e}") raise HTTPException(status_code=500, detail=str(e)) @app.on_event("shutdown") async def shutdown_event(): """Cleanup connections on shutdown""" global neo4j_driver, http_client, nc # Close Memory Retrieval if MEMORY_RETRIEVAL_AVAILABLE and memory_retrieval: try: await memory_retrieval.close() logger.info("🔌 Memory Retrieval closed") except Exception as e: logger.warning(f"⚠️ Memory Retrieval close error: {e}") if neo4j_driver: await neo4j_driver.close() logger.info("🔌 Neo4j connection closed") if http_client: await http_client.aclose() logger.info("🔌 HTTP client closed") if nc: await nc.close() logger.info("🔌 NATS connection closed")