vent_agent_2.0/agents/model_factory.py

"""模型实例构建。

三个要点：
1. **思考级别烧进实例**——级别是构造参数，运行时改级别等于换模型，需要
   重建 agent（见 agents/agent_registry.py 的失效重建）。
2. **必须显式声明上下文窗口**——内置 SummarizationMiddleware 的
   ``trigger=("fraction", x)`` 依赖模型 ``profile["max_input_tokens"]``；
   不声明则框架在首次触发压缩时直接报错，而不是降级。
3. **不信任进程环境**——用 ``config`` 读 .env 的值，而不是让 SDK 自己去
   捡 ``OPENAI_API_KEY`` 等环境变量，避免被外部环境污染。
"""

from __future__ import annotations

import functools
from ipaddress import ip_address
from typing import Any
from urllib.parse import urlsplit

import config
from obs import get_logger

logger = get_logger(__name__)

# 已知模型的上下文窗口（token）。来源是各家的公开文档；命中不了就用兜底值。
KNOWN_CONTEXT_WINDOWS: dict[str, int] = {
    "deepseek-flash": 128_000,
    "deepseek-v4-flash": 128_000,
    "deepseek-v4-pro": 128_000,
    "deepseek-chat": 64_000,
    "deepseek-reasoner": 64_000,
    "gpt-4o": 128_000,
    "gpt-4o-mini": 128_000,
    # 用户提供的平台上限：非思考最大输入 260K，总上下文 262K；声明输入额度。
    "qwen3.5-35b-a3b": 260_000,
}

# 兜底窗口：按最保守的假设给，宁可早点压缩也不要超窗被服务端截断
FALLBACK_CONTEXT_WINDOW = 64_000

VALID_THINKING_LEVELS = ("off", "high", "highest")


def _is_qwen35_model(model_name: str) -> bool:
    """兼容组织前缀和 FP8 等部署后缀，保留实际请求中的完整模型名。"""
    return model_name.strip().rsplit("/", 1)[-1].lower().startswith("qwen3.5-")


def _qwen_thinking_payload(level: str, extra_body: dict | None = None) -> dict[str, Any]:
    """Qwen3.5/vLLM 使用模板思考开关；high 和 highest 均开启思考。"""
    options = dict(extra_body or {})
    template_kwargs = dict(options.get("chat_template_kwargs") or {})
    template_kwargs["enable_thinking"] = level != "off"
    # 顶层 enable_thinking 是另一种接口的参数，避免同时发送相互冲突的开关。
    options.pop("enable_thinking", None)
    options["chat_template_kwargs"] = template_kwargs
    return options


def context_window(model_name: str | None = None) -> int:
    """解析模型的上下文窗口。

    ``CONTEXT_LIMIT`` 显式配置时优先——这是给「模型真实窗口比文档小」或
    「想留更多余量」的场景留的手动闸。
    """
    if config.CONTEXT_LIMIT > 0:
        return config.CONTEXT_LIMIT
    name = model_name or config.MODEL
    if name in KNOWN_CONTEXT_WINDOWS:
        return KNOWN_CONTEXT_WINDOWS[name]
    # 前缀匹配：deepseek-xxx-v5 这种没登记过的新版本，沿用同族窗口
    for known, window in KNOWN_CONTEXT_WINDOWS.items():
        family = known.split("-")[0]
        if name.startswith(family):
            return window
    logger.warning(
        "模型 %s 未登记上下文窗口，按兜底值 %d 处理（可用 CONTEXT_LIMIT 显式覆盖）",
        name, FALLBACK_CONTEXT_WINDOW,
    )
    return FALLBACK_CONTEXT_WINDOW


def normalize_thinking(level: str | None) -> str:
    """把思考级别归一到合法值；非法值回落 off 并告警。"""
    value = (level or config.THINKING_LEVEL or "off").strip().lower()
    if value not in VALID_THINKING_LEVELS:
        logger.warning("非法思考级别 %r，回落 off", level)
        return "off"
    return value


def _thinking_payload(level: str) -> dict[str, Any] | None:
    """思考级别 → 透传给 provider 的 extra_body。

    ``off`` 返回 None（**什么都不发**）：不同 provider 对「关闭思考」的
    表达方式不一致，发错字段会被服务端拒；不发才是真正中立的默认。
    """
    if level == "off":
        return None
    if level == "high":
        return {"thinking": {"type": "enabled"}, "reasoning_effort": "high"}
    return {"thinking": {"type": "enabled"}, "reasoning_effort": "highest"}


def build_model(
    model_name: str | None = None,
    thinking: str | None = None,
    timeout: int | None = None,
    context_limit: int | None = None,
    api_key: str | None = None,
    base_url: str | None = None,
    max_tokens: int | None = None,
    temperature: float | None = None,
    extra_body: dict | None = None,
) -> Any:
    """构造一个聊天模型实例。

    deepseek 系列优先用 ``langchain-deepseek``（它解析 ``reasoning_content``，
    官方 OpenAI 兼容层会把它丢掉）；其余走 ``init_chat_model`` 的 openai 通道。
    """
    name = model_name or config.MODEL
    level = normalize_thinking(thinking)
    window = context_limit or context_window(name)
    if config.OPENAI_MAX_RETRIES < 0:
        raise ValueError("OPENAI_MAX_RETRIES 必须大于或等于 0")
    provider_options = _thinking_payload(level)
    # 当前 DeepSeek 默认开启思考；off 必须显式关闭才能支持框架的强制工具选择。
    if name.lower().startswith("deepseek") and level == "off":
        provider_options = {"thinking": {"type": "disabled"}}
    if extra_body is not None:
        provider_options = extra_body
    is_qwen35 = _is_qwen35_model(name)
    if is_qwen35:
        # 使用 vLLM 支持的模板参数；保留文档提取的 JSON 格式等其他选项。
        # 不发送通用 thinking/reasoning_effort=highest，后者不是 vLLM 的合法档位。
        provider_options = _qwen_thinking_payload(level, extra_body)

    common: dict[str, Any] = {
        "model": name,
        "api_key": (config.OPENAI_API_KEY if api_key is None else api_key) or "sk-placeholder",
        "timeout": timeout or config.OPENAI_TIMEOUT,
        # 由 SDK 对单次 HTTP 请求退避重试，流已开始后不会重放 Agent 或工具。
        "max_retries": config.OPENAI_MAX_RETRIES,
        # 框架靠这个字段决定何时压缩上下文，必须显式给
        "profile": {"max_input_tokens": window},
    }
    if not config.OPENAI_TOOL_CALL_STREAMING:
        # 部分兼容网关会丢失 SSE 中的 delta.tool_calls；完整 JSON 响应仍可正常解析。
        # 仅带工具的模型请求改走非流式，Agent 的消息和进度事件仍照常返回。
        common["disable_streaming"] = "tool_calling"
    if is_qwen35:
        # 业务工具存在先后依赖，最终结构化结果也应单独提交。
        common["model_kwargs"] = {"parallel_tool_calls": False}
    endpoint = config.OPENAI_BASE_URL if base_url is None else base_url
    if endpoint:
        common["base_url"] = endpoint
        hostname = urlsplit(endpoint).hostname or ""
        try:
            private_endpoint = ip_address(hostname).is_private
        except ValueError:
            private_endpoint = hostname == "localhost"
        if private_endpoint:
            # httpx在macOS也会读取系统代理；单位内网模型应直接连接。
            import httpx
            common["http_client"] = httpx.Client(trust_env=False)
            common["http_async_client"] = httpx.AsyncClient(trust_env=False)
    if provider_options:
        common["extra_body"] = provider_options
    if max_tokens is not None:
        common["max_tokens"] = max_tokens
    if temperature is not None:
        common["temperature"] = temperature

    if common["api_key"] == "sk-placeholder":
        logger.warning("OPENAI_API_KEY 未配置，模型实例无法真正发起请求（仅可用于结构验证）")

    model = _instantiate(name, common)
    logger.info(
        "模型就绪 model=%s thinking=%s window=%d base_url=%s max_retries=%d tool_call_streaming=%s",
        name, level, window, endpoint or "(默认)", common["max_retries"], config.OPENAI_TOOL_CALL_STREAMING,
    )
    if is_qwen35:
        logger.info(
            "Qwen3.5 思考请求参数 chat_template_kwargs.enable_thinking=%s；配置级别=%s，high/highest 均开启思考",
            provider_options["chat_template_kwargs"]["enable_thinking"], level,
        )
    return model


def _instantiate(name: str, common: dict[str, Any]) -> Any:
    """按模型名选择集成包。导入失败时逐级回落，并说明回落原因。"""
    if name.lower().startswith("deepseek"):
        try:
            from langchain_deepseek import ChatDeepSeek

            return ChatDeepSeek(**common)
        except ImportError:
            logger.warning("langchain-deepseek 未安装，回落到 ChatOpenAI（会丢失 reasoning_content）")

    try:
        from langchain_openai import ChatOpenAI

        if common.get("base_url"):
            from agents.openai_compat import GatewayChatOpenAI

            return GatewayChatOpenAI(**common)
        return ChatOpenAI(**common)
    except ImportError:
        pass

    from langchain.chat_models import init_chat_model

    init_kwargs = {
        "api_key": common["api_key"],
        "timeout": common["timeout"],
        "max_retries": common["max_retries"],
    }
    if common.get("base_url"):
        init_kwargs["base_url"] = common["base_url"]
    for option in ("max_tokens", "temperature", "extra_body", "model_kwargs", "http_client", "http_async_client", "disable_streaming"):
        if option in common:
            init_kwargs[option] = common[option]
    return init_chat_model(f"openai:{name}", **init_kwargs)


# ============================================================
# 进程级缓存
# ============================================================
# 没有「摘要模型」这一档：压缩用的 SummarizationMiddleware 由 deepagents 内部
# 构造，拿到的就是传进 create_deep_agent 的那个实例，没有注入另一个模型的入口。
# 所以思考级别同样作用于压缩调用；要单独给压缩配便宜模型，得先能替换掉那个
# 中间件（同名冲突见 AGENTS.md 不变式 5）。

@functools.lru_cache(maxsize=8)
def _cached_main(model_name: str, thinking: str) -> Any:
    return build_model(model_name, thinking)


def get_model(model_name: str | None = None, thinking: str | None = None) -> Any:
    """取主模型实例（按 model+thinking 缓存，同配置复用同一实例）。"""
    name = model_name or config.MODEL
    level = normalize_thinking(thinking)
    return _cached_main(name, level)


def clear_cache() -> None:
    """丢弃模型实例缓存（思考级别切换后调用）。"""
    _cached_main.cache_clear()


def describe() -> dict[str, Any]:
    """当前模型配置摘要，供 GET /api/model 展示。"""
    name = config.MODEL
    level = normalize_thinking(None)
    return {
        "model": name,
        "thinking": level,
        "thinking_levels": list(VALID_THINKING_LEVELS),
        **({
            "thinking_parameters": _qwen_thinking_payload(level),
            "thinking_levels_note": "Qwen3.5 当前接口使用思考开关，high/highest 均对应开启思考。",
        } if _is_qwen35_model(name) else {}),
        "context_window": context_window(name),
        "context_limit_override": config.CONTEXT_LIMIT,
        "base_url": config.OPENAI_BASE_URL or "(默认)",
        "configured": bool(config.OPENAI_API_KEY),
    }