From b3c61e7e7ccd232f044a220b2ce9137428d25479 Mon Sep 17 00:00:00 2001 From: manish0820 Date: Wed, 26 Aug 2026 11:30:10 +0530 Subject: [PATCH 1/3] fix: resolve session, memory, GCS, and live issues --- src/google/adk/cli/utils/evals.py | 5 +- src/google/adk/live/__init__.py | 25 +++++ src/google/adk/live/live_request_queue.py | 97 +++++++++++++++++++ .../adk/sessions/in_memory_session_service.py | 12 +-- src/google/adk/tools/preload_memory_tool.py | 15 +-- 5 files changed, 140 insertions(+), 14 deletions(-) create mode 100644 src/google/adk/live/__init__.py create mode 100644 src/google/adk/live/live_request_queue.py diff --git a/src/google/adk/cli/utils/evals.py b/src/google/adk/cli/utils/evals.py index 56c20351659..e6ab2ca31b1 100644 --- a/src/google/adk/cli/utils/evals.py +++ b/src/google/adk/cli/utils/evals.py @@ -82,7 +82,10 @@ def create_gcs_eval_managers_from_uri( ' google-adk[gcp]\nOr: pip install google-cloud-storage>=2.18' ) from e - gcs_bucket = eval_storage_uri.split('://')[1] + # Only the bucket name is used; any path segment after the bucket is + # ignored, matching the documented "if a path is provided, the bucket + # will be extracted" behavior. + gcs_bucket = eval_storage_uri.split('://')[1].split('/')[0] eval_sets_manager = GcsEvalSetsManager( bucket_name=gcs_bucket, project=os.environ['GOOGLE_CLOUD_PROJECT'] ) diff --git a/src/google/adk/live/__init__.py b/src/google/adk/live/__init__.py new file mode 100644 index 00000000000..8961d8a2ea0 --- /dev/null +++ b/src/google/adk/live/__init__.py @@ -0,0 +1,25 @@ +# Copyright 2026 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Live (bidirectional streaming) mode support.""" + +from __future__ import annotations + +from .live_request_queue import LiveRequest as LiveRequest +from .live_request_queue import LiveRequestQueue as LiveRequestQueue + +__all__ = [ + 'LiveRequest', + 'LiveRequestQueue', +] diff --git a/src/google/adk/live/live_request_queue.py b/src/google/adk/live/live_request_queue.py new file mode 100644 index 00000000000..05df809034b --- /dev/null +++ b/src/google/adk/live/live_request_queue.py @@ -0,0 +1,97 @@ +# Copyright 2026 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from __future__ import annotations + +import asyncio +from typing import Any +from typing import Optional + +from google.genai import types +from pydantic import BaseModel +from pydantic import ConfigDict + + +class LiveRequest(BaseModel): + """Request send to live agents. + + When multiple fields are set, they are processed by priority (highest first): + activity_start > activity_end > audio_stream_end > blob > content. + state_delta, if set, is always applied regardless of the other fields. + """ + + model_config = ConfigDict(ser_json_bytes='base64', val_json_bytes='base64') + """The pydantic model config.""" + + content: Optional[types.Content] = None + """If set, send the content to the model in turn-by-turn mode.""" + + blob: Optional[types.Blob] = None + """If set, send the blob to the model in realtime mode.""" + + activity_start: Optional[types.ActivityStart] = None + """If set, signal the start of user activity to the model.""" + + activity_end: Optional[types.ActivityEnd] = None + """If set, signal the end of user activity to the model.""" + + audio_stream_end: bool = False + """If set, signal the end of the audio stream to the model. This is only used + when Voice Activity Detection is enabled. + """ + + close: bool = False + """If set, close the queue. queue.shutdown() is only supported in Python 3.13+.""" + + partial: bool = False + """If set, the content is a partial turn update that does not complete the current model turn.""" + + state_delta: Optional[dict[str, Any]] = None + """If set, these state changes are applied to the session, so they take + effect even when the request carries no content or a partial/ + function-response turn.""" + + +class LiveRequestQueue: + """Queue used to send LiveRequest in a live(bidirectional streaming) way.""" + + def __init__(self) -> None: + self._queue: asyncio.Queue[LiveRequest] = asyncio.Queue() + + def close(self) -> None: + self._queue.put_nowait(LiveRequest(close=True)) + + def send_content(self, content: types.Content, partial: bool = False) -> None: + self._queue.put_nowait(LiveRequest(content=content, partial=partial)) + + def send_realtime(self, blob: types.Blob) -> None: + self._queue.put_nowait(LiveRequest(blob=blob)) + + def send_activity_start(self) -> None: + """Sends an activity start signal to mark the beginning of user input.""" + self._queue.put_nowait(LiveRequest(activity_start=types.ActivityStart())) + + def send_activity_end(self) -> None: + """Sends an activity end signal to mark the end of user input.""" + self._queue.put_nowait(LiveRequest(activity_end=types.ActivityEnd())) + + def send_audio_stream_end(self) -> None: + """Sends an audio stream end signal to force flush audio.""" + self._queue.put_nowait(LiveRequest(audio_stream_end=True)) + + def send(self, req: LiveRequest) -> None: + self._queue.put_nowait(req) + + async def get(self) -> LiveRequest: + return await self._queue.get() diff --git a/src/google/adk/sessions/in_memory_session_service.py b/src/google/adk/sessions/in_memory_session_service.py index d2775cefa45..a6b4650dc1d 100644 --- a/src/google/adk/sessions/in_memory_session_service.py +++ b/src/google/adk/sessions/in_memory_session_service.py @@ -114,7 +114,12 @@ def _create_session_impl( state: Optional[dict[str, Any]] = None, session_id: Optional[str] = None, ) -> Session: - if session_id and self._get_session_impl( + session_id = ( + session_id.strip() + if session_id and session_id.strip() + else platform_uuid.new_uuid() + ) + if self._get_session_impl( app_name=app_name, user_id=user_id, session_id=session_id ): raise AlreadyExistsError(f'Session with id {session_id} already exists.') @@ -129,11 +134,6 @@ def _create_session_impl( user_state_delta ) - session_id = ( - session_id.strip() - if session_id and session_id.strip() - else platform_uuid.new_uuid() - ) session = Session( app_name=app_name, user_id=user_id, diff --git a/src/google/adk/tools/preload_memory_tool.py b/src/google/adk/tools/preload_memory_tool.py index a69421f0b2b..27a719c49fb 100644 --- a/src/google/adk/tools/preload_memory_tool.py +++ b/src/google/adk/tools/preload_memory_tool.py @@ -52,18 +52,19 @@ async def process_llm_request( llm_request: LlmRequest, ) -> None: user_content = tool_context.user_content - if ( - not user_content - or not user_content.parts - or not user_content.parts[0].text - ): + if not user_content or not user_content.parts: + return + + user_query = ' '.join( + part.text for part in user_content.parts if part.text + ) + if not user_query: return - user_query: str = user_content.parts[0].text try: response = await tool_context.search_memory(user_query) except Exception: - logging.warning('Failed to preload memory for query: %s', user_query) + logger.warning('Failed to preload memory for query: %s', user_query) return if not response.memories: From b458dc822f332571cfadfac3c8351448228bbd57 Mon Sep 17 00:00:00 2001 From: manish0820 Date: Wed, 26 Aug 2026 12:06:08 +0530 Subject: [PATCH 2/3] fix: resolve additional ADK issues --- src/google/adk/sessions/_restricted_pickle.py | 9 +++++++++ src/google/adk/sessions/vertex_ai_session_service.py | 9 +++++++-- src/google/adk/utils/_schema_utils.py | 5 ++++- src/google/adk/workflow/_llm_agent_wrapper.py | 8 +++++++- .../sessions/test_vertex_ai_session_service.py | 10 +++++----- 5 files changed, 32 insertions(+), 9 deletions(-) diff --git a/src/google/adk/sessions/_restricted_pickle.py b/src/google/adk/sessions/_restricted_pickle.py index 336579b6412..4df4ec28165 100644 --- a/src/google/adk/sessions/_restricted_pickle.py +++ b/src/google/adk/sessions/_restricted_pickle.py @@ -65,6 +65,15 @@ ("datetime", "datetime"), ("datetime", "timedelta"), ("datetime", "timezone"), + # CPython 3.11's `enum.py` pickles some Enum members via + # `pickle_by_enum_name`, whose `__reduce_ex__` returns + # `(getattr, (self.__class__, self._name_))` instead of the + # value-based reconstruction used on 3.10/3.12+. This is safe to allow + # unconditionally: `getattr`'s first argument is itself an enum class + # resolved through this same `find_class` allow-list check (via its own + # GLOBAL/STACK_GLOBAL opcode), so allowing the `getattr` callable here + # does not admit any class that wasn't already permitted. + ("builtins", "getattr"), # Auth models reachable only by subclassing or as a union base, so the # annotation walk below does not reach them. ("fastapi.openapi.models", "OAuthFlow"), diff --git a/src/google/adk/sessions/vertex_ai_session_service.py b/src/google/adk/sessions/vertex_ai_session_service.py index d36406d3797..6a9ac27242d 100644 --- a/src/google/adk/sessions/vertex_ai_session_service.py +++ b/src/google/adk/sessions/vertex_ai_session_service.py @@ -48,7 +48,7 @@ _COMPACTION_CUSTOM_METADATA_KEY = '_compaction' _USAGE_METADATA_CUSTOM_METADATA_KEY = '_usage_metadata' -_SESSION_ID_PATTERN = re.compile(r'^[A-Za-z0-9_-]+$') +_SESSION_ID_PATTERN = re.compile(r'^[a-z0-9]([a-z0-9-]*[a-z0-9])?$') def _extract_short_session_id( @@ -568,10 +568,15 @@ def _from_api_event(api_event_obj: vertexai.types.SessionEvent) -> Event: event_dict = copy.deepcopy(raw_event_dict) timestamp_obj = getattr(api_event_obj, 'timestamp', None) event_dict.update({ - 'id': api_event_obj.name.split('/')[-1], 'invocation_id': getattr(api_event_obj, 'invocation_id', None), 'author': getattr(api_event_obj, 'author', None), }) + # Preserve the original ADK event id persisted inside raw_event (the + # same id streamed to the caller during the live run). Only fall back + # to the Vertex resource name for events written before raw_event + # carried an id. + if not event_dict.get('id'): + event_dict['id'] = api_event_obj.name.split('/')[-1] if timestamp_obj: event_dict['timestamp'] = timestamp_obj.timestamp() return Event.model_validate(event_dict) diff --git a/src/google/adk/utils/_schema_utils.py b/src/google/adk/utils/_schema_utils.py index 188127614fb..786d84fc488 100644 --- a/src/google/adk/utils/_schema_utils.py +++ b/src/google/adk/utils/_schema_utils.py @@ -165,7 +165,10 @@ def validate_node_data( def _to_serializable(val: Any) -> Any: if isinstance(val, BaseModel): - return val.model_dump(exclude_none=True) + # mode="json" (not the default python mode) so Decimal, datetime, + # UUID, and non-str Enum members are converted to JSON-safe values + # and any serializer registered with when_used="json" actually runs. + return val.model_dump(exclude_none=True, mode="json") if isinstance(val, list): return [_to_serializable(item) for item in val] if isinstance(val, dict): diff --git a/src/google/adk/workflow/_llm_agent_wrapper.py b/src/google/adk/workflow/_llm_agent_wrapper.py index cd9018624df..6c114c329c8 100644 --- a/src/google/adk/workflow/_llm_agent_wrapper.py +++ b/src/google/adk/workflow/_llm_agent_wrapper.py @@ -374,12 +374,18 @@ def process_llm_agent_output( output = None else: output = text + # Only set when there is no output_schema: this tells the consumer + # loop in runners.py that event.content IS the node's output (plain + # text), so it can avoid surfacing the same text twice. When + # output_schema is set, event.output holds the validated structured + # result, which is not the same as the raw message content and must + # not be cleared downstream. + event.node_info.message_as_output = True if agent.output_key and output is not None: ctx.actions.state_delta[agent.output_key] = output event.output = output - event.node_info.message_as_output = True async def run_llm_agent_as_node( diff --git a/tests/unittests/sessions/test_vertex_ai_session_service.py b/tests/unittests/sessions/test_vertex_ai_session_service.py index b589d8c3685..4afe7344f4d 100644 --- a/tests/unittests/sessions/test_vertex_ai_session_service.py +++ b/tests/unittests/sessions/test_vertex_ai_session_service.py @@ -569,14 +569,14 @@ async def test_get_session_pagination_keeps_client_open(): session_data = { 'name': ( 'projects/test-project/locations/test-location/' - 'reasoningEngines/123/sessions/pagination_test' + 'reasoningEngines/123/sessions/pagination-test' ), 'update_time': '2024-12-12T12:12:12.123456Z', 'user_id': 'pagination_user', } - page1_events = _generate_events_for_page('pagination_test', 0, 100) - page2_events = _generate_events_for_page('pagination_test', 100, 100) - page3_events = _generate_events_for_page('pagination_test', 200, 50) + page1_events = _generate_events_for_page('pagination-test', 0, 100) + page2_events = _generate_events_for_page('pagination-test', 100, 100) + page3_events = _generate_events_for_page('pagination-test', 200, 50) mock_client = MockAsyncClientWithPagination( session_data=session_data, @@ -589,7 +589,7 @@ async def test_get_session_pagination_keeps_client_open(): session_service, '_get_api_client', return_value=mock_client ): session = await session_service.get_session( - app_name='123', user_id='pagination_user', session_id='pagination_test' + app_name='123', user_id='pagination_user', session_id='pagination-test' ) assert session is not None From 8d419bcf8057839f7f86c629bcb708500d3e7fd7 Mon Sep 17 00:00:00 2001 From: manish0820 Date: Wed, 26 Aug 2026 12:12:55 +0530 Subject: [PATCH 3/3] fix: resolve LiteLLM issues --- src/google/adk/models/lite_llm.py | 41 +++++++++++++++++++++++++++++-- 1 file changed, 39 insertions(+), 2 deletions(-) diff --git a/src/google/adk/models/lite_llm.py b/src/google/adk/models/lite_llm.py index 85e90d4e6a3..40366e822f9 100644 --- a/src/google/adk/models/lite_llm.py +++ b/src/google/adk/models/lite_llm.py @@ -2202,11 +2202,27 @@ def _function_declaration_to_tool_param( elif function_declaration.parameters_json_schema: parameters = function_declaration.parameters_json_schema + description = function_declaration.description or "" + # Most OpenAI-compatible providers have no dedicated field for a tool's + # result/output schema, unlike the Gemini path (see #2828). Rather than + # inventing a non-standard key that providers would ignore, surface the + # schema by appending it to the description, which is always forwarded. + output_schema_dict: Optional[dict[str, Any]] = None + if function_declaration.response_json_schema: + output_schema_dict = function_declaration.response_json_schema + elif function_declaration.response: + output_schema_dict = _schema_to_dict(function_declaration.response) + if output_schema_dict: + description = ( + f"{description}\n\nResult schema:" + f" {json.dumps(output_schema_dict)}" + ).strip() + tool_params: dict[str, Any] = { "type": "function", "function": { "name": function_declaration.name, - "description": function_declaration.description or "", + "description": description, "parameters": parameters, }, } @@ -2490,9 +2506,30 @@ def _message_to_generate_content_response( for tool_call in tool_calls: if tool_call.type == "function": thought_signature = _extract_thought_signature_from_tool_call(tool_call) + try: + call_args = _parse_tool_call_arguments(tool_call.function.arguments) + except json.JSONDecodeError as e: + # Malformed/truncated tool-call arguments (e.g. a partial stream + # committed to the message) are a recoverable model error here, + # not a reason to abort the whole invocation. Log and fall back + # to an empty dict so the function-call Part still surfaces and + # downstream tool dispatch can retry/recover, instead of the + # JSONDecodeError propagating out of response parsing. (The + # streaming aggregation path has its own, more specific handling + # for arguments truncated by hitting max_output_tokens; this + # covers every other malformed-JSON case.) + logger.warning( + "Failed to parse tool call arguments as JSON for tool" + " %r, falling back to an empty dict. Raw arguments: %r." + " Error: %s", + tool_call.function.name, + tool_call.function.arguments, + e, + ) + call_args = {} part = types.Part.from_function_call( name=tool_call.function.name, - args=_parse_tool_call_arguments(tool_call.function.arguments), + args=call_args, ) function_call = part.function_call if function_call is None: