diff --git a/README.md b/README.md index b625a2bc2..717fff8fd 100644 --- a/README.md +++ b/README.md @@ -259,6 +259,16 @@ curl -X POST "http://127.0.0.1:7861/v1/chat/completions" \ }' ``` +#### Chat Completions 消息转换范围 + +- 接受 `system`、`developer`、`user`、`assistant` 和 `tool`。所有位置的 `system` / `developer` 文本都会按顺序收集到 Gemini 的 `systemInstruction`；Gemini 只有一个顶层指令字段，因此无法完整保留这些指令在历史中的位置及各自的角色区别。 +- `user` 支持文本和 Base64 图片 data URL（`data:image/...;base64,...`）。远程图片 URL、音频和文件内容尚未实现转换；这些输入会明确报错，避免被忽略后继续回答。 +- `assistant` 支持文本和函数工具调用；拒绝内容块会转为文本，没有正文时也会保留 `refusal` 字段。assistant 音频历史、旧版 `function` 角色与 `function_call` 尚不支持。 +- 函数工具调用保留原始 `arguments`，由转换器解析字符串或对象参数。畸形 JSON 沿用既有容错：尝试恢复首个 JSON 对象，失败时保留空参数调用。`tool` 结果支持字符串和文本块数组；工具声明、调用、结果及指定工具的 `tool_choice` 使用一致的名称规范化。 +- `tool_choice` 支持 `auto`、`none`、`required` 及指定已声明的函数工具。指定函数时，`function` 应为包含非空 `name` 字符串的对象。规范化后的工具名称冲突、没有声明工具却使用 `required`、指定未声明的工具或命名选择类型错误等情况会返回本地错误。 + +请求模型的字段或角色校验失败时返回 HTTP 422；内容无法转换、图片 data URL 无效或工具选择无法表达时返回 HTTP 400。后一类错误通常包含 `{"detail":"具体原因"}`。这会将一些原先被忽略或交给远端报错的输入改为本地拒绝；本端点的转换范围未覆盖完整 Chat Completions 功能集合。 + ### Gemini 原生格式 ```bash diff --git a/src/converter/openai2gemini.py b/src/converter/openai2gemini.py index 980531bba..f2f311060 100644 --- a/src/converter/openai2gemini.py +++ b/src/converter/openai2gemini.py @@ -3,6 +3,8 @@ 被openai-router调用，负责OpenAI格式与Gemini格式的双向转换 """ +import base64 +import binascii import json import time import uuid @@ -946,7 +948,10 @@ def convert_openai_tools_to_gemini(openai_tools: List, model: str = "") -> List[ return [{"functionDeclarations": function_declarations}] -def convert_tool_choice_to_tool_config(tool_choice: Union[str, Dict[str, Any]]) -> Dict[str, Any]: +def convert_tool_choice_to_tool_config( + tool_choice: Union[str, Dict[str, Any]], + tool_names: Optional[set] = None, +) -> Dict[str, Any]: """ 将 OpenAI tool_choice 转换为 Gemini toolConfig @@ -962,21 +967,27 @@ def convert_tool_choice_to_tool_config(tool_choice: Union[str, Dict[str, Any]]) elif tool_choice == "none": return {"functionCallingConfig": {"mode": "NONE"}} elif tool_choice == "required": + if tool_names is not None and not tool_names: + raise ValueError("tool_choice 'required' needs at least one function tool") return {"functionCallingConfig": {"mode": "ANY"}} elif isinstance(tool_choice, dict): # {"type": "function", "function": {"name": "my_function"}} - if tool_choice.get("type") == "function": - function_name = tool_choice.get("function", {}).get("name") - if function_name: + function = tool_choice.get("function") + if tool_choice.get("type") == "function" and isinstance(function, dict): + function_name = function.get("name") + if isinstance(function_name, str) and function_name: + if tool_names is not None and function_name not in tool_names: + raise ValueError( + f"tool_choice refers to an undeclared function: {function_name}" + ) return { "functionCallingConfig": { "mode": "ANY", - "allowedFunctionNames": [function_name], + "allowedFunctionNames": [_normalize_function_name(function_name)], } } - # 默认返回 AUTO 模式 - return {"functionCallingConfig": {"mode": "AUTO"}} + raise ValueError("Unsupported tool_choice for Antigravity") def convert_tool_message_to_function_response(message, all_messages: List = None) -> Dict[str, Any]: @@ -1215,6 +1226,48 @@ def _sanitize_openai_roundtrip_signatures(contents: List[Dict[str, Any]]) -> Non parts[index] = sanitized_part +def _convert_openai_message_content(role: str, content: Any) -> List[Dict[str, Any]]: + """Convert supported Chat Completions content without silently losing parts.""" + if content is None: + return [] + if isinstance(content, str): + return [{"text": content}] if content else [] + if not isinstance(content, list): + raise ValueError(f"{role} message content must be text or content parts") + + parts = [] + for part in content: + if not isinstance(part, dict): + raise ValueError(f"{role} message contains an invalid content part") + part_type = part.get("type") + if part_type == "text" or (role == "assistant" and part_type == "refusal"): + value = part.get(part_type) + if not isinstance(value, str): + raise ValueError(f"{role} {part_type} content part must contain a string") + if value: + parts.append({"text": value}) + elif role == "user" and part_type == "image_url": + image_url = part.get("image_url") + url = image_url.get("url") if isinstance(image_url, dict) else None + if not isinstance(url, str) or not url.startswith("data:image/"): + raise ValueError( + "Antigravity requires image_url content as a base64 image data URL" + ) + header, separator, encoded = url.partition(",") + if not separator or not header.endswith(";base64"): + raise ValueError("Invalid base64 image data URL") + try: + base64.b64decode(encoded, validate=True) + except (binascii.Error, ValueError) as exc: + raise ValueError("Invalid base64 image data URL") from exc + parts.append({"inlineData": {"mimeType": header[5:-7], "data": encoded}}) + else: + raise ValueError( + f"{role} content part type {part_type!r} is unsupported by Antigravity" + ) + return parts + + async def convert_openai_to_gemini_request(openai_request: Dict[str, Any]) -> Dict[str, Any]: """ 将 OpenAI 格式请求体转换为 Gemini 格式请求体 @@ -1236,35 +1289,38 @@ async def convert_openai_to_gemini_request(openai_request: Dict[str, Any]) -> Di - systemInstruction: 系统指令 (如果有) - tools, toolConfig (如果有) """ - # 处理连续的system消息（兼容性模式） + # Keep instruction roles out of Gemini contents. openai_request = await merge_system_messages(openai_request) contents = [] # 提取消息列表 messages = openai_request.get("messages", []) - - # 构建 tool_call_id -> (name, original_id, signature) 的映射 + + # 与上游一致，按 ID 查找工具名，不因不完整或重复的历史提前拒绝请求。 tool_call_mapping = {} - for msg in messages: - if msg.get("role") == "assistant" and msg.get("tool_calls"): - for tc in msg["tool_calls"]: - encoded_id = tc.get("id", "") - func_name = tc.get("function", {}).get("name") or "" + for message in messages: + if message.get("role") == "assistant": + for tool_call in message.get("tool_calls") or []: + encoded_id = tool_call.get("id", "") if encoded_id: - # 解码获取原始ID和签名 original_id, _ = decode_tool_id_and_signature(encoded_id) - tool_call_mapping[encoded_id] = (func_name, original_id, None) - - # 构建工具名称到参数 schema 的映射（用于类型修正） + func_name = tool_call.get("function", {}).get("name") or "" + tool_call_mapping[encoded_id] = (func_name, original_id) + + # 一次收集工具声明，校验名称并保留参数 schema 供历史调用修正类型。 tool_schemas = {} - if "tools" in openai_request and openai_request["tools"]: - for tool in openai_request["tools"]: - if tool.get("type") == "function": - function = tool.get("function", {}) - func_name = function.get("name") - if func_name: - tool_schemas[func_name] = function.get("parameters", {}) + normalized_names = set() + for tool in openai_request.get("tools") or []: + if tool.get("type") == "function": + function = tool["function"] + func_name = function["name"] + normalized_name = _normalize_function_name(func_name) + if normalized_name in normalized_names: + raise ValueError("Function tool names collide after Antigravity normalization") + normalized_names.add(normalized_name) + tool_schemas[func_name] = function.get("parameters", {}) + tool_names = set(tool_schemas) # 用于累积连续的 tool message 的 functionResponse parts pending_tool_parts = [] @@ -1286,36 +1342,33 @@ def flush_pending_tool_parts(): # 处理工具消息（tool role）- 累积到 pending_tool_parts if role == "tool": tool_call_id = message.get("tool_call_id", "") - func_name = message.get("name") - - # 使用映射表查找 - if tool_call_id in tool_call_mapping: - func_name, original_id, _ = tool_call_mapping[tool_call_id] - else: - # 如果没有name,尝试从消息列表中查找 - if not func_name and tool_call_id: - for msg in messages: - if msg.get("role") == "assistant" and msg.get("tool_calls"): - for tc in msg["tool_calls"]: - if tc.get("id") == tool_call_id: - func_name = tc.get("function", {}).get("name") - break - if func_name: - break - - # 解码 tool_call_id 获取原始 ID - original_id, _ = decode_tool_id_and_signature(tool_call_id) - - # 最终兜底：确保 func_name 不为空 + original_id, _ = decode_tool_id_and_signature(tool_call_id) + func_name, original_id = tool_call_mapping.get( + tool_call_id, (message.get("name"), original_id) + ) if not func_name: func_name = "unknown_function" log.warning(f"Tool message missing function name for tool_call_id={tool_call_id}, using default: {func_name}") + func_name = _normalize_function_name(func_name) + + # OpenAI also permits an array of text content parts for tool results. + if isinstance(content, list): + if not all( + isinstance(part, dict) + and part.get("type") == "text" + and isinstance(part.get("text"), str) + for part in content + ): + raise ValueError("Tool message content may only contain text parts") + content = "".join(part.get("text", "") for part in content) + elif not isinstance(content, str): + raise ValueError("Tool message content must be text or text parts") # 解析响应数据 try: - response_data = json.loads(content) if isinstance(content, str) else content - except (json.JSONDecodeError, TypeError): - response_data = {"result": str(content)} + response_data = json.loads(content) + except json.JSONDecodeError: + response_data = {"result": content} # 确保 response_data 是字典类型（Gemini API 要求 response 必须是对象） if not isinstance(response_data, dict): @@ -1334,150 +1387,75 @@ def flush_pending_tool_parts(): # 遇到非 tool 消息时，先 flush 累积的 tool parts flush_pending_tool_parts() - # system/developer 消息已经由 merge_system_messages 处理，这里跳过 - if role in ("system", "developer"): - continue + if role not in ("user", "assistant"): + raise ValueError(f"Unsupported message role: {role}") + + parts = _convert_openai_message_content(role, content) + if role == "assistant" and not parts and message.get("refusal"): + parts = [{"text": message["refusal"]}] # 将OpenAI角色映射到Gemini角色 - if role == "assistant": - role = "model" + gemini_role = "model" if role == "assistant" else "user" # 检查是否有tool_calls tool_calls = message.get("tool_calls") if tool_calls: - parts = [] - - # 如果有文本内容,先添加文本 - # 注意: content 可能是 str、list（OpenAI content block 格式 [{"type":"text","text":"..."}]）、dict 或 None - # 必须解包为纯字符串，否则 text 字段会变成 list，触发 gemini_fix 的 str(dict) 产生嵌套字符串 - if content: - if isinstance(content, list): - for _part in content: - if isinstance(_part, dict): - if _part.get("type") == "text" or "text" in _part: - _t = _part.get("text", "") - if _t: - parts.append({"text": _t}) - elif isinstance(_part, str) and _part: - parts.append({"text": _part}) - elif isinstance(content, str): - parts.append({"text": content}) - elif isinstance(content, dict): - _t = content.get("text", "") - if _t: - parts.append({"text": _t}) - else: - parts.append({"text": str(content)}) + if role != "assistant": + raise ValueError("Only assistant messages may contain tool_calls") - # 添加每个工具调用 for tool_call in tool_calls: + encoded_id = tool_call.get("id", "") try: + function = tool_call["function"] args = ( - json.loads(tool_call["function"]["arguments"]) - if isinstance(tool_call["function"]["arguments"], str) - else tool_call["function"]["arguments"] + json.loads(function["arguments"]) + if isinstance(function["arguments"], str) + else function["arguments"] ) - - # 根据工具的 schema 修正参数类型 - func_name = tool_call["function"]["name"] - if func_name in tool_schemas: - args = fix_tool_call_args_types(args, tool_schemas[func_name]) - - # 解码工具ID和thoughtSignature - encoded_id = tool_call.get("id", "") - original_id, signature = decode_tool_id_and_signature(encoded_id) - - # 构建functionCall part - function_call_part = { - "functionCall": { - "id": original_id, - "name": func_name, - "args": args - } - } - - # OpenAI/RooCode 中转可能会改写或截断 tool_call_id，真实签名回传后容易触发 - # Corrupted thought signature。工具调用使用官方跳过校验占位符更稳。 - function_call_part["thoughtSignature"] = SKIP_THOUGHT_SIGNATURE_VALIDATOR - - parts.append(function_call_part) - except (json.JSONDecodeError, KeyError) as e: - # 不能直接丢掉这个 functionCall：历史里它后面必然跟着一条 - # functionResponse，父调用一消失它就成了孤儿，Gemini 会返回 - # 空 candidate（0 completion token、finishReason=STOP）。 - # 降级保留：能救出第一个 JSON 对象就用它，否则用空参数。 - degraded_name = "" - try: - degraded_name = tool_call["function"]["name"] - except (KeyError, TypeError): - degraded_name = "" - if not degraded_name: - log.error(f"Failed to parse tool call (dropped, no name): {e}") + func_name = function["name"] + except (json.JSONDecodeError, KeyError) as exc: + # 上游容错：保留调用，恢复首个 JSON 对象，无法恢复则使用空参数。 + function = tool_call.get("function") or {} + func_name = function.get("name") + if not func_name: + log.error(f"Failed to parse tool call (dropped, no name): {exc}") continue - - try: - raw_args = tool_call["function"]["arguments"] - except (KeyError, TypeError): - raw_args = "" - degraded_args = {} + args = {} + raw_args = function.get("arguments", "") if isinstance(raw_args, str) and raw_args: try: recovered, _ = json.JSONDecoder().raw_decode(raw_args.lstrip()) except ValueError: recovered = None if isinstance(recovered, dict): - degraded_args = recovered - if degraded_args and degraded_name in tool_schemas: - degraded_args = fix_tool_call_args_types( - degraded_args, tool_schemas[degraded_name] - ) - - original_id, _signature = decode_tool_id_and_signature( - tool_call.get("id", "") - ) + args = recovered log.warning( - f"Failed to parse tool call: {e} - keeping degraded functionCall " - f"name={degraded_name} " - f"args={'partial' if degraded_args else 'empty'}" + f"Failed to parse tool call: {exc} - keeping degraded functionCall " + f"name={func_name} args={'partial' if args else 'empty'}" ) - parts.append({ + + if func_name in tool_schemas: + args = fix_tool_call_args_types(args, tool_schemas[func_name]) + + original_id, _ = decode_tool_id_and_signature(encoded_id) + normalized_name = _normalize_function_name(func_name) + parts.append( + { "functionCall": { "id": original_id, - "name": degraded_name, - "args": degraded_args, + "name": normalized_name, + "args": args, }, + # 中转可能改写 tool_call_id，使用跳过校验占位符避免签名损坏。 "thoughtSignature": SKIP_THOUGHT_SIGNATURE_VALIDATOR, - }) + } + ) + if not parts: + continue - if parts: - contents.append({"role": role, "parts": parts}) - continue - - # 处理普通内容 - if isinstance(content, list): - parts = [] - for part in content: - if part.get("type") == "text": - parts.append({"text": part.get("text", "")}) - elif part.get("type") == "image_url": - image_url = part.get("image_url", {}).get("url") - if image_url: - try: - mime_type, base64_data = image_url.split(";") - _, mime_type = mime_type.split(":") - _, base64_data = base64_data.split(",") - parts.append({ - "inlineData": { - "mimeType": mime_type, - "data": base64_data, - } - }) - except ValueError: - continue - if parts: - contents.append({"role": role, "parts": parts}) - elif content: - contents.append({"role": role, "parts": [{"text": content}]}) + if not parts: + raise ValueError(f"{role} message has no content representable by Antigravity") + contents.append({"role": gemini_role, "parts": parts}) # 循环结束后，flush 剩余的 tool parts（如果消息列表以 tool 消息结尾） flush_pending_tool_parts() @@ -1497,7 +1475,7 @@ def flush_pending_tool_parts(): # 构建生成配置 generation_config = {} model = openai_request.get("model", "") - + # 基础参数映射 if "temperature" in openai_request: generation_config["temperature"] = openai_request["temperature"] @@ -1520,12 +1498,12 @@ def flush_pending_tool_parts(): generation_config["candidateCount"] = openai_request["n"] if "seed" in openai_request: generation_config["seed"] = openai_request["seed"] - + # 处理 response_format if "response_format" in openai_request and openai_request["response_format"]: response_format = openai_request["response_format"] format_type = response_format.get("type") - + if format_type == "json_schema": # JSON Schema 模式 if "json_schema" in response_format and "schema" in response_format["json_schema"]: @@ -1539,7 +1517,7 @@ def flush_pending_tool_parts(): elif format_type == "text": # Text 模式 generation_config["responseMimeType"] = "text/plain" - + # 如果contents为空,添加默认用户消息 if not contents: contents.append({"role": "user", "parts": [{"text": "请根据系统指令回答。"}]}) @@ -1555,13 +1533,14 @@ def flush_pending_tool_parts(): gemini_request["systemInstruction"] = openai_request["systemInstruction"] # 处理工具 - 传递 model 参数以便根据模型类型选择清理策略 - model = openai_request.get("model", "") if "tools" in openai_request and openai_request["tools"]: gemini_request["tools"] = convert_openai_tools_to_gemini(openai_request["tools"], model) # 处理tool_choice if "tool_choice" in openai_request and openai_request["tool_choice"]: - gemini_request["toolConfig"] = convert_tool_choice_to_tool_config(openai_request["tool_choice"]) + choice = openai_request["tool_choice"] + if tool_names or choice not in ("auto", "none"): + gemini_request["toolConfig"] = convert_tool_choice_to_tool_config(choice, tool_names) # 透传图片生成的 size 参数（如 "1024x1536"） if "size" in openai_request and openai_request["size"]: diff --git a/src/converter/utils.py b/src/converter/utils.py index 5632647c2..748c4a301 100644 --- a/src/converter/utils.py +++ b/src/converter/utils.py @@ -59,9 +59,11 @@ def extract_content_and_reasoning(parts: list) -> tuple: async def merge_system_messages(request_body: Dict[str, Any]) -> Dict[str, Any]: """ - 处理请求体中的 system 消息： - - 顶层 system 参数（Anthropic 格式）与开头的连续 system 消息合并为 systemInstruction - - 中间的 system 消息转换为 user 消息（保持顺序） + 将 system/developer 指令收集到 Gemini 的 systemInstruction。 + + Gemini contents 不支持这两种角色。即使指令出现在历史对话中间， + 也不能将其降级为 user 消息；按原有顺序收集指令，并从 contents 中移除。 + 顶层 system 参数用于 Anthropic 格式，排在消息中的指令之前。 Args: request_body: OpenAI 或 Claude 格式的请求体，包含 messages 字段 @@ -96,7 +98,8 @@ async def merge_system_messages(request_body: Dict[str, Any]) -> Dict[str, Any]: if not messages: return request_body - # 提取开头的连续 system 消息合并为 systemInstruction + # 提取所有 system/developer 消息。Gemini 只有一个顶层指令字段， + # 因而无法保留这些指令在对话历史中的原始位置。 system_parts = [] # 如果已经从顶层 system 参数创建了 systemInstruction，获取现有的 parts @@ -106,17 +109,14 @@ async def merge_system_messages(request_body: Dict[str, Any]) -> Dict[str, Any]: system_parts = existing_instruction.get("parts", []).copy() remaining_messages = [] - collecting_system = True - - # developer 角色与 system 同语义处理（OpenAI 新版指令角色），上游仅支持 user/model instruction_roles = ("system", "developer") for message in messages: role = message.get("role", "") content = message.get("content", "") - if role in instruction_roles and collecting_system: - # 提取 system 消息的文本内容 + if role in instruction_roles: + # 提取指令消息的文本内容 if isinstance(content, str): if content.strip(): system_parts.append({"text": content}) @@ -129,33 +129,14 @@ async def merge_system_messages(request_body: Dict[str, Any]) -> Dict[str, Any]: elif isinstance(item, str) and item.strip(): system_parts.append({"text": item}) else: - # 遇到非 system 消息，停止收集 - collecting_system = False - if role in instruction_roles: - # 将后续的 system 消息转换为 user 消息 - converted_message = message.copy() - converted_message["role"] = "user" - remaining_messages.append(converted_message) - else: - remaining_messages.append(message) - - # 如果没有收集到任何 systemInstruction 内容： - # 仅当 messages 未发生变化时才返回原始请求体； - # 存在被转换为 user 的中间 system 消息时，返回更新后的 messages - if not system_parts: - if remaining_messages == messages: - return request_body - result = request_body.copy() - result["messages"] = remaining_messages - return result - - # 构建新的请求体 - result = request_body.copy() + remaining_messages.append(message) - # 添加或更新 systemInstruction - result["systemInstruction"] = {"parts": system_parts} + if not system_parts and remaining_messages == messages: + return request_body - # 更新 messages 列表（移除已处理的 system 消息） + # 没有文本时也移除空指令消息，避免传给上游。 + result = request_body.copy() + if system_parts: + result["systemInstruction"] = {"parts": system_parts} result["messages"] = remaining_messages - return result diff --git a/src/models.py b/src/models.py index 63afd465a..ee9717b11 100644 --- a/src/models.py +++ b/src/models.py @@ -4,9 +4,9 @@ - 面板 API：登录、认证、凭证操作、配置保存模型 """ -from typing import Any, Dict, List, Optional +from typing import Any, Dict, List, Literal, Optional -from pydantic import BaseModel, ConfigDict +from pydantic import BaseModel, ConfigDict, model_validator def model_to_dict(model: BaseModel) -> Dict[str, Any]: @@ -35,23 +35,51 @@ class OpenAIToolFunction(BaseModel): parameters: Optional[Dict[str, Any]] = None +class OpenAIToolCallFunction(BaseModel): + """An assistant tool call carries arguments, not a tool parameter schema.""" + + name: str + arguments: Any = "" + + class OpenAIToolCall(BaseModel): id: str - type: str = "function" - function: OpenAIToolFunction + type: Literal["function"] = "function" + function: OpenAIToolCallFunction class OpenAITool(BaseModel): - type: str = "function" + type: Literal["function"] = "function" function: OpenAIToolFunction class OpenAIChatMessage(BaseModel): - role: str + role: Literal["system", "developer", "user", "assistant", "tool"] content: Any = None name: Optional[str] = None tool_calls: Optional[List[OpenAIToolCall]] = None tool_call_id: Optional[str] = None + refusal: Optional[str] = None + audio: Optional[Dict[str, Any]] = None + function_call: Optional[Dict[str, Any]] = None + + @model_validator(mode="after") + def validate_message(self): + if self.tool_calls and self.role != "assistant": + raise ValueError("only assistant messages may contain tool_calls") + if self.role == "assistant" and self.audio is not None: + raise ValueError("assistant audio history is unsupported by Antigravity") + if self.function_call is not None: + raise ValueError("legacy function_call history is unsupported; use tool_calls") + if self.role in ("system", "developer") and not isinstance(self.content, str): + if not isinstance(self.content, list) or not all( + isinstance(part, dict) + and part.get("type") == "text" + and isinstance(part.get("text"), str) + for part in self.content + ): + raise ValueError("system and developer messages require text content") + return self class OpenAIChatCompletionRequest(BaseModel): diff --git a/src/router/antigravity/openai.py b/src/router/antigravity/openai.py index 66e0e9e61..5bb52193f 100644 --- a/src/router/antigravity/openai.py +++ b/src/router/antigravity/openai.py @@ -87,7 +87,10 @@ async def chat_completions( # 转换为 Gemini 格式 (使用 converter) from src.converter.openai2gemini import convert_openai_to_gemini_request - gemini_dict = await convert_openai_to_gemini_request(normalized_dict) + try: + gemini_dict = await convert_openai_to_gemini_request(normalized_dict) + except ValueError as exc: + raise HTTPException(status_code=400, detail=str(exc)) from exc # convert_openai_to_gemini_request 不包含 model 字段，需要手动添加 gemini_dict["model"] = real_model @@ -96,6 +99,14 @@ async def chat_completions( from src.converter.antigravity_fix import normalize_antigravity_request gemini_dict = await normalize_antigravity_request(gemini_dict) + # This endpoint accepts OpenAI roles, but the Antigravity content schema + # accepts only USER/MODEL. Catch a conversion regression before HTTP send. + for content in gemini_dict.get("contents", []): + if not isinstance(content, dict) or content.get("role") not in ("user", "model"): + raise HTTPException( + status_code=400, detail="Unrepresentable content role for Antigravity" + ) + # 准备API请求格式 - 提取model并将其他字段放入request中 api_request = { "model": gemini_dict.pop("model"), @@ -306,4 +317,4 @@ async def normal_stream_generator(): log.info("启用流式抗截断功能") return await build_streaming_response_or_error(anti_truncation_generator()) else: - return await build_streaming_response_or_error(normal_stream_generator()) \ No newline at end of file + return await build_streaming_response_or_error(normal_stream_generator()) diff --git a/tests/test_openai_message_conversion.py b/tests/test_openai_message_conversion.py new file mode 100644 index 000000000..0b2662d84 --- /dev/null +++ b/tests/test_openai_message_conversion.py @@ -0,0 +1,387 @@ +"""Offline regression checks for the OpenAI request boundary and converter.""" + +import copy +import unittest +from unittest.mock import AsyncMock, Mock, patch + +from fastapi import FastAPI +from fastapi.responses import JSONResponse +from fastapi.testclient import TestClient + +from src.converter.openai2gemini import convert_openai_to_gemini_request +from src.converter.utils import merge_system_messages +from src.models import OpenAIChatCompletionRequest, model_to_dict + + +_MISSING = object() + + +def tool_history(arguments="{}", name="lookup", result='{"value":"OK"}'): + function = {"name": name} + if arguments is not _MISSING: + function["arguments"] = copy.deepcopy(arguments) + return [ + {"role": "user", "content": "Look up a value."}, + { + "role": "assistant", + "tool_calls": [{"id": "call_1", "type": "function", "function": function}], + }, + {"role": "tool", "tool_call_id": "call_1", "content": copy.deepcopy(result)}, + ] + + +def content_parts(request, key): + return [ + part[key] for content in request["contents"] for part in content["parts"] if key in part + ] + + +class TestOpenAIMessageConversion(unittest.IsolatedAsyncioTestCase): + async def convert(self, messages, **options): + request = OpenAIChatCompletionRequest.model_validate( + {"model": "gemini-3.8-flash-high", "messages": messages, **options} + ) + return await convert_openai_to_gemini_request(model_to_dict(request)) + + async def test_tool_arguments_survive_request_validation(self): + for arguments in ('{"key":"test"}', {"key": "test"}): + with self.subTest(arguments=arguments): + converted = await self.convert(tool_history(arguments)) + self.assertEqual( + content_parts(converted, "functionCall")[0]["args"], {"key": "test"} + ) + self.assertEqual(content_parts(converted, "functionResponse")[0]["id"], "call_1") + + async def test_malformed_arguments_keep_upstream_recovery(self): + cases = [ + (' {"key":"first"}{"key":"second"}', {"key": "first"}), + ("{oops", {}), + (_MISSING, {}), + ] + for arguments, expected in cases: + with ( + self.subTest(arguments=arguments), + patch("src.converter.openai2gemini.log.warning"), + ): + converted = await self.convert(tool_history(arguments)) + self.assertEqual(content_parts(converted, "functionCall")[0]["args"], expected) + self.assertEqual(len(content_parts(converted, "functionResponse")), 1) + + async def test_interleaved_instructions_remain_instructions(self): + for role in ("system", "developer"): + with self.subTest(role=role): + converted = await self.convert( + [ + {"role": "system", "content": "Global rule."}, + {"role": "user", "content": "Hello."}, + {"role": "assistant", "content": "Hi."}, + {"role": role, "content": [{"type": "text", "text": "Answer briefly."}]}, + {"role": "user", "content": "Continue."}, + ] + ) + self.assertEqual( + converted["systemInstruction"]["parts"], + [{"text": "Global rule."}, {"text": "Answer briefly."}], + ) + self.assertEqual( + [content["role"] for content in converted["contents"]], + ["user", "model", "user"], + ) + self.assertNotIn("Answer briefly.", content_parts(converted, "text")) + + async def test_tool_result_text_parts_are_joined(self): + converted = await self.convert( + tool_history( + result=[ + {"type": "text", "text": "first "}, + {"type": "text", "text": "second"}, + ] + ) + ) + self.assertEqual( + content_parts(converted, "functionResponse")[0]["response"], + {"result": "first second"}, + ) + + async def test_tool_names_match_declarations_calls_results_and_choice(self): + converted = await self.convert( + tool_history('{"key":"test"}', name="1lookup"), + tools=[{"type": "function", "function": {"name": "1lookup"}}], + tool_choice={"type": "function", "function": {"name": "1lookup"}}, + ) + names = [ + converted["tools"][0]["functionDeclarations"][0]["name"], + content_parts(converted, "functionCall")[0]["name"], + content_parts(converted, "functionResponse")[0]["name"], + converted["toolConfig"]["functionCallingConfig"]["allowedFunctionNames"][0], + ] + self.assertEqual(names, ["_1lookup"] * 4) + + async def test_incomplete_tool_history_keeps_upstream_tolerance(self): + duplicate = tool_history() + duplicate[1]["tool_calls"].append(copy.deepcopy(duplicate[1]["tool_calls"][0])) + missing_result = tool_history()[:2] + [{"role": "user", "content": "Continue."}] + orphan_result = [ + {"role": "user", "content": "Continue."}, + {"role": "tool", "name": "lookup", "tool_call_id": "unknown", "content": "OK"}, + ] + cases = [ + (duplicate, 2, 1), + (missing_result, 1, 0), + (orphan_result, 0, 1), + (tool_history()[:2], 1, 0), + ] + for messages, calls, results in cases: + with self.subTest(messages=messages): + converted = await self.convert(messages) + self.assertEqual(len(content_parts(converted, "functionCall")), calls) + self.assertEqual(len(content_parts(converted, "functionResponse")), results) + + async def test_assistant_refusal_is_preserved_as_text(self): + for fields in ( + {"refusal": "I cannot help."}, + {"content": [{"type": "refusal", "refusal": "I cannot help."}]}, + ): + with self.subTest(fields=fields): + converted = await self.convert( + [ + {"role": "user", "content": "First request."}, + {"role": "assistant", **fields}, + {"role": "user", "content": "Next request."}, + ] + ) + self.assertEqual( + converted["contents"][1], + {"role": "model", "parts": [{"text": "I cannot help."}]}, + ) + + async def test_image_encoding_and_content_boundaries(self): + png = ( + "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mP8/x8" + "AAwMCAO+jB0QAAAAASUVORK5CYII=" + ) + converted = await self.convert( + [ + { + "role": "user", + "content": [ + {"type": "text", "text": "Describe this image."}, + {"type": "image_url", "image_url": {"url": "data:image/png;base64," + png}}, + ], + } + ] + ) + self.assertEqual( + content_parts(converted, "inlineData"), [{"mimeType": "image/png", "data": png}] + ) + unsupported = [ + [{"type": "image_url", "image_url": {"url": "data:image/png,aGVsbG8="}}], + [{"type": "image_url", "image_url": {"url": "data:image/png;base64,???"}}], + [{"type": "input_audio", "input_audio": {"data": "aGVsbG8=", "format": "wav"}}], + [{"type": "file", "file": {"filename": "example.txt", "file_data": "aGVsbG8="}}], + [{"type": "unknown", "text": "unsupported"}], + {"text": "not a content array"}, + "", + ] + for content in unsupported: + with self.subTest(content=content), self.assertRaises(ValueError): + await self.convert([{"role": "user", "content": content}]) + + async def test_tool_choice_modes_and_invalid_choices(self): + messages = [{"role": "user", "content": "Hello."}] + tools = [{"type": "function", "function": {"name": "lookup"}}] + for choice, mode in (("auto", "AUTO"), ("none", "NONE"), ("required", "ANY")): + with self.subTest(choice=choice): + converted = await self.convert(messages, tools=tools, tool_choice=choice) + self.assertEqual(converted["toolConfig"]["functionCallingConfig"]["mode"], mode) + for choice in ("auto", "none"): + with self.subTest(choice=choice, tools=None): + self.assertNotIn("toolConfig", await self.convert(messages, tool_choice=choice)) + invalid = [ + {"tool_choice": "required"}, + {"tools": tools, "tool_choice": "unknown"}, + {"tools": tools, "tool_choice": {"type": "function", "function": {"name": "missing"}}}, + {"tools": tools, "tool_choice": {"type": "function", "function": "lookup"}}, + {"tools": tools, "tool_choice": {"type": "function", "function": {"name": ["lookup"]}}}, + { + "tools": [ + {"type": "function", "function": {"name": "1lookup"}}, + {"type": "function", "function": {"name": "_1lookup"}}, + ] + }, + ] + for options in invalid: + with self.subTest(options=options), self.assertRaises(ValueError): + await self.convert(messages, **options) + + async def test_anthropic_top_level_system_is_preserved(self): + for system in ("Top-level rule.", [{"type": "text", "text": "Top-level rule."}]): + with self.subTest(system=system): + messages = [{"role": "user", "content": "Hello."}] + converted = await merge_system_messages({"system": system, "messages": messages}) + self.assertEqual( + converted["systemInstruction"]["parts"], [{"text": "Top-level rule."}] + ) + self.assertEqual(converted["messages"], messages) + + +class TestOpenAIRoute(unittest.TestCase): + @classmethod + def setUpClass(cls): + # Importing the router initializes the CLI version. Keep that lookup offline. + release = Mock(json=lambda: {"tag_name": "v1.4.7"}) + with patch("httpx.get", return_value=release): + from src.router.antigravity.openai import router + from src.utils import authenticate_bearer + + cls.app = FastAPI() + cls.app.include_router(router) + cls.app.dependency_overrides[authenticate_bearer] = lambda: "test" + + def setUp(self): + self.backend = AsyncMock( + return_value=JSONResponse( + content={ + "response": { + "candidates": [ + { + "content": {"role": "model", "parts": [{"text": "OK"}]}, + "finishReason": "STOP", + } + ] + }, + } + ) + ) + + def request(self, messages, **options): + with ( + patch("src.api.antigravity.non_stream_request", new=self.backend), + patch( + "src.converter.antigravity_fix._get_model_rewrite_rules", + new=AsyncMock(return_value=None), + ), + patch( + "src.converter.antigravity_fix._get_sensitive_word_patterns", + new=AsyncMock(return_value=None), + ), + TestClient(self.app) as client, + ): + return client.post( + "/v1/chat/completions", + json={ + "model": "gemini-3.8-flash-high", + "messages": messages, + **options, + }, + ) + + def test_instructions_and_tool_arguments_reach_backend(self): + messages = [ + {"role": "system", "content": "Global rule."}, + {"role": "user", "content": "Hello."}, + {"role": "assistant", "content": "Hi."}, + {"role": "developer", "content": "Answer briefly."}, + *tool_history('{"key":"test"}'), + ] + response = self.request(messages) + self.assertEqual(response.status_code, 200) + self.assertEqual(response.json()["choices"][0]["message"]["role"], "assistant") + self.backend.assert_awaited_once() + payload = self.backend.await_args.kwargs["body"]["request"] + self.assertEqual( + payload["systemInstruction"]["parts"], + [{"text": "Global rule."}, {"text": "Answer briefly."}], + ) + self.assertTrue( + all(content["role"] in ("user", "model") for content in payload["contents"]) + ) + self.assertEqual(content_parts(payload, "functionCall")[0]["args"], {"key": "test"}) + + def test_remote_image_is_rejected_before_backend(self): + response = self.request( + [ + { + "role": "user", + "content": [ + {"type": "text", "text": "Describe this image."}, + { + "type": "image_url", + "image_url": {"url": "https://example.com/image.png"}, + }, + ], + } + ] + ) + self.assertEqual(response.status_code, 400) + self.assertEqual( + response.json(), + { + "detail": "Antigravity requires image_url content as a base64 image data URL", + }, + ) + self.backend.assert_not_awaited() + + def test_request_model_rejects_unsupported_roles_and_fields(self): + cases = [ + ([{"role": "unknown", "content": "Hello."}], {}), + ([{"role": "function", "name": "lookup", "content": "OK"}], {}), + ( + [ + { + "role": "developer", + "content": [{"type": "image_url", "image_url": {"url": "x"}}], + } + ], + {}, + ), + ([{"role": "user", "tool_calls": tool_history()[1]["tool_calls"]}], {}), + ([{"role": "assistant", "audio": {"id": "audio_example"}}], {}), + ([{"role": "assistant", "function_call": {"name": "lookup", "arguments": "{}"}}], {}), + ( + [{"role": "user", "content": "Hello."}], + { + "tools": [{"type": "custom", "function": {"name": "lookup"}}], + }, + ), + ] + for messages, options in cases: + with self.subTest(messages=messages, options=options): + self.assertEqual(self.request(messages, **options).status_code, 422) + self.backend.assert_not_awaited() + + def test_outgoing_role_guard_stops_invalid_normalized_role(self): + normalized = { + "model": "gemini-3.8-flash", + "contents": [ + {"role": "developer", "parts": [{"text": "Unexpected normalized role."}]}, + ], + } + with patch( + "src.converter.antigravity_fix.normalize_antigravity_request", + new=AsyncMock(return_value=normalized), + ): + response = self.request([{"role": "user", "content": "Hello."}]) + self.assertEqual(response.status_code, 400) + self.assertEqual( + response.json(), {"detail": "Unrepresentable content role for Antigravity"} + ) + self.backend.assert_not_awaited() + + def test_malformed_tool_choice_is_rejected_before_backend(self): + for function in ("lookup", {"name": ["lookup"]}): + with self.subTest(function=function): + response = self.request( + [{"role": "user", "content": "Hello."}], + tools=[{"type": "function", "function": {"name": "lookup"}}], + tool_choice={"type": "function", "function": function}, + ) + self.assertEqual(response.status_code, 400) + self.assertEqual( + response.json(), {"detail": "Unsupported tool_choice for Antigravity"} + ) + self.backend.assert_not_awaited() + + +if __name__ == "__main__": + unittest.main()