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

refactor party tonight

Browse Source
This commit is contained in:
Moishe Lettvin
2024-01-17 18:42:08 -05:00
parent a3ac0d84e8
commit 13f2f792af
10 changed files with 187 additions and 118 deletions
Download Patch File Download Diff File Expand all files Collapse all files

157
src/dailyai/services/ai_services.py
View File

@@ -2,6 +2,8 @@ import asyncio
import logging
import re
from httpx import request
from dailyai.queue_frame import QueueFrame, FrameType
from abc import abstractmethod
@@ -13,9 +15,7 @@ from collections.abc import Iterable, AsyncIterable
class AIService:
def __init__(
self
):
def __init__(self):
self.logger = logging.getLogger("dailyai")
def stop(self):
@@ -27,30 +27,60 @@ class AIService:
def possible_output_frame_types(self) -> set[FrameType]:
return set()
async def run_to_queue(self, queue: asyncio.Queue, frames, add_end_of_stream=False) -> None:
async for frame in self.run(frames):
print("got frame", frame.frame_type)
await queue.put(frame)
if add_end_of_stream:
await queue.put(QueueFrame(FrameType.END_STREAM, None))
async def run(
self,
requested_frame_types:set[FrameType],
frames:Iterable[QueueFrame] | AsyncIterable[QueueFrame]
) -> AsyncGenerator[QueueFrame, None]:
if self.possible_output_frame_types().intersection(requested_frame_types) == set():
self,
frames: Iterable[QueueFrame]
| AsyncIterable[QueueFrame]
| asyncio.Queue[QueueFrame],
requested_frame_types: set[FrameType] | None=None,
) -> AsyncGenerator[QueueFrame, None]:
if requested_frame_types and self.possible_output_frame_types().intersection(requested_frame_types) == set():
raise Exception(f"Requested frame types {requested_frame_types} are not supported by this service.")
if not requested_frame_types:
requested_frame_types = self.possible_output_frame_types()
print("running", self.__class__.__name__, "with frame types", requested_frame_types)
if isinstance(frames, AsyncIterable):
async for frame in frames:
output_frame: QueueFrame | None = await self.process_frame(requested_frame_types, frame)
if output_frame:
async for output_frame in self.process_frame(requested_frame_types, frame):
print(
"yielding frame", self.__class__.__name__, output_frame.frame_type
)
yield output_frame
elif isinstance(frames, Iterable):
for frame in frames:
output_frame = await self.process_frame(requested_frame_types, frame)
if output_frame:
async for output_frame in self.process_frame(requested_frame_types, frame):
print(
"yielding frame", self.__class__.__name__, output_frame.frame_type
)
yield output_frame
elif isinstance(frames, asyncio.Queue):
while True:
frame = await frames.get()
async for output_frame in self.process_frame(requested_frame_types, frame):
print(
"yielding frame", self.__class__.__name__, output_frame.frame_type
)
yield output_frame
if frame.frame_type == FrameType.END_STREAM:
break
else:
raise Exception("Frames must be an iterable or async iterable")
@abstractmethod
async def process_frame(self, requested_frame_types:set[FrameType], frame:QueueFrame) -> QueueFrame | None:
pass
async def process_frame(self, requested_frame_types:set[FrameType], frame:QueueFrame) -> AsyncGenerator[QueueFrame, None]:
# Yield something so the linter can deduce what should happen here.
yield QueueFrame(FrameType.END_STREAM, None)
class SentenceAggregator(AIService):
def __init__(self, **kwargs):
@@ -63,29 +93,26 @@ class SentenceAggregator(AIService):
def possible_output_frame_types(self) -> set[FrameType]:
return set([FrameType.SENTENCE])
async def process_frame(self, requested_frame_types: set[FrameType], frame: QueueFrame) -> QueueFrame | None:
async def process_frame(self, requested_frame_types: set[FrameType], frame: QueueFrame) -> AsyncGenerator[QueueFrame, None]:
if not FrameType.SENTENCE in requested_frame_types:
return None
return
if frame.frame_type == FrameType.TEXT_CHUNK:
if type(frame.frame_data) != str:
raise Exception("Sentence aggregator requires a string for the data field")
raise Exception(
"Sentence aggregator requires a string for the data field"
)
self.current_sentence += frame.frame_data
if self.current_sentence.endswith((".", "?", "!")):
sentence = self.current_sentence
self.current_sentence = ""
return QueueFrame(FrameType.SENTENCE, sentence)
return None
yield QueueFrame(FrameType.SENTENCE, sentence)
elif frame.frame_type == FrameType.END_STREAM:
if self.current_sentence:
return QueueFrame(FrameType.SENTENCE, self.current_sentence)
else:
return None
yield QueueFrame(FrameType.SENTENCE, self.current_sentence)
elif frame.frame_type == FrameType.SENTENCE:
return frame
else:
return None
yield frame
class LLMService(AIService):
@@ -93,30 +120,29 @@ class LLMService(AIService):
return set([FrameType.LLM_MESSAGE, FrameType.SENTENCE, FrameType.TRANSCRIPTION])
def allowed_output_frame_types(self) -> set[FrameType]:
return set([FrameType.SENTENCE, FrameType.SENTENCE, FrameType.TEXT_CHUNK])
return set([FrameType.SENTENCE, FrameType.TEXT_CHUNK])
async def run_llm_async_sentences(self, messages) -> AsyncGenerator[str, None]:
current_text = ""
async for text in self.run_llm_async(messages):
current_text += text
if re.match(r"^.*[.!?]$", text):
yield current_text
current_text = ""
@abstractmethod
async def run_llm_async(self, messages) -> AsyncGenerator[str, None]:
yield ""
if current_text:
yield current_text
async def process_frame(self, frame:QueueFrame) -> QueueFrame | None:
if not self.output_queue:
raise Exception("Output queue must be set before using the run method.")
@abstractmethod
async def run_llm(self, messages) -> str:
pass
async def process_frame(self, requested_frame_types: set[FrameType], frame: QueueFrame) -> AsyncGenerator[QueueFrame, None]:
if frame.frame_type == FrameType.LLM_MESSAGE:
if type(frame.frame_data) != list:
raise Exception("LLM service requires a dict for the data field")
messages: list[dict[str, str]] = frame.frame_data
async for message in self.run_llm_async_sentences(messages):
await self.output_queue.put(QueueFrame(FrameType.SENTENCE, message))
if FrameType.SENTENCE in requested_frame_types:
yield QueueFrame(FrameType.SENTENCE, await self.run_llm(messages))
else:
async for text_chunk in self.run_llm_async(messages):
yield QueueFrame(FrameType.TEXT_CHUNK, text_chunk)
# TODO: handle other frame types! Need to aggregate into messages
class TTSService(AIService):
@@ -124,6 +150,12 @@ class TTSService(AIService):
def get_mic_sample_rate(self):
return 16000
def allowed_input_frame_types(self) -> set[FrameType]:
return set([FrameType.SENTENCE, FrameType.TRANSCRIPTION, FrameType.TEXT_CHUNK])
def possible_output_frame_types(self) -> set[FrameType]:
return set([FrameType.AUDIO])
# Converts the sentence to audio. Yields a list of audio frames that can
# be sent to the microphone device
@abstractmethod
@@ -131,25 +163,48 @@ class TTSService(AIService):
# yield empty bytes here, so linting can infer what this method does
yield bytes()
async def process_frame(self, frame:QueueFrame) -> QueueFrame | None:
if not self.output_queue:
raise Exception("Output queue must be set before using the run method.")
async def process_frame(self, requested_frame_types: set[FrameType], frame: QueueFrame) -> AsyncGenerator[QueueFrame, None]:
if not FrameType.AUDIO in requested_frame_types:
return
if frame.frame_type == FrameType.SENTENCE:
if type(frame.frame_data) != str:
raise Exception("TTS service requires a string for the data field")
if type(frame.frame_data) != str:
raise Exception("TTS service requires a string for the data field")
text = frame.frame_data
async for audio in self.run_tts(text):
await self.output_queue.put(QueueFrame(FrameType.AUDIO, audio))
async for audio_chunk in self.run_tts(frame.frame_data):
yield QueueFrame(FrameType.AUDIO, audio_chunk)
# Convenience function to send the audio for a sentence to the given queue
async def say(self, sentence, queue: asyncio.Queue):
async for audio_chunk in self.run_tts(sentence):
await queue.put(QueueFrame(FrameType.AUDIO, audio_chunk))
class ImageGenService(AIService):
def __init__(self, image_size, **kwargs):
super().__init__(**kwargs)
self.image_size = image_size
def allowed_input_frame_types(self) -> set[FrameType]:
return set([FrameType.SENTENCE, FrameType.TRANSCRIPTION, FrameType.TEXT_CHUNK, FrameType.IMAGE_DESCRIPTION])
def possible_output_frame_types(self) -> set[FrameType]:
return set([FrameType.IMAGE])
# Renders the image. Returns an Image object.
@abstractmethod
async def run_image_gen(self, sentence, size) -> tuple[str, bytes]:
async def run_image_gen(self, sentence) -> tuple[str, bytes]:
pass
async def process_frame(self, requested_frame_types: set[FrameType], frame: QueueFrame) -> AsyncGenerator[QueueFrame, None]:
if not FrameType.IMAGE in requested_frame_types:
return
if type(frame.frame_data) != str:
raise Exception("Image service requires a string for the data field")
(_, image_data) = await self.run_image_gen(frame.frame_data)
yield QueueFrame(FrameType.IMAGE, image_data)
@dataclass
class AIServiceConfig:

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

@@ -16,8 +16,8 @@ from PIL import Image
from azure.cognitiveservices.speech import SpeechSynthesizer, SpeechConfig, ResultReason, CancellationReason
class AzureTTSService(TTSService):
def __init__(self, input_queue=None, output_queue=None, speech_key=None, speech_region=None):
super().__init__(input_queue, output_queue)
def __init__(self, speech_key=None, speech_region=None):
super().__init__()
speech_key = speech_key or os.getenv("AZURE_SPEECH_SERVICE_KEY")
speech_region = speech_region or os.getenv("AZURE_SPEECH_SERVICE_REGION")
@@ -35,7 +35,10 @@ class AzureTTSService(TTSService):
"<prosody rate='1.05'>" \
f"{sentence}" \
"</prosody></mstts:express-as></voice></speak> "
result = await asyncio.to_thread(self.speech_synthesizer.speak_ssml, (ssml))
try:
result = await asyncio.to_thread(self.speech_synthesizer.speak_ssml, (ssml))
except Exception as e:
self.logger.error("Error in azure tts", e)
self.logger.info("Got azure tts result")
if result.reason == ResultReason.SynthesizingAudioCompleted:
self.logger.info("Returning result")
@@ -48,8 +51,8 @@ class AzureTTSService(TTSService):
self.logger.info("Error details: {}".format(cancellation_details.error_details))
class AzureLLMService(LLMService):
def __init__(self, input_queue=None, output_queue=None, api_key=None, azure_endpoint=None, api_version=None, model=None):
super().__init__(input_queue, output_queue)
def __init__(self, api_key=None, azure_endpoint=None, api_version=None, model=None):
super().__init__()
api_key = api_key or os.getenv("AZURE_CHATGPT_KEY")
azure_endpoint = azure_endpoint or os.getenv("AZURE_CHATGPT_ENDPOINT")
@@ -92,14 +95,14 @@ class AzureLLMService(LLMService):
class AzureImageGenServiceREST(ImageGenService):
def __init__(self, api_key=None, azure_endpoint=None, api_version=None, model=None):
super().__init__()
def __init__(self, image_size:str, 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")
async def run_image_gen(self, sentence, size) -> tuple[str, bytes]:
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}"
@@ -107,7 +110,7 @@ class AzureImageGenServiceREST(ImageGenService):
body = {
# Enter your prompt text here
"prompt": sentence,
"size": size,
"size": self.image_size,
"n": 1,
}
async with session.post(url, headers=headers, json=body) as submission:
@@ -153,14 +156,14 @@ class AzureImageGenService(ImageGenService):
api_version=api_version,
)
async def run_image_gen(self, sentence, size) -> tuple[str, bytes]:
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=size,
size=self.image_size,
)
url = image["data"][0]["url"]

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

@@ -206,6 +206,9 @@ class DailyTransportService(EventHandler):
if frame.frame_type == FrameType.END_STREAM:
break
def wait_for_send_queue_to_empty(self):
self.threadsafe_send_queue.join()
async def run(self) -> None:
self.configure_daily()

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

@@ -9,8 +9,8 @@ from dailyai.services.ai_services import TTSService
class ElevenLabsTTSService(TTSService):
def __init__(self, input_queue=None, output_queue=None, api_key=None, voice_id=None):
super().__init__(input_queue, output_queue)
def __init__(self, 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")

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

@@ -50,20 +50,20 @@ class OpenAILLMService(LLMService):
return None
class OpenAIImageGenService(ImageGenService):
def __init__(self, api_key=None, model=None):
super().__init__()
def __init__(self, image_size:str, 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)
async def run_image_gen(self, sentence, size) -> tuple[str, bytes]:
async def run_image_gen(self, sentence) -> tuple[str, bytes]:
self.logger.info("Generating OpenAI image", sentence)
image = await self.client.images.generate(
prompt=sentence,
model=self.model,
n=1,
size=size
size=self.image_size
)
image_url = image.data[0].url
if not image_url:

9
src/samples/theoretical-to-real/01-say-one-thing.py
View File

@@ -27,21 +27,16 @@ async def main(room_url):
# similarly, create a tts service
tts = AzureTTSService()
# 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.
audio_generator: AsyncGenerator[bytes, None] = tts.run_tts("hello world")
# Register an event handler so we can play the audio when the participant joins.
@transport.event_handler("on_participant_joined")
async def on_participant_joined(transport, participant):
if participant["info"]["isLocal"]:
return
async for audio in audio_generator:
transport.output_queue.put(QueueFrame(FrameType.AUDIO, audio))
await tts.say("Hello there, " + participant["info"]["userName"] + "!", transport.send_queue)
# wait for the output queue to be empty, then leave the meeting
transport.output_queue.join()
transport.wait_for_send_queue_to_empty()
transport.stop()
await transport.run()

27
src/samples/theoretical-to-real/02-llm-say-one-thing.py
View File

@@ -4,6 +4,7 @@ from typing import AsyncGenerator
from dailyai.queue_frame import QueueFrame, FrameType
from dailyai.services.daily_transport_service import DailyTransportService
from dailyai.services.ai_services import SentenceAggregator
from dailyai.services.azure_ai_services import AzureLLMService
from dailyai.services.elevenlabs_ai_service import ElevenLabsTTSService
@@ -17,29 +18,27 @@ async def main(room_url):
)
transport.mic_enabled = True
text_to_llm_queue = asyncio.Queue()
llm_to_tts_queue = asyncio.Queue()
tts = ElevenLabsTTSService(
llm_to_tts_queue, transport.get_async_send_queue(), voice_id="29vD33N1CtxCmqQRPOHJ"
)
llm = AzureLLMService(text_to_llm_queue, llm_to_tts_queue)
tts = ElevenLabsTTSService(voice_id="29vD33N1CtxCmqQRPOHJ")
llm = AzureLLMService()
messages = [{
"role": "system",
"content": "You are an LLM in a WebRTC session, and this is a 'hello world' demo. Say hello to the world."
}]
await text_to_llm_queue.put(QueueFrame(FrameType.LLM_MESSAGE, messages))
await text_to_llm_queue.put(QueueFrame(FrameType.END_STREAM, None))
llm_task = asyncio.create_task(llm.run())
tts_task = asyncio.create_task(
tts.run_to_queue(
transport.send_queue,
SentenceAggregator().run(
llm.run([QueueFrame(FrameType.LLM_MESSAGE, messages)])
)
)
)
@transport.event_handler("on_first_other_participant_joined")
async def on_first_other_participant_joined(transport):
await asyncio.gather(llm_task, tts.run())
await tts_task
# wait for the output queue to be empty, then leave the meeting
transport.output_queue.join()
transport.wait_for_send_queue_to_empty()
transport.stop()
await transport.run()

11
src/samples/theoretical-to-real/03-still-frame.py
View File

@@ -21,13 +21,14 @@ async def main(room_url):
transport.camera_width = 1024
transport.camera_height = 1024
imagegen = OpenAIImageGenService()
image_task = asyncio.create_task(imagegen.run_image_gen("a cat in the style of picasso", "1024x1024"))
imagegen = OpenAIImageGenService(image_size="1024x1024")
image_task = asyncio.create_task(
imagegen.run_to_queue(transport.send_queue, [QueueFrame(FrameType.IMAGE_DESCRIPTION, "a cat in the style of picasso")])
)
@transport.event_handler("on_participant_joined")
async def on_participant_joined(transport, participant):
(_, image_bytes) = await image_task
transport.output_queue.put(QueueFrame(FrameType.IMAGE, image_bytes))
await image_task
await transport.run()
@@ -38,6 +39,6 @@ if __name__ == "__main__":
"-u", "--url", type=str, required=True, help="URL of the Daily room to join"
)
args: argparse.Namespace = parser.parse_args()
args, unknown = parser.parse_known_args()
asyncio.run(main(args.url))

54
src/samples/theoretical-to-real/04-utterance-and-speech.py
View File

@@ -2,9 +2,11 @@ import argparse
import asyncio
import re
from dailyai.services.ai_services import SentenceAggregator
from dailyai.services.daily_transport_service import DailyTransportService
from dailyai.services.azure_ai_services import AzureLLMService, AzureTTSService
from dailyai.queue_frame import QueueFrame, FrameType
from dailyai.services.elevenlabs_ai_service import ElevenLabsTTSService
async def main(room_url:str):
global transport
@@ -22,34 +24,46 @@ async def main(room_url:str):
transport.camera_enabled = False
llm = AzureLLMService()
tts = AzureTTSService()
azure_tts = AzureTTSService()
elevenlabs_tts = ElevenLabsTTSService(voice_id="ErXwobaYiN019PkySvjV")
messages = [{"role": "system", "content": "tell the user a joke about llamas"}]
# Start a task to run the LLM to create a joke, and convert the LLM output to audio frames. This task
# will run in parallel with generating and speaking the audio for static text, so there's no delay to
# speak the LLM response.
buffer_queue = asyncio.Queue()
llm_response_task = asyncio.create_task(
elevenlabs_tts.run_to_queue(
buffer_queue,
SentenceAggregator().run(
llm.run([QueueFrame(FrameType.LLM_MESSAGE, messages)])
),
True,
)
)
@transport.event_handler("on_participant_joined")
async def on_joined(transport, participant):
if participant["id"] == transport.my_participant_id:
return
# queue two pieces of speech: one specified as a text literal,
# and one generated by an llm. We'll kick off the llm first, and let
# it generate a response while we're speaking the literal string.
#
# Note that in this case, we don't use `run_llm_async` because we're
# taking advantage of the time spent speaking the first phrase to generate
# the entire LLM response, and this happens asynchronously in a task.
llm_response_task = asyncio.create_task(llm.run_llm(
[{"role": "system", "content": "tell the user a joke about llamas"}]
))
await azure_tts.run_to_queue(
transport.send_queue,
[QueueFrame(FrameType.SENTENCE, "My friend the LLM is now going to tell a joke about llamas.")]
)
async for audio_chunk in tts.run_tts("My friend the LLM is now going to tell a joke about llamas."):
transport.output_queue.put(QueueFrame(FrameType.AUDIO, audio_chunk))
async def buffer_to_send_queue():
while True:
frame = await buffer_queue.get()
await transport.send_queue.put(frame)
buffer_queue.task_done()
if frame.frame_type == FrameType.END_STREAM:
break
llm_response = await llm_response_task
async for audio_chunk in tts.run_tts(llm_response):
transport.output_queue.put(QueueFrame(FrameType.AUDIO, audio_chunk))
await asyncio.gather(llm_response_task, buffer_to_send_queue())
# wait for the output queue to be empty, then leave the meeting
transport.output_queue.join()
transport.wait_for_send_queue_to_empty()
transport.stop()
await transport.run()
@@ -61,6 +75,6 @@ if __name__ == "__main__":
"-u", "--url", type=str, required=True, help="URL of the Daily room to join"
)
args: argparse.Namespace = parser.parse_args()
args, unknown = parser.parse_known_args()
asyncio.run(main(args.url))

7
src/samples/theoretical-to-real/05-sync-speech-and-text.py
View File

@@ -26,10 +26,9 @@ async def main(room_url):
transport.camera_height = 1024
llm = AzureLLMService()
#tts = ElevenLabsTTSService(voice_id="ErXwobaYiN019PkySvjV")
tts = ElevenLabsTTSService()
dalle = FalImageGenService()
# dalle = OpenAIImageGenService()
tts = ElevenLabsTTSService(voice_id="ErXwobaYiN019PkySvjV")
#dalle = OpenAIImageGenService(image_size="1024x1024")
# Get a complete audio chunk from the given text. Splitting this into its own
# coroutine lets us ensure proper ordering of the audio chunks on the output queue.
@@ -61,7 +60,7 @@ async def main(room_url):
tts_tasks.append(get_all_audio(sentence))
tts_tasks.insert(0, dalle.run_image_gen(image_text, "1024x1024"))
tts_tasks.insert(0, dalle.run_image_gen(image_text))
print(f"waiting for tasks to finish for {month}")
data = await asyncio.gather(