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

Initial vanessa code

Browse Source
This commit is contained in:
James Hush
2024-11-19 12:53:07 +08:00
parent 9409546f90
commit 3680b4af21
2 changed files with 630 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

106
examples/bigly-debug.patch Normal file
View File

@@ -0,0 +1,106 @@
diff --git a/src/pipecat/processors/frame_processor.py b/src/pipecat/processors/frame_processor.py
index 9f614ba..e7d3e0a 100644
--- a/src/pipecat/processors/frame_processor.py
+++ b/src/pipecat/processors/frame_processor.py
@@ -245,7 +245,8 @@ class FrameProcessor:
logger.trace(f"Pushing {frame} upstream from {self} to {self._prev}")
await self._prev.queue_frame(frame, direction)
except Exception as e:
- logger.exception(f"Uncaught exception in {self}: {e}")
+ logger.exception(f"<> Uncaught exception in {self}: {e}")
+ raise Exception("internal_push_frame") from e
def __create_input_task(self):
self.__input_queue = asyncio.Queue()
diff --git a/src/pipecat/services/ai_services.py b/src/pipecat/services/ai_services.py
index e0f16e2..5c374a0 100644
--- a/src/pipecat/services/ai_services.py
+++ b/src/pipecat/services/ai_services.py
@@ -289,6 +289,7 @@ class TTSService(AIService):
async def process_frame(self, frame: Frame, direction: FrameDirection):
await super().process_frame(frame, direction)
+ print(f"_---_ai_services.py process_frame * frame: {frame}")
if isinstance(frame, TextFrame):
await self._process_text_frame(frame)
@@ -343,6 +344,7 @@ class TTSService(AIService):
await self._push_tts_frames(text)
async def _push_tts_frames(self, text: str):
+ print(f"_____ai_services.py * push_tts_frames str: {str}")
# Don't send only whitespace. This causes problems for some TTS models. But also don't
# strip all whitespace, as whitespace can influence prosody.
if not text.strip():
diff --git a/src/pipecat/services/cartesia.py b/src/pipecat/services/cartesia.py
index 7fad02f..276b172 100644
--- a/src/pipecat/services/cartesia.py
+++ b/src/pipecat/services/cartesia.py
@@ -137,6 +137,7 @@ class CartesiaTTSService(WordTTSService):
def _build_msg(
self, text: str = "", continue_transcript: bool = True, add_timestamps: bool = True
):
+ print(f"_____cartesia.py * _build_msg str: {str}")
voice_config = {}
voice_config["mode"] = "id"
voice_config["id"] = self._voice_id
diff --git a/src/pipecat/services/openai.py b/src/pipecat/services/openai.py
index b6927e8..df9715b 100644
--- a/src/pipecat/services/openai.py
+++ b/src/pipecat/services/openai.py
@@ -166,6 +166,7 @@ class BaseOpenAILLMService(LLMService):
params.update(self._settings["extra"])
chunks = await self._client.chat.completions.create(**params)
+ print(f"_____openai.py get_chat_completions * chunks: {chunks}")
return chunks
async def _stream_chat_completions(
@@ -202,6 +203,7 @@ class BaseOpenAILLMService(LLMService):
function_name = ""
arguments = ""
tool_call_id = ""
+ print(f"_____openai.py * _process_context: tool_call_id", tool_call_id)
await self.start_ttfb_metrics()
@@ -210,6 +212,7 @@ class BaseOpenAILLMService(LLMService):
)
async for chunk in chunk_stream:
+ # print(f"___*__openai.py * chunk: {chunk}")
if chunk.usage:
tokens = LLMTokenUsage(
prompt_tokens=chunk.usage.prompt_tokens,
@@ -237,7 +240,7 @@ class BaseOpenAILLMService(LLMService):
# We accumulate all the arguments for the rest of the streamed response, then when
# the response is done, we package up all the arguments and the function name and
# yield a frame containing the function name and the arguments.
-
+ print(f"___________________________openai.py * tool_call_id: {tool_call_id}")
tool_call = chunk.choices[0].delta.tool_calls[0]
if tool_call.index != func_idx:
functions_list.append(function_name)
@@ -503,6 +506,7 @@ class OpenAIAssistantContextAggregator(LLMAssistantContextAggregator):
async def process_frame(self, frame, direction):
await super().process_frame(frame, direction)
+ print(f"___<1>__openai.py * OpenAIAssistantContextAggregator process_frame frame : {frame}")
# See note above about not calling push_frame() here.
if isinstance(frame, StartInterruptionFrame):
self._function_calls_in_progress.clear()
@@ -525,11 +529,14 @@ class OpenAIAssistantContextAggregator(LLMAssistantContextAggregator):
elif isinstance(frame, OpenAIImageMessageFrame):
self._pending_image_frame_message = frame
await self._push_aggregation()
+ else:
+ print(f"___<2>__openai.py * OpenAIAssistantContextAggregator process_frame frame : {frame}")
async def _push_aggregation(self):
if not (
self._aggregation or self._function_call_result or self._pending_image_frame_message
):
+ print(f"_____openai.py * return from push_aggregation__________")
return
run_llm = False

524
examples/bigly.py Normal file
View File

@@ -0,0 +1,524 @@
#### run from in pipecat/examples/foundational
#
# Copyright (c) 2024, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import argparse
import asyncio
import os
import random
import sys
from functools import partial
import aiohttp
import requests
from dotenv import load_dotenv
from loguru import logger
from openai.types.chat import ChatCompletionToolParam
from runner import configure
from pipecat.audio.vad.silero import SileroVADAnalyzer, VADParams
from pipecat.frames.frames import EndFrame, LLMMessagesFrame, TextFrame
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineParams, PipelineTask
from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
from pipecat.services.cartesia import CartesiaTTSService
from pipecat.services.deepgram import DeepgramSTTService
from pipecat.services.openai import OpenAILLMService
from pipecat.transports.services.daily import DailyParams, DailyTransport
from pipecat.transports.services.helpers.daily_rest import DailyRESTHelper
from pipecat.transports.services.helpers.daily_rest import DailyRESTHelper
load_dotenv(override=True)
logger.remove(0)
logger.add(sys.stderr, level="DEBUG")
PARTICIPANT_ID = [""] # this will get filled in later
DEEPGRAM_API_KEY = os.getenv("DEEPGRAM_API_KEY")
DAILY_API_KEY = os.getenv("DAILY_API_KEY")
CARTESIA_API_KEY = os.getenv("CARTESIA_API_KEY")
DAILY_ROOM_NAME = "bigs" #### ENSURE THIS ROOM HAS enable_dialout SET (and exp)
TO = "+16104204483"
# FROM_DAILY_CALLER_ID = "+13373378501"
QUESTION_AFFIRMATION = [
"Excellent question!",
"That's a solid question!",
"Good question!",
"Great question!",
"I am glad you asked that!",
]
VOICEMAIL_EXAMPLES = [
"We are sorry, there is noone available to take your call...",
"Please leave a message for ",
"Please leave your name and phone and I'll get back to you as soon as I can",
"Your call has been forwarded to voicemail, the person you are trying to reach is not available",
"The person you're trying to reach is not available",
"Hey, it's (user's name) leave a message",
"Hey you reached (user's name), please leave a message",
"Hi/Hey I'm not available please leave a message",
"The number you're trying to reach...",
"I'll get back to you as soon as possible",
"This voicemail is not receiving any messages",
"My VM is full so If you need to reach me, please text me",
"Leave a message and I'll call you back",
"You've reached my cell phone, I'm not available",
"We are sorry, there is noneone available to take your call...",
]
async def configure(aiohttp_session: aiohttp.ClientSession):
(url, token, _) = await configure_with_args(aiohttp_session)
return (url, token)
async def configure_with_args(
aiohttp_session: aiohttp.ClientSession, parser: argparse.ArgumentParser | None = None
):
if not parser:
parser = argparse.ArgumentParser(description="Daily AI SDK Bot Sample")
parser.add_argument(
"-u", "--url", type=str, required=False, help="URL of the Daily room to join"
)
parser.add_argument(
"-k",
"--apikey",
type=str,
required=False,
help="Daily API Key (needed to create an owner token for the room)",
)
args, unknown = parser.parse_known_args()
#####
url = f"https://pc-34b1bdc94a7741719b57b2efb82d658e.daily.co/{DAILY_ROOM_NAME}"
# url = f"https://biglysales-team.daily.co/{DAILY_ROOM_NAME}"
key = DAILY_API_KEY
print(f"_____bigly.py * key: {key}")
if not url:
raise Exception(
"No Daily room specified. use the -u/--url option from the command line, or set DAILY_SAMPLE_ROOM_URL in your environment to specify a Daily room URL."
)
if not key:
raise Exception(
"No Daily API key specified. use the -k/--apikey option from the command line, or set DAILY_API_KEY in your environment to specify a Daily API key, available from https://dashboard.daily.co/developers."
)
daily_rest_helper = DailyRESTHelper(
daily_api_key=key,
daily_api_url=os.getenv("DAILY_API_URL", "https://api.daily.co/v1"),
aiohttp_session=aiohttp_session,
)
# Create a meeting token for the given room with an expiration 1 hour in
# the future.
expiry_time: float = 60 * 60
token = await daily_rest_helper.get_token(url, expiry_time)
print(f"_____bigly.py * token: {token}")
return (url, token, args)
async def cancel_transport_input_task(transport_input) -> None:
transport_input._audio_task.cancel()
await transport_input._audio_task
def get_cartesia_static_response(text: str, voice_id: str, model_id: str, **kwargs) -> bytes:
"""
Makes an API call to Cartesia to generate TTS audio bytes.
Args:
text (str): The transcript text.
voice_id (str): The ID of the voice to be used.
model_id (str): The model ID for the TTS request.
**kwargs: Additional parameters like output format.
Returns:
bytes: The audio bytes returned by the API.
Raises:
ValueError: If the API request fails.
"""
try:
output_format = kwargs.get("output_format")
response = requests.post(
"https://api.cartesia.ai/tts/bytes",
headers={
"X-API-Key": CARTESIA_API_KEY,
"Cartesia-Version": "2024-06-10",
"Content-Type": "application/json",
},
json={
"model_id": model_id,
"transcript": text,
"voice": {"mode": "id", "id": voice_id},
"output_format": output_format,
},
)
response.raise_for_status()
logger.info("Cartesia TTS response cached.")
return response.content
except Exception as error:
logger.opt(exception=True).error("Error Occurred while getting Audio Bytes from Cartesia")
raise error
async def say_agent_response(
end_response,
transport,
audio_end_buffer: float,
cancel_input_audio: bool = True,
) -> None:
"""
Sends the agent's audio response via the specified transport and adds a buffer delay after playback.
Optionally cancels any ongoing audio input tasks before playing the response.
Args:
end_response (bytes): The audio response to be played.
transport (Union[DailyTransport, FastAPIWebsocketTransport]): The transport handling audio I/O.
audio_end_buffer (int): Time (in seconds) to wait after playing the audio.
cancel_input_audio (bool, optional): Whether to cancel ongoing audio input tasks. Defaults to True.
Retries:
Retries up to 3 times in case of an exception, with a 2-second delay between attempts.
"""
try:
if cancel_input_audio and transport:
logger.info("Canceling the Audio Input Task")
await cancel_transport_input_task(transport._input)
await transport._output.write_raw_audio_frames(end_response)
await asyncio.sleep(audio_end_buffer)
except AttributeError:
logger.info("Audio Input Transport already cancelled")
except Exception as error:
logger.opt(exception=True).error("Error Occurred while Uttering Agent Response.")
raise error
async def get_meeting_dates(
function_name: str,
tool_call_id: str,
_args,
llm,
context: OpenAILLMContext,
result_callback,
) -> None:
logger.info("Invoking `get_meeting_dates` tool with argument {_args}", _args=_args)
await result_callback("""
"1- 8AM Eastern on November 16 2024\n\n2- 9AM Eastern on November 16 2024\n\n3- 10AM Eastern on November 16 2024\n\n4- 11AM Eastern on November 16 2024\n\n5- 12PM Eastern on November 16 2024\n\n6- 1PM Eastern on November 16 2024\n\n7- 2PM Eastern on November 16 2024\n\n8- 3PM Eastern on November 16 2024\n\n9- 4PM Eastern on November 16 2024\n\n10- 5PM Eastern on November 16 2024\n\n11- 6PM Eastern on November 16 2024\n\n12- 7PM Eastern on November 16 2024\n\n"
""")
async def transfer_call(
function_name: str,
tool_call_id: str,
_args,
llm,
context: OpenAILLMContext,
result_callback,
) -> None:
await result_callback("CALL TRANSFERED")
async def voicemail(
function_name: str, tool_call_id: str, _args, llm, context, result_callback, transport
) -> None:
logger.info("Invoking `voicemail` tool with argument {_args}", _args=_args)
await transport.stop_dialout(PARTICIPANT_ID[0])
await result_callback("VOICEMAIL DETECTED")
async def get_knowledge_base(
function_name: str, tool_call_id: str, _args, llm, context: OpenAILLMContext, result_callback
) -> None:
logger.info("Invoking `get_knowledge_base` tool with argument {_args}", _args=_args)
kb_call_reason = _args.get("reason", "GENERAL_QUESTION")
if kb_call_reason == "GENERAL_QUESTION":
# await llm.push_frame("aldkjfls oaogi8ovs(*YVSDY*( &*Tqr))")
await llm.push_frame(TextFrame(random.choice(QUESTION_AFFIRMATION)))
await result_callback(
"""Sure! Here are some random facts that could be associated with a solar panel company: 1. **History**: The company was founded in 2008 by a group of renewable energy enthusiasts who wanted to make solar power more accessible to homeowners and businesses. 2. **Headquarters**: Their headquarters is a net-zero energy building powered entirely by their own solar panels, showcasing their commitment to sustainability. 3. **Products**: They produce three main types of solar panels: monocrystalline, polycrystalline, and thin-film, catering to different customer needs and budgets. 4. **Innovation**: The company holds patents for advanced solar cell technology that increases efficiency by 20% compared to industry standards. 5. **Global Reach**: They have installed solar systems in over 40 countries and have manufacturing plants on three continents.6. **Community Impact**: For every 100 solar panels sold, they donate a panel to a school or community center in underprivileged areas. 7. **Workforce**: The company employs over 5,000 people, 40% of whom are in research and development roles. 8. **Recognition**: They won the “Green Energy Innovator of the Year” award in 2022 for their work on solar panels made from recycled materials. 9. **Sustainability**: Their panels are designed to last 25+ years and are 95% recyclable at the end of their life cycle. 10. **Customer Perks**: They offer a 25-year warranty and real-time monitoring systems that allow users to track energy production via an app. 11. **Mission Statement**: "Empowering the world with clean energy, one panel at a time." 12. **Energy Production**: The combined output of all their installations generates enough electricity to power over 2 million homes annually. 13. **R&D Efforts**: They are actively working on integrating solar panels into everyday items like backpacks and electric vehicles. 14. **Solar Farms**: The company has partnered with governments to develop large-scale solar farms, including one that spans over 10,000 acres. 15. **Future Goals**: By 2030, they aim to make solar power the most affordable energy source worldwide. Would you like these customized for a specific scenario?"""
)
# raise RuntimeError("<><><><><>get_knowledge_base")
async def end_call(
function_name: str,
tool_call_id: str,
_args,
llm,
context: OpenAILLMContext,
result_callback,
voice_provider: str,
voice_id: str,
transport: DailyTransport,
) -> None:
logger.info("Invoking `end_call` tool with argument {_args}", _args=_args)
end_call_sentence = _args.get("end_call_sentence", "Thank you for your time have a great day")
estimated_time_end_call_sentence = 15
logger.info(
"Estimated End Call Sentence Time is {estimated_time_end_call_sentence}",
estimated_time_end_call_sentence=estimated_time_end_call_sentence,
)
if voice_provider == "cartesia":
await say_agent_response(
get_cartesia_static_response(
end_call_sentence,
voice_id,
"sonic-english",
output_format={"container": "raw", "encoding": "pcm_s16le", "sample_rate": 16000},
),
transport,
estimated_time_end_call_sentence,
True,
)
await transport.stop_dialout(PARTICIPANT_ID[0])
await result_callback("CALL ENDED BY ASSISTANT")
async def main():
async with aiohttp.ClientSession() as session:
(room_url, token) = await configure(session)
stt = DeepgramSTTService(
api_key=DEEPGRAM_API_KEY,
audio_passthrough=False,
)
transport = DailyTransport(
room_url,
token,
"Test-Bot",
DailyParams(
api_key=DAILY_API_KEY,
audio_in_enabled=True,
audio_out_enabled=True,
vad_audio_passthrough=True,
vad_enabled=True,
transcription_enabled=False,
vad_analyzer=SileroVADAnalyzer(params=VADParams(start_secs=0.2)),
),
)
print(f"_____bigly.py * transport: {transport}")
tts = CartesiaTTSService(
api_key=CARTESIA_API_KEY,
voice_id="7360f116-6306-4e9a-b487-1235f35a0f21",
)
llm = OpenAILLMService(api_key=os.environ["OPENAI_API_KEY"], model="gpt-4o")
llm.register_function("get_knowledge_base", get_knowledge_base)
llm.register_function(
"end_call",
partial(
end_call,
transport=transport,
voice_provider="cartesia",
voice_id="7360f116-6306-4e9a-b487-1235f35a0f21",
),
)
llm.register_function("transfer_call", transfer_call)
llm.register_function("voicemail", partial(voicemail, transport=transport))
llm.register_function("get_meeting_dates", get_meeting_dates)
tools = [
ChatCompletionToolParam(
type="function",
function={
"name": "get_knowledge_base",
"description": """""Always Use this tool:
- When User have Questions about Company
- If user is not showing interest and you need to convince them by using information about Company
- When User have any General Query and Questions
""",
"strict": True, # type: ignore[typeddict-unknown-key]
"parameters": {
"type": "object",
"properties": {
"question": {
"type": "string",
"description": "User Question/Concern about Company",
},
"reason": {
"type": "string",
"enum": ["CONVINCE", "GENERAL_QUESTION"],
"description": "Reason why you are using this tool. This can only be one of the category: `CONVINCE`, `GENERAL_QUESTION`.\
`CONVINCE` would be in case you are convincing the user otherwise it is `GENERAL_QUESTION`",
},
},
"additionalProperties": False,
"required": ["question", "reason"],
},
},
),
ChatCompletionToolParam(
type="function",
function={
"name": "end_call",
"description": "Use this tool to end the call",
"strict": True, # type: ignore[typeddict-unknown-key]
"parameters": {
"type": "object",
"properties": {
"end_call_sentence": {
"type": "string",
"description": "The End Call Sentence that needs to be utter by the AI",
}
},
"additionalProperties": False,
"required": ["end_call_sentence"],
},
},
),
ChatCompletionToolParam(
type="function",
function={
"name": "transfer_call",
"description": "Use this tool to transfer the call",
"strict": True, # type: ignore[typeddict-unknown-key]
"parameters": {
"type": "object",
"properties": {
"transfer_number_sentence": {
"type": "string",
"description": "The sentence that AI needs to speak before transfering the call",
},
"transfer_number": {
"type": "string",
"description": "The number to use to which we need transfer the call",
},
},
"additionalProperties": False,
"required": ["transfer_number", "transfer_number_sentence"],
},
},
),
ChatCompletionToolParam(
type="function",
function={
"name": "voicemail",
"description": f"Use this tool if you reach voicemail. Here is some examples: {'\n'.join(VOICEMAIL_EXAMPLES)}",
"parameters": {},
},
),
ChatCompletionToolParam(
type="function",
function={
"name": "get_meeting_dates",
"description": "Use this tool to get the meeting dates in order to schedule a meeting date with the user. The output of this\
tool are slots which you need to recommend to the user. The slots are given in numerical list in ascending order\
(earlier slots to later slots)",
"parameters": {},
},
),
]
messages = [
{
"role": "system",
"content": """Main Prompt""",
},
]
context = OpenAILLMContext(messages, tools)
context_aggregator = llm.create_context_aggregator(context)
pipeline = Pipeline(
[
transport.input(), # Transport user input
stt,
context_aggregator.user(), # User responses
llm, # LLM
tts, # TTS
transport.output(), # Transport bot output
context_aggregator.assistant(), # Assistant spoken responses
]
)
task = PipelineTask(
pipeline,
PipelineParams(
allow_interruptions=True,
enable_metrics=False, ####
enable_usage_metrics=False, #####
report_only_initial_ttfb=True,
),
)
@transport.event_handler("on_call_state_updated")
async def on_call_state_updated(transport, state: str) -> None:
logger.info("Call State Updated, state: {state}", state=state)
async def _dialout_retry_handler() -> None:
try:
for i in range(3):
logger.info("Attempting a Dial-Out, Attempt: {attempt}", attempt=i + 1)
await transport.start_dialout(
{"phoneNumber": TO, "video": False}
# {"phoneNumber": TO, "callerId": FROM_DAILY_CALLER_ID, "video": False}
)
await asyncio.sleep(15)
current_participant_count = transport.participant_counts()
if current_participant_count["present"] >= 2:
return
raise Exception("Unable to perform a dial-out for Daily-Co ROOM")
except Exception as e:
raise e
async def _dialout_task_exception(task: asyncio.Task) -> None:
if task.exception():
await task.queue_frames([EndFrame()])
else:
logger.info("Dial-out completed successfully.")
def _handle_dialout_completion(task: asyncio.Task) -> None:
asyncio.create_task(_dialout_task_exception(task))
if state == "joined":
task = transport.input().get_event_loop().create_task(_dialout_retry_handler())
task.add_done_callback(_handle_dialout_completion)
# Event handler for on_first_participant_joined
@transport.event_handler("on_first_participant_joined")
async def on_first_participant_joined(transport, participant) -> None:
logger.info(
"First Participant Joined with ID {participant_id}",
participant_id=participant["id"],
)
PARTICIPANT_ID[0] = participant["id"]
await transport.capture_participant_transcription(participant["id"])
@transport.event_handler("on_dialout_error")
async def on_dialout_error(transport, cdata) -> None:
logger.info("Dial-Out error: {data}", data=cdata)
await task.queue_frames([LLMMessagesFrame(messages)])
# Event handler for on_dialout_answered
@transport.event_handler("on_dialout_answered")
async def on_dialout_answered(transport, cdata) -> None:
logger.info("Dial-Out Answered with data as follow: {data}", data=cdata)
await task.queue_frames([LLMMessagesFrame(messages)])
# Event handler for on_participant_left
@transport.event_handler("on_participant_left")
async def on_participant_left(transport, participant, reason) -> None:
logger.info("Following Participant Left {participant}", participant=participant)
await task.queue_frames([EndFrame()])
runner = PipelineRunner()
await runner.run(task)
if __name__ == "__main__":
asyncio.run(main())