Extract and prepare content for Gemini API from OpenAI message content. Args: content: Either a string or a list of content items (text/image_url) Returns: tuple: (gemini_parts, combined_text_prompt, image_count)
(self, content)
| 336 | return system_instruction, contents |
| 337 | |
| 338 | def _prepare_gemini_content(self, content): |
| 339 | """ |
| 340 | Extract and prepare content for Gemini API from OpenAI message content. |
| 341 | |
| 342 | Args: |
| 343 | content: Either a string or a list of content items (text/image_url) |
| 344 | |
| 345 | Returns: |
| 346 | tuple: (gemini_parts, combined_text_prompt, image_count) |
| 347 | """ |
| 348 | gemini_parts = [] |
| 349 | combined_text_prompt = "" |
| 350 | image_count = 0 |
| 351 | |
| 352 | if isinstance(content, str): |
| 353 | text = content.strip() |
| 354 | if text: |
| 355 | gemini_parts.append({"text": text}) |
| 356 | combined_text_prompt = text |
| 357 | elif isinstance(content, list): |
| 358 | text_parts_log = [] |
| 359 | for item in content: |
| 360 | if not isinstance(item, dict): continue |
| 361 | item_type = item.get("type") |
| 362 | if item_type == "text": |
| 363 | text = item.get("text", "").strip() |
| 364 | if text: |
| 365 | gemini_parts.append({"text": text}) |
| 366 | text_parts_log.append(text) |
| 367 | elif item_type == "image_url": |
| 368 | image_url_data = item.get("image_url") |
| 369 | if isinstance(image_url_data, dict) and "url" in image_url_data: |
| 370 | url = image_url_data["url"] |
| 371 | if url.startswith("data:"): |
| 372 | try: |
| 373 | header, base64_data = url.split(",", 1) |
| 374 | mime_match = re.match(r"data:(image\/[a-zA-Z+.-]+);base64", header) |
| 375 | if mime_match: |
| 376 | mime_type = mime_match.group(1) |
| 377 | gemini_parts.append({ |
| 378 | "inline_data": {"mime_type": mime_type, "data": base64_data} |
| 379 | }) |
| 380 | image_count += 1 |
| 381 | except Exception as e: |
| 382 | logger.error(f"Error processing data URI image: {e}") |
| 383 | combined_text_prompt = " ".join(text_parts_log) + f" ({image_count} images)" if image_count else " ".join(text_parts_log) |
| 384 | |
| 385 | return gemini_parts, combined_text_prompt, image_count |
| 386 | |
| 387 | def _convert_gemini_chunk_to_openai(self, gemini_chunk: dict, chunk_id: str, index: int, model: str): |
| 388 | """Convert a Gemini streaming chunk to OpenAI format.""" |
no test coverage detected