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

Merge pull request #14 from daily-co/aiosessions

Browse Source 
Don't create aiohttp sessions inside services
This commit is contained in:
Moishe Lettvin
2024-01-26 14:01:24 -05:00
committed by GitHub
parent 8baf137511 e81f247845
commit d4c4d49035
13 changed files with 401 additions and 408 deletions
Download Patch File Download Diff File Expand all files Collapse all files

61
src/dailyai/services/azure_ai_services.py
View File

@@ -101,28 +101,30 @@ class AzureImageGenServiceREST(ImageGenService):
def __init__(
self,
image_size: str,
aiohttp_session:aiohttp.ClientSession,
api_key=None,
azure_endpoint=None,
api_version=None,
model=None):
super().__init__(image_size=image_size)
self.api_key = api_key or os.getenv("AZURE_DALLE_KEY")
self.azure_endpoint = azure_endpoint or os.getenv("AZURE_DALLE_ENDPOINT")
self.api_version = api_version or "2023-06-01-preview"
self.model = model or os.getenv("AZURE_DALLE_DEPLOYMENT_ID")
self._api_key = api_key or os.getenv("AZURE_DALLE_KEY")
self._azure_endpoint = azure_endpoint or os.getenv("AZURE_DALLE_ENDPOINT")
self._api_version = api_version or "2023-06-01-preview"
self._model = model or os.getenv("AZURE_DALLE_DEPLOYMENT_ID")
self._aiohttp_session = aiohttp_session
async def run_image_gen(self, sentence) -> tuple[str, bytes]:
# TODO hoist the session to app-level
async with aiohttp.ClientSession() as session:
url = f"{self.azure_endpoint}openai/images/generations:submit?api-version={self.api_version}"
headers = {"api-key": self.api_key, "Content-Type": "application/json"}
url = f"{self._azure_endpoint}openai/images/generations:submit?api-version={self._api_version}"
headers = {"api-key": self._api_key, "Content-Type": "application/json"}
body = {
# Enter your prompt text here
"prompt": sentence,
"size": self.image_size,
"n": 1,
}
async with session.post(url, headers=headers, json=body) as submission:
async with self._aiohttp_session.post(
url, headers=headers, json=body
) as submission:
operation_location = submission.headers['operation-location']
status = ""
@@ -134,7 +136,9 @@ class AzureImageGenServiceREST(ImageGenService):
raise Exception("Image generation timed out")
await asyncio.sleep(1)
response = await session.get(operation_location, headers=headers)
response = await self._aiohttp_session.get(
operation_location, headers=headers
)
json_response = await response.json()
status = json_response["status"]
@@ -143,42 +147,7 @@ class AzureImageGenServiceREST(ImageGenService):
raise Exception("Image generation failed")
# Load the image from the url
async with session.get(image_url) as response:
async with self._aiohttp_session.get(image_url) as response:
image_stream = io.BytesIO(await response.content.read())
image = Image.open(image_stream)
return (image_url, image.tobytes())
class AzureImageGenService(ImageGenService):
def __init__(self, api_key=None, azure_endpoint=None, api_version=None, model=None):
super().__init__()
api_key = api_key or os.getenv("AZURE_DALLE_KEY")
azure_endpoint = azure_endpoint or os.getenv("AZURE_DALLE_ENDPOINT")
api_version = api_version or "2023-06-01-preview"
self.model = model or os.getenv("AZURE_DALLE_DEPLOYMENT_ID")
self.client = AzureOpenAI(
api_key=api_key,
azure_endpoint=azure_endpoint,
api_version=api_version,
)
async def run_image_gen(self, sentence) -> tuple[str, bytes]:
self.logger.info("Generating azure image", sentence)
image = self.client.images.generate(
model=self.model,
prompt=sentence,
n=1,
size=self.image_size,
)
url = image["data"][0]["url"]
response = requests.get(url)
dalle_stream = io.BytesIO(response.content)
dalle_im = Image.open(dalle_stream.tobytes())
return (url, dalle_im)

4
src/dailyai/services/daily_transport_service.py
View File

@@ -15,7 +15,6 @@ from dailyai.queue_frame import (
ImageQueueFrame,
QueueFrame,
StartStreamQueueFrame,
TextQueueFrame,
TranscriptionQueueFrame,
)
@@ -413,3 +412,6 @@ class DailyTransportService(EventHandler):
raise e
b = bytearray()
except Exception as e:
print("!!!!", e)
raise e

14
src/dailyai/services/deepgram_ai_services.py
View File

@@ -9,11 +9,12 @@ from dailyai.services.ai_services import TTSService
class DeepgramTTSService(TTSService):
def __init__(self, speech_key=None, voice=None):
def __init__(self, aiohttp_session, speech_key=None, voice=None):
super().__init__()
self.voice = voice or os.getenv("DEEPGRAM_VOICE") or "alpha-asteria-en-v2"
self.speech_key = speech_key or os.getenv("DEEPGRAM_API_KEY")
self._voice = voice or os.getenv("DEEPGRAM_VOICE") or "alpha-asteria-en-v2"
self._speech_key = speech_key or os.getenv("DEEPGRAM_API_KEY")
self._aiohttp_session = aiohttp_session
def get_mic_sample_rate(self):
return 24000
@@ -21,10 +22,9 @@ class DeepgramTTSService(TTSService):
async def run_tts(self, sentence) -> AsyncGenerator[bytes, None]:
self.logger.info(f"Running deepgram tts for {sentence}")
base_url = "https://api.beta.deepgram.com/v1/speak"
request_url = f"{base_url}?model={self.voice}&encoding=linear16&container=none&sample_rate=16000"
headers = {"authorization": f"token {self.speech_key}"}
request_url = f"{base_url}?model={self._voice}&encoding=linear16&container=none&sample_rate=16000"
headers = {"authorization": f"token {self._speech_key}"}
body = {"text": sentence}
async with aiohttp.ClientSession() as session:
async with session.post(request_url, headers=headers, json=body) as r:
async with self._aiohttp_session.post(request_url, headers=headers, json=body) as r:
async for data in r.content:
yield data

22
src/dailyai/services/elevenlabs_ai_service.py
View File

@@ -9,22 +9,30 @@ from dailyai.services.ai_services import TTSService
class ElevenLabsTTSService(TTSService):
def __init__(self, api_key=None, voice_id=None):
def __init__(
self,
aiohttp_session: aiohttp.ClientSession,
api_key=None,
voice_id=None,
):
super().__init__()
self.api_key = api_key or os.getenv("ELEVENLABS_API_KEY")
self.voice_id = voice_id or os.getenv("ELEVENLABS_VOICE_ID")
self._api_key = api_key or os.getenv("ELEVENLABS_API_KEY")
self._voice_id = voice_id or os.getenv("ELEVENLABS_VOICE_ID")
self._aiohttp_session = aiohttp_session
async def run_tts(self, sentence) -> AsyncGenerator[bytes, None]:
async with aiohttp.ClientSession() as session:
url = f"https://api.elevenlabs.io/v1/text-to-speech/{self.voice_id}/stream"
url = f"https://api.elevenlabs.io/v1/text-to-speech/{self._voice_id}/stream"
payload = {"text": sentence, "model_id": "eleven_turbo_v2"}
querystring = {"output_format": "pcm_16000", "optimize_streaming_latency": 2}
headers = {
"xi-api-key": self.api_key,
"xi-api-key": self._api_key,
"Content-Type": "application/json",
}
async with session.post(url, json=payload, headers=headers, params=querystring) as r:
async with self._aiohttp_session.post(
url, json=payload, headers=headers, params=querystring
) as r:
if r.status != 200:
self.logger.error(
f"audio fetch status code: {r.status}, error: {r.text}"

15
src/dailyai/services/fal_ai_services.py
View File

@@ -11,23 +11,21 @@ from dailyai.services.ai_services import LLMService, TTSService, ImageGenService
class FalImageGenService(ImageGenService):
def __init__(self, image_size):
def __init__(self, image_size, aiohttp_session:aiohttp.ClientSession):
super().__init__(image_size)
self._aiohttp_session = aiohttp_session
async def run_image_gen(self, sentence) -> tuple[str, bytes]:
def get_image_url(sentence, size):
print("starting fal submit...")
handler = fal.apps.submit(
"110602490-fast-sdxl",
arguments={
"prompt": sentence
},
)
print("past fal handler init, about to wait for iter_events...")
for event in handler.iter_events():
if isinstance(event, fal.apps.InProgress):
print('Request in progress')
print(event.logs)
pass
result = handler.get()
@@ -36,15 +34,10 @@ class FalImageGenService(ImageGenService):
raise Exception("Image generation failed")
return image_url
print(f"fetching image url...")
image_url = await asyncio.to_thread(get_image_url, sentence, self.image_size)
print(f"got image url, downloading image...")
# Load the image from the url
async with aiohttp.ClientSession() as session:
async with session.get(image_url) as response:
print("got image response")
async with self._aiohttp_session.get(image_url) as response:
image_stream = io.BytesIO(await response.content.read())
print("read image stream")
image = Image.open(image_stream)
return (image_url, image.tobytes())

23
src/dailyai/services/open_ai_services.py
View File

@@ -51,18 +51,26 @@ class OpenAILLMService(LLMService):
class OpenAIImageGenService(ImageGenService):
def __init__(self, image_size: str, api_key=None, model=None):
def __init__(
self,
image_size: str,
aiohttp_session: aiohttp.ClientSession,
api_key=None,
model=None,
):
super().__init__(image_size=image_size)
api_key = api_key or os.getenv("OPEN_AI_KEY")
self.model = model or os.getenv("OPEN_AI_IMAGE_MODEL") or "dall-e-3"
self.client = AsyncOpenAI(api_key=api_key)
self._model = model or os.getenv("OPEN_AI_IMAGE_MODEL") or "dall-e-3"
self._client = AsyncOpenAI(api_key=api_key)
self._aiohttp_session = aiohttp_session
async def run_image_gen(self, sentence) -> tuple[str, bytes]:
self.logger.info("Generating OpenAI image", sentence)
image = await self.client.images.generate(
image = await self._client.images.generate(
prompt=sentence,
model=self.model,
model=self._model,
n=1,
size=self.image_size
)
@@ -71,10 +79,7 @@ class OpenAIImageGenService(ImageGenService):
raise Exception("No image provided in response", image)
# Load the image from the url
async with aiohttp.ClientSession() as session:
async with session.get(image_url) as response:
async with self._aiohttp_session.get(image_url) as response:
image_stream = io.BytesIO(await response.content.read())
image = Image.open(image_stream)
return (image_url, image.tobytes())
return (image_url, dalle_im.tobytes())

5
src/samples/foundational/01-say-one-thing.py
View File

@@ -1,11 +1,14 @@
import argparse
import asyncio
import aiohttp
from dailyai.services.daily_transport_service import DailyTransportService
from dailyai.services.elevenlabs_ai_service import ElevenLabsTTSService
async def main(room_url):
async with aiohttp.ClientSession() as session:
# create a transport service object using environment variables for
# the transport service's API key, room url, and any other configuration.
# services can all define and document the environment variables they use.
@@ -22,7 +25,7 @@ async def main(room_url):
meeting_duration_minutes,
)
transport.mic_enabled = True
tts = ElevenLabsTTSService(voice_id="ErXwobaYiN019PkySvjV")
tts = ElevenLabsTTSService(session, voice_id="ErXwobaYiN019PkySvjV")
# Register an event handler so we can play the audio when the participant joins.
@transport.event_handler("on_participant_joined")

10
src/samples/foundational/01a-greet-user.py
View File

@@ -2,13 +2,15 @@ import asyncio
import time
from typing import AsyncGenerator
from dailyai.queue_frame import QueueFrame, FrameType
import aiohttp
from dailyai.queue_frame import AudioQueueFrame
from dailyai.services.daily_transport_service import DailyTransportService
from dailyai.services.azure_ai_services import AzureTTSService
from dailyai.services.deepgram_ai_services import DeepgramTTSService
async def main(room_url):
async with aiohttp.ClientSession() as session:
# create a transport service object using environment variables for
# the transport service's API key, room url, and any other configuration.
# services can all define and document the environment variables they use.
@@ -27,7 +29,7 @@ async def main(room_url):
transport.mic_enabled = True
# similarly, create a tts service
tts = DeepgramTTSService()
tts = DeepgramTTSService(session)
# Get the generator for the audio. This will start running in the background,
# and when we ask the generator for its items, we'll get what it's generated.
@@ -44,7 +46,7 @@ async def main(room_url):
f"Hello there, {participant['info']['userName']}!")
async for audio in audio_generator:
transport.output_queue.put(QueueFrame(FrameType.AUDIO, audio))
await transport.send_queue.put(AudioQueueFrame(audio))
print("setting up call state handler")

9
src/samples/foundational/02-llm-say-one-thing.py
View File

@@ -1,5 +1,8 @@
import argparse
import asyncio
import logging
import aiohttp
from dailyai.queue_frame import LLMMessagesQueueFrame
from dailyai.services.daily_transport_service import DailyTransportService
@@ -8,6 +11,10 @@ from dailyai.services.elevenlabs_ai_service import ElevenLabsTTSService
async def main(room_url):
async with aiohttp.ClientSession() as session:
logger = logging.getLogger("dailyai")
logger.setLevel(logging.DEBUG)
meeting_duration_minutes = 1
transport = DailyTransportService(
room_url,
@@ -17,7 +24,7 @@ async def main(room_url):
)
transport.mic_enabled = True
tts = ElevenLabsTTSService(voice_id="29vD33N1CtxCmqQRPOHJ")
tts = ElevenLabsTTSService(session, voice_id="29vD33N1CtxCmqQRPOHJ")
llm = AzureLLMService()
messages = [{

5
src/samples/foundational/03-still-frame.py
View File

@@ -1,6 +1,8 @@
import argparse
import asyncio
import aiohttp
from dailyai.queue_frame import TextQueueFrame
from dailyai.services.daily_transport_service import DailyTransportService
from dailyai.services.open_ai_services import OpenAIImageGenService
@@ -10,6 +12,7 @@ participant_joined = False
async def main(room_url):
async with aiohttp.ClientSession() as session:
meeting_duration_minutes = 1
transport = DailyTransportService(
room_url,
@@ -22,7 +25,7 @@ async def main(room_url):
transport.camera_width = 1024
transport.camera_height = 1024
imagegen = OpenAIImageGenService(image_size="1024x1024")
imagegen = OpenAIImageGenService(image_size="1024x1024", aiohttp_session=session)
image_task = asyncio.create_task(
imagegen.run_to_queue(
transport.send_queue, [

9
src/samples/foundational/04-utterance-and-speech.py
View File

@@ -2,6 +2,8 @@ import argparse
import asyncio
import re
import aiohttp
from dailyai.services.daily_transport_service import DailyTransportService
from dailyai.services.azure_ai_services import AzureLLMService, AzureTTSService
from dailyai.queue_frame import EndStreamQueueFrame, LLMMessagesQueueFrame
@@ -9,10 +11,7 @@ from dailyai.services.elevenlabs_ai_service import ElevenLabsTTSService
async def main(room_url: str):
global transport
global llm
global tts
async with aiohttp.ClientSession() as session:
transport = DailyTransportService(
room_url,
None,
@@ -25,7 +24,7 @@ async def main(room_url: str):
llm = AzureLLMService()
azure_tts = AzureTTSService()
elevenlabs_tts = ElevenLabsTTSService(voice_id="ErXwobaYiN019PkySvjV")
elevenlabs_tts = ElevenLabsTTSService(session, voice_id="ErXwobaYiN019PkySvjV")
messages = [{"role": "system", "content": "tell the user a joke about llamas"}]

8
src/samples/foundational/05-sync-speech-and-text.py
View File

@@ -1,6 +1,8 @@
import argparse
import asyncio
import aiohttp
from dailyai.queue_frame import AudioQueueFrame, ImageQueueFrame
from dailyai.services.azure_ai_services import AzureLLMService
from dailyai.services.elevenlabs_ai_service import ElevenLabsTTSService
@@ -9,6 +11,7 @@ from dailyai.services.fal_ai_services import FalImageGenService
async def main(room_url):
async with aiohttp.ClientSession() as session:
meeting_duration_minutes = 5
transport = DailyTransportService(
room_url,
@@ -23,8 +26,8 @@ async def main(room_url):
transport.camera_height = 1024
llm = AzureLLMService()
dalle = FalImageGenService(image_size="1024x1024")
tts = ElevenLabsTTSService(voice_id="ErXwobaYiN019PkySvjV")
dalle = FalImageGenService(aiohttp_session=session, image_size="1024x1024")
tts = ElevenLabsTTSService(aiohttp_session=session, voice_id="ErXwobaYiN019PkySvjV")
# dalle = OpenAIImageGenService(image_size="1024x1024")
# Get a complete audio chunk from the given text. Splitting this into its own
@@ -85,6 +88,7 @@ async def main(room_url):
# likely no delay between months, but the months won't display in order.
for month_data_task in asyncio.as_completed(month_tasks):
data = await month_data_task
if data:
await transport.send_queue.put(
[
ImageQueueFrame(data["image_url"], data["image"]),

8
src/samples/foundational/07-interruptible.py
View File

@@ -1,5 +1,6 @@
import argparse
import asyncio
import aiohttp
import requests
import time
import urllib.parse
@@ -12,10 +13,7 @@ from dailyai.services.elevenlabs_ai_service import ElevenLabsTTSService
async def main(room_url: str, token):
global transport
global llm
global tts
async with aiohttp.ClientSession() as session:
transport = DailyTransportService(
room_url,
token,
@@ -28,7 +26,7 @@ async def main(room_url: str, token):
transport.start_transcription = True
llm = AzureLLMService()
tts = ElevenLabsTTSService(voice_id="ErXwobaYiN019PkySvjV")
tts = ElevenLabsTTSService(session, voice_id="ErXwobaYiN019PkySvjV")
async def run_response(user_speech, tma_in, tma_out):
await tts.run_to_queue(