diff --git a/src/converter/antigravity_fix.py b/src/converter/antigravity_fix.py index b5769c506..d130cfcee 100644 --- a/src/converter/antigravity_fix.py +++ b/src/converter/antigravity_fix.py @@ -12,6 +12,13 @@ # ==================== Gemini API 配置 ==================== +_BLOCKED_HERMES_IDENTITY = ( + "You are Hermes Agent, an intelligent AI assistant created by Nous Research." +) +_COMPATIBLE_HERMES_IDENTITY = ( + "Hermes Agent is an intelligent and helpful software assistant from Nous Research." +) + DEFAULT_SAFETY_SETTINGS = [ {"category": "HARM_CATEGORY_HARASSMENT", "threshold": "BLOCK_NONE"}, {"category": "HARM_CATEGORY_HATE_SPEECH", "threshold": "BLOCK_NONE"}, @@ -34,6 +41,33 @@ ] +def _rewrite_blocked_system_identity(value: Any) -> tuple[Any, bool]: + """Reword the Hermes identity fingerprint that Antigravity rejects as a 429.""" + if isinstance(value, str): + rewritten = value.replace(_BLOCKED_HERMES_IDENTITY, _COMPATIBLE_HERMES_IDENTITY) + return rewritten, rewritten != value + + if isinstance(value, list): + rewritten_items = [] + changed = False + for item in value: + rewritten_item, item_changed = _rewrite_blocked_system_identity(item) + rewritten_items.append(rewritten_item) + changed = changed or item_changed + return rewritten_items, changed + + if isinstance(value, dict): + rewritten_dict = {} + changed = False + for key, item in value.items(): + rewritten_item, item_changed = _rewrite_blocked_system_identity(item) + rewritten_dict[key] = rewritten_item + changed = changed or item_changed + return rewritten_dict, changed + + return value, False + + def _append_schema_hint(schema: Dict[str, Any], hint: str) -> None: """Move fragile validation details into description instead of sending them raw.""" if not hint: @@ -716,7 +750,19 @@ async def normalize_antigravity_request( model = result.get("model", "") generation_config = (result.get("generationConfig") or {}).copy() # 创建副本避免修改原对象 tools = result.get("tools") - system_instruction = result.get("systemInstruction") or result.get("system_instructions") + + for system_key in ("systemInstruction", "system_instructions"): + if system_key not in result: + continue + rewritten_system, identity_rewritten = _rewrite_blocked_system_identity( + result[system_key] + ) + if identity_rewritten: + result[system_key] = rewritten_system + log.warning( + "[ANTIGRAVITY] Reworded a Hermes system identity fingerprint " + "rejected by the upstream service" + ) # 记录原始请求 log.debug(f"[ANTIGRAVITY_FIX] 原始请求 - 模型: {model}, generationConfig: {generation_config}") diff --git a/tests/test_gemini_fix.py b/tests/test_gemini_fix.py index c1600ff20..6cc471f23 100644 --- a/tests/test_gemini_fix.py +++ b/tests/test_gemini_fix.py @@ -1,4 +1,7 @@ -from src.converter.gemini_fix import _ensure_empty_tool_schema_for_claude +from src.converter.antigravity_fix import ( + _ensure_empty_tool_schema_for_claude, + _rewrite_blocked_system_identity, +) def test_antigravity_claude_tools_keep_schema_in_parameters(): @@ -22,3 +25,33 @@ def test_antigravity_claude_tools_keep_schema_in_parameters(): assert declaration["parameters"]["type"] == "object" assert "parametersJsonSchema" not in declaration + + +def test_rewrites_blocked_hermes_identity_without_mutating_input(): + original = { + "parts": [ + { + "text": ( + "You are Hermes Agent, an intelligent AI assistant created by " + "Nous Research. Keep helping the user." + ) + } + ] + } + + rewritten, changed = _rewrite_blocked_system_identity(original) + + assert changed is True + assert rewritten["parts"][0]["text"].startswith( + "Hermes Agent is an intelligent and helpful software assistant from Nous Research." + ) + assert original["parts"][0]["text"].startswith("You are Hermes Agent") + + +def test_leaves_unrelated_system_identity_unchanged(): + original = {"parts": [{"text": "You are a concise assistant."}]} + + rewritten, changed = _rewrite_blocked_system_identity(original) + + assert changed is False + assert rewritten == original