from __future__ import annotations from .shared import * from ..runtime_audit_context import current_agent_audit_context 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() default_model = str(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)) return candidates 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() runtime["enabled"] = bool(runtime.get("enabled", True)) if not runtime.get("id"): runtime["id"] = str(uuid4()) 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, user_id: int | None = None, provider_request_payload: Any = None, provider_response_payload: Any = None, provider_attempts: Any = None, ) -> None: provider_usage = self._extract_provider_usage(provider_response_payload) tokens_used = int(provider_usage.get("total_tokens") or 0) if tokens_used <= 0: 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) audit_context = current_agent_audit_context() agent_run_id = str(audit_context.get("agent_run_id") or "").strip() or None agent_step_key = str(audit_context.get("agent_step_key") or "").strip() or None if agent_run_id: meta["agent_run_id"] = agent_run_id if agent_step_key: meta["agent_step_key"] = agent_step_key if provider_usage: meta["provider_usage"] = provider_usage trace_id = str(meta.get("trace_id") or meta.get("run_id") or "").strip() or f"ai-{uuid4().hex}" row_payload = { "tenant_id": tenant_id, "user_id": int(user_id) if user_id is not None else None, "function_name": function_name, "prompt": prompt[:8000], "response": self._trim_ai_call_response(response), "tokens_used": tokens_used, "cost": cost, "status": status, "trace_id": trace_id, "agent_run_id": agent_run_id, "agent_step_key": agent_step_key, "provider": str(provider or meta.get("provider") or "").strip() or None, "model": str(model or meta.get("model") or "").strip() or None, "latency_ms": self._safe_int(meta.get("latency_ms") or meta.get("duration_ms")), "routing_meta": meta or None, "prompt_key": str(meta.get("prompt_key") or "").strip() or None, "prompt_release_id": self._safe_int(meta.get("prompt_release_id")), "prompt_source": str(meta.get("prompt_source") or "").strip() or None, "provider_request_payload": self._json_safe_payload(provider_request_payload), "provider_response_payload": self._json_safe_payload(provider_response_payload), "provider_attempts": self._json_safe_payload(provider_attempts), "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, "slot_key": meta.get("slot_key") or None, "slot_version": meta.get("slot_version") or None, "slot_runtime_source": meta.get("slot_runtime_source") or None, "slot_fallback_index": meta.get("slot_fallback_index"), "slot_degraded": bool(meta.get("slot_degraded", False)), "prompt_key": meta.get("prompt_key") or None, "prompt_release_id": meta.get("prompt_release_id"), "prompt_source": meta.get("prompt_source") or None, "prompt_format_error": bool(meta.get("prompt_format_error", False)), "provider_usage": provider_usage, } 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 _safe_int(value: Any) -> int | None: if value is None or value == "": return None try: return int(float(value)) except (TypeError, ValueError): return None @classmethod def _json_safe_payload(cls, value: Any) -> Any: if value is None: return None try: json.dumps(value, ensure_ascii=False) return value except Exception: try: return json.loads(json.dumps(value, ensure_ascii=False, default=str)) except Exception: return {"repr": str(value)} @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, "dashscope": 0.0006, "openai": 0.0012, "ollama": 0.0, }.get(provider, 0.0005) return round((tokens / 1000.0) * per_1k, 6) @staticmethod def _extract_provider_usage(payload: Any) -> dict[str, int]: if not isinstance(payload, dict): return {} usage = payload.get("usage") if isinstance(payload.get("usage"), dict) else {} if not usage: return {} result: dict[str, int] = {} for source_key, target_key in ( ("prompt_tokens", "prompt_tokens"), ("completion_tokens", "completion_tokens"), ("total_tokens", "total_tokens"), ("input_tokens", "prompt_tokens"), ("output_tokens", "completion_tokens"), ): if source_key not in usage: continue try: value = int(usage.get(source_key) or 0) except (TypeError, ValueError): continue if value > 0: result[target_key] = value if "total_tokens" not in result and (result.get("prompt_tokens") or result.get("completion_tokens")): result["total_tokens"] = int(result.get("prompt_tokens") or 0) + int(result.get("completion_tokens") or 0) return result @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()