Replace httpx with aiohttp

examples/foundational/38-smart-turn.py

@@ -6,6 +6,7 @@
 
 import os
 
+import aiohttp
 from dotenv import load_dotenv
 from loguru import logger
 
@@ -31,78 +32,79 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection):
 
     remote_smart_turn_url = os.getenv("REMOTE_SMART_TURN_URL")
 
-    transport = SmallWebRTCTransport(
-        webrtc_connection=webrtc_connection,
-        params=TransportParams(
-            audio_in_enabled=True,
-            audio_out_enabled=True,
-            vad_enabled=True,
-            vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.2)),
-            vad_audio_passthrough=True,
-            turn_analyzer=SmartTurnAnalyzer(url=remote_smart_turn_url),
-        ),
-    )
+    async with aiohttp.ClientSession() as session:
+        transport = SmallWebRTCTransport(
+            webrtc_connection=webrtc_connection,
+            params=TransportParams(
+                audio_in_enabled=True,
+                audio_out_enabled=True,
+                vad_enabled=True,
+                vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.2)),
+                vad_audio_passthrough=True,
+                turn_analyzer=SmartTurnAnalyzer(url=remote_smart_turn_url, aiohttp_session=session),
+            ),
+        )
 
-    stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))
+        stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))
 
-    tts = CartesiaTTSService(
-        api_key=os.getenv("CARTESIA_API_KEY"),
-        voice_id="71a7ad14-091c-4e8e-a314-022ece01c121",  # British Reading Lady
-    )
+        tts = CartesiaTTSService(
+            api_key=os.getenv("CARTESIA_API_KEY"),
+            voice_id="71a7ad14-091c-4e8e-a314-022ece01c121",  # British Reading Lady
+        )
 
-    llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"))
+        llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"))
 
-    messages = [
-        {
-            "role": "system",
-            "content": "You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio so don't include special characters in your answers. Respond to what the user said in a creative and helpful way.",
-        },
-    ]
-
-    context = OpenAILLMContext(messages)
-    context_aggregator = llm.create_context_aggregator(context)
-
-    pipeline = Pipeline(
-        [
-            transport.input(),  # Transport user input
-            stt,
-            context_aggregator.user(),  # User responses
-            llm,  # LLM
-            tts,  # TTS
-            transport.output(),  # Transport bot output
-            context_aggregator.assistant(),  # Assistant spoken responses
+        messages = [
+            {
+                "role": "system",
+                "content": "You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio so don't include special characters in your answers. Respond to what the user said in a creative and helpful way.",
+            },
         ]
-    )
 
-    task = PipelineTask(
-        pipeline,
-        params=PipelineParams(
-            allow_interruptions=True,
-            enable_metrics=True,
-            enable_usage_metrics=True,
-            report_only_initial_ttfb=True,
-        ),
-    )
+        context = OpenAILLMContext(messages)
+        context_aggregator = llm.create_context_aggregator(context)
 
-    @transport.event_handler("on_client_connected")
-    async def on_client_connected(transport, client):
-        logger.info(f"Client connected")
-        # Kick off the conversation.
-        messages.append({"role": "system", "content": "Please introduce yourself to the user."})
-        await task.queue_frames([context_aggregator.user().get_context_frame()])
+        pipeline = Pipeline(
+            [
+                transport.input(),  # Transport user input
+                stt,
+                context_aggregator.user(),  # User responses
+                llm,  # LLM
+                tts,  # TTS
+                transport.output(),  # Transport bot output
+                context_aggregator.assistant(),  # Assistant spoken responses
+            ]
+        )
 
-    @transport.event_handler("on_client_disconnected")
-    async def on_client_disconnected(transport, client):
-        logger.info(f"Client disconnected")
+        task = PipelineTask(
+            pipeline,
+            params=PipelineParams(
+                allow_interruptions=True,
+                enable_metrics=True,
+                enable_usage_metrics=True,
+                report_only_initial_ttfb=True,
+            ),
+        )
 
-    @transport.event_handler("on_client_closed")
-    async def on_client_closed(transport, client):
-        logger.info(f"Client closed connection")
-        await task.cancel()
+        @transport.event_handler("on_client_connected")
+        async def on_client_connected(transport, client):
+            logger.info(f"Client connected")
+            # Kick off the conversation.
+            messages.append({"role": "system", "content": "Please introduce yourself to the user."})
+            await task.queue_frames([context_aggregator.user().get_context_frame()])
 
-    runner = PipelineRunner(handle_sigint=False)
+        @transport.event_handler("on_client_disconnected")
+        async def on_client_disconnected(transport, client):
+            logger.info(f"Client disconnected")
 
-    await runner.run(task)
+        @transport.event_handler("on_client_closed")
+        async def on_client_closed(transport, client):
+            logger.info(f"Client closed connection")
+            await task.cancel()
+
+        runner = PipelineRunner(handle_sigint=False)
+
+        await runner.run(task)
 
 
 if __name__ == "__main__":

src/pipecat/audio/turn/smart_turn.py

@@ -5,10 +5,11 @@
 #
 
 
+import asyncio
 import io
 from typing import Dict
 
-import httpx
+import aiohttp
 import numpy as np
 from loguru import logger
 
@@ -16,19 +17,15 @@ from pipecat.audio.turn.base_smart_turn import BaseSmartTurn, SmartTurnTimeoutEx
 
 
 class SmartTurnAnalyzer(BaseSmartTurn):
-    def __init__(self, url: str, **kwargs):
+    def __init__(self, url: str, aiohttp_session: aiohttp.ClientSession, **kwargs):
         super().__init__(**kwargs)
         self.remote_smart_turn_url = url
+        self._aiohttp_session = aiohttp_session
 
         if not self.remote_smart_turn_url:
             logger.error("remote_smart_turn_url is not set.")
             raise Exception("remote_smart_turn_url must be provided.")
 
-        self.client = httpx.AsyncClient(
-            headers={"Connection": "keep-alive"},
-            timeout=httpx.Timeout(self._params.stop_secs),
-        )
-
     def _serialize_array(self, audio_array: np.ndarray) -> bytes:
         logger.trace("Serializing NumPy array to bytes...")
         buffer = io.BytesIO()
@@ -43,33 +40,36 @@ class SmartTurnAnalyzer(BaseSmartTurn):
             f"Sending {len(data_bytes)} bytes as raw body to {self.remote_smart_turn_url}..."
         )
         try:
-            response = await self.client.post(
-                self.remote_smart_turn_url,
-                content=data_bytes,
-                headers=headers,
-            )
+            timeout = aiohttp.ClientTimeout(total=self._params.stop_secs)
 
-            logger.trace("\n--- Response ---")
-            logger.trace(f"Status Code: {response.status_code}")
+            async with self._aiohttp_session.post(
+                self.remote_smart_turn_url, data=data_bytes, headers=headers, timeout=timeout
+            ) as response:
+                logger.trace("\n--- Response ---")
+                logger.trace(f"Status Code: {response.status}")
 
-            if response.is_success:
-                try:
-                    json_data = response.json()
-                    logger.trace("Response JSON:")
-                    logger.trace(json_data)
-                    return json_data
-                except httpx.DecodingError:
-                    logger.trace("Response Content (non-JSON):")
-                    logger.trace(response.text)
-            else:
-                logger.trace("Response Content (Error):")
-                logger.trace(response.text)
-                response.raise_for_status()
+                if response.status == 200:
+                    try:
+                        json_data = await response.json()
+                        logger.trace("Response JSON:")
+                        logger.trace(json_data)
+                        return json_data
+                    except aiohttp.ContentTypeError:
+                        # Non-JSON response
+                        text = await response.text()
+                        logger.trace("Response Content (non-JSON):")
+                        logger.trace(text)
+                        raise Exception(f"Non-JSON response: {text}")
+                else:
+                    error_text = await response.text()
+                    logger.trace("Response Content (Error):")
+                    logger.trace(error_text)
+                    response.raise_for_status()
 
-        except httpx.TimeoutException:
+        except asyncio.TimeoutError:
             logger.error(f"Request timed out after {self._params.stop_secs} seconds")
             raise SmartTurnTimeoutException(f"Request exceeded {self._params.stop_secs} seconds.")
-        except httpx.RequestError as e:
+        except aiohttp.ClientError as e:
             logger.error(f"Failed to send raw request to Daily Smart Turn: {e}")
             raise Exception("Failed to send raw request to Daily Smart Turn.")