vent_deep_agent/api/model_config.py

# -*- coding: utf-8 -*-
"""
运行时模型配置管理器 —— 支持 API 动态切换模型（仅影响 chat_routes 下的智能体）

click_routes 下的智能体和 fast 模式不受影响，继续从 .env 读取固定模型。

设计：
- 模块级变量存储当前模型名，从 .env 初始化
- 切换时仅更新内存中的配置，不写回 .env（重启恢复默认）
- 线程安全：使用 threading.Lock 保护读写
"""

import threading
from pathlib import Path as _Path

# 可用模型列表
AVAILABLE_MODELS = ["deepseek-v4-pro", "deepseek-v4-flash"]

# ── 思考级别配置 ──
THINKING_LEVELS = ["off", "high", "highest"]

# 各思考级别对应的 extra_body（透传至 DeepSeek API）
# 参考：https://api-docs.deepseek.com/guides/thinking_mode
# - thinking: {"type": "disabled"} → 关闭思考链
# - thinking: {"type": "enabled"} → 开启思考链（deepseek-chat V3 系列）
# - reasoning_effort: "low"/"high"/"max" → 思考深度（deepseek-reasoner R1 系列）
# 注：V3 系列同时支持 thinking.type 和 reasoning_effort（新版 API）
THINKING_KWARGS_MAP = {
    "off":      {"thinking": {"type": "disabled"}},
    "high":     {"thinking": {"type": "enabled"}},
    "highest":  {"thinking": {"type": "enabled"}, "reasoning_effort": "max"},
}

# 从 .env 读取初始模型名
_env_file = _Path(__file__).parent.parent / ".env"
try:
    from dotenv import dotenv_values
    _cfg = dotenv_values(str(_env_file))
    _initial_model = _cfg.get("DEEPAGENT_MODEL", "deepseek-v4-pro").strip().strip('"').strip("'")
    if _initial_model not in AVAILABLE_MODELS:
        _initial_model = "deepseek-v4-pro"
except Exception:
    _initial_model = "deepseek-v4-pro"

_current_model: str = _initial_model
_thinking_level: str = "off"
_lock = threading.Lock()

# ── 模型实例缓存（切换后自动失效）──
_cached_model_instance = None
_cached_model_name: str | None = None
_cached_thinking_level: str | None = None


def get_current_model() -> str:
    """获取当前运行时模型名（线程安全）。"""
    with _lock:
        return _current_model


def set_current_model(model_name: str) -> str:
    """设置当前运行时模型名（线程安全）。

    校验模型名在可用列表中，更新配置并清除模型实例缓存。

    Args:
        model_name: 模型名称，如 "deepseek-v4-pro" 或 "deepseek-v4-flash"

    Returns:
        设置后的模型名

    Raises:
        ValueError: 模型名不在可用列表中
    """
    global _current_model, _cached_model_instance, _cached_model_name

    model_name = model_name.strip().strip('"').strip("'")

    if model_name not in AVAILABLE_MODELS:
        raise ValueError(
            f"不支持的模型: {model_name}，可用模型: {', '.join(AVAILABLE_MODELS)}"
        )

    with _lock:
        _current_model = model_name
        # 清除模型实例缓存，下次调用 get_model_instance() 时重新创建
        _cached_model_instance = None
        _cached_model_name = None

    return model_name


def get_available_models() -> list[str]:
    """返回可用模型列表。"""
    return list(AVAILABLE_MODELS)


def get_thinking_level() -> str:
    """获取当前思考级别（线程安全）。"""
    with _lock:
        return _thinking_level


def set_thinking_level(level: str) -> str:
    """设置思考级别（线程安全）。

    Args:
        level: "off" / "high" / "highest"

    Returns:
        设置后的级别名

    Raises:
        ValueError: 级别不在可用列表中
    """
    global _thinking_level

    level = level.strip().lower()
    if level not in THINKING_LEVELS:
        raise ValueError(
            f"不支持的思考级别: {level}，可用级别: {', '.join(THINKING_LEVELS)}"
        )

    # 清除模型实例缓存（因为 model_kwargs 变了）
    invalidate_model_cache()

    with _lock:
        _thinking_level = level

    return level


def get_available_thinking_levels() -> list[str]:
    """返回可用思考级别列表。"""
    return list(THINKING_LEVELS)


def invalidate_model_cache():
    """清除模型实例缓存（由各 Agent 模块切换时调用）。"""
    global _cached_model_instance, _cached_model_name, _cached_thinking_level
    with _lock:
        _cached_model_instance = None
        _cached_model_name = None
        _cached_thinking_level = None


def get_model_instance():
    """获取当前运行时模型的 LangChain ChatModel 实例。

    复用 _get_model() 的 Base URL / API Key / 超时逻辑，
    但模型名来自运行时配置而非 .env。
    同时根据当前思考级别传入对应的 model_kwargs。

    返回的实例会被缓存，直到模型或思考级别切换时自动失效。
    """
    global _cached_model_instance, _cached_model_name, _cached_thinking_level

    current = get_current_model()
    think_level = get_thinking_level()

    # 缓存命中：模型未变 + 思考级别未变 + 实例已创建
    if (_cached_model_instance is not None
            and _cached_model_name == current
            and _cached_thinking_level == think_level):
        print(f"[模型] 缓存命中: {current} | 思考级别: {think_level}")
        return _cached_model_instance

    # 读取 .env 中的 API 配置（Base URL / Key / Timeout 固定不变）
    from dotenv import dotenv_values
    from langchain.chat_models import init_chat_model

    _cfg = dotenv_values(str(_env_file))

    base_url = _cfg.get("OPENAI_BASE_URL", "")
    api_key = _cfg.get("OPENAI_API_KEY", "")
    timeout_str = _cfg.get("OPENAI_TIMEOUT", "180")

    # 清理可能带入的引号
    api_key = api_key.strip().strip('"').strip("'") if api_key else ""
    base_url = base_url.strip().strip('"').strip("'") if base_url else ""

    # 解析超时时间
    try:
        timeout = float(timeout_str.strip())
    except (ValueError, TypeError):
        timeout = 180.0

    # 确保模型字符串带有 provider 前缀
    model_str = current
    if ":" not in model_str:
        model_str = f"openai:{model_str}"

    # 构建思考级别对应的参数
    # ChatOpenAI 有独立的 reasoning_effort 字段（直接透传至 API 顶层）
    # thinking 等 DeepSeek 特有参数必须通过 extra_body 传输
    thinking_cfg = dict(THINKING_KWARGS_MAP.get(think_level, {}))
    
    # 分离：reasoning_effort → 直接传参，其余 → extra_body
    direct_kwargs = {}
    extra_body_params = {}
    if "reasoning_effort" in thinking_cfg:
        direct_kwargs["reasoning_effort"] = thinking_cfg.pop("reasoning_effort")
    extra_body_params = thinking_cfg  # 剩余的（如 thinking）
    
    if direct_kwargs or extra_body_params:
        print(f"[模型] 创建实例: {model_str} | 思考级别: {think_level} | direct: {direct_kwargs} | extra_body: {extra_body_params}")

    if base_url and api_key:
        instance = init_chat_model(
            model_str,
            openai_api_key=api_key,
            openai_api_base=base_url,
            temperature=0,
            timeout=timeout,
            **direct_kwargs,
            model_kwargs={"extra_body": extra_body_params} if extra_body_params else {},
        )
    else:
        instance = init_chat_model(
            model_str,
            temperature=0,
            timeout=timeout,
            **direct_kwargs,
            model_kwargs={"extra_body": extra_body_params} if extra_body_params else {},
        )

    # 缓存实例
    with _lock:
        _cached_model_instance = instance
        _cached_model_name = current
        _cached_thinking_level = think_level

    return instance