Stream Gemini API response and convert to OpenAI SSE format.
(self, request_data: dict, target_model: str)
| 223 | return openai_response |
| 224 | |
| 225 | async def _stream_gemini_response(self, request_data: dict, target_model: str): |
| 226 | """Stream Gemini API response and convert to OpenAI SSE format.""" |
| 227 | # Convert messages to Gemini format |
| 228 | messages = request_data.get("messages", []) |
| 229 | system_instruction, contents = self._convert_messages_to_gemini_format(messages) |
| 230 | |
| 231 | # Prepare Gemini API call |
| 232 | gemini_url = f"https://generativelanguage.googleapis.com/v1beta/models/{target_model}:streamGenerateContent?alt=sse" |
| 233 | payload = {"contents": contents} |
| 234 | |
| 235 | # Add system_instruction if present |
| 236 | if system_instruction: |
| 237 | payload["system_instruction"] = system_instruction |
| 238 | |
| 239 | # Add generationConfig if needed |
| 240 | generation_config = {} |
| 241 | if "temperature" in request_data: generation_config["temperature"] = request_data["temperature"] |
| 242 | if "max_tokens" in request_data: generation_config["maxOutputTokens"] = request_data["max_tokens"] |
| 243 | if generation_config: payload["generationConfig"] = generation_config |
| 244 | |
| 245 | headers = { |
| 246 | "Content-Type": "application/json", |
| 247 | "x-goog-api-key": self.api_key, |
| 248 | "User-Agent": "ObserverAI-FastAPI-Client/1.0" |
| 249 | } |
| 250 | |
| 251 | logger.info(f"Streaming Gemini Pro API: model={target_model}, messages={len(contents)}, system_instruction={system_instruction is not None}") |
| 252 | |
| 253 | try: |
| 254 | client = get_http_client() |
| 255 | async with client.stream("POST", gemini_url, headers=headers, json=payload) as response: |
| 256 | response.raise_for_status() |
| 257 | |
| 258 | chunk_id = "gemini-pro-chatcmpl-" + secrets.token_hex(12) |
| 259 | chunk_index = 0 |
| 260 | |
| 261 | async for line in response.aiter_lines(): |
| 262 | if line.startswith("data: "): |
| 263 | chunk_data = line[6:] # Remove "data: " prefix |
| 264 | if chunk_data.strip(): |
| 265 | try: |
| 266 | gemini_chunk = json.loads(chunk_data) |
| 267 | # Convert Gemini chunk to OpenAI format |
| 268 | openai_chunk = self._convert_gemini_chunk_to_openai( |
| 269 | gemini_chunk, chunk_id, chunk_index, target_model |
| 270 | ) |
| 271 | if openai_chunk: |
| 272 | yield f"data: {json.dumps(openai_chunk)}\n\n" |
| 273 | chunk_index += 1 |
| 274 | except json.JSONDecodeError: |
| 275 | # Skip invalid JSON chunks |
| 276 | continue |
| 277 | |
| 278 | # Send [DONE] when finished |
| 279 | yield f"data: [DONE]\n\n" |
| 280 | |
| 281 | except httpx.RequestError as exc: |
| 282 | logger.error(f"Gemini Pro streaming API request failed: {exc}") |
no test coverage detected