Spaces:
Sleeping
Sleeping
File size: 6,492 Bytes
5868187 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 |
"""
Format detection utilities for supporting both OpenAI and Gemini request formats
"""
from typing import Dict, Any
from log import log
def detect_request_format(data: Dict[str, Any]) -> str:
"""
Detect whether the request is in OpenAI or Gemini format.
Returns:
"openai" or "gemini"
"""
# OpenAI format indicators:
# - Has "messages" field with array of {role, content} objects
# - Role values are "system", "user", "assistant"
if "messages" in data and isinstance(data["messages"], list):
if data["messages"] and isinstance(data["messages"][0], dict):
# Check for OpenAI role values
first_role = data["messages"][0].get("role", "")
if first_role in ["system", "user", "assistant"]:
return "openai"
# Gemini format indicators:
# - Has "contents" field with array of {role, parts} objects
# - Role values are "user", "model"
# - May have "systemInstruction" field
if "contents" in data and isinstance(data["contents"], list):
if data["contents"] and isinstance(data["contents"][0], dict):
# Check for Gemini structure
if "parts" in data["contents"][0]:
return "gemini"
# Additional Gemini indicators
if "systemInstruction" in data or "generationConfig" in data:
return "gemini"
# Default to OpenAI if unclear (for backwards compatibility)
log.debug(f"Unable to definitively detect format, defaulting to OpenAI. Keys present: {list(data.keys())}")
return "openai"
def gemini_request_to_openai(gemini_request: Dict[str, Any]) -> Dict[str, Any]:
"""
Convert a Gemini format request to OpenAI format.
Args:
gemini_request: Request in Gemini API format
Returns:
Dictionary in OpenAI API format
"""
openai_request = {
"model": gemini_request.get("model", "gemini-2.5-pro"),
"messages": []
}
# Convert system instruction if present
if "systemInstruction" in gemini_request:
system_content = ""
if isinstance(gemini_request["systemInstruction"], dict):
parts = gemini_request["systemInstruction"].get("parts", [])
for part in parts:
if "text" in part:
system_content += part["text"]
elif isinstance(gemini_request["systemInstruction"], str):
system_content = gemini_request["systemInstruction"]
if system_content:
openai_request["messages"].append({
"role": "system",
"content": system_content
})
# Convert contents to messages
contents = gemini_request.get("contents", [])
for content in contents:
role = content.get("role", "user")
# Map Gemini roles to OpenAI roles
if role == "model":
role = "assistant"
# Convert parts to content
parts = content.get("parts", [])
if len(parts) == 1 and "text" in parts[0]:
# Simple text message
openai_request["messages"].append({
"role": role,
"content": parts[0]["text"]
})
elif len(parts) > 0:
# Multi-part message (could include images)
content_parts = []
for part in parts:
if "text" in part:
content_parts.append({
"type": "text",
"text": part["text"]
})
elif "inlineData" in part:
# Convert Gemini inline data to OpenAI image format
inline_data = part["inlineData"]
mime_type = inline_data.get("mimeType", "image/jpeg")
data = inline_data.get("data", "")
content_parts.append({
"type": "image_url",
"image_url": {
"url": f"data:{mime_type};base64,{data}"
}
})
if content_parts:
# If only one text part, use simple string format
if len(content_parts) == 1 and content_parts[0]["type"] == "text":
openai_request["messages"].append({
"role": role,
"content": content_parts[0]["text"]
})
else:
openai_request["messages"].append({
"role": role,
"content": content_parts
})
# Convert generation config if present
if "generationConfig" in gemini_request:
config = gemini_request["generationConfig"]
if "temperature" in config:
openai_request["temperature"] = config["temperature"]
if "topP" in config:
openai_request["top_p"] = config["topP"]
if "topK" in config:
openai_request["top_k"] = config["topK"]
if "maxOutputTokens" in config:
openai_request["max_tokens"] = config["maxOutputTokens"]
if "stopSequences" in config:
openai_request["stop"] = config["stopSequences"]
if "frequencyPenalty" in config:
openai_request["frequency_penalty"] = config["frequencyPenalty"]
if "presencePenalty" in config:
openai_request["presence_penalty"] = config["presencePenalty"]
if "candidateCount" in config:
openai_request["n"] = config["candidateCount"]
if "seed" in config:
openai_request["seed"] = config["seed"]
# Preserve stream setting if present
if "stream" in gemini_request:
openai_request["stream"] = gemini_request["stream"]
return openai_request
def validate_and_normalize_request(data: Dict[str, Any]) -> Dict[str, Any]:
"""
Validate and normalize the request to OpenAI format.
Automatically detects format and converts if necessary.
Args:
data: Raw request data
Returns:
Normalized request in OpenAI format
"""
format_type = detect_request_format(data)
log.info(f"Detected request format: {format_type}")
if format_type == "gemini":
# Convert Gemini format to OpenAI format
return gemini_request_to_openai(data)
else:
# Already in OpenAI format
return data |