Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 9 additions & 0 deletions src/google/adk/flows/llm_flows/base_llm_flow.py
Original file line number Diff line number Diff line change
Expand Up @@ -557,9 +557,18 @@ async def run_live(
# initial_history_in_client_content to True. This tells the Live server
# that the provided history already includes the model's past responses,
# preventing the server from generating duplicate responses for those replayed turns.
#
# ``history_config`` is only supported by the Gemini Developer API
# backend; the Vertex AI / Gemini Enterprise Agent Platform backend has
# no such field on its live setup message and rejects it. On Vertex,
# history is instead seeded by ``send_history`` below
# (``send_client_content`` with prior turns), so we skip
# ``history_config`` for that backend.
if (
llm_request.contents
and not invocation_context.live_session_resumption_handle
and isinstance(llm, Gemini)
and llm._api_backend == GoogleLLMVariant.GEMINI_API # pylint: disable=protected-access
):
if not llm_request.live_connect_config:
llm_request.live_connect_config = types.LiveConnectConfig()
Expand Down
82 changes: 72 additions & 10 deletions tests/unittests/flows/llm_flows/test_base_llm_flow.py
Original file line number Diff line number Diff line change
Expand Up @@ -1390,15 +1390,8 @@ async def mock_receive_2():


@pytest.mark.asyncio
@pytest.mark.parametrize(
'api_backend',
[
GoogleLLMVariant.GEMINI_API,
GoogleLLMVariant.VERTEX_AI,
],
)
async def test_run_live_history_config_set_for_all_backends(api_backend):
"""Test that run_live sets history_config for all backends."""
async def test_run_live_history_config_set_for_gemini_api_backend():
"""history_config is auto-set when seeding history on the Gemini API backend."""

real_model = Gemini(model='gemini-3.1-flash-live-preview')
mock_connection = mock.AsyncMock()
Expand Down Expand Up @@ -1445,7 +1438,7 @@ async def mock_receive():
Gemini,
'_api_backend',
new_callable=mock.PropertyMock,
return_value=api_backend,
return_value=GoogleLLMVariant.GEMINI_API,
):
try:
async for _ in flow.run_live(invocation_context):
Expand All @@ -1463,6 +1456,75 @@ async def mock_receive():
)


@pytest.mark.asyncio
async def test_run_live_history_config_not_set_for_vertex_backend():
"""history_config is NOT auto-set on the Vertex backend (it rejects it).

The Vertex AI / Gemini Enterprise Agent Platform live setup message has no
``history``/``history_config`` field. ADK seeds Vertex history via
``send_history`` (``send_client_content``) instead, so the auto-injection of
``history_config`` must be skipped for this backend.
"""

real_model = Gemini(model='gemini-3.1-flash-live-preview')
mock_connection = mock.AsyncMock()

class StopTestError(Exception):
pass

async def mock_receive():
yield LlmResponse(
content=types.Content(parts=[types.Part.from_text(text='hi')])
)
raise StopTestError('stop')

mock_connection.receive = mock.Mock(side_effect=mock_receive)

agent = Agent(name='test_agent', model=real_model)
invocation_context = await testing_utils.create_invocation_context(
agent=agent
)
invocation_context.live_request_queue = LiveRequestQueue()

flow = BaseLlmFlowForTesting()

with mock.patch.object(flow, '_send_to_model', new_callable=AsyncMock):

async def mock_preprocess(ctx, req):
req.contents = [
types.Content(parts=[types.Part.from_text(text='history')])
]
yield Event(id=Event.new_id(), author='test')

with mock.patch.object(
flow, '_preprocess_async', side_effect=mock_preprocess
):
with mock.patch.object(
Gemini, '_api_backend', new_callable=mock.PropertyMock
) as mock_backend:
mock_backend.return_value = GoogleLLMVariant.VERTEX_AI
with mock.patch(
'google.adk.models.google_llm.Gemini.connect'
) as mock_connect:
mock_connect.return_value.__aenter__.return_value = mock_connection

try:
async for _ in flow.run_live(invocation_context):
pass
except StopTestError:
pass

assert mock_connect.call_count == 1
called_req = mock_connect.call_args[0][0]
# history_config must NOT be auto-injected on Vertex.
assert (
called_req.live_connect_config is None
or called_req.live_connect_config.history_config is None
)
# History is still seeded via send_history (send_client_content).
mock_connection.send_history.assert_awaited_once()


@pytest.mark.asyncio
async def test_run_live_respects_explicit_initial_history_in_client_content_false():
"""Test that run_live respects explicit initial_history_in_client_content=False in RunConfig."""
Expand Down
Loading