from __future__ import annotations from .shared import * AI_CALL_LOG_RESPONSE_MAX_CHARS = 10_000 AI_CALL_LOG_RESPONSE_TAIL_CHARS = 2_000 AI_CALL_LOG_TRUNCATION_MARKER = "\n...[TRUNCATED]...\n" class LLMPromptProfileRuntimeMixin: async def _resolve_active_prompt_profile(self, db: AsyncSession, *, tenant_id: int) -> dict[str, str]: cached = self._prompt_profile_cache.get(int(tenant_id)) now = time.time() if cached and cached[0] > now: return dict(cached[1]) role_name = "" system_prompt = "" role_key = "" try: row = ( await db.execute( select(Integration).where( and_( Integration.tenant_id == int(tenant_id), Integration.type == "ai_prompt_profiles", ) ) ) ).scalar_one_or_none() bundle = _safe_json(row.config if row else {}) active_key = str(bundle.get("active_role_key") or "").strip() roles = bundle.get("roles") if isinstance(bundle.get("roles"), list) else [] selected = None for item in roles: if not isinstance(item, dict): continue item_key = str(item.get("role_key") or "").strip() if item_key == active_key: selected = item break if selected is None: for item in roles: if isinstance(item, dict) and bool(item.get("enabled", True)): selected = item break if isinstance(selected, dict): role_name = str(selected.get("role_name") or "").strip() system_prompt = str(selected.get("system_prompt") or "").strip() role_key = str(selected.get("role_key") or "").strip() except Exception: role_name = "" system_prompt = "" role_key = "" if len(system_prompt) < 20: fallback = build_default_role_prompt_bundle() fallback_key = str(fallback.get("active_role_key") or "").strip() roles = fallback.get("roles") if isinstance(fallback.get("roles"), list) else [] role = None for item in roles: if not isinstance(item, dict): continue if str(item.get("role_key") or "").strip() == fallback_key: role = item break if role is None and roles: role = roles[0] if isinstance(roles[0], dict) else None if isinstance(role, dict): role_name = str(role.get("role_name") or role_name or "默认角色").strip() system_prompt = str(role.get("system_prompt") or "").strip() role_key = str(role.get("role_key") or role_key or fallback_key).strip() result = { "role_key": role_key, "role_name": role_name, "system_prompt": system_prompt, } self._prompt_profile_cache[int(tenant_id)] = (now + PROMPT_PROFILE_CACHE_TTL_SECONDS, result) return dict(result) def _build_fallback_engines(self, primary_engine: dict[str, Any]) -> list[dict[str, Any]]: candidates: list[dict[str, Any]] = [] provider = _normalize_provider(primary_engine.get("provider")) current_model = str(primary_engine.get("model") or "").strip() provider_runtime = self._env_provider_runtime(provider) default_model = str(provider_runtime.get("model") or PROVIDER_DEFAULTS.get(provider, {}).get("model") or "").strip() if default_model and default_model != current_model: same_provider = dict(primary_engine) same_provider["model"] = default_model candidates.append(self._normalize_engine_runtime(same_provider)) fallback_provider_raw = str(os.getenv("LLM_FALLBACK_PROVIDER", "ollama")).strip() if fallback_provider_raw: fallback_provider = _normalize_provider(fallback_provider_raw) if fallback_provider != provider: candidates.append(self._env_provider_runtime(fallback_provider)) return candidates def _env_provider_runtime(self, provider: str) -> dict[str, Any]: provider_name = _normalize_provider(provider) defaults = PROVIDER_DEFAULTS[provider_name] api_env = str(defaults.get("api_env") or "").strip() api_key = os.getenv(api_env, "") if api_env else "" if provider_name == "glm": api_key = os.getenv("DASHSCOPE_API_KEY", api_key) return { "id": f"env-{provider_name}", "name": f"{defaults['name']} (ENV)", "provider": provider_name, "api_env": api_env, "api_key": str(api_key or "").strip(), "base_url": str(defaults["base_url"]).strip().rstrip("/"), "model": str(defaults["model"]).strip(), } def _env_engine_runtime(self) -> dict[str, Any]: provider = _normalize_provider(os.getenv("LLM_PROVIDER", "glm")) runtime = self._env_provider_runtime(provider) runtime["id"] = "env-default" return runtime def _normalize_engine_runtime(self, engine: dict[str, Any]) -> dict[str, Any]: provider = _normalize_provider(engine.get("provider")) defaults = PROVIDER_DEFAULTS[provider] runtime = dict(engine) runtime["provider"] = provider runtime["base_url"] = str(runtime.get("base_url") or defaults["base_url"]).strip().rstrip("/") runtime["model"] = str(runtime.get("model") or defaults["model"]).strip() runtime["api_key"] = str(runtime.get("api_key") or "").strip() runtime["name"] = str(runtime.get("name") or defaults["name"]).strip() if not runtime.get("id"): runtime["id"] = str(uuid4()) runtime = _overlay_engine_api_key_from_env(runtime) return runtime async def _log_ai_call( self, *, tenant_id: int, function_name: str, prompt: str, response: str, status: str, provider: str = "", model: str = "", routing_meta: dict[str, Any] | None = None, ) -> None: tokens_used = max(len(prompt) // 4, 1) + max(len(response) // 4, 0) cost = self._estimate_cost(provider=provider, model=model, tokens=tokens_used) meta = _normalize_routing_meta(routing_meta) row_payload = { "tenant_id": tenant_id, "user_id": None, "function_name": function_name, "prompt": prompt[:8000], "response": self._trim_ai_call_response(response), "tokens_used": tokens_used, "cost": cost, "status": status, "planned_tier": meta.get("planned_tier") or None, "executed_tier": meta.get("executed_tier") or None, "routing_decision": meta.get("routing_decision") or None, "route_source": meta.get("route_source") or None, "route_rule_name": meta.get("route_rule_name") or None, "routing_config_version": meta.get("routing_config_version") or None, "routing_config_updated_at": meta.get("routing_config_updated_at") or None, } self._last_ai_call_summary = { "function_name": function_name, "status": status, "planned_tier": meta.get("planned_tier") or None, "executed_tier": meta.get("executed_tier") or None, "route_source": meta.get("route_source") or None, "route_rule_name": meta.get("route_rule_name") or None, "routing_decision": meta.get("routing_decision") or None, "routing_config_version": meta.get("routing_config_version") or None, } async with AsyncSessionLocal() as session: for attempt in range(2): try: session.add(AICallLog(**row_payload)) await session.commit() return except OperationalError as exc: await session.rollback() if "database is locked" not in str(exc).lower() or attempt >= 1: return await asyncio.sleep(0.05 * (attempt + 1)) except Exception: await session.rollback() return @staticmethod def _trim_ai_call_response(response: str) -> str: text = str(response or "") if len(text) <= AI_CALL_LOG_RESPONSE_MAX_CHARS: return text tail_chars = min(AI_CALL_LOG_RESPONSE_TAIL_CHARS, AI_CALL_LOG_RESPONSE_MAX_CHARS // 3) head_budget = AI_CALL_LOG_RESPONSE_MAX_CHARS - tail_chars - len(AI_CALL_LOG_TRUNCATION_MARKER) if head_budget <= 0: return text[:AI_CALL_LOG_RESPONSE_MAX_CHARS] return f"{text[:head_budget]}{AI_CALL_LOG_TRUNCATION_MARKER}{text[-tail_chars:]}" @staticmethod def _estimate_cost(provider: str, model: str, tokens: int) -> float: _ = model per_1k = { "deepseek": 0.0003, "glm": 0.0006, "openai": 0.0012, "ollama": 0.0, }.get(provider, 0.0005) return round((tokens / 1000.0) * per_1k, 6) @staticmethod def _parse_json_object(text: str) -> dict[str, Any]: raw = (text or "").strip() if raw.startswith("```"): parts = raw.split("```") if len(parts) >= 2: raw = parts[1].strip() if raw.startswith("json"): raw = raw[4:].strip() try: parsed = json.loads(raw) return parsed if isinstance(parsed, dict) else {} except Exception: return {} @staticmethod def _parse_json_array(text: str) -> list[dict]: raw = (text or "").strip() if raw.startswith("```"): parts = raw.split("```") if len(parts) >= 2: raw = parts[1].strip() if raw.startswith("json"): raw = raw[4:].strip() try: parsed = json.loads(raw) if isinstance(parsed, list): return [item for item in parsed if isinstance(item, dict)] except Exception: return [] return [] @staticmethod def _extract_text_from_value(value: Any) -> str: if value is None: return "" if isinstance(value, str): return value if isinstance(value, (int, float, bool)): return str(value) if isinstance(value, dict): parts: list[str] = [] for key in ("text", "content", "value", "output_text", "input_text", "arguments", "json"): if key in value: parts.append(LLMService._extract_text_from_value(value.get(key))) return "".join(parts) if isinstance(value, list): return "".join(LLMService._extract_text_from_value(item) for item in value) return str(value) @staticmethod def _sanitize_visible_answer(text: str) -> str: cleaned = str(text or "") if not cleaned: return "" cleaned = re.sub(r".*?", "", cleaned, flags=re.IGNORECASE | re.DOTALL) return cleaned.strip()