Method _create_stream_response

app/core/upstream.py:1725–1900

创建流式响应，并在首包前支持双池重试。

(
        self,
        request: OpenAIRequest,
        transformed: Dict[str, Any]
    )

            return handle_error(e, "请求处理")

    async def _create_stream_response(
        self,
        request: OpenAIRequest,
        transformed: Dict[str, Any]
    ) -> AsyncGenerator[str, None]:
        """创建流式响应，并在首包前支持双池重试。"""
        max_attempts = self._get_total_retry_limit()
        excluded_tokens: Set[str] = set()
        excluded_guest_user_ids: Set[str] = set()
        current_token = str(transformed.get("token") or "")

        try:
            proxies = self._get_proxy_config()

            async with httpx.AsyncClient(
                timeout=self._build_timeout(read_timeout=180.0),
                http2=True,
                limits=self._build_limits(),
                proxy=proxies,
            ) as client:
                for attempt in range(max_attempts):
                    self.logger.info(f"🎯 发送请求到上游: {transformed['url']}")
                    async with client.stream(
                        "POST",
                        transformed["url"],
                        json=transformed["body"],
                        headers=transformed["headers"],
                    ) as response:
                        error_text = await response.aread() if response.status_code != 200 else b""
                        error_msg = error_text.decode("utf-8", errors="ignore")
                        error_code, parsed_error_message = (
                            self._extract_upstream_error_details(
                                response.status_code,
                                error_msg,
                            )
                            if response.status_code != 200
                            else (None, "")
                        )
                        is_concurrency_limited = self._is_concurrency_limited(
                            response.status_code,
                            error_code,
                            parsed_error_message,
                        )

                        if self._should_retry_guest_session(
                            response.status_code,
                            is_concurrency_limited,
                            attempt,
                            max_attempts,
                            transformed,
                        ):
                            guest_user_id = str(
                                transformed.get("guest_user_id")
                                or transformed.get("user_id")
                                or ""
                            )
                            if guest_user_id:
                                excluded_guest_user_ids.add(guest_user_id)