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

wip, committing before importing my conversation wrapper

Browse Source
This commit is contained in:
Chad Bailey
2024-02-29 14:16:00 -06:00
parent a8d618ede1
commit af4ab95713
9 changed files with 221 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
src/dailyai/conversation_wrappers.py
View File

@@ -3,7 +3,7 @@ import copy
import functools
from typing import AsyncGenerator, Awaitable, Callable
from dailyai.queue_aggregators import LLMAssistantContextAggregator, LLMContextAggregator, LLMUserContextAggregator
from dailyai.queue_frame import EndStreamQueueFrame, QueueFrame, TranscriptionQueueFrame
from dailyai.queue_frame import EndStreamQueueFrame, QueueFrame, TranscriptionQueueFrame, UserStartedSpeakingFrame
class InterruptibleConversationWrapper:
@@ -63,9 +63,10 @@ class InterruptibleConversationWrapper:
if frame.participantId == self._my_participant_id:
continue
if current_response_task:
if current_response_task and isinstance(frame, UserStartedSpeakingFrame):
current_response_task.cancel()
self._interrupt()
self._current_phrase += " " + frame.text
current_llm_messages = copy.deepcopy(self._messages)

10
src/dailyai/queue_aggregators.py
View File

@@ -52,7 +52,7 @@ class LLMContextAggregator(AIService):
if isinstance(frame, TranscriptionQueueFrame):
if frame.participantId == self.bot_participant_id:
return
print(f"@@@ tma got a frame: {frame.text}")
# The common case for "pass through" is receiving frames from the LLM that we'll
# use to update the "assistant" LLM messages, but also passing the text frames
# along to a TTS service to be spoken to the user.
@@ -68,8 +68,8 @@ class LLMContextAggregator(AIService):
self.messages.append(
{"role": self.role, "content": self.sentence})
self.sentence = ""
for message in self.messages:
print(f"{message['role']}: {message['content']}")
# for message in self.messages:
# print(f"{message['role']}: {message['content']}")
yield LLMMessagesQueueFrame(self.messages)
else:
# type: ignore -- the linter thinks this isn't a TextQueueFrame, even
@@ -81,8 +81,8 @@ class LLMContextAggregator(AIService):
# Send any dangling words that weren't finished with punctuation.
if self.complete_sentences and self.sentence:
self.messages.append({"role": self.role, "content": self.sentence})
for message in self.messages:
print(f"{message['role']}: {message['content']}")
# for message in self.messages:
# print(f"{message['role']}: {message['content']}")
yield LLMMessagesQueueFrame(self.messages)

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

@@ -43,7 +43,7 @@ class AIService:
frames: Iterable[QueueFrame]
| AsyncIterable[QueueFrame]
| asyncio.Queue[QueueFrame],
) -> AsyncGenerator[QueueFrame, None]:
**kwargs) -> AsyncGenerator[QueueFrame, None]:
try:
if isinstance(frames, AsyncIterable):
async for frame in frames:

7
src/dailyai/services/base_transport_service.py
View File

@@ -217,7 +217,6 @@ class BaseTransportService():
new_confidence = model(
torch.from_numpy(audio_float32), 16000).item()
speaking = new_confidence > 0.5
if speaking:
match self._vad_state:
case VADState.QUIET:
@@ -240,14 +239,16 @@ class BaseTransportService():
self._vad_stopping_count += 1
if self._vad_state == VADState.STARTING and self._vad_starting_count >= self._vad_start_frames:
print("##### VAD START")
asyncio.run_coroutine_threadsafe(
self.receive_queue.put(
UserStartedSpeakingFrame()), self._loop
)
# self.interrupt()
self.interrupt()
self._vad_state = VADState.SPEAKING
self._vad_starting_count = 0
if self._vad_state == VADState.STOPPING and self._vad_stopping_count >= self._vad_stop_frames:
print("##### VAD STOP")
asyncio.run_coroutine_threadsafe(
self.receive_queue.put(
UserStoppedSpeakingFrame()), self._loop
@@ -264,6 +265,7 @@ class BaseTransportService():
break
def interrupt(self):
print(f"!!!!! INTERRUPT")
self._is_interrupted.set()
async def get_receive_frames(self) -> AsyncGenerator[QueueFrame, None]:
@@ -333,7 +335,6 @@ class BaseTransportService():
raise Exception("Unknown type in output queue")
for frame in frames:
print(f"got frame of type: {type(frame)}")
if isinstance(frame, EndStreamQueueFrame):
self._logger.info("Stopping frame consumer thread")
self._threadsafe_send_queue.task_done()

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

@@ -13,8 +13,8 @@ class OpenAILLMService(LLMService):
def __init__(self, *, api_key, model="gpt-4", tools=None):
super().__init__()
self._model = model
self._client = AsyncOpenAI(api_key=api_key)
self._tools = tools
self._client = AsyncOpenAI(api_key=api_key)
async def get_response(self, messages, stream):
return await self._client.chat.completions.create(
@@ -27,8 +27,16 @@ class OpenAILLMService(LLMService):
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, tools=self._tools, tool_choice=tool_choice)
print("---")
print(f"tools: {self._tools}")
print("---")
print(f"messages: {messages_for_log}")
print("-----")
if self._tools:
tools = self._tools
else:
tools = None
chunks = await self._client.chat.completions.create(model=self._model, stream=True, messages=messages, tools=tools, tool_choice=tool_choice)
async for chunk in chunks:
if len(chunk.choices) == 0:
continue

225
src/examples/foundational/06b-patient-intake.py
View File

@@ -1,7 +1,9 @@
import aiohttp
import asyncio
import json
import random
import os
import wave
from typing import AsyncGenerator
from dailyai.services.daily_transport_service import DailyTransportService
@@ -11,13 +13,31 @@ from dailyai.services.deepgram_ai_services import DeepgramTTSService
from dailyai.services.elevenlabs_ai_service import ElevenLabsTTSService
from dailyai.queue_aggregators import LLMAssistantContextAggregator, LLMContextAggregator, LLMUserContextAggregator
from support.runner import configure
from dailyai.queue_frame import LLMMessagesQueueFrame, TranscriptionQueueFrame, QueueFrame, TextQueueFrame, LLMFunctionCallFrame, LLMResponseEndQueueFrame, StartStreamQueueFrame
from dailyai.queue_frame import LLMMessagesQueueFrame, TranscriptionQueueFrame, QueueFrame, TextQueueFrame, LLMFunctionCallFrame, LLMResponseEndQueueFrame, StartStreamQueueFrame, AudioQueueFrame
from dailyai.services.ai_services import FrameLogger, AIService
from dailyai.conversation_wrappers import InterruptibleConversationWrapper
import logging
logging.basicConfig(level=logging.ERROR)
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)
tools = [
{
"type": "function",
@@ -137,6 +157,152 @@ tools = [
}
]
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, double-check to make sure they don't have any other health concerns. After that, 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):
@@ -151,27 +317,18 @@ class TranscriptFilter(AIService):
class ChecklistProcessor(AIService):
def __init__(self, messages, llm, *args, **kwargs):
def __init__(self, messages, llm, tools, *args, **kwargs):
super().__init__(*args, **kwargs)
self._current_step = 0
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 Advanced Optimum Health Solution Specialists. Your job is to collect important information from the user before they visit a doctor. 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."
self._id = "You are Jessica, an agent for a company called Tri-County Advanced Optimum Health Solution Specialists. Your job is to collect important information from the user before they visit a doctor. 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."]
self._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."},
{"prompt": "You've already confirmed the user's birthday, so don't call the verify_birthday function. Ask the user to list their current prescriptions. If the user responds with one or two prescriptions, ask them to confirm it's the complete list. Make sure each medication also includes the dosage. Once the user has provided all their prescriptions along with dosages, call the list_prescriptions function. Do not call list_prescriptions with any unknown dosages", "run_async": True},
{"prompt": "Don't call the verify_birthday or list_prescription functions. 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},
{"prompt": "Don't call the verify_birthday, list_allergies, or list_prescriptions functions. 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},
{"prompt": "Ask the user the reason for their doctor visit today. Once they answer, double-check to make sure they don't have any other health concerns. After that, call the list_visit_reasons function.", "run_async": True},
{"prompt": "Now, thank the user and end the conversation.", "run_async": True},
{"prompt": "", "run_async": True}
]
messages.append(
{"role": "system", "content": f"{self._id} {self._steps[0]}"})
{"role": "system", "content": f"{self._id} {steps[0]['prompt']}"})
def verify_birthday(self, args):
return args['birthday'] == "1983-08-19"
@@ -189,26 +346,35 @@ class ChecklistProcessor(AIService):
print(f"Visit Reasons: {args['visit_reasons']}")
async def process_frame(self, frame: QueueFrame) -> AsyncGenerator[QueueFrame, None]:
this_step = self._steps[self._current_step]
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"FUNCTION CALL: {frame}")
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
self._current_step += 1
# yield TextQueueFrame(f"We should move on to Step {self._current_step}.")
current_step += 1
# yield TextQueueFrame(f"We should move on to Step {current_step}.")
self._messages.append({
"role": "system", "content": self._steps[self._current_step]['prompt']})
"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 AudioQueueFrame(sounds["clack-short-quiet.wav"])
elif isinstance(frame, LLMFunctionCallFrame) and frame.arguments:
self._arguments += frame.arguments
elif isinstance(frame, LLMResponseEndQueueFrame):
print(
f"got a response end. function_name is {self._function_name}, arguments is {self._arguments}")
f"%%% got a response end. function_name is {self._function_name}, arguments is {self._arguments}")
print(f"%%%% messages is {self._messages}")
if self._function_name and self._arguments:
fn = getattr(self, self._function_name)
@@ -218,10 +384,10 @@ class ChecklistProcessor(AIService):
self._arguments = ""
if not this_step['run_async']:
if result:
self._current_step += 1
# yield TextQueueFrame(f"We should move on to Step {self._current_step}.")
current_step += 1
# yield TextQueueFrame(f"We should move on to Step {current_step}.")
self._messages.append({
"role": "system", "content": self._steps[self._current_step]['prompt']})
"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"):
@@ -253,24 +419,27 @@ async def main(room_url: str, token):
mic_enabled=True,
mic_sample_rate=16000,
camera_enabled=False,
start_transcription=True
start_transcription=True,
vad_enabled=True
)
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", tools=tools)
"OPENAI_CHATGPT_API_KEY"), model="gpt-4-turbo-preview", tools=tools)
# 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=os.getenv("DEEPGRAM_VOICE"))
messages = [
]
tma_in = LLMUserContextAggregator(
messages, transport._my_participant_id)
tma_out = LLMAssistantContextAggregator(
messages, transport._my_participant_id)
checklist = ChecklistProcessor(messages, llm)
checklist = ChecklistProcessor(messages, llm, tools)
fl = FrameLogger("got transcript")
fl2 = FrameLogger("just above the checklist")

BIN
src/examples/foundational/assets/clack-short-quiet.wav Normal file
View File

Binary file not shown.

BIN
src/examples/foundational/assets/clack-short.wav Normal file
View File

Binary file not shown.

BIN
src/examples/foundational/assets/clack.wav Normal file
View File

Binary file not shown.