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

Refactoring how we are reconnecting the STT.

Browse Source
This commit is contained in:
filipi87
2026-05-12 18:20:19 -03:00
parent 7984556692
commit 5d61763987
1 changed files with 26 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files

47
src/pipecat/services/nvidia/sagemaker/stt.py
View File

@@ -194,19 +194,31 @@ class NvidiaSageMakerSTTService(STTService):
# ── Connection management ───────────────────────────────────────────────── # ── Connection management ─────────────────────────────────────────────────
async def _open_client_session(self):
self._client = SageMakerBidiClient(
endpoint_name=self._endpoint_name,
region=self._region,
model_query_string=None,
model_invocation_path=None,
)
await self._client.start_session()
await self._send_session_config()
async def _close_client_session(self):
if self._client and self._client.is_active:
try:
await self._client.send_json({"type": "session.end"})
except Exception as e:
logger.warning(f"{self}: error sending session.end: {e}")
await self._client.close_session()
self._client = None
async def _connect(self): async def _connect(self):
logger.debug( logger.debug(
f"{self}: connecting to SageMaker bidi-stream endpoint '{self._endpoint_name}'" f"{self}: connecting to SageMaker bidi-stream endpoint '{self._endpoint_name}'"
) )
try: try:
self._client = SageMakerBidiClient( await self._open_client_session()
endpoint_name=self._endpoint_name,
region=self._region,
model_query_string=None,
model_invocation_path=None,
)
await self._client.start_session()
await self._send_session_config()
self._response_task = self.create_task(self._process_responses()) self._response_task = self.create_task(self._process_responses())
logger.debug(f"{self}: connected") logger.debug(f"{self}: connected")
await self._call_event_handler("on_connected") await self._call_event_handler("on_connected")
@@ -219,19 +231,13 @@ class NvidiaSageMakerSTTService(STTService):
if self._response_task and not self._response_task.done(): if self._response_task and not self._response_task.done():
await self.cancel_task(self._response_task) await self.cancel_task(self._response_task)
self._response_task = None self._response_task = None
await self._close_client_session()
if self._client and self._client.is_active:
logger.debug(f"{self}: disconnecting")
try:
await self._client.send_json({"type": "session.end"})
except Exception as e:
logger.warning(f"{self}: error sending session.end: {e}")
await self._client.close_session()
logger.debug(f"{self}: disconnected")
self._client = None
await self._call_event_handler("on_disconnected") await self._call_event_handler("on_disconnected")
async def _do_reconnect(self):
await self._close_client_session()
await self._open_client_session()
async def _send_session_config(self): async def _send_session_config(self):
"""Send transcription_session.update to configure audio format and params. """Send transcription_session.update to configure audio format and params.
@@ -333,8 +339,7 @@ class NvidiaSageMakerSTTService(STTService):
): ):
await self.push_error(error_msg=f"NIM ASR error: {msg}") await self.push_error(error_msg=f"NIM ASR error: {msg}")
# In case of error we need to reconnect, otherwise we are not going to receive from the STT service anymore # In case of error we need to reconnect, otherwise we are not going to receive from the STT service anymore
await self._disconnect() await self._request_reconnect()
await self._connect()
except asyncio.CancelledError: except asyncio.CancelledError:
logger.debug(f"{self}: response processor cancelled") logger.debug(f"{self}: response processor cancelled")