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

Refactoring to use keep alive sessions.

Browse Source
This commit is contained in:
Filipi Fuchter
2025-04-16 06:44:57 -03:00
parent 3ebef9346f
commit 0e4115049b
1 changed files with 13 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

24
src/pipecat/audio/turn/remote_smart_turn.py
View File

@@ -13,9 +13,7 @@ import numpy as np
import requests import requests
from loguru import logger from loguru import logger
from pipecat.audio.turn.base_turn_analyzer import ( from pipecat.audio.turn.base_turn_analyzer import BaseEndOfTurnAnalyzer
BaseEndOfTurnAnalyzer,
)
class RemoteSmartTurnAnalyzer(BaseEndOfTurnAnalyzer): class RemoteSmartTurnAnalyzer(BaseEndOfTurnAnalyzer):
@@ -27,30 +25,34 @@ class RemoteSmartTurnAnalyzer(BaseEndOfTurnAnalyzer):
logger.error("REMOTE_SMART_TURN_URL is not set.") logger.error("REMOTE_SMART_TURN_URL is not set.")
raise Exception("REMOTE_SMART_TURN_URL environment variable must be provided.") raise Exception("REMOTE_SMART_TURN_URL environment variable must be provided.")
# Use a session to reuse connections (keep-alive)
self.session = requests.Session()
self.session.headers.update({"Connection": "keep-alive"})
def _serialize_array(self, audio_array: np.ndarray) -> bytes: def _serialize_array(self, audio_array: np.ndarray) -> bytes:
"""Serializes a NumPy array into bytes using np.save."""
logger.debug("Serializing NumPy array to bytes...") logger.debug("Serializing NumPy array to bytes...")
buffer = io.BytesIO() buffer = io.BytesIO()
np.save(buffer, audio_array) # Saves in npy format np.save(buffer, audio_array)
serialized_bytes = buffer.getvalue() serialized_bytes = buffer.getvalue()
logger.debug(f"Serialized size: {len(serialized_bytes)} bytes") logger.debug(f"Serialized size: {len(serialized_bytes)} bytes")
return serialized_bytes return serialized_bytes
def _send_raw_request(self, data_bytes: bytes): def _send_raw_request(self, data_bytes: bytes):
"""Sends the bytes as the raw request body."""
headers = {"Content-Type": "application/octet-stream"} headers = {"Content-Type": "application/octet-stream"}
logger.debug( logger.debug(
f"Sending {len(data_bytes)} bytes as raw body to {self.remote_smart_turn_url}..." f"Sending {len(data_bytes)} bytes as raw body to {self.remote_smart_turn_url}..."
) )
try: try:
response = requests.post( response = self.session.post(
self.remote_smart_turn_url, data=data_bytes, headers=headers, timeout=60 self.remote_smart_turn_url,
) # Added timeout data=data_bytes,
headers=headers,
timeout=60,
)
logger.debug("\n--- Response ---") logger.debug("\n--- Response ---")
logger.debug(f"Status Code: {response.status_code}") logger.debug(f"Status Code: {response.status_code}")
# Try to logger.debug JSON if successful, otherwise logger.debug text
if response.ok: if response.ok:
try: try:
logger.debug("Response JSON:") logger.debug("Response JSON:")
@@ -62,7 +64,7 @@ class RemoteSmartTurnAnalyzer(BaseEndOfTurnAnalyzer):
else: else:
logger.debug("Response Content (Error):") logger.debug("Response Content (Error):")
logger.debug(response.text) logger.debug(response.text)
response.raise_for_status() # Raise an exception for bad status codes response.raise_for_status()
except requests.exceptions.RequestException as e: except requests.exceptions.RequestException as e:
logger.debug(f"Failed to send raw request to Daily Smart Turn: {e}") logger.debug(f"Failed to send raw request to Daily Smart Turn: {e}")