diff --git a/src/dailyai/pipeline/aggregators.py b/src/dailyai/pipeline/aggregators.py index d229b7178..aea02d789 100644 --- a/src/dailyai/pipeline/aggregators.py +++ b/src/dailyai/pipeline/aggregators.py @@ -13,34 +13,62 @@ from dailyai.pipeline.frames import ( LLMResponseStartFrame, TextFrame, TranscriptionQueueFrame, + UserStartedSpeakingFrame, + UserStoppedSpeakingFrame ) from dailyai.pipeline.pipeline import Pipeline from dailyai.services.ai_services import AIService from typing import AsyncGenerator, Coroutine, List -class LLMResponseAggregator(FrameProcessor): - def __init__(self, messages: list[dict]): +class ResponseAggregator(FrameProcessor): + def __init__(self, *, messages: list[dict], role: str, start_frame, end_frame, accumulator_frame, pass_through=True): self.aggregation = "" self.aggregating = False self.messages = messages + self._role = role + self._start_frame = start_frame + self._end_frame = end_frame + self._accumulator_frame = accumulator_frame + self._pass_through = pass_through async def process_frame( self, frame: Frame ) -> AsyncGenerator[Frame, None]: - if isinstance(frame, LLMResponseStartFrame): + if isinstance(frame, self._start_frame): self.aggregating = True - elif isinstance(frame, LLMResponseEndFrame): + elif isinstance(frame, self._end_frame): self.aggregating = False - self.messages.append({"role": "assistant", "content": self.aggregation}) + self.messages.append({"role": self._role, "content": self.aggregation}) self.aggregation = "" yield LLMMessagesQueueFrame(self.messages) - elif isinstance(frame, TextFrame) and self.aggregating: - self.aggregation += frame.text - yield frame + elif isinstance(frame, self._accumulator_frame) and self.aggregating: + self.aggregation += f" {frame.text}" + if self._pass_through: + yield frame else: yield frame +class LLMResponseAggregator(ResponseAggregator): + def __init__(self, messages: list[dict]): + super().__init__( + messages=messages, + role="assistant", + start_frame=LLMResponseStartFrame, + end_frame=LLMResponseEndFrame, + accumulator_frame=TextFrame + ) + +class UserResponseAggregator(ResponseAggregator): + def __init__(self, messages: list[dict]): + super().__init__( + messages=messages, + role="user", + start_frame=UserStartedSpeakingFrame, + end_frame=UserStoppedSpeakingFrame, + accumulator_frame=TranscriptionQueueFrame, + pass_through=False + ) class LLMContextAggregator(AIService): def __init__( diff --git a/src/dailyai/pipeline/frames.py b/src/dailyai/pipeline/frames.py index aa20b27ea..dd348478d 100644 --- a/src/dailyai/pipeline/frames.py +++ b/src/dailyai/pipeline/frames.py @@ -72,3 +72,8 @@ class UserStartedSpeakingFrame(Frame): class UserStoppedSpeakingFrame(Frame): pass + +@dataclass() +class LLMFunctionCallFrame(Frame): + function_name: str + arguments: str \ No newline at end of file diff --git a/src/dailyai/services/ai_services.py b/src/dailyai/services/ai_services.py index 6a692fb52..e738b6cdb 100644 --- a/src/dailyai/services/ai_services.py +++ b/src/dailyai/services/ai_services.py @@ -12,9 +12,11 @@ from dailyai.pipeline.frames import ( LLMMessagesQueueFrame, LLMResponseEndFrame, LLMResponseStartFrame, + LLMFunctionCallFrame, Frame, TextFrame, TranscriptionQueueFrame, + UserStoppedSpeakingFrame ) from abc import abstractmethod @@ -65,6 +67,11 @@ class AIService(FrameProcessor): class LLMService(AIService): + def __init__(self, messages=None, tools=None): + super().__init__() + self._tools = tools + self._messages = messages + @abstractmethod async def run_llm_async(self, messages) -> AsyncGenerator[str, None]: yield "" @@ -73,12 +80,27 @@ class LLMService(AIService): async def run_llm(self, messages) -> str: pass - async def process_frame(self, frame: Frame) -> AsyncGenerator[Frame, None]: + async def process_frame(self, frame: Frame, tool_choice: str = None) -> AsyncGenerator[Frame, None]: if isinstance(frame, LLMMessagesQueueFrame): - yield LLMResponseStartFrame() - async for text_chunk in self.run_llm_async(frame.messages): - yield TextFrame(text_chunk) - yield LLMResponseEndFrame() + function_name = "" + arguments = "" + if isinstance(frame, LLMMessagesQueueFrame): + yield LLMResponseStartFrame() + async for text_chunk in self.run_llm_async(frame.messages, tool_choice): + if isinstance(text_chunk, str): + yield TextFrame(text_chunk) + elif text_chunk.function: + if text_chunk.function.name: + # function_name += text_chunk.function.name + yield LLMFunctionCallFrame(function_name=text_chunk.function.name, arguments=None) + if text_chunk.function.arguments: + # arguments += text_chunk.function.arguments + yield LLMFunctionCallFrame(function_name=None, arguments=text_chunk.function.arguments) + + if (function_name and arguments): + function_name = "" + arguments = "" + yield LLMResponseEndFrame() else: yield frame @@ -129,7 +151,7 @@ class TTSService(AIService): # Convenience function to send the audio for a sentence to the given queue async def say(self, sentence, queue: asyncio.Queue): - await self.run_to_queue(queue, [TextFrame(sentence)]) + await self.run_to_queue(queue, [LLMResponseStartFrame(), TextFrame(sentence), LLMResponseEndFrame()]) class ImageGenService(AIService): diff --git a/src/dailyai/services/azure_ai_services.py b/src/dailyai/services/azure_ai_services.py index a69095d02..627032a7e 100644 --- a/src/dailyai/services/azure_ai_services.py +++ b/src/dailyai/services/azure_ai_services.py @@ -2,6 +2,7 @@ import aiohttp import asyncio import io import json +import time from openai import AsyncAzureOpenAI import os @@ -48,8 +49,8 @@ class AzureTTSService(TTSService): class AzureLLMService(LLMService): - def __init__(self, *, api_key, endpoint, api_version="2023-12-01-preview", model): - super().__init__() + def __init__(self, *, api_key, endpoint, api_version="2023-12-01-preview", model, tools=None, messages=None): + super().__init__(tools=tools, messages=messages) self._model: str = model self._client = AsyncAzureOpenAI( @@ -58,16 +59,22 @@ class AzureLLMService(LLMService): api_version=api_version, ) - async def run_llm_async(self, messages) -> AsyncGenerator[str, None]: + async def run_llm_async(self, messages, tool_choice=None) -> AsyncGenerator[str, None]: messages_for_log = json.dumps(messages) self.logger.debug(f"Generating chat via azure: {messages_for_log}") - - chunks = await self._client.chat.completions.create(model=self._model, stream=True, messages=messages) + if self._tools: + tools = self._tools + else: + tools = None + start_time = time.time() + chunks = await self._client.chat.completions.create(model=self._model, stream=True, messages=messages, tools=tools, tool_choice=tool_choice) + self.logger.info(f"=== Azure OpenAI LLM TTFB: {time.time() - start_time}") async for chunk in chunks: if len(chunk.choices) == 0: continue - - if chunk.choices[0].delta.content: + if chunk.choices[0].delta.tool_calls: + yield chunk.choices[0].delta.tool_calls[0] + elif chunk.choices[0].delta.content: yield chunk.choices[0].delta.content async def run_llm(self, messages) -> str | None: diff --git a/src/dailyai/services/base_transport_service.py b/src/dailyai/services/base_transport_service.py index 3cec85d8e..4ca554645 100644 --- a/src/dailyai/services/base_transport_service.py +++ b/src/dailyai/services/base_transport_service.py @@ -220,7 +220,6 @@ class BaseTransportService(): self.interrupt() pipeline_task = asyncio.create_task(pipeline.run_pipeline()) started = False - continue if not started: await self.send_queue.put(StartFrame()) diff --git a/src/dailyai/services/daily_transport_service.py b/src/dailyai/services/daily_transport_service.py index c2a47c2dc..868c296b2 100644 --- a/src/dailyai/services/daily_transport_service.py +++ b/src/dailyai/services/daily_transport_service.py @@ -249,8 +249,9 @@ class DailyTransportService(BaseTransportService, EventHandler): participantId = message["participantId"] elif "session_id" in message: participantId = message["session_id"] - frame = TranscriptionQueueFrame(message["text"], participantId, message["timestamp"]) - asyncio.run_coroutine_threadsafe(self.receive_queue.put(frame), self._loop) + if self._my_participant_id and participantId != self._my_participant_id: + frame = TranscriptionQueueFrame(message["text"], participantId, message["timestamp"]) + asyncio.run_coroutine_threadsafe(self.receive_queue.put(frame), self._loop) def on_transcription_stopped(self, stopped_by, stopped_by_error): pass diff --git a/src/dailyai/services/open_ai_services.py b/src/dailyai/services/open_ai_services.py index 05e1e9860..13457979b 100644 --- a/src/dailyai/services/open_ai_services.py +++ b/src/dailyai/services/open_ai_services.py @@ -1,6 +1,7 @@ import aiohttp from PIL import Image import io +import time from openai import AsyncOpenAI import json @@ -10,8 +11,8 @@ from dailyai.services.ai_services import LLMService, ImageGenService class OpenAILLMService(LLMService): - def __init__(self, *, api_key, model="gpt-4"): - super().__init__() + def __init__(self, *, api_key, model="gpt-4", tools=None, messages=None): + super().__init__(tools=tools, messages=messages) self._model = model self._client = AsyncOpenAI(api_key=api_key) @@ -19,19 +20,26 @@ class OpenAILLMService(LLMService): return await self._client.chat.completions.create( stream=stream, messages=messages, - model=self._model + model=self._model, + tools=self._tools ) - async def run_llm_async(self, messages) -> AsyncGenerator[str, None]: + async def run_llm_async(self, messages, tool_choice=None) -> AsyncGenerator[str, None]: messages_for_log = json.dumps(messages) self.logger.debug(f"Generating chat via openai: {messages_for_log}") - - chunks = await self._client.chat.completions.create(model=self._model, stream=True, messages=messages) + if self._tools: + tools = self._tools + else: + tools = None + start_time = time.time() + chunks = await self._client.chat.completions.create(model=self._model, stream=True, messages=messages, tools=tools, tool_choice=tool_choice) + self.logger.info(f"=== OpenAI LLM TTFB: {time.time() - start_time}") async for chunk in chunks: if len(chunk.choices) == 0: continue - - if chunk.choices[0].delta.content: + if chunk.choices[0].delta.tool_calls: + yield chunk.choices[0].delta.tool_calls[0] + elif chunk.choices[0].delta.content: yield chunk.choices[0].delta.content async def run_llm(self, messages) -> str | None: diff --git a/src/examples/foundational/07-interruptible.py b/src/examples/foundational/07-interruptible.py index 9c22a1f17..2f33503b3 100644 --- a/src/examples/foundational/07-interruptible.py +++ b/src/examples/foundational/07-interruptible.py @@ -1,7 +1,7 @@ import asyncio import aiohttp import os -from dailyai.pipeline.aggregators import LLMAssistantContextAggregator, LLMResponseAggregator, LLMUserContextAggregator +from dailyai.pipeline.aggregators import LLMAssistantContextAggregator, LLMResponseAggregator, LLMUserContextAggregator, UserResponseAggregator from dailyai.pipeline.pipeline import Pipeline from dailyai.services.ai_services import FrameLogger @@ -49,8 +49,8 @@ async def main(room_url: str, token): post_processor=LLMResponseAggregator( messages ), - pre_processor=LLMUserContextAggregator( - messages, transport._my_participant_id, complete_sentences=False + pre_processor=UserResponseAggregator( + messages ), ) diff --git a/src/examples/foundational/14-patient-intake.py b/src/examples/foundational/14-patient-intake.py new file mode 100644 index 000000000..7ddb287d9 --- /dev/null +++ b/src/examples/foundational/14-patient-intake.py @@ -0,0 +1,376 @@ +import aiohttp +import asyncio +import json +import random +import os +import re +import wave +from typing import AsyncGenerator +from PIL import Image + +from dailyai.pipeline.pipeline import Pipeline +from dailyai.services.daily_transport_service import DailyTransportService +from dailyai.services.azure_ai_services import AzureLLMService, AzureTTSService +from dailyai.services.open_ai_services import OpenAILLMService +from dailyai.services.deepgram_ai_services import DeepgramTTSService +from dailyai.services.elevenlabs_ai_service import ElevenLabsTTSService +from dailyai.pipeline.aggregators import LLMAssistantContextAggregator, LLMContextAggregator, LLMUserContextAggregator, UserResponseAggregator, LLMResponseAggregator +from support.runner import configure +from dailyai.pipeline.frames import LLMMessagesQueueFrame, TranscriptionQueueFrame, Frame, TextFrame, LLMFunctionCallFrame, LLMResponseEndFrame, StartFrame, AudioFrame, SpriteFrame, ImageFrame +from dailyai.services.ai_services import FrameLogger, AIService + +import logging +logging.basicConfig(level=logging.DEBUG) + +sounds = {} +sound_files = [ + 'clack-short.wav', + 'clack.wav', + 'clack-short-quiet.wav' +] + +script_dir = os.path.dirname(__file__) + +for file in sound_files: + # Build the full path to the image file + full_path = os.path.join(script_dir, "assets", file) + # Get the filename without the extension to use as the dictionary key + filename = os.path.splitext(os.path.basename(full_path))[0] + # Open the image and convert it to bytes + with wave.open(full_path) as audio_file: + sounds[file] = audio_file.readframes(-1) + + +steps = [ + { + "prompt": "Start by introducing yourself. Then, ask the user to confirm their identity by telling you their birthday, including the year. When they answer with their birthday, call the verify_birthday function.", + "run_async": False, + "failed": "The user provided an incorrect birthday. Ask them for their birthday again. When they answer, call the verify_birthday function.", "tools": [{ + "type": "function", + "function": { + "name": "verify_birthday", + "description": "Use this function to verify the user has provided their correct birthday.", + "parameters": { + "type": "object", + "properties": { + "birthday": { + "type": "string", + "description": "The user's birthdate, including the year. The user can provide it in any format, but convert it to YYYY-MM-DD format to call this function." + } + } + } + } + }]}, + { + "prompt": "Next, thank the user for confirming their identity, then ask the user to list their current prescriptions. Each prescription needs to have a medication name and a dosage. Do not call the list_prescriptions function with any unknown dosages.", + "run_async": True, + "tools": [{ + "type": "function", + "function": { + "name": "list_prescriptions", + "description": "Once the user has provided a list of their prescription medications, call this function.", + "parameters": { + "type": "object", + "properties": { + "prescriptions": { + "type": "array", + "items": { + "type": "object", + "properties": { + "medication": { + "type": "string", + "description": "The medication's name" + }, + "dosage": { + "type": "string", + "description": "The prescription's dosage" + } + } + } + } + } + } + } + }] + }, + { + "prompt": "Next, ask the user if they have any allergies. Once they have listed their allergies or confirmed they don't have any, call the list_allergies function.", + "run_async": True, + "tools": [ + { + "type": "function", + "function": { + "name": "list_allergies", + "description": "Once the user has provided a list of their allergies, call this function.", + "parameters": { + "type": "object", + "properties": { + "allergies": { + "type": "array", + "items": { + "type": "object", + "properties": { + "name": { + "type": "string", + "description": "What the user is allergic to" + } + } + } + } + } + } + } + } + ] + }, + { + "prompt": "Now ask the user if they have any medical conditions the doctor should know about. Once they've answered the question, call the list_conditions function.", + "run_async": True, + "tools": [ + { + "type": "function", + "function": { + "name": "list_conditions", + "description": "Once the user has provided a list of their medical conditions, call this function.", + "parameters": { + "type": "object", + "properties": { + "conditions": { + "type": "array", + "items": { + "type": "object", + "properties": { + "name": { + "type": "string", + "description": "The user's medical condition" + } + } + } + } + } + } + } + }, + ], + }, + { + "prompt": "Finally, ask the user the reason for their doctor visit today. Once they answer, call the list_visit_reasons function.", + "run_async": True, + "tools": [ + { + "type": "function", + "function": { + "name": "list_visit_reasons", + "description": "Once the user has provided a list of the reasons they are visiting a doctor today, call this function.", + "parameters": { + "type": "object", + "properties": { + "visit_reasons": { + "type": "array", + "items": { + "type": "object", + "properties": { + "name": { + "type": "string", + "description": "The user's reason for visiting the doctor" + } + } + } + } + } + } + } + } + ] + }, + {"prompt": "Now, thank the user and end the conversation.", + "run_async": True, "tools": []}, + {"prompt": "", "run_async": True, "tools": []} +] +current_step = 0 + + +class TranscriptFilter(AIService): + def __init__(self, bot_participant_id=None): + super().__init__() + self.bot_participant_id = bot_participant_id + + async def process_frame(self, frame: Frame) -> AsyncGenerator[Frame, None]: + if isinstance(frame, TranscriptionQueueFrame): + if frame.participantId != self.bot_participant_id: + yield frame + + +class ChecklistProcessor(AIService): + def __init__(self, messages, llm, tools, *args, **kwargs): + super().__init__(*args, **kwargs) + self._messages = messages + self._llm = llm + self._tools = tools + self._function_name = "" + self._arguments = "" + self._id = "You are Jessica, an agent for a company called Tri-County Health Services. Your job is to collect important information from the user before their doctor visit. You're talking to Chad Bailey. You should address the user by their first name and be polite and professional. You're not a medical professional, so you shouldn't provide any advice. Keep your responses short. Your job is to collect information to give to a doctor. Don't make assumptions about what values to plug into functions. Ask for clarification if a user response is ambiguous." + self._acks = ["One sec.", "Let me confirm that.", "Thanks.", "OK."] + + messages.append( + {"role": "system", "content": f"{self._id} {steps[0]['prompt']}"}) + + def verify_birthday(self, args): + return args['birthday'] == "1983-01-01" + + def list_prescriptions(self, args): + # print(f"--- Prescriptions: {args['prescriptions']}\n") + pass + + def list_allergies(self, args): + # print(f"--- Allergies: {args['allergies']}\n") + pass + + def list_conditions(self, args): + # print(f"--- Medical Conditions: {args['conditions']}") + pass + + def list_visit_reasons(self, args): + # print(f"Visit Reasons: {args['visit_reasons']}") + pass + + async def process_frame(self, frame: Frame) -> AsyncGenerator[Frame, None]: + global current_step + this_step = steps[current_step] + # TODO-CB: forcing a global here :/ + self._tools.clear() + self._tools.extend(this_step['tools']) + if isinstance(frame, LLMFunctionCallFrame) and frame.function_name: + print(f"... Preparing function call: {frame.function_name}") + self._function_name = frame.function_name + if this_step['run_async']: + # Get the LLM talking about the next step before getting the rest + # of the function call completion + current_step += 1 + # yield TextFrame(f"We should move on to Step {current_step}.") + self._messages.append({ + "role": "system", "content": steps[current_step]['prompt']}) + # yield LLMMessagesQueueFrame(self._messages) + yield LLMMessagesQueueFrame(self._messages) + async for frame in llm.process_frame(LLMMessagesQueueFrame(self._messages), tool_choice="none"): + yield frame + else: + # Insert a quick response while we run the function + # yield AudioFrame(sounds["clack-short-quiet.wav"]) + pass + elif isinstance(frame, LLMFunctionCallFrame) and frame.arguments: + self._arguments += frame.arguments + elif isinstance(frame, LLMResponseEndFrame): + + if self._function_name and self._arguments: + print( + f"--> Calling function: {self._function_name} with arguments:") + pretty_json = re.sub("\n", "\n ", json.dumps( + json.loads(self._arguments), indent=2)) + print(f"--> {pretty_json}\n") + fn = getattr(self, self._function_name) + result = fn(json.loads(self._arguments)) + self._function_name = "" + self._arguments = "" + if not this_step['run_async']: + if result: + current_step += 1 + # yield TextFrame(f"We should move on to Step {current_step}.") + self._messages.append({ + "role": "system", "content": steps[current_step]['prompt']}) + # yield LLMMessagesQueueFrame(self._messages) + yield LLMMessagesQueueFrame(self._messages) + async for frame in llm.process_frame(LLMMessagesQueueFrame(self._messages), tool_choice="none"): + yield frame + else: + self._messages.append({ + "role": "system", "content": this_step['failed']}) + # yield LLMMessagesQueueFrame(self._messages) + yield LLMMessagesQueueFrame(self._messages) + async for frame in llm.process_frame(LLMMessagesQueueFrame(self._messages), tool_choice="none"): + yield frame + print(f"<-- Verify result: {result}\n") + + else: + yield frame + + +async def main(room_url: str, token): + async with aiohttp.ClientSession() as session: + global transport + global llm + global tts + + transport = DailyTransportService( + room_url, + token, + "Intake Bot", + 5, + mic_enabled=True, + mic_sample_rate=16000, + camera_enabled=False, + start_transcription=True, + vad_enabled=True + ) + # TODO-CB: Go back to vad_enabled + + messages = [] + tools = [] + + # llm = AzureLLMService(api_key=os.getenv("AZURE_CHATGPT_API_KEY"), endpoint=os.getenv( + # "AZURE_CHATGPT_ENDPOINT"), model=os.getenv("AZURE_CHATGPT_MODEL")) + llm = OpenAILLMService(api_key=os.getenv( + "OPENAI_CHATGPT_API_KEY"), model="gpt-4-1106-preview", tools=tools) # gpt-4-1106-preview + # tts = AzureTTSService(api_key=os.getenv( + # "AZURE_SPEECH_API_KEY"), region=os.getenv("AZURE_SPEECH_REGION")) + tts = ElevenLabsTTSService(aiohttp_session=session, api_key=os.getenv( + "ELEVENLABS_API_KEY"), voice_id="XrExE9yKIg1WjnnlVkGX") # matilda + # tts = DeepgramTTSService(aiohttp_session=session, api_key=os.getenv( + # "DEEPGRAM_API_KEY"), voice="aura-asteria-en") + + # lca = LLMContextAggregator( + # messages=messages, bot_participant_id=transport._my_participant_id) + checklist = ChecklistProcessor(messages, llm, tools) + fl = FrameLogger("FRAME LOGGER 1:") + fl2 = FrameLogger("FRAME LOGGER 2:") + + @transport.event_handler("on_first_other_participant_joined") + async def on_first_other_participant_joined(transport): + fl = FrameLogger("first other participant") + # TODO-CB: Make sure this message gets into the context somehow + await tts.run_to_queue( + transport.send_queue, + llm.run([LLMMessagesQueueFrame(messages)]), + + ) + + async def handle_intake(): + pipeline = Pipeline( + processors=[ + fl, + llm, + fl2, + checklist, + tts + ] + ) + await transport.run_interruptible_pipeline(pipeline, + post_processor=LLMResponseAggregator( + messages + ), + pre_processor=UserResponseAggregator(messages) + ) + + + transport.transcription_settings["extra"]["endpointing"] = True + transport.transcription_settings["extra"]["punctuate"] = True + try: + await asyncio.gather(transport.run(), handle_intake()) + except (asyncio.CancelledError, KeyboardInterrupt): + print('whoops') + transport.stop() + + +if __name__ == "__main__": + (url, token) = configure() + asyncio.run(main(url, token)) diff --git a/src/examples/foundational/assets/clack-short-quiet.wav b/src/examples/foundational/assets/clack-short-quiet.wav new file mode 100644 index 000000000..f0580d11e Binary files /dev/null and b/src/examples/foundational/assets/clack-short-quiet.wav differ diff --git a/src/examples/foundational/assets/clack-short.wav b/src/examples/foundational/assets/clack-short.wav new file mode 100644 index 000000000..864994b28 Binary files /dev/null and b/src/examples/foundational/assets/clack-short.wav differ diff --git a/src/examples/foundational/assets/clack.wav b/src/examples/foundational/assets/clack.wav new file mode 100644 index 000000000..2f36164b3 Binary files /dev/null and b/src/examples/foundational/assets/clack.wav differ