From cc9901a82f9d108c21b5d9172a498acf09e3c890 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Tue, 22 Apr 2025 17:14:19 -0400 Subject: [PATCH] Replace httpx with aiohttp --- examples/foundational/38-smart-turn.py | 124 +++++++++++++------------ src/pipecat/audio/turn/smart_turn.py | 58 ++++++------ 2 files changed, 92 insertions(+), 90 deletions(-) diff --git a/examples/foundational/38-smart-turn.py b/examples/foundational/38-smart-turn.py index 6bace018b..929a22e68 100644 --- a/examples/foundational/38-smart-turn.py +++ b/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__": diff --git a/src/pipecat/audio/turn/smart_turn.py b/src/pipecat/audio/turn/smart_turn.py index b540fb187..8ac165943 100644 --- a/src/pipecat/audio/turn/smart_turn.py +++ b/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.")