From 434772dc23e61eca22af5a0db0f26e811b30b3fe Mon Sep 17 00:00:00 2001 From: Moishe Lettvin Date: Sun, 3 Mar 2024 19:50:13 -0500 Subject: [PATCH] Update sample 5! --- src/dailyai/pipeline/aggregators.py | 22 ++++- src/dailyai/services/ai_services.py | 2 + src/dailyai/tests/test_aggregators.py | 6 +- .../foundational/05-sync-speech-and-image.py | 99 ++++++------------- 4 files changed, 54 insertions(+), 75 deletions(-) diff --git a/src/dailyai/pipeline/aggregators.py b/src/dailyai/pipeline/aggregators.py index 742555006..e1873e981 100644 --- a/src/dailyai/pipeline/aggregators.py +++ b/src/dailyai/pipeline/aggregators.py @@ -9,6 +9,7 @@ from dailyai.pipeline.frames import ( EndParallelPipeQueueFrame, EndStreamQueueFrame, LLMMessagesQueueFrame, + LLMResponseEndQueueFrame, QueueFrame, TextQueueFrame, TranscriptionQueueFrame, @@ -16,7 +17,7 @@ from dailyai.pipeline.frames import ( from dailyai.pipeline.pipeline import Pipeline from dailyai.services.ai_services import AIService -from typing import AsyncGenerator, Coroutine, List +from typing import AsyncGenerator, Coroutine, List, Text class LLMContextAggregator(AIService): @@ -122,6 +123,23 @@ class SentenceAggregator(FrameProcessor): yield frame +class LLMFullResponseAggregator(FrameProcessor): + def __init__(self): + self.aggregation = "" + + async def process_frame( + self, frame: QueueFrame + ) -> AsyncGenerator[QueueFrame, None]: + if isinstance(frame, TextQueueFrame): + self.aggregation += frame.text + elif isinstance(frame, LLMResponseEndQueueFrame): + yield TextQueueFrame(self.aggregation) + self.aggregation = "" + else: + yield frame + + + class StatelessTextTransformer(FrameProcessor): def __init__(self, transform_fn): self.transform_fn = transform_fn @@ -158,7 +176,7 @@ class ParallelPipeline(FrameProcessor): if not isinstance(frame, EndParallelPipeQueueFrame): yield frame -class GatedAccumulator(FrameProcessor): +class GatedAggregator(FrameProcessor): def __init__(self, gate_open_fn, gate_close_fn, start_open): self.gate_open_fn = gate_open_fn self.gate_close_fn = gate_close_fn diff --git a/src/dailyai/services/ai_services.py b/src/dailyai/services/ai_services.py index d6248c060..4860f245d 100644 --- a/src/dailyai/services/ai_services.py +++ b/src/dailyai/services/ai_services.py @@ -12,6 +12,7 @@ from dailyai.pipeline.frames import ( ImageQueueFrame, LLMMessagesQueueFrame, LLMResponseEndQueueFrame, + LLMResponseStartQueueFrame, QueueFrame, TextQueueFrame, TranscriptionQueueFrame, @@ -78,6 +79,7 @@ class LLMService(AIService): async def process_frame(self, frame: QueueFrame) -> AsyncGenerator[QueueFrame, None]: if isinstance(frame, LLMMessagesQueueFrame): + yield LLMResponseStartQueueFrame() async for text_chunk in self.run_llm_async(frame.messages): yield TextQueueFrame(text_chunk) yield LLMResponseEndQueueFrame() diff --git a/src/dailyai/tests/test_aggregators.py b/src/dailyai/tests/test_aggregators.py index 986b28226..2dfe574e4 100644 --- a/src/dailyai/tests/test_aggregators.py +++ b/src/dailyai/tests/test_aggregators.py @@ -3,7 +3,7 @@ import functools import unittest from dailyai.pipeline.aggregators import ( - GatedAccumulator, + GatedAggregator, ParallelPipeline, SentenceAggregator, StatelessTextTransformer, @@ -43,7 +43,7 @@ class TestDailyFrameAggregators(unittest.IsolatedAsyncioTestCase): self.assertEqual(expected_sentences, []) async def test_gated_accumulator(self): - gated_accumulator = GatedAccumulator( + gated_aggregator = GatedAggregator( gate_open_fn=lambda frame: isinstance(frame, ImageQueueFrame), gate_close_fn=lambda frame: isinstance(frame, LLMResponseStartQueueFrame), start_open=False, @@ -69,7 +69,7 @@ class TestDailyFrameAggregators(unittest.IsolatedAsyncioTestCase): LLMResponseEndQueueFrame(), ] for frame in frames: - async for out_frame in gated_accumulator.process_frame(frame): + async for out_frame in gated_aggregator.process_frame(frame): self.assertEqual(out_frame, expected_output_frames.pop(0)) self.assertEqual(expected_output_frames, []) diff --git a/src/examples/foundational/05-sync-speech-and-image.py b/src/examples/foundational/05-sync-speech-and-image.py index ba8540a69..0ab448275 100644 --- a/src/examples/foundational/05-sync-speech-and-image.py +++ b/src/examples/foundational/05-sync-speech-and-image.py @@ -1,8 +1,11 @@ import asyncio +from re import S import aiohttp import os +from dailyai.pipeline.aggregators import GatedAggregator, LLMFullResponseAggregator, ParallelPipeline, SentenceAggregator -from dailyai.pipeline.frames import AudioQueueFrame, ImageQueueFrame +from dailyai.pipeline.frames import AudioQueueFrame, EndStreamQueueFrame, ImageQueueFrame, LLMMessagesQueueFrame, LLMResponseStartQueueFrame +from dailyai.pipeline.pipeline import Pipeline from dailyai.services.azure_ai_services import AzureLLMService, AzureImageGenServiceREST, AzureTTSService from dailyai.services.elevenlabs_ai_service import ElevenLabsTTSService from dailyai.services.daily_transport_service import DailyTransportService @@ -35,98 +38,54 @@ async def main(room_url): aiohttp_session=session, api_key=os.getenv("ELEVENLABS_API_KEY"), voice_id="ErXwobaYiN019PkySvjV") - # tts = AzureTTSService(api_key=os.getenv("AZURE_SPEECH_API_KEY"), region=os.getenv("AZURE_SPEECH_REGION")) dalle = FalImageGenService( image_size="1024x1024", aiohttp_session=session, key_id=os.getenv("FAL_KEY_ID"), key_secret=os.getenv("FAL_KEY_SECRET")) - # dalle = OpenAIImageGenService(aiohttp_session=session, api_key=os.getenv("OPENAI_DALLE_API_KEY"), image_size="1024x1024") - # dalle = AzureImageGenServiceREST(image_size="1024x1024", aiohttp_session=session, api_key=os.getenv("AZURE_DALLE_API_KEY"), endpoint=os.getenv("AZURE_DALLE_ENDPOINT"), model=os.getenv("AZURE_DALLE_MODEL")) - # 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 send queue. - async def get_all_audio(text): - all_audio = bytearray() - async for audio in tts.run_tts(text): - all_audio.extend(audio) + source_queue = asyncio.Queue() - return all_audio - - async def get_month_data(month): + for month in ["January", "February"]: messages = [ { "role": "system", "content": f"Describe a nature photograph suitable for use in a calendar, for the month of {month}. Include only the image description with no preamble. Limit the description to one sentence, please.", } ] + await source_queue.put(LLMMessagesQueueFrame(messages)) - image_description = await llm.run_llm(messages) - if not image_description: - return + await source_queue.put(EndStreamQueueFrame()) - to_speak = f"{month}: {image_description}" - audio_task = asyncio.create_task(get_all_audio(to_speak)) - image_task = asyncio.create_task(dalle.run_image_gen(image_description)) - print(f"about to gather tasks for {month}") - (audio, image_data) = await asyncio.gather( - audio_task, image_task - ) - print(f"about to return from get_month_data for {month}") - return { - "month": month, - "text": image_description, - "image_url": image_data[0], - "image": image_data[1], - "audio": audio, - } + gated_aggregator = GatedAggregator( + gate_open_fn=lambda frame: isinstance(frame, ImageQueueFrame), + gate_close_fn=lambda frame: isinstance(frame, LLMResponseStartQueueFrame), + start_open=False, + ) + + sentence_aggregator = SentenceAggregator() + llm_full_response_aggregator = LLMFullResponseAggregator() + + pipeline = Pipeline( + source=source_queue, + sink=transport.send_queue, + processors=[ + llm, + sentence_aggregator, + ParallelPipeline([[tts], [llm_full_response_aggregator, dalle]]), + gated_aggregator, + ], + ) + pipeline_task = pipeline.run_pipeline() - months: list[str] = [ - "January", - "February", - "March", - "April", - "May", - "June" - ] - """ - "February", - "March", - "April", - "May", - "June", - "July", - "August", - "September", - "October", - "November", - "December", - """ @transport.event_handler("on_first_other_participant_joined") async def on_first_other_participant_joined(transport): - # This will play the months in the order they're completed. The benefit - # is we'll have as little delay as possible before the first month, and - # likely no delay between months, but the months won't display in order. - for month_data_task in asyncio.as_completed(month_tasks): - print(f"month_data_task: {month_data_task}") - try: - data = await month_data_task - except Exception: - print("OMG EXCEPTION!!!!") - if data: - await transport.send_queue.put( - [ - ImageQueueFrame(data["image_url"], data["image"]), - AudioQueueFrame(data["audio"]), - ] - ) + await pipeline_task # wait for the output queue to be empty, then leave the meeting await transport.stop_when_done() - month_tasks = [asyncio.create_task(get_month_data(month)) for month in months] - await transport.run() if __name__ == "__main__":