wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

In AWSBedrockAgentCoreProcessor use self.create_task()/self.cancel_task() instead of using asyncio directly.

Browse Source
This commit is contained in:
Paul Kompfner
2025-11-24 17:24:28 -05:00
parent 9e4ec4f7f3
commit 5907b51c7d
1 changed files with 3 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
src/pipecat/services/aws/agent_core.py
View File

@@ -178,11 +178,7 @@ class AWSAgentCoreProcessor(FrameProcessor):
"""Push a text frame, managing output response bookends.""" """Push a text frame, managing output response bookends."""
# Cancel any pending close task # Cancel any pending close task
if self._close_task and not self._close_task.done(): if self._close_task and not self._close_task.done():
self._close_task.cancel() await self.cancel_task(self._close_task)
try:
await self._close_task
except asyncio.CancelledError:
pass
# Open output response if needed # Open output response if needed
if not self._output_response_open: if not self._output_response_open:
@@ -194,7 +190,7 @@ class AWSAgentCoreProcessor(FrameProcessor):
self._last_text_frame_time = asyncio.get_event_loop().time() self._last_text_frame_time = asyncio.get_event_loop().time()
# Schedule closing the output response after timeout # Schedule closing the output response after timeout
self._close_task = asyncio.create_task(self._close_output_response_after_timeout()) self._close_task = self.create_task(self._close_output_response_after_timeout())
async def process_frame(self, frame: Frame, direction: FrameDirection): async def process_frame(self, frame: Frame, direction: FrameDirection):
"""Process incoming frames and handle LLM message frames. """Process incoming frames and handle LLM message frames.
@@ -255,11 +251,7 @@ class AWSAgentCoreProcessor(FrameProcessor):
# Final close if output response is still open after all parts processed # Final close if output response is still open after all parts processed
if self._output_response_open: if self._output_response_open:
if self._close_task and not self._close_task.done(): if self._close_task and not self._close_task.done():
self._close_task.cancel() await self.cancel_task(self._close_task)
try:
await self._close_task
except asyncio.CancelledError:
pass
self._output_response_open = False self._output_response_open = False
await self.push_frame(LLMFullResponseEndFrame()) await self.push_frame(LLMFullResponseEndFrame())
else: else: