diff --git a/astrbot/core/provider/sources/gemini_source.py b/astrbot/core/provider/sources/gemini_source.py index f4cba79805..6d57880ad4 100644 --- a/astrbot/core/provider/sources/gemini_source.py +++ b/astrbot/core/provider/sources/gemini_source.py @@ -331,6 +331,10 @@ def append_or_extend( else: contents.append(content_cls(parts=part)) + model_name = str( + payloads.get("model", getattr(self, "model_name", "")), + ).lower() + is_gemini3_model = model_name.rsplit("/", 1)[-1].startswith("gemini-3") gemini_contents: list[types.Content] = [] for message in payloads["messages"]: role, content = message["role"], message.get("content") @@ -351,6 +355,14 @@ def append_or_extend( elif role == "assistant": parts = [] + tool_calls = message.get("tool_calls") or [] + tool_calls_have_thought_signature = any( + isinstance(tool, dict) + and isinstance(tool.get("extra_content"), dict) + and isinstance(tool["extra_content"].get("google"), dict) + and tool["extra_content"]["google"].get("thought_signature") + for tool in tool_calls + ) if isinstance(content, str): parts.append(types.Part.from_text(text=content)) elif isinstance(content, list): @@ -373,17 +385,19 @@ def append_or_extend( ) thinking_signature = None + if ( + is_gemini3_model + and tool_calls + and not tool_calls_have_thought_signature + ): + # An unsigned tool step came from another provider, so + # its opaque reasoning signature is not valid for Gemini. + thinking_signature = None + if ( not text and thinking_signature - and "tool_calls" in message - and any( - isinstance(tool, dict) - and isinstance(tool.get("extra_content"), dict) - and isinstance(tool["extra_content"].get("google"), dict) - and tool["extra_content"]["google"].get("thought_signature") - for tool in message["tool_calls"] - ) + and tool_calls_have_thought_signature ): # If the main content is empty but tool calls have thought signatures, # skip adding an empty text part to deduplicate the thinking signature in the main content and tool calls. @@ -396,8 +410,8 @@ def append_or_extend( ) ) - if "tool_calls" in message: - for tool in message["tool_calls"]: + if tool_calls: + for index, tool in enumerate(tool_calls): part = types.Part.from_function_call( name=tool["function"]["name"], args=json.loads(tool["function"]["arguments"]), @@ -405,14 +419,24 @@ def append_or_extend( # we should set thought_signature back to part if exists # for more info about thought_signature, see: # https://ai.google.dev/gemini-api/docs/thought-signatures + ts_bs64 = None if "extra_content" in tool and tool["extra_content"]: ts_bs64 = ( tool["extra_content"] .get("google", {}) .get("thought_signature") ) - if ts_bs64: - part.thought_signature = base64.b64decode(ts_bs64) + if ts_bs64: + part.thought_signature = base64.b64decode(ts_bs64) + elif ( + is_gemini3_model + and not tool_calls_have_thought_signature + and index == 0 + ): + # Cross-provider histories do not carry Gemini's opaque + # signature. Gemini 3 accepts this documented sentinel + # on the first function call in the step. + part.thought_signature = b"skip_thought_signature_validator" parts.append(part) if not parts: diff --git a/tests/test_gemini_source.py b/tests/test_gemini_source.py index 4c33de6bca..df2473e584 100644 --- a/tests/test_gemini_source.py +++ b/tests/test_gemini_source.py @@ -1,3 +1,4 @@ +import base64 from types import SimpleNamespace import httpx @@ -107,6 +108,152 @@ async def test_gemini_prepare_conversation_resolves_local_history_image(tmp_path assert image_part.inline_data.data == image_bytes +@pytest.mark.asyncio +async def test_gemini3_prepare_conversation_adds_signature_to_cross_provider_tool_calls(): + provider = ProviderGoogleGenAI.__new__(ProviderGoogleGenAI) + + contents = await provider._prepare_conversation( + { + "model": "gemini-3.1-pro-preview", + "messages": [ + {"role": "user", "content": "find the latest result"}, + { + "role": "assistant", + "content": None, + "tool_calls": [ + { + "function": { + "name": "search", + "arguments": '{"query": "result"}', + }, + "extra_content": {"openai": {"item_id": "item_1"}}, + }, + { + "function": { + "name": "read_page", + "arguments": '{"id": "page"}', + } + }, + ], + }, + ], + } + ) + + assert contents[-1].parts is not None + assert ( + contents[-1].parts[0].thought_signature == b"skip_thought_signature_validator" + ) + assert contents[-1].parts[1].thought_signature is None + + +@pytest.mark.asyncio +async def test_gemini3_prepare_conversation_drops_foreign_tool_step_signature(): + provider = ProviderGoogleGenAI.__new__(ProviderGoogleGenAI) + foreign_signature = base64.b64encode(b"foreign-provider-signature").decode("utf-8") + + contents = await provider._prepare_conversation( + { + "model": "gemini-3.1-pro-preview", + "messages": [ + {"role": "user", "content": "find the latest result"}, + { + "role": "assistant", + "content": [ + { + "type": "think", + "text": "provider-specific reasoning", + "encrypted": foreign_signature, + }, + {"type": "text", "text": "I will check."}, + ], + "tool_calls": [ + { + "function": { + "name": "search", + "arguments": '{"query": "result"}', + } + } + ], + }, + ], + } + ) + + assert contents[-1].parts is not None + assert contents[-1].parts[0].text == "I will check." + assert contents[-1].parts[0].thought_signature is None + assert ( + contents[-1].parts[1].thought_signature == b"skip_thought_signature_validator" + ) + + +@pytest.mark.asyncio +async def test_gemini3_prepare_conversation_preserves_real_tool_signature(): + provider = ProviderGoogleGenAI.__new__(ProviderGoogleGenAI) + signature = b"real-gemini-signature" + + contents = await provider._prepare_conversation( + { + "model": "gemini-3-flash-preview", + "messages": [ + {"role": "user", "content": "find the latest result"}, + { + "role": "assistant", + "content": None, + "tool_calls": [ + { + "function": { + "name": "search", + "arguments": '{"query": "result"}', + }, + "extra_content": { + "google": { + "thought_signature": base64.b64encode( + signature + ).decode("utf-8") + } + }, + } + ], + }, + ], + } + ) + + assert contents[-1].parts is not None + assert contents[-1].parts[0].thought_signature == signature + + +@pytest.mark.asyncio +async def test_gemini25_prepare_conversation_keeps_unsigned_tool_calls_unchanged(): + provider = ProviderGoogleGenAI.__new__(ProviderGoogleGenAI) + + contents = await provider._prepare_conversation( + { + "model": "gemini-2.5-pro", + "messages": [ + {"role": "user", "content": "find the latest result"}, + { + "role": "assistant", + "content": None, + "tool_calls": [ + { + "function": { + "name": "search", + "arguments": '{"query": "result"}', + } + } + ], + }, + ], + } + ) + + assert contents[-1].parts is not None + assert contents[-1].parts[0].thought_signature is None + + def test_gemini_empty_output_raises_empty_model_output_error(): llm_response = LLMResponse(role="assistant")