Skip to content

fix: remove live event buffering in runner #6151

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
wuliang229 merged 2 commits into from
Jun 17, 2026
Merged

fix: remove live event buffering in runner #6151

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
a98a7a0
fix: remove live event buffering in runner
wuliang229 Jun 17, 2026
1de0881
Modify tests.
wuliang229 Jun 17, 2026
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
87 changes: 8 additions & 79 deletions src/google/adk/runners.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -66,10 +66,6 @@
logger = logging.getLogger('google_adk.' + __name__)


def _is_tool_call_or_response(event: Event) -> bool:
return bool(event.get_function_calls() or event.get_function_responses())


def _get_function_responses_from_content(
content: types.Content,
) -> list[types.FunctionResponse]:
Expand All @@ -80,21 +76,6 @@ def _get_function_responses_from_content(
]


def _is_transcription(event: Event) -> bool:
return (
event.input_transcription is not None
or event.output_transcription is not None
)


def _has_non_empty_transcription_text(
transcription: types.Transcription,
) -> bool:
return bool(
transcription and transcription.text and transcription.text.strip()
)


def _apply_run_config_custom_metadata(
event: Event, run_config: RunConfig | None
) -> None:
Expand Down Expand Up @@ -873,22 +854,6 @@ async def _exec_with_plugin(
yield early_exit_event
else:
# Step 2: Otherwise continue with normal execution
# Note for live/bidi:
# the transcription may arrive later than the action(function call
# event and thus function response event). In this case, the order of
# transcription and function call event will be wrong if we just
# append as it arrives. To address this, we should check if there is
# transcription going on. If there is transcription going on, we
# should hold on appending the function call event until the
# transcription is finished. The transcription in progress can be
# identified by checking if the transcription event is partial. When
# the next transcription event is not partial, it means the previous
# transcription is finished. Then if there is any buffered function
# call event, we should append them after this finished(non-partial)
# transcription event.
buffered_events: list[Event] = []
is_transcribing: bool = False

async with Aclosing(execute_fn(invocation_context)) as agen:
async for event in agen:
_apply_run_config_custom_metadata(
Expand All @@ -906,50 +871,14 @@ async def _exec_with_plugin(
)

if is_live_call:
if event.partial and _is_transcription(event):
is_transcribing = True
if is_transcribing and _is_tool_call_or_response(event):
# only buffer function call and function response event which is
# non-partial
buffered_events.append(output_event)
continue
# Note for live/bidi: for audio response, it's considered as
# non-partial event(event.partial=None)
# event.partial=False and event.partial=None are considered as
# non-partial event; event.partial=True is considered as partial
# event.
if event.partial is not True:
if _is_transcription(event) and (
_has_non_empty_transcription_text(event.input_transcription)
or _has_non_empty_transcription_text(
event.output_transcription
)
):
# transcription end signal, append buffered events
is_transcribing = False
logger.debug(
'Appending transcription finished event: %s', event
)
if self._should_append_event(event, is_live_call):
await self.session_service.append_event(
session=invocation_context.session, event=output_event
)

for buffered_event in buffered_events:
logger.debug('Appending buffered event: %s', buffered_event)
await self.session_service.append_event(
session=invocation_context.session, event=buffered_event
)
yield buffered_event # yield buffered events to caller
buffered_events = []
else:
# non-transcription event or empty transcription event, for
# example, event that stores blob reference, should be appended.
if self._should_append_event(event, is_live_call):
logger.debug('Appending non-buffered event: %s', event)
await self.session_service.append_event(
session=invocation_context.session, event=output_event
)
# Skip partial transcriptions for Live
if event.partial is not True and self._should_append_event(
event, is_live_call
):
logger.debug('Appending live event: %s', output_event)
await self.session_service.append_event(
session=invocation_context.session, event=output_event
)
else:
if event.partial is not True:
await self.session_service.append_event(
Expand Down
Loading
Loading