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

Method _stream_gemini_response

api/gemini_pro_handler.py:225–290  ·  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

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}")

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