(feat): Improve bidi event loop by mehtarac · Pull Request #10 · mehtarac/sdk-python

mehtarac · 2025-10-21T15:23:43Z

Bidirectional Agent Loop Implementation

This implementation represents a significant architectural simplification of the bidirectional streaming event loop, reducing complexity from 3 background tasks to 1 while maintaining all original functionality and performance benefits.

The refactoring eliminates complex supervisor pattern in favor of standard Python error handling, removes coordinator function layers, and provides tool execution concurrent operations.

What has changed?

1. Simplified Background Task Architecture

Before:

# 3 separate background tasks with complex coordination
session = BidirectionalConnection(model_session=model_session, agent=agent)
session.background_tasks = [
    asyncio.create_task(_process_model_events(session)),
    asyncio.create_task(_process_tool_execution(session)),
]
session.main_cycle_task = asyncio.create_task(bidirectional_event_loop_cycle(session))

# Queue-based tool execution with polling delays
async def _process_tool_execution(session: BidirectionalConnection) -> None:
    while session.active:
        tool_use = await asyncio.wait_for(session.tool_queue.get(), timeout=0.5)
        # 0-500ms polling delay before tool execution

After:

# Single background task with integrated error handling
agent_loop = BidirectionalAgentLoop(model_session=model_session, agent=agent)
self.background_tasks = [
    asyncio.create_task(self._process_model_events())
]

# Tool execution with concurrent task creation
def schedule_tool_execution(self, tool_use: ToolUse) -> None:
    task = asyncio.create_task(self._execute_tool(tool_use))  # 0ms delay
    self.pending_tool_tasks[task_id] = task

2. Eliminated Complex Supervision Pattern

Before:

# Event-driven supervision with asyncio.wait monitoring
async def _supervise_session(self) -> None:
    while self.active and tasks_to_supervise:
        done, pending = await asyncio.wait(
            tasks_to_supervise,
            return_when=asyncio.FIRST_COMPLETED,
            timeout=1.0
        )
        # Complex task failure detection logic

After:

# Built-in error handling in main processor
async def _process_model_events(self) -> None:
    try:
        async for provider_event in self.model_session.receive_events():
            # Event processing logic...
    except Exception as e:
        logger.error("Agent loop processor failed: %s", e)
        self.active = False  # Trigger shutdown

3. Removed Coordinator Function Layer

Before:

# Agent uses coordinator functions as intermediary layer
from ..event_loop.bidirectional_event_loop import start_bidirectional_connection, stop_bidirectional_connection

async def start(self):
    self._session = await start_bidirectional_connection(self)

async def end(self):
    await stop_bidirectional_connection(self._session)

After:

# Agent creates and manages BidirectionalAgentLoop directly
from ..event_loop.bidirectional_event_loop import BidirectionalAgentLoop

async def start(self):
    model_session = await self.model.create_bidirectional_connection(...)
    self._session = BidirectionalAgentLoop(model_session=model_session, agent=self)
    await self._session.start()

async def end(self):
    await self._session.stop()

4. Enhanced Tool Execution

Before:

# Tools could be cancelled during interruptions
async def _handle_interruption(session: BidirectionalConnection) -> None:
    # Cancel all pending tool tasks regardless of execution state
    for task_id, task in list(session.pending_tool_tasks.items()):
        if not task.done():
            task.cancel()

After:

# Tools are protected from interruption once started
async def handle_interruption(self) -> None:
    active_tool_tasks = [task for task in self.pending_tool_tasks.values() if not task.done()]
    if active_tool_tasks:
        logger.debug("Tools are protected - %d tools currently executing", len(active_tool_tasks))
        # Clear audio but don't cancel running tools
    else:
        logger.debug("No active tools - full interruption handling")
    # Always clear audio queues for responsive interruption

Performance Improvements

Tool execution latency: Reduced from 0-500ms (queue polling) to 0ms (immediate scheduling)
Background tasks: Reduced from 3 tasks to 1 task (67% reduction)
Architecture: Simplified from multi-function coordination to single class

Breaking Changes

None. All changes maintain full backward compatibility:

Same public interface for bidirectional agent methods
Same functionality and behavior

Related Issues

None

mehtarac · 2025-10-21T16:11:21Z

+        # Synchronization primitives
        self.interrupted = False
        self.interruption_lock = asyncio.Lock()
+        self.conversation_lock = asyncio.Lock()  # Race condition prevention


This lock is used when adding to conversation history. Using a lock here to ensure the conversation history is not corrupted or overwritten. For example if multiple tools complete at the same time, then a lock will help ensure that all the results are written to the conversation history

Does order of writing tool results matter? I guess tools will execute as they appear, and the model will continue to stream its response.

Technically speaking we shouldn't need a lock for the following reasons:

async tasks run one at a time but in a cooperative manner. But because they run one at a time, their is no concern of resource conflicts when operating in memory.

The messages array is first updated in memory, which can only be altered by one task at a time.

The session managers operate in hooks, which run synchronously. Assuming hooks could run asynchronously, I would say it is up to them to place in locks.

mehtarac · 2025-10-21T16:13:50Z

+        # Thread-safe counter increment
+        current_tool_number = self.tool_count + 1
+        self.tool_count = current_tool_number
+        print(f"\nTool #{current_tool_number}: {tool_name}")


similar to PrintingCallbackHandler in sdk-python/src/strands/handlers/callback_handler.py

Given that we are getting closer to release, we should not have any prints in the code

mehtarac · 2025-10-21T16:17:29Z

+                            break
+
+                # Remove completed tasks from supervision list
+                tasks_to_supervise = [task for task in tasks_to_supervise if not task.done()]


In this case the only task being supervised will be the _process_model_events task in normal connection/session and thus tasks_to_supervise list remains unchanged - no tasks removed. However, The removal logic exists as a safety net in case the model events task unexpectedly completes (network failure) and we need graceful shutdown.

mehtarac · 2025-10-21T16:18:42Z

+    return event_loop


+async def stop_bidirectional_connection(event_loop: "BidirectionalEventLoop") -> None:


just a wrapper that calls event_loop.stop(). Mainly to preserve existing api of agent.start_bidirectional_connection and agent.stop_bidirectional_connection in the agent class.

Unshure · 2025-10-21T16:29:10Z

+        # Synchronization primitives
        self.interrupted = False
        self.interruption_lock = asyncio.Lock()
+        self.conversation_lock = asyncio.Lock()  # Race condition prevention


Does order of writing tool results matter? I guess tools will execute as they appear, and the model will continue to stream its response.

JackYPCOnline · 2025-10-21T16:31:04Z

-while maintaining a simple interface for agent interaction.
-
-Features:
- Concurrent task management for model events and tool execution


_process_tool_execution I saw this function is removed but mentioned in doc.

this was just to compare the old architecture

JackYPCOnline · 2025-10-21T16:34:03Z

+                    tool_use_id = tool_result.get("toolUseId")
+                    await self.model_session.send_tool_result(tool_use_id, tool_result)
+                    logger.debug("Tool result sent: %s", tool_use_id)
+                elif isinstance(tool_event, ToolStreamEvent):


Just curious how we deal with ToolStreamEvent

should we also add these to the output queue? We want our customers to develop nice UI around this

mkmeral · 2025-11-03T13:19:40Z

-    "start_bidirectional_connection",
-    "stop_bidirectional_connection",
-    "bidirectional_event_loop_cycle",
+    "BidirectionalAgentLoop",


do we actually want to export this? Is there any use case where customers would like to access agentic loop directly?

mkmeral · 2025-11-03T13:24:05Z

-    logger.debug("Session cleanup starting")
-    session.active = False
+        # Cancel pending tool tasks
+        for task in self.pending_tool_tasks.values():


nit: instead of duplicating cancellation, you can move all_tasks = list(self.pending_tool_tasks.values()) + self.background_tasks to top, and just do task loop cancellation once.

mkmeral · 2025-11-03T13:24:38Z

-        self.tool_queue = asyncio.Queue()
-        self.audio_output_queue = asyncio.Queue()
+        # Task tracking
+        self.background_tasks: List[asyncio.Task] = []


given that there is a single background task, do we need it to be a list?

mkmeral · 2025-11-03T13:25:27Z

+        # Thread-safe counter increment
+        current_tool_number = self.tool_count + 1
+        self.tool_count = current_tool_number
+        print(f"\nTool #{current_tool_number}: {tool_name}")


Given that we are getting closer to release, we should not have any prints in the code

mkmeral · 2025-11-03T13:30:56Z

-async def bidirectional_event_loop_cycle(session: BidirectionalConnection) -> None:
-    """Main event loop coordinator that runs continuously during the session.
+        # Create task with UUID tracking
+        task_id = str(uuid.uuid4())


why not just use tool use id?

mkmeral · 2025-11-03T13:33:08Z

-        self.audio_output_queue = asyncio.Queue()
+        # Task tracking
+        self.background_tasks: List[asyncio.Task] = []
+        self.pending_tool_tasks: Dict[str, asyncio.Task] = {}


nit: pending tool tasks to me sounds like tool uses that we still haven't started executing. From the code, it looks like currently running tool tasks right?

mkmeral · 2025-11-03T13:36:03Z

-            logger.debug("Interruption already in progress")
-            return
+            if active_tool_tasks:
+                logger.debug("Tools are protected - %d tools currently executing", len(active_tool_tasks))


nit: warn or info?

mkmeral · 2025-11-03T13:38:01Z

-            logger.debug("Tool tasks cancelled: %d", cancelled_tools)
+            # Clear output queue
+            cleared_count = 0
+            while True:


nit: use queue.empty()

mkmeral · 2025-11-03T13:45:55Z

+                if not isinstance(provider_event, dict):
+                    continue
+
+                strands_event = provider_event


mkmeral · 2025-11-03T13:47:37Z

+                    tool_use_id = tool_result.get("toolUseId")
+                    await self.model_session.send_tool_result(tool_use_id, tool_result)
+                    logger.debug("Tool result sent: %s", tool_use_id)
+                elif isinstance(tool_event, ToolStreamEvent):


should we also add these to the output queue? We want our customers to develop nice UI around this

(feat): Improve bidi event loop

7dff976

mehtarac had a problem deploying to auto-approve October 21, 2025 15:24 — with GitHub Actions Failure

mehtarac commented Oct 21, 2025

View reviewed changes

Comment thread src/strands/experimental/bidirectional_streaming/event_loop/bidirectional_event_loop.py Outdated

mehtarac commented Oct 21, 2025

View reviewed changes

Comment thread src/strands/experimental/bidirectional_streaming/event_loop/bidirectional_event_loop.py Outdated

mehtarac commented Oct 21, 2025

View reviewed changes

JackYPCOnline reviewed Oct 21, 2025

View reviewed changes

Comment thread src/strands/experimental/bidirectional_streaming/event_loop/bidirectional_event_loop.py

JackYPCOnline reviewed Oct 21, 2025

View reviewed changes

Comment thread src/strands/experimental/bidirectional_streaming/event_loop/bidirectional_event_loop.py

mehtarac self-assigned this Oct 21, 2025

Unshure reviewed Oct 21, 2025

View reviewed changes

JackYPCOnline reviewed Oct 21, 2025

View reviewed changes

pgrayy reviewed Oct 21, 2025

View reviewed changes

Comment thread src/strands/experimental/bidirectional_streaming/event_loop/bidirectional_event_loop.py Outdated

pgrayy reviewed Oct 21, 2025

View reviewed changes

Comment thread src/strands/experimental/bidirectional_streaming/event_loop/bidirectional_event_loop.py Outdated

JackYPCOnline reviewed Oct 21, 2025

View reviewed changes

pgrayy reviewed Oct 21, 2025

View reviewed changes

Comment thread src/strands/experimental/bidirectional_streaming/event_loop/bidirectional_event_loop.py Outdated

pgrayy reviewed Oct 21, 2025

View reviewed changes

Comment thread src/strands/experimental/bidirectional_streaming/event_loop/bidirectional_event_loop.py Outdated

pgrayy reviewed Oct 21, 2025

View reviewed changes

Comment thread src/strands/experimental/bidirectional_streaming/event_loop/bidirectional_event_loop.py Outdated

(feat): Improve bidi event loop

1944d0d

mehtarac had a problem deploying to auto-approve October 28, 2025 13:07 — with GitHub Actions Failure

(feat): Improve bidi event loop

288d9fe

mehtarac had a problem deploying to auto-approve October 28, 2025 13:23 — with GitHub Actions Failure

(feat): Improve bidi event loop

11d7ad7

mehtarac had a problem deploying to auto-approve October 28, 2025 13:23 — with GitHub Actions Failure

(feat): Improve bidi event loop

70edc6a

mehtarac had a problem deploying to auto-approve October 28, 2025 13:34 — with GitHub Actions Failure

mkmeral reviewed Nov 3, 2025

View reviewed changes

Merge branch 'main' into bar_raise_eventloop

48430be

github-actions Bot added the size/l label Nov 7, 2025

mehtarac had a problem deploying to auto-approve November 7, 2025 14:31 — with GitHub Actions Failure

mehtarac mentioned this pull request Nov 10, 2025

feat: (Agent): Finalize Bidirectional Agent class #12

Merged

mehtarac closed this Nov 19, 2025

		return event_loop


		async def stop_bidirectional_connection(event_loop: "BidirectionalEventLoop") -> None:

Conversation

mehtarac commented Oct 21, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Bidirectional Agent Loop Implementation

What has changed?

1. Simplified Background Task Architecture

2. Eliminated Complex Supervision Pattern

3. Removed Coordinator Function Layer

4. Enhanced Tool Execution

Performance Improvements

Breaking Changes

Related Issues

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants

mehtarac commented Oct 21, 2025 •

edited

Loading