fix(gemini realtime): bound _tool_call_id_to_name with an LRU; exercise modality forwarding test

Two minor follow-ups from review:

* Switch _tool_call_id_to_name to a 256-entry LRU OrderedDict so a long
  session with many tool calls doesn't grow the dict without bound,
  while retried function_call_output lookups still hit for recently-seen
  call_ids.
* Fix test_gemini_realtime_transformation_session_created to wrap the
  cached session config in {"setup": ...} so the modality lookup in
  transform_session_created_event actually exercises responseModalities
  forwarding (the prior payload was silently treated as empty).
This commit is contained in:
mateo-berri 2026-05-22 22:23:37 +00:00
parent 20764dd342
commit e5ffd021f7
No known key found for this signature in database
2 changed files with 77 additions and 12 deletions

View file

@ -3,6 +3,7 @@ This file contains the transformation logic for the Gemini realtime API.
"""
import json
from collections import OrderedDict
from typing import Any, Dict, List, Optional, Union, cast
import litellm
@ -72,10 +73,16 @@ MAP_GEMINI_FIELD_TO_OPENAI_EVENT: Dict[
class GeminiRealtimeConfig(BaseRealtimeConfig):
# Cap the LRU of in-flight tool calls so long sessions with many tool
# calls don't grow the dict without bound. Sized large enough to cover
# bursts of pending tool responses; the oldest entry is evicted when a
# new call beyond the cap arrives.
_TOOL_CALL_ID_TO_NAME_MAX = 256
def __init__(self):
super().__init__()
# Store call_id → function_name mapping for tool call round-trip
self._tool_call_id_to_name: Dict[str, str] = {}
self._tool_call_id_to_name: "OrderedDict[str, str]" = OrderedDict()
def validate_environment(
self, headers: dict, model: str, api_key: Optional[str] = None
@ -439,9 +446,12 @@ class GeminiRealtimeConfig(BaseRealtimeConfig):
# Look up the function name from stored mapping. Keep the entry so a
# client SDK that retries function_call_output (or sends it twice for
# the same tool call) still produces a Gemini toolResponse with the
# required ``name`` field.
# required ``name`` field; refresh the LRU position so an active
# call_id stays warm across long sessions.
function_name = self._tool_call_id_to_name.get(call_id)
if not function_name:
if function_name:
self._tool_call_id_to_name.move_to_end(call_id)
else:
verbose_logger.warning(
f"Gemini Realtime: Function name not found for call_id={call_id}. "
"This may cause Gemini to reject the response."
@ -857,9 +867,14 @@ class GeminiRealtimeConfig(BaseRealtimeConfig):
call_id = fc.get("id", "")
name = fc.get("name", "")
# Store call_id → name mapping for round-trip
# Store call_id → name mapping for round-trip. Use an LRU so
# repeated function_call_output lookups (retries) still hit, while
# sessions with many tool calls don't grow the dict unboundedly.
if call_id and name:
self._tool_call_id_to_name[call_id] = name
self._tool_call_id_to_name.move_to_end(call_id)
while len(self._tool_call_id_to_name) > self._TOOL_CALL_ID_TO_NAME_MAX:
self._tool_call_id_to_name.popitem(last=False)
events.append(
OpenAIRealtimeFunctionCallArgumentsDone(

View file

@ -20,8 +20,10 @@ def test_gemini_realtime_transformation_session_created():
assert config is not None
session_configuration_request = {
"model": "gemini-1.5-flash",
"generationConfig": {"responseModalities": ["TEXT"]},
"setup": {
"model": "gemini-1.5-flash",
"generationConfig": {"responseModalities": ["TEXT"]},
}
}
session_configuration_request_str = json.dumps(session_configuration_request)
session_created_message = {"setupComplete": {}}
@ -45,8 +47,11 @@ def test_gemini_realtime_transformation_session_created():
},
)
print(transformed_message)
assert transformed_message["response"][0]["type"] == "session.created"
session_created = transformed_message["response"][0]
assert session_created["type"] == "session.created"
# Verify the setup-wrapped configuration reaches the modality lookup so
# the synthetic session.created reflects the cached responseModalities.
assert session_created["session"]["modalities"] == ["text"]
def test_session_created_does_not_overwrite_session_configuration_request():
@ -1050,10 +1055,6 @@ def test_gemini_follow_up_session_update_preserves_response_modalities_on_partia
def test_gemini_subsequent_session_update_preserves_automatic_activity_detection_subfields():
"""A follow-up turn_detection update that only sets ``create_response``
(mapped to ``disabled``) must not drop ``silenceDurationMs`` /
``prefixPaddingMs`` from the original ``automaticActivityDetection``
block."""
config = GeminiRealtimeConfig()
original_setup = {
@ -1087,3 +1088,52 @@ def test_gemini_subsequent_session_update_preserves_automatic_activity_detection
assert automatic_activity_detection["disabled"] is True
assert automatic_activity_detection["silenceDurationMs"] == 500
assert automatic_activity_detection["prefixPaddingMs"] == 100
def test_gemini_tool_call_id_to_name_evicts_oldest_when_capped():
"""The call_id → name LRU must evict the oldest entry once the cap is
reached so long sessions with many tool calls don't grow unboundedly,
while keeping recently-seen call_ids resolvable for retried
function_call_output messages."""
config = GeminiRealtimeConfig()
logging_obj = MagicMock()
logging_obj.litellm_trace_id = "trace_lru"
config._TOOL_CALL_ID_TO_NAME_MAX = 4
for idx in range(8):
config.transform_realtime_response(
json.dumps(
{
"toolCall": {
"functionCalls": [
{
"id": f"call_{idx}",
"name": f"fn_{idx}",
"args": {},
}
]
}
}
),
"gemini-2.5-flash",
logging_obj,
realtime_response_transform_input={
"session_configuration_request": None,
"current_output_item_id": None,
"current_response_id": None,
"current_conversation_id": None,
"current_delta_chunks": [],
"current_item_chunks": [],
"current_delta_type": None,
},
)
assert len(config._tool_call_id_to_name) == 4
# Most recent 4 retained; oldest 4 evicted.
assert list(config._tool_call_id_to_name) == [
"call_4",
"call_5",
"call_6",
"call_7",
]