Method transform_request

app/core/upstream.py:1300–1599

转换 OpenAI 请求为上游格式。

(
        self,
        request: OpenAIRequest,
        excluded_tokens: Optional[Set[str]] = None,
        excluded_guest_user_ids: Optional[Set[str]] = None,
    )

            return None

    async def transform_request(
        self,
        request: OpenAIRequest,
        excluded_tokens: Optional[Set[str]] = None,
        excluded_guest_user_ids: Optional[Set[str]] = None,
    ) -> Dict[str, Any]:
        """转换 OpenAI 请求为上游格式。"""
        self.logger.info(f"🔄 转换 OpenAI 请求到上游格式: {request.model}")

        raw_messages = [
            message.model_dump(exclude_none=True)
            for message in request.messages
        ]
        normalized_messages = _preprocess_openai_messages(raw_messages)

        auth_info = await self.get_auth_info(
            excluded_tokens=excluded_tokens,
            excluded_guest_user_ids=excluded_guest_user_ids,
        )
        token = str(auth_info.get("token") or "")
        if not token:
            raise RuntimeError("无法获取上游认证令牌")

        user_id = str(auth_info.get("user_id") or _extract_user_id_from_token(token))
        auth_mode = str(auth_info.get("auth_mode") or "authenticated")
        token_source = str(auth_info.get("token_source") or "unknown")
        guest_user_id = auth_info.get("guest_user_id")
        # 确定请求的模型特性
        last_user_text = _extract_last_user_text(raw_messages)
        requested_model = request.model
        is_thinking_model = "-thinking" in requested_model.casefold()
        is_search_model = "-search" in requested_model.casefold()
        is_advanced_search = requested_model == settings.GLM47_ADVANCED_SEARCH_MODEL
        upstream_model_id = self.model_mapping.get(requested_model, "0727-360B-API")
        tools = request.tools if settings.TOOL_SUPPORT and request.tools else None
        tool_choice = getattr(request, "tool_choice", None)
        model_profile = self._get_model_request_profile(upstream_model_id)
        enable_thinking = request.enable_thinking
        if enable_thinking is None:
            default_enable_thinking = model_profile["default_enable_thinking"]
            enable_thinking = (
                default_enable_thinking
                if default_enable_thinking is not None
                else is_thinking_model
            )

        web_search = request.web_search
        if web_search is None:
            web_search = is_search_model or is_advanced_search

        use_persisted_chat = bool(model_profile["use_persisted_chat"])
        preview_mode = bool(model_profile["preview_mode"])
        feature_entries = list(model_profile["feature_entries"])
        persisted_user_message_id = generate_uuid() if use_persisted_chat else None
        persisted_assistant_message_id = generate_uuid() if use_persisted_chat else None

        mcp_servers = list(model_profile["mcp_servers"])
        if is_advanced_search and "advanced-search" not in mcp_servers: