Files
Aether/_deprecated_py_src/services/provider/adapters/antigravity/envelope.py

1119 lines
42 KiB
Python
Raw Normal View History

"""Antigravity v1internal request/response envelope helpers.
Antigravity reuses the `gemini:chat` endpoint signature but wraps the actual
wire format:
- Request: V1InternalRequest (top-level metadata + nested GeminiRequest)
- Response: V1InternalResponse (top-level responseId + nested GeminiResponse)
对齐 Antigravity-Manager wrapper.rs 的处理逻辑
- Claude model tool ID 注入request + response
- Thinking budget capping (Auto cap 24576)
- [undefined] 字符串深度清理
- parametersJsonSchema parameters 重命名
- JSON Schema 禁止字段清洗
- Antigravity System Instruction 注入
- Signature 错误检测
- Model alias mappingpreview physical
- Google Search (grounding) 注入
- thoughtSignature 注入到 functionCall parts
- Image generation config 注入aspectRatio / imageSize
"""
from __future__ import annotations
import uuid
from typing import Any
from src.services.provider.adapters.antigravity.constants import (
ANTIGRAVITY_SYSTEM_INSTRUCTION,
ASPECT_RATIO_TABLE,
IMAGE_ASPECT_RATIO_SUFFIXES,
IMAGE_GEN_UPSTREAM_MODEL,
MODEL_MAX_OUTPUT_LIMIT,
NETWORKING_TOOL_KEYWORDS,
OUTPUT_OVERHEAD,
OUTPUT_OVERHEAD_IMAGE,
)
from src.services.provider.adapters.antigravity.constants import (
REQUEST_USER_AGENT as ANTIGRAVITY_REQUEST_USER_AGENT,
)
from src.services.provider.adapters.antigravity.constants import (
SIGNATURE_ERROR_KEYWORDS,
STANDARD_ASPECT_RATIOS,
THINKING_BUDGET_AUTO_CAP,
THINKING_BUDGET_DEFAULT_INJECT,
THINKING_MODELS_AUTO_INJECT_KEYWORDS,
WEB_SEARCH_MODEL,
)
from src.services.provider.adapters.antigravity.url_availability import url_availability
from src.services.provider.request_context import get_selected_base_url
# ---------------------------------------------------------------------------
# Key normalization: snake_case → camelCase
# ---------------------------------------------------------------------------
# The Gemini normalizer (gemini.py) outputs snake_case keys that mirror protobuf
# field names (e.g. "generation_config", "function_declarations"). However, the
# Antigravity v1internal JSON protocol (like the REST Gemini API) uses camelCase
# for all field names. A mismatch causes downstream processing functions in this
# module to silently skip keys or create duplicate entries.
#
# We normalise once at the entry point of wrap_v1internal_request so that every
# subsequent helper can safely assume camelCase.
_TOP_LEVEL_KEY_RENAMES: dict[str, str] = {
"system_instruction": "systemInstruction",
"generation_config": "generationConfig",
"tool_config": "toolConfig",
# safety_settings 在 wrap_v1internal_request 入口处已 pop无需映射
}
_GENERATION_CONFIG_KEY_RENAMES: dict[str, str] = {
"max_output_tokens": "maxOutputTokens",
"stop_sequences": "stopSequences",
"top_p": "topP",
"top_k": "topK",
"thinking_config": "thinkingConfig",
"response_modalities": "responseModalities",
"response_mime_type": "responseMimeType",
}
def _normalize_to_camel_case(body: dict[str, Any]) -> None:
"""In-place normalise known Gemini snake_case keys to their camelCase form.
This must be called **before** any other processing so that all helpers
in this module can consistently use camelCase lookups.
"""
# 1. Top-level keys
for snake, camel in _TOP_LEVEL_KEY_RENAMES.items():
if snake in body and camel not in body:
body[camel] = body.pop(snake)
# 2. Inside tools: function_declarations → functionDeclarations
tools = body.get("tools")
if isinstance(tools, list):
for tool in tools:
if isinstance(tool, dict):
if "function_declarations" in tool and "functionDeclarations" not in tool:
tool["functionDeclarations"] = tool.pop("function_declarations")
# 3. Inside generationConfig: normalise sub-keys
gc = body.get("generationConfig")
if isinstance(gc, dict):
for snake, camel in _GENERATION_CONFIG_KEY_RENAMES.items():
if snake in gc and camel not in gc:
gc[camel] = gc.pop(snake)
# 4. Inside contents → parts: function_call/function_response → camelCase
contents = body.get("contents")
if isinstance(contents, list):
for content in contents:
if not isinstance(content, dict):
continue
parts = content.get("parts")
if not isinstance(parts, list):
continue
for part in parts:
if not isinstance(part, dict):
continue
if "function_call" in part and "functionCall" not in part:
part["functionCall"] = part.pop("function_call")
if "function_response" in part and "functionResponse" not in part:
part["functionResponse"] = part.pop("function_response")
# ---------------------------------------------------------------------------
# Request body 预处理工具函数(对齐 AM wrapper.rs / common_utils.rs
# ---------------------------------------------------------------------------
def _deep_clean_undefined(obj: Any) -> None:
"""In-place 递归清理 '[undefined]' 字符串值。
Cherry Studio 等客户端会在请求中注入 '[undefined]' 字符串
可能导致上游 API 解析错误
"""
if isinstance(obj, dict):
keys_to_remove = [k for k, v in obj.items() if v == "[undefined]"]
for k in keys_to_remove:
del obj[k]
for v in obj.values():
if isinstance(v, (dict, list)):
_deep_clean_undefined(v)
elif isinstance(obj, list):
i = 0
while i < len(obj):
if obj[i] == "[undefined]":
obj.pop(i)
else:
if isinstance(obj[i], (dict, list)):
_deep_clean_undefined(obj[i])
i += 1
def _inject_claude_tool_ids_request(inner_request: dict[str, Any], model: str) -> None:
"""为 Claude 模型注入 functionCall/functionResponse 的 id 字段。
Google v1internal 在目标模型为 Claude 时要求 functionCall 带有 id 字段
但标准 Gemini 协议不包含此字段对齐 AM wrapper.rs #1522。
算法
1. 第一遍扫描所有 functionCall为没有 id 的生成 id并记录 (name, id) 队列
2. 第二遍扫描所有 functionResponse从对应 name 的队列中取出 id 使用
"""
if "claude" not in model.lower():
return
contents = inner_request.get("contents")
if not isinstance(contents, list):
return
# 第一遍:收集所有 functionCall 的 ID按 name 分组,保持顺序)
# name -> [id1, id2, ...] 每个调用的 ID 按出现顺序排列
call_ids_by_name: dict[str, list[str]] = {}
# 所有 call ID 按原始出现顺序(用于 fallback 匹配)
all_call_ids_ordered: list[str] = []
# 用于生成新 ID 的计数器(全局,确保唯一性)
name_counters: dict[str, int] = {}
for content in contents:
if not isinstance(content, dict):
continue
parts = content.get("parts")
if not isinstance(parts, list):
continue
for part in parts:
if not isinstance(part, dict):
continue
fc = part.get("functionCall") or part.get("function_call")
if isinstance(fc, dict):
name = fc.get("name", "")
if not isinstance(name, str) or not name:
name = "unknown"
fc_id = fc.get("id")
if not (isinstance(fc_id, str) and fc_id):
# 生成新 ID
count = name_counters.get(name, 0)
fc_id = f"call_{name}_{count}"
fc["id"] = fc_id
name_counters[name] = count + 1
# 记录这个 call 的 ID供后续 response 使用
if name not in call_ids_by_name:
call_ids_by_name[name] = []
call_ids_by_name[name].append(fc_id)
# 同时按原始出现顺序记录(用于 fallback
all_call_ids_ordered.append(fc_id)
# 第二遍:为 functionResponse 分配匹配的 ID
# 使用索引追踪每个 name 已消费到第几个 ID
response_index_by_name: dict[str, int] = {}
# 已使用的 call ID 集合
used_call_ids: set[str] = set()
# fallback 用的有序索引
fallback_index = 0
for content in contents:
if not isinstance(content, dict):
continue
parts = content.get("parts")
if not isinstance(parts, list):
continue
for part in parts:
if not isinstance(part, dict):
continue
fr = part.get("functionResponse") or part.get("function_response")
if isinstance(fr, dict) and not (isinstance(fr.get("id"), str) and fr.get("id")):
name = fr.get("name", "")
if not isinstance(name, str) or not name:
name = "unknown"
assigned_id: str | None = None
# 优先:从对应 name 的 call ID 队列中取出下一个 ID
call_ids = call_ids_by_name.get(name, [])
idx = response_index_by_name.get(name, 0)
while idx < len(call_ids):
cid = call_ids[idx]
idx += 1
if cid not in used_call_ids:
assigned_id = cid
used_call_ids.add(cid)
break
response_index_by_name[name] = idx
# Fallback如果 name 是 unknown原本为空尝试使用下一个未使用的 call ID按原始出现顺序
if assigned_id is None and name == "unknown":
while fallback_index < len(all_call_ids_ordered):
cid = all_call_ids_ordered[fallback_index]
fallback_index += 1
if cid not in used_call_ids:
assigned_id = cid
used_call_ids.add(cid)
break
if assigned_id is not None:
fr["id"] = assigned_id
else:
# 没有匹配的 call ID异常情况生成一个新 ID
count = name_counters.get(name, 0)
fr["id"] = f"call_{name}_{count}"
name_counters[name] = count + 1
def _process_thinking_budget(inner_request: dict[str, Any], model: str) -> None:
"""处理 Thinking Budget自动注入 + Auto Cap + maxOutputTokens 约束。
对齐 AM wrapper.rs + CLIProxyAPI 混合方案
- flash/pro/thinking 模型处理 thinkingConfig
- 自动注入 thinkingConfig对已知需要 thinking 的模型
- Auto Capbudget 超过 24576 时裁剪
- Claude 要求maxOutputTokens 必须 > thinkingBudget
- 优先增加 maxOutputTokens超限时才减少 budget
"""
lower_model = model.lower()
if not any(kw in lower_model for kw in ("flash", "pro", "thinking")):
return
# 确保 generationConfig 存在
gen_config = inner_request.setdefault("generationConfig", {})
if not isinstance(gen_config, dict):
return
# 自动注入 thinkingConfig对已知需要 thinking 的模型)
if gen_config.get("thinkingConfig") is None:
should_inject = any(kw in lower_model for kw in THINKING_MODELS_AUTO_INJECT_KEYWORDS)
if should_inject:
gen_config["thinkingConfig"] = {
"includeThoughts": True,
"thinkingBudget": THINKING_BUDGET_DEFAULT_INJECT,
}
# Auto Cap
thinking_config = gen_config.get("thinkingConfig")
if not isinstance(thinking_config, dict):
return
budget = thinking_config.get("thinkingBudget")
if not isinstance(budget, int) or budget <= 0:
return
# 1. 限制 budget 上限
if budget > THINKING_BUDGET_AUTO_CAP:
budget = THINKING_BUDGET_AUTO_CAP
thinking_config["thinkingBudget"] = budget
# 2. 确保 maxOutputTokens > thinkingBudget
current_max = gen_config.get("maxOutputTokens")
# 根据模型类型选择增量(对齐 Antigravity-Manager
overhead = OUTPUT_OVERHEAD_IMAGE if "-image" in lower_model else OUTPUT_OVERHEAD
# 计算理想的 maxOutputTokens
ideal_max = budget + overhead
# 3. 确保不超过模型限制
if ideal_max > MODEL_MAX_OUTPUT_LIMIT:
# 超限时:减少 budget 而不是超限(对齐 CLIProxyAPI 策略)
ideal_max = MODEL_MAX_OUTPUT_LIMIT
# 确保 budget < ideal_max保留 overhead 空间给输出
max_budget = ideal_max - overhead
if budget > max_budget:
thinking_config["thinkingBudget"] = max_budget
# 4. 应用修正
if current_max is None or (isinstance(current_max, int) and current_max <= budget):
gen_config["maxOutputTokens"] = ideal_max
def _clean_tool_declarations(inner_request: dict[str, Any]) -> None:
"""清洗工具声明:重命名字段 + 移除禁止的 Schema 字段 + 过滤搜索声明。
对齐 AM wrapper.rs
- parametersJsonSchema parametersGemini CLI 兼容
- 移除 Gemini 不支持的 Schema 字段multipleOf
- 过滤 web_search / google_search 工具声明
"""
tools = inner_request.get("tools")
if not isinstance(tools, list):
return
for tool in tools:
if not isinstance(tool, dict):
continue
# 支持 camelCase + snake_case
decls_key = (
"functionDeclarations"
if "functionDeclarations" in tool
else "function_declarations" if "function_declarations" in tool else None
)
if decls_key is None:
continue
decls = tool.get(decls_key)
if not isinstance(decls, list):
continue
# 1. 过滤搜索关键字函数(对齐 NETWORKING_TOOL_KEYWORDS
decls[:] = [
d
for d in decls
if not (
isinstance(d, dict)
and isinstance(d.get("name"), str)
and d["name"] in NETWORKING_TOOL_KEYWORDS
)
]
# 2. 重命名 + 清洗
for decl in decls:
if not isinstance(decl, dict):
continue
# parametersJsonSchema → parameters
if "parametersJsonSchema" in decl:
params = decl.pop("parametersJsonSchema")
if isinstance(params, dict):
_clean_json_schema(params)
decl["parameters"] = params
elif "parameters" in decl:
params = decl["parameters"]
if isinstance(params, dict):
_clean_json_schema(params)
def _clean_json_schema(schema: dict[str, Any]) -> None:
"""递归移除 Gemini 不支持的 JSON Schema 字段。"""
from src.core.api_format.schema_utils import clean_gemini_schema
clean_gemini_schema(schema)
# ---------------------------------------------------------------------------
# Google Search (Grounding) 检测与注入(对齐 AM common_utils.rs
# ---------------------------------------------------------------------------
def _detect_networking_tools(inner_request: dict[str, Any]) -> bool:
"""检测请求中是否包含联网/搜索工具声明。
对齐 AM common_utils.rs detects_networking_tool支持多种声明风格
1. Claude/Anthropic 直发风格: {"name": "web_search"} / {"type": "web_search_20250305"}
2. OpenAI 嵌套风格: {"type": "function", "function": {"name": "web_search"}}
3. Gemini 原生风格: {"functionDeclarations": [{"name": "web_search"}]}
4. Gemini googleSearch 声明: {"googleSearch": {}}
"""
tools = inner_request.get("tools")
if not isinstance(tools, list):
return False
for tool in tools:
if not isinstance(tool, dict):
continue
# 1. 直发风格: name / type 字段
name = tool.get("name")
if isinstance(name, str) and name in NETWORKING_TOOL_KEYWORDS:
return True
type_ = tool.get("type")
if isinstance(type_, str) and type_ in NETWORKING_TOOL_KEYWORDS:
return True
# 2. OpenAI 嵌套风格
func = tool.get("function")
if isinstance(func, dict):
fn_name = func.get("name")
if isinstance(fn_name, str) and fn_name in NETWORKING_TOOL_KEYWORDS:
return True
# 3. Gemini functionDeclarations 风格(支持 camelCase + snake_case
decls = tool.get("functionDeclarations")
if decls is None:
decls = tool.get("function_declarations")
if isinstance(decls, list):
for decl in decls:
if isinstance(decl, dict):
decl_name = decl.get("name")
if isinstance(decl_name, str) and decl_name in NETWORKING_TOOL_KEYWORDS:
return True
# 4. Gemini googleSearch / googleSearchRetrieval 声明
if tool.get("googleSearch") is not None or tool.get("googleSearchRetrieval") is not None:
return True
return False
def _detect_online_suffix(model: str) -> bool:
"""检测模型名是否含 -online 后缀(联网意图)。"""
return model.rstrip().endswith("-online")
def _inject_google_search_tool(inner_request: dict[str, Any]) -> None:
"""注入 googleSearch tool 到请求中。
对齐 AM common_utils.rs inject_google_search_tool
- 如果已有 functionDeclarations跳过v1internal 不支持混用
- 先清理已有的 googleSearch / googleSearchRetrieval
- 注入 {"googleSearch": {}}
"""
tools = inner_request.setdefault("tools", [])
if not isinstance(tools, list):
inner_request["tools"] = [{"googleSearch": {}}]
return
# 如果已有 functionDeclarations不注入v1internal 不支持混用 search 和 functions
has_functions = any(
isinstance(t, dict) and ("functionDeclarations" in t or "function_declarations" in t)
for t in tools
)
if has_functions:
return
# 清理已存在的 googleSearch / googleSearchRetrieval避免重复
tools[:] = [
t
for t in tools
if not (isinstance(t, dict) and ("googleSearch" in t or "googleSearchRetrieval" in t))
]
# 注入
tools.append({"googleSearch": {}})
# ---------------------------------------------------------------------------
# thoughtSignature 注入到 functionCall parts对齐 AM wrapper.rs
# ---------------------------------------------------------------------------
def _inject_thought_signatures(inner_request: dict[str, Any], session_id: str | None) -> None:
"""Inject thought signatures into functionCall parts.
Prefer tool-specific signatures (tool_use_id -> thoughtSignature), then fall back
to a session-level signature when available.
"""
try:
from src.core.api_format.conversion.thinking_cache import signature_cache
except Exception:
return
session_sig: str | None = None
if session_id:
session_sig = signature_cache.get_session_signature(session_id)
contents = inner_request.get("contents")
if not isinstance(contents, list):
return
for content in contents:
if not isinstance(content, dict):
continue
parts = content.get("parts")
if not isinstance(parts, list):
continue
for part in parts:
if not isinstance(part, dict):
continue
fc = part.get("functionCall") or part.get("function_call")
if not isinstance(fc, dict):
continue
# Normalize existing signature aliases to thoughtSignature (if present).
sig_val: str | None = None
ts = part.get("thoughtSignature")
if isinstance(ts, str) and ts.strip():
sig_val = ts.strip()
if ts != sig_val:
part["thoughtSignature"] = sig_val
else:
tss = part.get("thought_signature")
if isinstance(tss, str) and tss.strip():
sig_val = tss.strip()
part["thoughtSignature"] = sig_val
part.pop("thought_signature", None)
else:
legacy = part.get("signature")
if isinstance(legacy, str) and legacy.strip():
sig_val = legacy.strip()
part["thoughtSignature"] = sig_val
part.pop("signature", None)
if sig_val:
continue
tool_id = fc.get("id")
tool_sig: str | None = None
if isinstance(tool_id, str) and tool_id:
tool_sig = signature_cache.get_tool_signature(tool_id)
chosen = tool_sig or session_sig
if not chosen:
continue
part["thoughtSignature"] = chosen
# Avoid sending duplicate aliases once we inject.
part.pop("thought_signature", None)
if part.get("signature") in (None, ""):
part.pop("signature", None)
# ---------------------------------------------------------------------------
# Image Generation Config对齐 AM common_utils.rs parse_image_config_with_params
# ---------------------------------------------------------------------------
def _calculate_aspect_ratio(size: str) -> str:
""""WIDTHxHEIGHT""W:H" 字符串计算宽高比。
对齐 AM common_utils.rs calculate_aspect_ratio_from_size
1. 先检查是否已是标准比例字符串 ( "16:9")
2. 解析 WIDTHxHEIGHT 并容差匹配
3. 默认返回 "1:1"
"""
if size in STANDARD_ASPECT_RATIOS:
return size
if "x" in size:
try:
w_str, h_str = size.split("x", 1)
width, height = float(w_str), float(h_str)
if width > 0 and height > 0:
ratio = width / height
for target_ratio, label in ASPECT_RATIO_TABLE:
if abs(ratio - target_ratio) < 0.05:
return label
except (ValueError, ZeroDivisionError):
pass
return "1:1"
def _parse_image_config(
model: str,
inner_request: dict[str, Any],
) -> tuple[dict[str, Any], str]:
"""解析图像生成配置,返回 (imageConfig, clean_model_name)。
对齐 AM common_utils.rs parse_image_config_with_params + resolve_request_config
1. 从请求体中提取 OpenAI 风格 size / quality 参数优先
2. 回退到模型后缀解析 ( -16x9, -4k)
3. 合并请求体中的 generationConfig.imageConfig如果存在
4. 上游模型固定为 "gemini-3-pro-image"
"""
# 提取 OpenAI 风格参数(可能由跨格式转换层注入到请求根部)
size = inner_request.pop("size", None)
quality = inner_request.pop("quality", None)
if not isinstance(size, str):
size = None
if not isinstance(quality, str):
quality = None
# --- 解析 aspectRatio ---
aspect_ratio = "1:1"
if size:
aspect_ratio = _calculate_aspect_ratio(size)
else:
lower_model = model.lower()
for suffix, ratio in IMAGE_ASPECT_RATIO_SUFFIXES.items():
if suffix in lower_model:
aspect_ratio = ratio
break
config: dict[str, Any] = {"aspectRatio": aspect_ratio}
# --- 解析 imageSize ---
if quality:
q_lower = quality.lower()
if q_lower in ("hd", "4k"):
config["imageSize"] = "4K"
elif q_lower in ("medium", "2k"):
config["imageSize"] = "2K"
elif q_lower in ("standard", "1k"):
config["imageSize"] = "1K"
else:
lower_model = model.lower()
if "-4k" in lower_model or "-hd" in lower_model:
config["imageSize"] = "4K"
elif "-2k" in lower_model:
config["imageSize"] = "2K"
# --- 合并请求体中已有的 imageConfigbody 可以覆盖除 imageSize 降级外的字段) ---
gen_config = inner_request.get("generationConfig")
if isinstance(gen_config, dict):
body_image_config = gen_config.get("imageConfig")
if isinstance(body_image_config, dict):
for key, value in body_image_config.items():
# 防止 body 降级 inferred imageSize对齐 AM 的 shield 逻辑)
if (
key == "imageSize"
and (value == "1K" or value is None)
and "imageSize" in config
):
continue
config[key] = value
return config, IMAGE_GEN_UPSTREAM_MODEL
def _apply_image_gen_config(inner_request: dict[str, Any], image_config: dict[str, Any]) -> None:
"""将 imageConfig 应用到请求的 generationConfig 中。
对齐 AM wrapper.rs 的图像生成处理
- 移除 tools / systemInstruction
- 确保 contents 中每个 content role 字段
- 清理 generationConfig 中与图像生成冲突的字段
- 注入 imageConfig
- 处理图像思维模式默认 disabled
"""
# 移除不兼容字段_normalize_to_camel_case 已统一 key仅需 camelCase
for key in ("tools", "toolConfig", "systemInstruction"):
inner_request.pop(key, None)
# 确保 contents 中每个 content 有 role 字段
contents = inner_request.get("contents")
if isinstance(contents, list):
for content in contents:
if isinstance(content, dict) and "role" not in content:
content["role"] = "user"
# 清理 generationConfig
gen_config = inner_request.setdefault("generationConfig", {})
if not isinstance(gen_config, dict):
gen_config = {}
inner_request["generationConfig"] = gen_config
# 移除与图像生成冲突的字段_normalize_to_camel_case 已统一 key
for key in ("responseMimeType", "responseModalities"):
gen_config.pop(key, None)
# 注入 imageConfig
gen_config["imageConfig"] = image_config
# 图像思维模式:默认 disabled对齐 AM wrapper.rs image_thinking_mode
gen_config["thinkingConfig"] = {"includeThoughts": False}
def _compact_contents(inner_request: dict[str, Any]) -> None:
"""Strip invalid parts, drop empty contents, merge consecutive same-role.
Delegates to :func:`compact_gemini_contents` from the Gemini normalizer to
avoid duplicating the validation logic.
This function modifies *inner_request* in-place.
"""
from src.core.api_format.conversion.normalizers.gemini import compact_gemini_contents
contents = inner_request.get("contents")
if not isinstance(contents, list):
return
result = compact_gemini_contents(contents)
inner_request["contents"] = result
def _inject_system_instruction(inner_request: dict[str, Any]) -> None:
"""注入 Antigravity 身份系统指令。
对齐 AM wrapper.rs
- 如果已有 systemInstruction在前面插入避免重复
- 如果没有创建新的
- 补全 role: userGemini API 要求
注意Gemini API 使用 protobuf oneof`system_instruction`snake_case
`systemInstruction`camelCase是同一个字段只能设置其中一个
需要先统一到 camelCase 再处理
"""
# 统一 snake_case 到 camelCase避免 oneof 冲突)
if "system_instruction" in inner_request:
snake_value = inner_request.pop("system_instruction")
# 仅当 camelCase 不存在时才迁移
if "systemInstruction" not in inner_request:
inner_request["systemInstruction"] = snake_value
system_instruction = inner_request.get("systemInstruction")
if isinstance(system_instruction, dict):
# 补全 role
if "role" not in system_instruction:
system_instruction["role"] = "user"
parts = system_instruction.get("parts")
if isinstance(parts, list):
# 检查是否已包含 Antigravity 身份(避免重复注入)
has_antigravity = False
if parts and isinstance(parts[0], dict):
text = parts[0].get("text", "")
if isinstance(text, str) and "You are Antigravity" in text:
has_antigravity = True
if not has_antigravity:
parts.insert(0, {"text": ANTIGRAVITY_SYSTEM_INSTRUCTION})
else:
# 没有 systemInstruction创建新的
inner_request["systemInstruction"] = {
"role": "user",
"parts": [{"text": ANTIGRAVITY_SYSTEM_INSTRUCTION}],
}
# ---------------------------------------------------------------------------
# Response 后处理工具函数
# ---------------------------------------------------------------------------
def _inject_claude_tool_ids_response(response: dict[str, Any], model: str) -> None:
"""为 Claude 模型的响应注入 functionCall 的 id 字段。
对齐 AM wrapper.rs inject_ids_to_response
让下游客户端 OpenCode/Vercel AI SDK能感知 tool call ID
并在下一轮对话中原样带回
"""
if "claude" not in model.lower():
return
candidates = response.get("candidates")
if not isinstance(candidates, list):
return
for candidate in candidates:
if not isinstance(candidate, dict):
continue
content = candidate.get("content")
if not isinstance(content, dict):
continue
parts = content.get("parts")
if not isinstance(parts, list):
continue
name_counters: dict[str, int] = {}
for part in parts:
if not isinstance(part, dict):
continue
fc = part.get("functionCall") or part.get("function_call")
if isinstance(fc, dict) and not (isinstance(fc.get("id"), str) and fc.get("id")):
name = fc.get("name", "unknown")
if not isinstance(name, str):
name = "unknown"
count = name_counters.get(name, 0)
fc["id"] = f"call_{name}_{count}"
name_counters[name] = count + 1
# ---------------------------------------------------------------------------
# Core wrap / unwrap 函数
# ---------------------------------------------------------------------------
def _generate_stable_session_id(inner_request: dict[str, Any]) -> str:
"""生成稳定的 sessionId对齐 CLIProxyAPI generateStableSessionID
基于请求内容的哈希生成确定性 sessionId相同内容的请求会得到相同的 sessionId
有助于上游维持会话上下文
"""
import hashlib
import json
try:
# 使用 contents 字段生成稳定哈希(与 CLIProxyAPI 对齐)
contents = inner_request.get("contents")
if contents:
raw = json.dumps(contents, sort_keys=True, separators=(",", ":"))
else:
raw = json.dumps(inner_request, sort_keys=True, separators=(",", ":"))
digest = hashlib.sha256(raw.encode()).hexdigest()[:32]
return f"session-{digest}"
except Exception:
return f"session-{uuid.uuid4().hex[:32]}"
def wrap_v1internal_request(
gemini_request: dict[str, Any],
*,
project_id: str,
model: str,
request_type: str = "agent",
) -> dict[str, Any]:
"""Wrap a GeminiRequest into Antigravity V1InternalRequest.
处理流程对齐 AM wrapper.rs + common_utils.rs
1. 移除 model / safetySettings
2. 深度清理 [undefined] 字符串
3. 模型别名映射preview physical
4. 联网检测 + -online 后缀检测
5. 图像生成检测 + imageConfig 解析
6. Claude model tool ID 注入
7. thoughtSignature 注入到 functionCall parts
8. Thinking budget 处理
9. 工具声明清洗
10. Google Search 注入联网请求
11. System Instruction 注入
12. 清理空 parts contents 并合并连续同角色条目
13. 注入 sessionId
14. 构建 v1internal 信封
"""
from src.core.video_utils import is_image_gen_model
inner_request = dict(gemini_request)
inner_request.pop("model", None)
inner_request.pop("safetySettings", None)
inner_request.pop("safety_settings", None)
# 0. 统一 snake_case → camelCaseGemini normalizer 输出 snake_case
# v1internal 以及本模块所有 helper 均使用 camelCase
_normalize_to_camel_case(inner_request)
# 1. 深度清理 [undefined]
_deep_clean_undefined(inner_request)
# 2. 剥离 -online 后缀(联网意图由 tools 检测,不依赖后缀传递到上游)
has_online_suffix = _detect_online_suffix(model)
final_model = model.rstrip().removesuffix("-online")
# 3. 联网检测(对齐 AM-online 后缀或客户端声明了联网工具)
has_networking = has_online_suffix or _detect_networking_tools(inner_request)
# 4. 图像生成检测 + imageConfig 解析
is_image_gen = is_image_gen_model(final_model)
if is_image_gen:
# 解析 imageConfig 并确定上游模型名
image_config, final_model = _parse_image_config(final_model, inner_request)
_apply_image_gen_config(inner_request, image_config)
request_type = "image_gen"
# 图像生成不需要联网
has_networking = False
else:
# 5. Claude tool ID 注入
_inject_claude_tool_ids_request(inner_request, final_model)
# 6. thoughtSignature 注入到 functionCall parts对齐 AM wrapper.rs
session_id = inner_request.get("sessionId")
if not isinstance(session_id, str):
# 提前生成 sessionId 用于 signature 查找
session_id = _generate_stable_session_id(inner_request)
inner_request["sessionId"] = session_id
_inject_thought_signatures(inner_request, session_id)
# 7. Thinking budget 处理(图像生成和普通请求都需要)
_process_thinking_budget(inner_request, final_model)
if not is_image_gen:
# 8. 工具声明清洗
_clean_tool_declarations(inner_request)
# 9. Google Search 注入(对齐 AM common_utils.rs
if has_networking:
# 仅 gemini-2.5-flash 支持 googleSearch对齐 AM其他模型降级到 2.5-flash
if final_model != WEB_SEARCH_MODEL:
final_model = WEB_SEARCH_MODEL
_inject_google_search_tool(inner_request)
request_type = "web_search"
# 10. System Instruction 注入
_inject_system_instruction(inner_request)
# 11. 清理空 parts 的 contents 并合并连续同角色条目
# 跨格式转换(如 Responses API reasoning 块)可能产生空 parts 的 content
# Gemini API 要求每个 content 至少有一个有效 part并且严格交替 user/model 角色。
_compact_contents(inner_request)
# 12. 注入 sessionId如果还没有的话
if "sessionId" not in inner_request:
inner_request["sessionId"] = _generate_stable_session_id(inner_request)
return {
"project": project_id,
"requestId": f"agent-{uuid.uuid4()}",
"request": inner_request,
"model": final_model,
"userAgent": ANTIGRAVITY_REQUEST_USER_AGENT,
"requestType": request_type,
}
def unwrap_v1internal_response(response: dict[str, Any]) -> dict[str, Any]:
"""Unwrap Antigravity V1InternalResponse into a GeminiResponse-like dict."""
inner = response.get("response")
if isinstance(inner, dict):
unwrapped = dict(inner)
resp_id = response.get("responseId")
if resp_id is not None:
unwrapped["_v1internal_response_id"] = resp_id
return unwrapped
return response
def cache_thought_signatures(model: str, response: dict[str, Any]) -> None:
"""Best-effort cache for Antigravity thought signatures.
同时缓存到 legacy (text) 层和 tool (Layer 1)
"""
try:
from src.core.api_format.conversion.thinking_cache import signature_cache
except Exception:
return
try:
candidates = response.get("candidates")
if not isinstance(candidates, list):
return
for cand in candidates:
if not isinstance(cand, dict):
continue
content = cand.get("content")
if not isinstance(content, dict):
continue
parts = content.get("parts")
if not isinstance(parts, list):
continue
for part in parts:
if not isinstance(part, dict):
continue
# 缓存 thinking signaturelegacy text 层)
text = part.get("text")
sig = (
part.get("thoughtSignature")
or part.get("thought_signature")
or part.get("signature")
)
if isinstance(text, str) and text and isinstance(sig, str) and sig:
signature_cache.cache(model, text, sig)
# 缓存 tool call signatureLayer 1
fc = part.get("functionCall") or part.get("function_call")
if isinstance(fc, dict) and isinstance(sig, str) and sig:
tool_id = fc.get("id")
if isinstance(tool_id, str) and tool_id:
signature_cache.cache_tool_signature(tool_id, sig)
except Exception:
# Never fail request path due to cache issues.
return
def is_signature_error(status_code: int, error_body: str) -> bool:
"""检测 400 错误是否为 thinking signature 相关错误。
对齐 AM handlers/common.rs用于判断是否应该移除 thinking 配置后重试
"""
if status_code != 400:
return False
return any(kw in error_body for kw in SIGNATURE_ERROR_KEYWORDS)
# ---------------------------------------------------------------------------
# Envelope 类Provider Hook 接口)
# ---------------------------------------------------------------------------
class AntigravityV1InternalEnvelope:
"""Provider envelope hooks for Antigravity v1internal wrapper."""
name = "antigravity:v1internal"
def extra_headers(self) -> dict[str, str] | None:
from src.services.provider.adapters.antigravity.constants import (
get_v1internal_extra_headers,
)
return get_v1internal_extra_headers()
def wrap_request(
self,
request_body: dict[str, Any],
*,
model: str,
url_model: str | None,
decrypted_auth_config: dict[str, Any] | None,
) -> tuple[dict[str, Any], str | None]:
from src.core.logger import logger as _envelope_logger
project_id = (decrypted_auth_config or {}).get("project_id")
if not isinstance(project_id, str) or not project_id:
from src.core.exceptions import ProviderNotAvailableException
raise ProviderNotAvailableException(
"Antigravity OAuth 配置缺少 project_id请重新授权",
provider_name="antigravity",
upstream_response="missing auth_config.project_id",
)
wrapped = wrap_v1internal_request(
request_body,
project_id=project_id,
model=model,
)
# Debug: 打印 v1internal 请求的关键字段(不打印完整 body 避免日志过大)
_envelope_logger.debug(
"[Antigravity Envelope] model={}, project_id={}, requestType={}, "
"userAgent={}, has_sessionId={}, has_systemInstruction={}, "
"has_contents={}, has_generationConfig={}",
wrapped.get("model"),
str(wrapped.get("project", ""))[:8] + "...",
wrapped.get("requestType"),
wrapped.get("userAgent"),
"sessionId" in (wrapped.get("request") or {}),
"systemInstruction" in (wrapped.get("request") or {}),
"contents" in (wrapped.get("request") or {}),
"generationConfig" in (wrapped.get("request") or {}),
)
# Antigravity's model lives in the request body, not the URL path.
return wrapped, None
def unwrap_response(self, data: Any) -> Any:
if isinstance(data, dict):
return unwrap_v1internal_response(data)
return data
def postprocess_unwrapped_response(self, *, model: str, data: Any) -> None:
if isinstance(data, dict):
# Claude model: 注入 tool call ID对齐 AM wrapper.rs inject_ids_to_response
_inject_claude_tool_ids_response(data, model)
cache_thought_signatures(model, data)
def capture_selected_base_url(self) -> str | None:
return get_selected_base_url()
def on_http_status(self, *, base_url: str | None, status_code: int) -> None:
if not base_url:
return
if 200 <= status_code < 300:
url_availability.mark_success(base_url)
elif status_code in (404, 408, 429) or 500 <= status_code < 600:
url_availability.mark_unavailable(base_url)
def on_connection_error(self, *, base_url: str | None, exc: Exception) -> None: # noqa: ARG002
if not base_url:
return
url_availability.mark_unavailable(base_url)
def force_stream_rewrite(self) -> bool:
# Streaming must be rewritten even when endpoint signature matches, because
# Antigravity wraps chunks in v1internal envelope.
return True
antigravity_v1internal_envelope = AntigravityV1InternalEnvelope()
__all__ = [
"AntigravityV1InternalEnvelope",
"antigravity_v1internal_envelope",
"cache_thought_signatures",
"is_signature_error",
"unwrap_v1internal_response",
"wrap_v1internal_request",
]