MCPcopy Create free account
hub / github.com/Roy3838/Observer / _stream_gemini_response

Method _stream_gemini_response

api/gemini_pro_handler.py:231–296  ·  view source on GitHub ↗

Stream Gemini API response and convert to OpenAI SSE format.

(self, request_data: dict, target_model: str)

Source from the content-addressed store, hash-verified

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

Callers 1

handle_requestMethod · 0.95

Calls 5

get_http_clientFunction · 0.90
infoMethod · 0.80
errorMethod · 0.80

Tested by

no test coverage detected