import json import os from pathlib import Path from typing import Any, Dict, List, Optional def _env(name: str, default: str = "") -> str: return os.environ.get(name, default) def _env_optional(name: str) -> Optional[str]: value = os.environ.get(name) if value is None: # 回退读取 .env（支持运行中更新） env_path = Path(__file__).resolve().parents[1] / ".env" if env_path.exists(): try: for raw_line in env_path.read_text(encoding="utf-8").splitlines(): line = raw_line.strip() if not line or line.startswith("#") or "=" not in line: continue key, val = line.split("=", 1) if key.strip() == name: value = val.strip().strip('"').strip("'") break except Exception: value = None if value is None: return None value = value.strip() return value or None # 模型上下文窗口（单位: token） CONTEXT_WINDOWS = { "kimi": 256_000, "kimi-k2.5": 256_000, "qwen3-max": 256_000, "qwen3-vl-plus": 256_000, "minimax-m2.5": 204_800, "deepseek": 128_000, } # 默认（Kimi） KIMI_BASE = _env("API_BASE_KIMI", _env("AGENT_API_BASE_URL", "https://api.moonshot.cn/v1")) KIMI_KEY = _env("API_KEY_KIMI", _env("AGENT_API_KEY", "")) KIMI_BASE_OFFICIAL = _env_optional("API_BASE_KIMI_OFFICIAL") KIMI_KEY_OFFICIAL = _env_optional("API_KEY_KIMI_OFFICIAL") KIMI_FAST_MODEL = _env("MODEL_KIMI_FAST", _env("AGENT_MODEL_ID", "kimi-k2-0905-preview")) KIMI_THINK_MODEL = _env("MODEL_KIMI_THINK", _env("AGENT_THINKING_MODEL_ID", "kimi-k2-thinking")) KIMI_25_MODEL = _env("MODEL_KIMI_25", "kimi-k2.5") # DeepSeek DEEPSEEK_BASE = _env("API_BASE_DEEPSEEK", "https://api.deepseek.com") DEEPSEEK_KEY = _env("API_KEY_DEEPSEEK", _env("AGENT_DEEPSEEK_API_KEY", "")) DEEPSEEK_FAST_MODEL = _env("MODEL_DEEPSEEK_FAST", "deepseek-chat") DEEPSEEK_THINK_MODEL = _env("MODEL_DEEPSEEK_THINK", "deepseek-reasoner") # Qwen QWEN_BASE = _env("API_BASE_QWEN", "https://dashscope.aliyuncs.com/compatible-mode/v1") QWEN_KEY = _env("API_KEY_QWEN", _env("DASHSCOPE_API_KEY", "")) QWEN_BASE_OFFICIAL = _env_optional("API_BASE_QWEN_OFFICIAL") QWEN_KEY_OFFICIAL = _env_optional("API_KEY_QWEN_OFFICIAL") QWEN_MAX_MODEL = _env("MODEL_QWEN_MAX", "qwen3-max") QWEN_VL_MODEL = _env("MODEL_QWEN_VL", "qwen3.5-plus") # MiniMax MINIMAX_BASE = _env("API_BASE_MINIMAX", "https://api.minimaxi.com/v1") MINIMAX_KEY = _env("API_KEY_MINIMAX", "") MINIMAX_BASE_OFFICIAL = _env_optional("API_BASE_MINIMAX_OFFICIAL") MINIMAX_KEY_OFFICIAL = _env_optional("API_KEY_MINIMAX_OFFICIAL") MINIMAX_MODEL = _env("MODEL_MINIMAX", "MiniMax-M2.5") MODEL_PROFILES = { "kimi": { "context_window": CONTEXT_WINDOWS["kimi"], "fast": { "base_url": KIMI_BASE, "api_key": KIMI_KEY, "model_id": KIMI_FAST_MODEL, "max_tokens": None, "context_window": CONTEXT_WINDOWS["kimi"], }, "thinking": { "base_url": KIMI_BASE, "api_key": KIMI_KEY, "model_id": KIMI_THINK_MODEL, "max_tokens": None, "context_window": CONTEXT_WINDOWS["kimi"], }, "supports_thinking": True, "fast_only": False, "name": "Kimi-k2", "description": "综合能力较强", "multimodal": "none", }, "kimi-k2.5": { "context_window": CONTEXT_WINDOWS["kimi-k2.5"], "fast": { "base_url": KIMI_BASE, "api_key": KIMI_KEY, "model_id": KIMI_25_MODEL, "max_tokens": None, "context_window": CONTEXT_WINDOWS["kimi-k2.5"], "extra_params": {"thinking": {"type": "disabled"}} }, "thinking": { "base_url": KIMI_BASE, "api_key": KIMI_KEY, "model_id": KIMI_25_MODEL, "max_tokens": None, "context_window": CONTEXT_WINDOWS["kimi-k2.5"], "extra_params": {"thinking": {"type": "enabled"}, "enable_thinking": True} }, "supports_thinking": True, "fast_only": False, "name": "Kimi-k2.5", "description": "新版 Kimi，支持图文 & 思考开关", "multimodal": "image,video", }, "deepseek": { "context_window": CONTEXT_WINDOWS["deepseek"], "fast": { "base_url": DEEPSEEK_BASE, "api_key": DEEPSEEK_KEY, "model_id": DEEPSEEK_FAST_MODEL, "max_tokens": 8192, "context_window": CONTEXT_WINDOWS["deepseek"] }, "thinking": { "base_url": DEEPSEEK_BASE, "api_key": DEEPSEEK_KEY, "model_id": DEEPSEEK_THINK_MODEL, "max_tokens": 65536, "context_window": CONTEXT_WINDOWS["deepseek"] }, "supports_thinking": True, "fast_only": False, "name": "DeepSeek", "description": "数学能力较强", "multimodal": "none", }, "qwen3-max": { "context_window": CONTEXT_WINDOWS["qwen3-max"], "fast": { "base_url": QWEN_BASE, "api_key": QWEN_KEY, "model_id": QWEN_MAX_MODEL, "max_tokens": 65536, "context_window": CONTEXT_WINDOWS["qwen3-max"] }, "thinking": None, # 不支持思考 "supports_thinking": False, "fast_only": True, "name": "Qwen3-Max", "description": "仅快速模式", "multimodal": "none", "hidden": True, }, "qwen3-vl-plus": { "context_window": CONTEXT_WINDOWS["qwen3-vl-plus"], "fast": { "base_url": QWEN_BASE, "api_key": QWEN_KEY, "model_id": QWEN_VL_MODEL, "max_tokens": 32768, "context_window": CONTEXT_WINDOWS["qwen3-vl-plus"], "extra_params": {} }, "thinking": { "base_url": QWEN_BASE, "api_key": QWEN_KEY, "model_id": QWEN_VL_MODEL, "max_tokens": 32768, "context_window": CONTEXT_WINDOWS["qwen3-vl-plus"], "extra_params": {"enable_thinking": True} }, "supports_thinking": True, "fast_only": False, "name": "Qwen3.5", "description": "图文视频多模态 + 深度思考", "multimodal": "image,video", }, "minimax-m2.5": { "context_window": CONTEXT_WINDOWS["minimax-m2.5"], "fast": { "base_url": MINIMAX_BASE, "api_key": MINIMAX_KEY, "model_id": MINIMAX_MODEL, "max_tokens": 65536, "context_window": CONTEXT_WINDOWS["minimax-m2.5"], "extra_params": {"reasoning_split": True} }, "thinking": { "base_url": MINIMAX_BASE, "api_key": MINIMAX_KEY, "model_id": MINIMAX_MODEL, "max_tokens": 65536, "context_window": CONTEXT_WINDOWS["minimax-m2.5"], "extra_params": {"reasoning_split": True} }, "supports_thinking": True, "fast_only": False, "deep_only": True, "name": "MiniMax-M2.5", "description": "仅深度思考，超长上下文", "multimodal": "none", } } MODEL_PROMPT_OVERRIDES = { "kimi": { "model_description": "你的基础模型是 Kimi-k2，由月之暗面公司开发，是一个开源的 MoE 架构模型，拥有 1T 参数和 32B 激活参数，当前智能助手应用由火山引擎提供 API 服务。", "thinking_model_line": "思考模式时，第一次请求的模型不是 Kimi-k2，而是 Kimi-k2-Thinking，一个更善于分析复杂问题、规划复杂流程的模型，在后续请求时模型会换回 Kimi-k2。", "deep_thinking_line": "在深度思考模式中，请求的模型是 Kimi-k2-Thinking，一个更善于分析复杂问题、规划复杂流程的模型。" }, "kimi-k2.5": { "model_description": "你的基础模型是 Kimi-k2.5，支持图文多模态，并通过 thinking 参数开启/关闭思考能力。", "thinking_model_line": "思考模式时使用同一个 Kimi-k2.5 模型，但会在请求中注入 thinking={\"type\": \"enabled\"} 来开启思考；快速模式则传递 thinking={\"type\": \"disabled\"}。", "deep_thinking_line": "深度思考模式下，所有请求都会携带 thinking={\"type\": \"enabled\"}，以获得持续的推理能力。" }, "deepseek": { "model_description": "你的基础模型是 DeepSeek-V3.2（deepseek-chat），由 DeepSeek 提供，数学与推理能力较强，当前通过官方 API 调用。", "thinking_model_line": "思考模式时，第一次请求使用 DeepSeek-Reasoner，一个强化推理的模型，后续请求会切回 DeepSeek-V3.2。", "deep_thinking_line": "在深度思考模式中，请求的模型是 DeepSeek-Reasoner，用于深入分析复杂问题并规划步骤。" }, "qwen3-max": { "model_description": "你的基础模型是 Qwen3-Max，由通义千问提供，当前仅支持快速模式，不提供思考或深度思考能力。", "thinking_model_line": "Qwen3-Max 仅支持快速模式，思考模式会被自动关闭。", "deep_thinking_line": "Qwen3-Max 不支持深度思考模式，将保持快速模式。" }, "qwen3-vl-plus": { "model_description": "你的基础模型是 Qwen3.5，由通义千问提供，支持图文多模态理解。", "thinking_model_line": "思考模式时仍使用 Qwen3.5，并开启思考能力。", "deep_thinking_line": "深度思考模式下，所有请求都将启用思考能力，以获得更强的分析表现。" }, "minimax-m2.5": { "model_description": "你的基础模型是 MiniMax-M2.5，支持超长上下文，当前仅以深度思考模式运行。", "thinking_model_line": "MiniMax-M2.5 为思考模型，快速模式不会使用。", "deep_thinking_line": "深度思考模式下，所有请求持续输出思考过程并给出最终回答。" } } def get_model_profile(key: str) -> dict: profiles = get_registered_model_profiles() if key not in profiles: raise ValueError(f"未知模型 key: {key}") profile = profiles[key] try: from utils.aliyun_fallback import is_fallback_active except Exception: is_fallback_active = None if is_fallback_active and is_fallback_active(key): if key == "kimi-k2.5": kimi_base_official = _env_optional("API_BASE_KIMI_OFFICIAL") or KIMI_BASE_OFFICIAL kimi_key_official = _env_optional("API_KEY_KIMI_OFFICIAL") or KIMI_KEY_OFFICIAL if kimi_base_official and kimi_key_official: profile = dict(profile) fast = dict(profile.get("fast") or {}) thinking = dict(profile.get("thinking") or fast) fast.update({"base_url": kimi_base_official, "api_key": kimi_key_official}) thinking.update({"base_url": kimi_base_official, "api_key": kimi_key_official}) profile["fast"] = fast profile["thinking"] = thinking elif key == "qwen3-vl-plus": qwen_base_official = _env_optional("API_BASE_QWEN_OFFICIAL") or QWEN_BASE_OFFICIAL qwen_key_official = _env_optional("API_KEY_QWEN_OFFICIAL") or QWEN_KEY_OFFICIAL if qwen_base_official and qwen_key_official: profile = dict(profile) fast = dict(profile.get("fast") or {}) thinking = dict(profile.get("thinking") or fast) fast.update({"base_url": qwen_base_official, "api_key": qwen_key_official}) thinking.update({"base_url": qwen_base_official, "api_key": qwen_key_official}) profile["fast"] = fast profile["thinking"] = thinking elif key == "minimax-m2.5": minimax_base_official = _env_optional("API_BASE_MINIMAX_OFFICIAL") or MINIMAX_BASE_OFFICIAL minimax_key_official = _env_optional("API_KEY_MINIMAX_OFFICIAL") or MINIMAX_KEY_OFFICIAL if minimax_base_official and minimax_key_official: profile = dict(profile) fast = dict(profile.get("fast") or {}) thinking = dict(profile.get("thinking") or fast) fast.update({"base_url": minimax_base_official, "api_key": minimax_key_official}) thinking.update({"base_url": minimax_base_official, "api_key": minimax_key_official}) profile["fast"] = fast profile["thinking"] = thinking # 基础校验：必须有 fast 段且有 key fast = profile.get("fast") or {} if not fast.get("api_key"): raise ValueError(f"模型 {key} 缺少 API Key 配置") return profile def _normalize_multimodal(value: Any) -> str: text = str(value or "none").strip().lower() if text in {"image", "video", "none", "image,video", "video,image"}: return "image,video" if "image" in text and "video" in text else text return "none" def _parse_env_ref(raw: Any) -> Optional[str]: text = str(raw or "").strip() if not text: return None # 1) ${ENV_KEY} if text.startswith("${") and text.endswith("}") and len(text) > 3: key = text[2:-1].strip() return _env_optional(key) # 2) env:ENV_KEY if text.lower().startswith("env:") and len(text) > 4: key = text[4:].strip() resolved = _env_optional(key) return resolved if resolved is not None else None # 3) $ENV_KEY if text.startswith("$") and len(text) > 1: key = text[1:].strip() resolved = _env_optional(key) return resolved if resolved is not None else None # 4) 直接写 ENV_KEY（若环境/.env存在同名键则读取其值，否则按字面量） if text.replace("_", "").isalnum() and text.upper() == text: resolved = _env_optional(text) if resolved is not None: return resolved return text def _reasoning_flags(capability: str) -> Dict[str, bool]: normalized = str(capability or "fast").replace(" ", "").lower() if normalized == "thinking": return {"supports_thinking": True, "fast_only": False, "deep_only": True} if normalized == "fast,thinking": return {"supports_thinking": True, "fast_only": False, "deep_only": False} return {"supports_thinking": False, "fast_only": True, "deep_only": False} def _build_custom_profile(item: Dict[str, Any]) -> Optional[Dict[str, Any]]: key = str(item.get("model_name") or "").strip() if not key: return None base_url = _parse_env_ref(item.get("url")) api_key = _parse_env_ref(item.get("apikey")) if not base_url or not api_key: return None context_window = item.get("context_window") max_output_tokens = item.get("max_output_tokens") try: context_window = int(context_window) if context_window is not None else None except (TypeError, ValueError): context_window = None try: max_output_tokens = int(max_output_tokens) if max_output_tokens is not None else None except (TypeError, ValueError): max_output_tokens = None capability = str(item.get("reasoning_capability") or "fast").replace(" ", "").lower() flags = _reasoning_flags(capability) thinkmode = item.get("thinkmode_status") or {} mode_type = str((thinkmode.get("type") or thinkmode.get("mode") or "")).strip().lower() extra_global = item.get("extra_parameter") if isinstance(item.get("extra_parameter"), dict) else {} fast_extra = thinkmode.get("fast_extra_parameter") if isinstance(thinkmode.get("fast_extra_parameter"), dict) else {} if not fast_extra and isinstance(thinkmode.get("fast_parameter"), dict): fast_extra = thinkmode.get("fast_parameter") thinking_extra = thinkmode.get("thinking_extra_parameter") if isinstance(thinkmode.get("thinking_extra_parameter"), dict) else {} if not thinking_extra and isinstance(thinkmode.get("thinking_parameter"), dict): thinking_extra = thinkmode.get("thinking_parameter") fast_model_id = None thinking_model_id = None if mode_type == "switch_model": fast_model_id = str(thinkmode.get("fast_model_id") or "").strip() thinking_model_id = str(thinkmode.get("thinking_model_id") or "").strip() elif mode_type == "param_toggle": common_model = str(thinkmode.get("model_id") or "").strip() fast_model_id = common_model thinking_model_id = common_model else: common_model = str(thinkmode.get("model_id") or "").strip() if not common_model: common_model = str(item.get("model_id") or "").strip() fast_model_id = common_model thinking_model_id = common_model if not fast_model_id: fast_model_id = thinking_model_id if not thinking_model_id: thinking_model_id = fast_model_id if not fast_model_id: return None fast_config = { "base_url": base_url, "api_key": api_key, "model_id": fast_model_id, "max_tokens": max_output_tokens, "context_window": context_window, "extra_params": {**extra_global, **(fast_extra or {})}, } profile: Dict[str, Any] = { "name": key, "description": str(item.get("description") or ""), "hidden": not bool(item.get("visible", True)), "multimodal": _normalize_multimodal(item.get("multimodal")), "context_window": context_window, "fast": fast_config, "supports_thinking": bool(flags["supports_thinking"]), "fast_only": bool(flags["fast_only"]), } if flags["deep_only"]: profile["deep_only"] = True if flags["supports_thinking"]: profile["thinking"] = { "base_url": base_url, "api_key": api_key, "model_id": thinking_model_id or fast_model_id, "max_tokens": max_output_tokens, "context_window": context_window, "extra_params": {**extra_global, **(thinking_extra or {})}, } else: profile["thinking"] = None return profile def _load_custom_models() -> Dict[str, Dict[str, Any]]: custom_path = Path(__file__).resolve().parent / "custom_models.json" if not custom_path.exists(): return {} try: data = json.loads(custom_path.read_text(encoding="utf-8")) except Exception: return {} items = data if isinstance(data, list) else data.get("models", []) if not isinstance(items, list): return {} output: Dict[str, Dict[str, Any]] = {} for item in items: if not isinstance(item, dict): continue profile = _build_custom_profile(item) if not profile: continue model_name = str(item.get("model_name") or "").strip() if not model_name: continue output[model_name] = profile return output def get_registered_model_profiles() -> Dict[str, Dict[str, Any]]: merged = dict(MODEL_PROFILES) merged.update(_load_custom_models()) # 同名覆盖内置，不报错 return merged def get_registered_model_keys(visible_only: bool = False) -> List[str]: keys: List[str] = [] for key, profile in get_registered_model_profiles().items(): if visible_only and profile.get("hidden"): continue keys.append(key) return keys def get_model_capabilities(key: str) -> Dict[str, Any]: profile = get_model_profile(key) multimodal = _normalize_multimodal(profile.get("multimodal")) supports_image = multimodal in {"image", "image,video"} supports_video = multimodal in {"video", "image,video"} return { "supports_image": supports_image, "supports_video": supports_video, "multimodal": multimodal, "supports_thinking": bool(profile.get("supports_thinking", False)), "fast_only": bool(profile.get("fast_only", False)), "deep_only": bool(profile.get("deep_only", False)), } def model_supports_image(key: str) -> bool: try: return bool(get_model_capabilities(key).get("supports_image")) except Exception: return False def model_supports_video(key: str) -> bool: try: return bool(get_model_capabilities(key).get("supports_video")) except Exception: return False def get_model_prompt_replacements(key: str) -> dict: """获取模型相关的提示词替换字段，若缺失则回退到 Kimi 版本。""" fallback = MODEL_PROMPT_OVERRIDES.get("kimi", {}) overrides = MODEL_PROMPT_OVERRIDES.get(key) or {} return { "model_description": overrides.get("model_description") or fallback.get("model_description") or "", "thinking_model_line": overrides.get("thinking_model_line") or fallback.get("thinking_model_line") or "", "deep_thinking_line": overrides.get("deep_thinking_line") or fallback.get("deep_thinking_line") or "" } def get_model_context_window(key: str) -> Optional[int]: """ 获取模型的最大上下文窗口（token 数）。 优先使用 profile.context_window，其次回退到 fast.context_window。 """ profile = get_model_profile(key) ctx = profile.get("context_window") if ctx: return ctx fast = profile.get("fast") or {} return fast.get("context_window")