Method _stream_gemini_response

api/gemini_handler.py:263–328

Stream Gemini API response and convert to OpenAI SSE format.

(self, request_data: dict, target_model: str)

        return openai_response

    async def _stream_gemini_response(self, request_data: dict, target_model: str):
        """Stream Gemini API response and convert to OpenAI SSE format."""
        # Convert messages to Gemini format
        messages = request_data.get("messages", [])
        system_instruction, contents = self._convert_messages_to_gemini_format(messages)

        # Prepare Gemini API call
        gemini_url = f"https://generativelanguage.googleapis.com/v1beta/models/{target_model}:streamGenerateContent?alt=sse"
        payload = {"contents": contents}

        # Add system_instruction if present
        if system_instruction:
            payload["system_instruction"] = system_instruction

        # Add generationConfig if needed
        generation_config = {}
        if "temperature" in request_data: generation_config["temperature"] = request_data["temperature"]
        if "max_tokens" in request_data: generation_config["maxOutputTokens"] = request_data["max_tokens"]
        if generation_config: payload["generationConfig"] = generation_config

        headers = {
            "Content-Type": "application/json",
            "x-goog-api-key": self.api_key,
            "User-Agent": "ObserverAI-FastAPI-Client/1.0"
        }

        logger.info(f"Streaming Gemini API: model={target_model}, messages={len(contents)}, system_instruction={system_instruction is not None}")

        try:
            client = get_http_client()
            async with client.stream("POST", gemini_url, headers=headers, json=payload) as response:
                response.raise_for_status()

                chunk_id = "gemini-chatcmpl-" + secrets.token_hex(12)
                chunk_index = 0

                async for line in response.aiter_lines():
                    if line.startswith("data: "):
                        chunk_data = line[6:]  # Remove "data: " prefix
                        if chunk_data.strip():
                            try:
                                gemini_chunk = json.loads(chunk_data)
                                # Convert Gemini chunk to OpenAI format
                                openai_chunk = self._convert_gemini_chunk_to_openai(
                                    gemini_chunk, chunk_id, chunk_index, target_model
                                )
                                if openai_chunk:
                                    yield f"data: {json.dumps(openai_chunk)}\n\n"
                                    chunk_index += 1
                            except json.JSONDecodeError:
                                # Skip invalid JSON chunks
                                continue

                # Send [DONE] when finished
                yield f"data: [DONE]\n\n"

        except httpx.RequestError as exc:
            logger.error(f"Gemini streaming API request failed: {exc}")