diff --git a/CHANGELOG.md b/CHANGELOG.md index 919bde5ef..1324e2dbb 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,11 +5,22 @@ All notable changes to **Pipecat** will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). - ## [Unreleased] ### Added +- `DeepgramTTSService` accepts `base_url` argument again, allowing you to + connect to an on-prem service. + +- Added `LLMUserAggregatorParams` and `LLMAssistantAggregatorParams` which allow + you to control aggregator settings. You can now pass these arguments when + creating aggregator pairs with `create_context_aggregator()`. + +- Added `previous_text` context support to ElevenLabsHttpTTSService, improving + speech consistency across sentences within an LLM response. + +- Added word/timestamp pairs to `ElevenLabsHttpTTSService`. + - It is now possible to disable `SoundfileMixer` when created. You can then use `MixerEnableFrame` to dynamically enable it when necessary. @@ -21,13 +32,37 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Changed +- Daily's REST helpers now include an `eject_at_token_exp` param, which ejects + the user when their token expires. This new parameter defaults to False. + Also, the default value for `enable_prejoin_ui` changed to False and + `eject_at_room_exp` changed to False. + +- `OpenAILLMService` and `OpenPipeLLMService` now use `gpt-4.1` as their + default model. + - `SoundfileMixer` constructor arguments need to be keywords. +### Deprecated + +- `DeepgramSTTService` parameter `url` is now deprecated, use `base_url` + instead. + +### Removed + +- Parameters `user_kwargs` and `assistant_kwargs` when creating a context + aggregator pair using `create_context_aggregator()` have been removed. Use + `user_params` and `assistant_params` instead. + ### Fixed +- Fixed a `TavusVideoService` issue that was causing audio choppiness. + - Fixed an issue in `SmallWebRTCTransport` where an error was thrown if the client did not create a video transceiver. +- Fixed an issue where LLM input parameters were not working and applied correctly in `GoogleVertexLLMService`, causing + unexpected behavior during inference. + ## [0.0.63] - 2025-04-11 ### Added diff --git a/examples/canonical-metrics/bot.py b/examples/canonical-metrics/bot.py index 513de9b14..d14f4149e 100644 --- a/examples/canonical-metrics/bot.py +++ b/examples/canonical-metrics/bot.py @@ -72,7 +72,7 @@ async def main(): # voice_id="gD1IexrzCvsXPHUuT0s3", ) - llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o") + llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) messages = [ { diff --git a/examples/chatbot-audio-recording/bot.py b/examples/chatbot-audio-recording/bot.py index 92644301e..c3e20ecfe 100644 --- a/examples/chatbot-audio-recording/bot.py +++ b/examples/chatbot-audio-recording/bot.py @@ -95,7 +95,7 @@ async def main(): # voice_id="gD1IexrzCvsXPHUuT0s3", ) - llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o") + llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) messages = [ { diff --git a/examples/deployment/flyio-example/bot.py b/examples/deployment/flyio-example/bot.py index 2e610cb23..7e5bef3ee 100644 --- a/examples/deployment/flyio-example/bot.py +++ b/examples/deployment/flyio-example/bot.py @@ -53,7 +53,7 @@ async def main(room_url: str, token: str): voice_id=os.getenv("ELEVENLABS_VOICE_ID", ""), ) - llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o") + llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) messages = [ { diff --git a/examples/deployment/modal-example/bot.py b/examples/deployment/modal-example/bot.py index 49e1acc0a..327c6a5e4 100644 --- a/examples/deployment/modal-example/bot.py +++ b/examples/deployment/modal-example/bot.py @@ -43,7 +43,7 @@ async def main(room_url: str, token: str): api_key=os.getenv("CARTESIA_API_KEY", ""), voice_id="71a7ad14-091c-4e8e-a314-022ece01c121" ) - llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o") + llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) messages = [ { diff --git a/examples/deployment/pipecat-cloud-example/bot.py b/examples/deployment/pipecat-cloud-example/bot.py index a0e3f5cfe..d86c7f575 100644 --- a/examples/deployment/pipecat-cloud-example/bot.py +++ b/examples/deployment/pipecat-cloud-example/bot.py @@ -61,7 +61,7 @@ async def main(room_url: str, token: str): api_key=os.getenv("CARTESIA_API_KEY"), voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22" ) - llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o") + llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) messages = [ { diff --git a/examples/foundational/02-llm-say-one-thing.py b/examples/foundational/02-llm-say-one-thing.py index 510e9291d..94d2b27f7 100644 --- a/examples/foundational/02-llm-say-one-thing.py +++ b/examples/foundational/02-llm-say-one-thing.py @@ -38,7 +38,7 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection): voice_id="71a7ad14-091c-4e8e-a314-022ece01c121", # British Reading Lady ) - llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o") + llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) messages = [ { diff --git a/examples/foundational/05-sync-speech-and-image.py b/examples/foundational/05-sync-speech-and-image.py index 6778b6946..9021dbf54 100644 --- a/examples/foundational/05-sync-speech-and-image.py +++ b/examples/foundational/05-sync-speech-and-image.py @@ -85,7 +85,7 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection): # Create an HTTP session for API calls async with aiohttp.ClientSession() as session: - llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o") + llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) tts = CartesiaHttpTTSService( api_key=os.getenv("CARTESIA_API_KEY"), diff --git a/examples/foundational/05a-local-sync-speech-and-image.py b/examples/foundational/05a-local-sync-speech-and-image.py index dee4c3c6d..77486d247 100644 --- a/examples/foundational/05a-local-sync-speech-and-image.py +++ b/examples/foundational/05a-local-sync-speech-and-image.py @@ -93,7 +93,7 @@ async def main(): self.frame = frame await self.push_frame(frame, direction) - llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o") + llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) tts = CartesiaHttpTTSService( api_key=os.getenv("CARTESIA_API_KEY"), diff --git a/examples/foundational/06-listen-and-respond.py b/examples/foundational/06-listen-and-respond.py index 6df1b82b8..c41c39748 100644 --- a/examples/foundational/06-listen-and-respond.py +++ b/examples/foundational/06-listen-and-respond.py @@ -73,7 +73,7 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection): voice_id="71a7ad14-091c-4e8e-a314-022ece01c121", # British Reading Lady ) - llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o") + llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) ml = MetricsLogger() diff --git a/examples/foundational/06a-image-sync.py b/examples/foundational/06a-image-sync.py index 1117ae940..8ebe5400c 100644 --- a/examples/foundational/06a-image-sync.py +++ b/examples/foundational/06a-image-sync.py @@ -91,7 +91,7 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection): voice_id="71a7ad14-091c-4e8e-a314-022ece01c121", # British Reading Lady ) - llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o") + llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) messages = [ { diff --git a/examples/foundational/07-interruptible.py b/examples/foundational/07-interruptible.py index 9ec3b0215..a94dfad4b 100644 --- a/examples/foundational/07-interruptible.py +++ b/examples/foundational/07-interruptible.py @@ -45,7 +45,7 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection): voice_id="71a7ad14-091c-4e8e-a314-022ece01c121", # British Reading Lady ) - llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o") + llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) messages = [ { diff --git a/examples/foundational/07a-interruptible-vad.py b/examples/foundational/07a-interruptible-vad.py index ab70e5013..a146ad636 100644 --- a/examples/foundational/07a-interruptible-vad.py +++ b/examples/foundational/07a-interruptible-vad.py @@ -44,7 +44,7 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection): voice_id="71a7ad14-091c-4e8e-a314-022ece01c121", # British Reading Lady ) - llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o") + llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) messages = [ { diff --git a/examples/foundational/07b-interruptible-langchain.py b/examples/foundational/07b-interruptible-langchain.py index 8f72375c7..6a4c67b82 100644 --- a/examples/foundational/07b-interruptible-langchain.py +++ b/examples/foundational/07b-interruptible-langchain.py @@ -74,7 +74,7 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection): ("human", "{input}"), ] ) - chain = prompt | ChatOpenAI(model="gpt-4o", temperature=0.7) + chain = prompt | ChatOpenAI(model="gpt-4.1", temperature=0.7) history_chain = RunnableWithMessageHistory( chain, get_session_history, diff --git a/examples/foundational/07c-interruptible-deepgram-vad.py b/examples/foundational/07c-interruptible-deepgram-vad.py index 224002752..366644be6 100644 --- a/examples/foundational/07c-interruptible-deepgram-vad.py +++ b/examples/foundational/07c-interruptible-deepgram-vad.py @@ -48,7 +48,7 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection): tts = DeepgramTTSService(api_key=os.getenv("DEEPGRAM_API_KEY"), voice="aura-helios-en") - llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o") + llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) messages = [ { diff --git a/examples/foundational/07c-interruptible-deepgram.py b/examples/foundational/07c-interruptible-deepgram.py index 7cd1ea140..12081480d 100644 --- a/examples/foundational/07c-interruptible-deepgram.py +++ b/examples/foundational/07c-interruptible-deepgram.py @@ -42,7 +42,7 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection): tts = DeepgramTTSService(api_key=os.getenv("DEEPGRAM_API_KEY"), voice="aura-helios-en") - llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o") + llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) messages = [ { diff --git a/examples/foundational/07d-interruptible-elevenlabs-http.py b/examples/foundational/07d-interruptible-elevenlabs-http.py index c28361e8b..4984897bc 100644 --- a/examples/foundational/07d-interruptible-elevenlabs-http.py +++ b/examples/foundational/07d-interruptible-elevenlabs-http.py @@ -49,7 +49,7 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection): aiohttp_session=session, ) - llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o") + llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) messages = [ { diff --git a/examples/foundational/07d-interruptible-elevenlabs.py b/examples/foundational/07d-interruptible-elevenlabs.py index 1d51e287b..fe31c12b7 100644 --- a/examples/foundational/07d-interruptible-elevenlabs.py +++ b/examples/foundational/07d-interruptible-elevenlabs.py @@ -45,7 +45,7 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection): voice_id=os.getenv("ELEVENLABS_VOICE_ID", ""), ) - llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o") + llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) messages = [ { diff --git a/examples/foundational/07e-interruptible-playht-http.py b/examples/foundational/07e-interruptible-playht-http.py index cc4f581ef..394fff2ce 100644 --- a/examples/foundational/07e-interruptible-playht-http.py +++ b/examples/foundational/07e-interruptible-playht-http.py @@ -46,7 +46,7 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection): voice_url="s3://voice-cloning-zero-shot/d9ff78ba-d016-47f6-b0ef-dd630f59414e/female-cs/manifest.json", ) - llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o") + llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) messages = [ { diff --git a/examples/foundational/07e-interruptible-playht.py b/examples/foundational/07e-interruptible-playht.py index cd8d70051..a1b5dddf9 100644 --- a/examples/foundational/07e-interruptible-playht.py +++ b/examples/foundational/07e-interruptible-playht.py @@ -48,7 +48,7 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection): params=PlayHTTTSService.InputParams(language=Language.EN), ) - llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o") + llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) messages = [ { diff --git a/examples/foundational/07g-interruptible-openai.py b/examples/foundational/07g-interruptible-openai.py index a55f249f0..3bc019ad1 100644 --- a/examples/foundational/07g-interruptible-openai.py +++ b/examples/foundational/07g-interruptible-openai.py @@ -46,7 +46,7 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection): tts = OpenAITTSService(api_key=os.getenv("OPENAI_API_KEY"), voice="ballad") - llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o") + llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) messages = [ { diff --git a/examples/foundational/07h-interruptible-openpipe.py b/examples/foundational/07h-interruptible-openpipe.py index 9335b4a20..633e0cf15 100644 --- a/examples/foundational/07h-interruptible-openpipe.py +++ b/examples/foundational/07h-interruptible-openpipe.py @@ -50,7 +50,6 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection): llm = OpenPipeLLMService( api_key=os.getenv("OPENAI_API_KEY"), openpipe_api_key=os.getenv("OPENPIPE_API_KEY"), - model="gpt-4o", tags={"conversation_id": f"pipecat-{timestamp}"}, ) diff --git a/examples/foundational/07i-interruptible-xtts.py b/examples/foundational/07i-interruptible-xtts.py index f6aaf80cb..aa64425fd 100644 --- a/examples/foundational/07i-interruptible-xtts.py +++ b/examples/foundational/07i-interruptible-xtts.py @@ -49,7 +49,7 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection): base_url="http://localhost:8000", ) - llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o") + llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) messages = [ { diff --git a/examples/foundational/07j-interruptible-gladia.py b/examples/foundational/07j-interruptible-gladia.py index 3174c4bee..4ed903ea6 100644 --- a/examples/foundational/07j-interruptible-gladia.py +++ b/examples/foundational/07j-interruptible-gladia.py @@ -54,7 +54,7 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection): voice_id="71a7ad14-091c-4e8e-a314-022ece01c121", # British Reading Lady ) - llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY", ""), model="gpt-4o") + llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY", "")) messages = [ { diff --git a/examples/foundational/07k-interruptible-lmnt.py b/examples/foundational/07k-interruptible-lmnt.py index 0c9bec12e..092e4758f 100644 --- a/examples/foundational/07k-interruptible-lmnt.py +++ b/examples/foundational/07k-interruptible-lmnt.py @@ -42,7 +42,7 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection): tts = LmntTTSService(api_key=os.getenv("LMNT_API_KEY"), voice_id="morgan") - llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o") + llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) messages = [ { diff --git a/examples/foundational/07m-interruptible-polly.py b/examples/foundational/07m-interruptible-polly.py index d832a7a06..6fe09619d 100644 --- a/examples/foundational/07m-interruptible-polly.py +++ b/examples/foundational/07m-interruptible-polly.py @@ -48,7 +48,7 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection): params=PollyTTSService.InputParams(engine="neural", language="en-GB", rate="1.05"), ) - llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o") + llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) messages = [ { diff --git a/examples/foundational/07o-interruptible-assemblyai.py b/examples/foundational/07o-interruptible-assemblyai.py index bc132860e..137060651 100644 --- a/examples/foundational/07o-interruptible-assemblyai.py +++ b/examples/foundational/07o-interruptible-assemblyai.py @@ -47,7 +47,7 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection): voice_id="71a7ad14-091c-4e8e-a314-022ece01c121", # British Reading Lady ) - llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o") + llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) messages = [ { diff --git a/examples/foundational/07p-interruptible-krisp.py b/examples/foundational/07p-interruptible-krisp.py index ab8e4617d..1f650363e 100644 --- a/examples/foundational/07p-interruptible-krisp.py +++ b/examples/foundational/07p-interruptible-krisp.py @@ -44,7 +44,7 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection): tts = DeepgramTTSService(api_key=os.getenv("DEEPGRAM_API_KEY"), voice="aura-helios-en") - llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o") + llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) messages = [ { diff --git a/examples/foundational/07q-interruptible-rime-http.py b/examples/foundational/07q-interruptible-rime-http.py index b6ee068c0..c5dedf63a 100644 --- a/examples/foundational/07q-interruptible-rime-http.py +++ b/examples/foundational/07q-interruptible-rime-http.py @@ -49,7 +49,7 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection): aiohttp_session=session, ) - llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o") + llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) messages = [ { diff --git a/examples/foundational/07q-interruptible-rime.py b/examples/foundational/07q-interruptible-rime.py index b297177f3..3df39ef6c 100644 --- a/examples/foundational/07q-interruptible-rime.py +++ b/examples/foundational/07q-interruptible-rime.py @@ -45,7 +45,7 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection): voice_id="rex", ) - llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o") + llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) messages = [ { diff --git a/examples/foundational/07t-interruptible-fish.py b/examples/foundational/07t-interruptible-fish.py index db00b1095..160bbddaf 100644 --- a/examples/foundational/07t-interruptible-fish.py +++ b/examples/foundational/07t-interruptible-fish.py @@ -45,7 +45,7 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection): model="4ce7e917cedd4bc2bb2e6ff3a46acaa1", # Barack Obama ) - llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o") + llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) messages = [ { diff --git a/examples/foundational/07v-interruptible-neuphonic-http.py b/examples/foundational/07v-interruptible-neuphonic-http.py index 66170bcf9..f1db00454 100644 --- a/examples/foundational/07v-interruptible-neuphonic-http.py +++ b/examples/foundational/07v-interruptible-neuphonic-http.py @@ -45,7 +45,7 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection): voice_id="fc854436-2dac-4d21-aa69-ae17b54e98eb", # Emily ) - llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o") + llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) messages = [ { diff --git a/examples/foundational/07v-interruptible-neuphonic.py b/examples/foundational/07v-interruptible-neuphonic.py index d72b9327d..25df4e2e6 100644 --- a/examples/foundational/07v-interruptible-neuphonic.py +++ b/examples/foundational/07v-interruptible-neuphonic.py @@ -45,7 +45,7 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection): voice_id="fc854436-2dac-4d21-aa69-ae17b54e98eb", # Emily ) - llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o") + llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) messages = [ { diff --git a/examples/foundational/07w-interruptible-fal.py b/examples/foundational/07w-interruptible-fal.py index 2b8df2aaf..6c1b71937 100644 --- a/examples/foundational/07w-interruptible-fal.py +++ b/examples/foundational/07w-interruptible-fal.py @@ -47,7 +47,7 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection): voice_id="71a7ad14-091c-4e8e-a314-022ece01c121", # British Reading Lady ) - llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o") + llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) messages = [ { diff --git a/examples/foundational/07x-interruptible-local.py b/examples/foundational/07x-interruptible-local.py index 989af1883..5027e8bcb 100644 --- a/examples/foundational/07x-interruptible-local.py +++ b/examples/foundational/07x-interruptible-local.py @@ -45,7 +45,7 @@ async def main(): voice_id="71a7ad14-091c-4e8e-a314-022ece01c121", # British Reading Lady ) - llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o") + llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) messages = [ { diff --git a/examples/foundational/10-wake-phrase.py b/examples/foundational/10-wake-phrase.py index 24397bfa5..4fb19343d 100644 --- a/examples/foundational/10-wake-phrase.py +++ b/examples/foundational/10-wake-phrase.py @@ -47,7 +47,7 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection): voice_id="71a7ad14-091c-4e8e-a314-022ece01c121", # British Reading Lady ) - llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o") + llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) messages = [ { diff --git a/examples/foundational/11-sound-effects.py b/examples/foundational/11-sound-effects.py index f9d179867..4a70d15d1 100644 --- a/examples/foundational/11-sound-effects.py +++ b/examples/foundational/11-sound-effects.py @@ -93,7 +93,7 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection): stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY")) - llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o") + llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) tts = CartesiaTTSService( api_key=os.getenv("CARTESIA_API_KEY"), diff --git a/examples/foundational/12b-describe-video-gpt-4o.py b/examples/foundational/12b-describe-video-gpt-4o.py index 8e1dc99dc..c676390cc 100644 --- a/examples/foundational/12b-describe-video-gpt-4o.py +++ b/examples/foundational/12b-describe-video-gpt-4o.py @@ -74,7 +74,7 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection): stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY")) # OpenAI GPT-4o for vision analysis - openai = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o") + openai = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) tts = CartesiaTTSService( api_key=os.getenv("CARTESIA_API_KEY"), diff --git a/examples/foundational/14-function-calling.py b/examples/foundational/14-function-calling.py index 70d3c1667..31a6417f4 100644 --- a/examples/foundational/14-function-calling.py +++ b/examples/foundational/14-function-calling.py @@ -53,7 +53,7 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection): voice_id="71a7ad14-091c-4e8e-a314-022ece01c121", # British Reading Lady ) - llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o") + llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) # You can also register a function_name of None to get all functions # sent to the same callback with an additional function_name parameter. diff --git a/examples/foundational/14d-function-calling-video.py b/examples/foundational/14d-function-calling-video.py index f8c86b8e7..e51dc96dd 100644 --- a/examples/foundational/14d-function-calling-video.py +++ b/examples/foundational/14d-function-calling-video.py @@ -82,7 +82,7 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection): voice_id="71a7ad14-091c-4e8e-a314-022ece01c121", # British Reading Lady ) - llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o") + llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) llm.register_function("get_weather", get_weather) llm.register_function("get_image", get_image) diff --git a/examples/foundational/15-switch-voices.py b/examples/foundational/15-switch-voices.py index 85e063f21..14aeab76a 100644 --- a/examples/foundational/15-switch-voices.py +++ b/examples/foundational/15-switch-voices.py @@ -83,7 +83,7 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection): voice_id="a0e99841-438c-4a64-b679-ae501e7d6091", # Barbershop Man ) - llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o") + llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) llm.register_function("switch_voice", switch_voice) tools = [ diff --git a/examples/foundational/15a-switch-languages.py b/examples/foundational/15a-switch-languages.py index 0e7cb7c3c..de89721db 100644 --- a/examples/foundational/15a-switch-languages.py +++ b/examples/foundational/15a-switch-languages.py @@ -73,7 +73,7 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection): voice_id="d4db5fb9-f44b-4bd1-85fa-192e0f0d75f9", # Spanish-speaking Lady ) - llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o") + llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) llm.register_function("switch_language", switch_language) tools = [ diff --git a/examples/foundational/16-gpu-container-local-bot.py b/examples/foundational/16-gpu-container-local-bot.py index d23b29271..e4b1b34d9 100644 --- a/examples/foundational/16-gpu-container-local-bot.py +++ b/examples/foundational/16-gpu-container-local-bot.py @@ -6,7 +6,6 @@ import os -import aiohttp from dotenv import load_dotenv from loguru import logger @@ -40,105 +39,101 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection): ), ) - # Create an HTTP session - async with aiohttp.ClientSession() as session: - stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY")) + stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY")) - tts = DeepgramTTSService( - aiohttp_session=session, - api_key=os.getenv("DEEPGRAM_API_KEY"), - voice="aura-asteria-en", - base_url="http://0.0.0.0:8080/v1/speak", - ) + tts = DeepgramTTSService( + api_key=os.getenv("DEEPGRAM_API_KEY"), + voice="aura-asteria-en", + base_url="http://0.0.0.0:8080", + ) - llm = OpenAILLMService( - # To use OpenAI - # api_key=os.getenv("OPENAI_API_KEY"), - # model="gpt-4o" - # Or, to use a local vLLM (or similar) api server - model="meta-llama/Meta-Llama-3-8B-Instruct", - base_url="http://0.0.0.0:8000/v1", - ) + llm = OpenAILLMService( + # To use OpenAI + # api_key=os.getenv("OPENAI_API_KEY"), + # Or, to use a local vLLM (or similar) api server + model="meta-llama/Meta-Llama-3-8B-Instruct", + base_url="http://0.0.0.0:8000/v1", + ) - messages = [ - { - "role": "system", - "content": "You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio so don't include special characters in your answers. Respond to what the user said in a creative and helpful way.", - }, + messages = [ + { + "role": "system", + "content": "You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio so don't include special characters in your answers. Respond to what the user said in a creative and helpful way.", + }, + ] + + context = OpenAILLMContext(messages) + context_aggregator = llm.create_context_aggregator(context) + + pipeline = Pipeline( + [ + transport.input(), # Transport user input + stt, # STT + context_aggregator.user(), + llm, # LLM + tts, # TTS + transport.output(), # Transport bot output + context_aggregator.assistant(), ] + ) - context = OpenAILLMContext(messages) - context_aggregator = llm.create_context_aggregator(context) + task = PipelineTask( + pipeline, + params=PipelineParams( + allow_interruptions=True, + enable_metrics=True, + ), + ) - pipeline = Pipeline( - [ - transport.input(), # Transport user input - stt, # STT - context_aggregator.user(), - llm, # LLM - tts, # TTS - transport.output(), # Transport bot output - context_aggregator.assistant(), - ] - ) + # When the first participant joins, the bot should introduce itself. + @transport.event_handler("on_client_connected") + async def on_client_connected(transport, client): + logger.info(f"Client connected") + # Kick off the conversation. + messages.append({"role": "system", "content": "Please introduce yourself to the user."}) + await task.queue_frames([context_aggregator.user().get_context_frame()]) - task = PipelineTask( - pipeline, - params=PipelineParams( - allow_interruptions=True, - enable_metrics=True, - ), - ) - - # When the first participant joins, the bot should introduce itself. - @transport.event_handler("on_client_connected") - async def on_client_connected(transport, client): - logger.info(f"Client connected") - # Kick off the conversation. - messages.append({"role": "system", "content": "Please introduce yourself to the user."}) - await task.queue_frames([context_aggregator.user().get_context_frame()]) - - # Handle "latency-ping" messages. The client will send app messages that look like - # this: - # { "latency-ping": { ts: }} - # - # We want to send an immediate pong back to the client from this handler function. - # Also, we will push a frame into the top of the pipeline and send it after the - # - @transport.event_handler("on_app_message") - async def on_app_message(transport, message, sender): - try: - if "latency-ping" in message: - logger.debug(f"Received latency ping app message: {message}") - ts = message["latency-ping"]["ts"] - # Send immediately - transport.output().send_message( - DailyTransportMessageFrame( - message={"latency-pong-msg-handler": {"ts": ts}}, participant_id=sender - ) + # Handle "latency-ping" messages. The client will send app messages that look like + # this: + # { "latency-ping": { ts: }} + # + # We want to send an immediate pong back to the client from this handler function. + # Also, we will push a frame into the top of the pipeline and send it after the + # + @transport.event_handler("on_app_message") + async def on_app_message(transport, message, sender): + try: + if "latency-ping" in message: + logger.debug(f"Received latency ping app message: {message}") + ts = message["latency-ping"]["ts"] + # Send immediately + transport.output().send_message( + DailyTransportMessageFrame( + message={"latency-pong-msg-handler": {"ts": ts}}, participant_id=sender ) - # And push to the pipeline for the Daily transport.output to send - await task.queue_frame( - DailyTransportMessageFrame( - message={"latency-pong-pipeline-delivery": {"ts": ts}}, - participant_id=sender, - ) + ) + # And push to the pipeline for the Daily transport.output to send + await task.queue_frame( + DailyTransportMessageFrame( + message={"latency-pong-pipeline-delivery": {"ts": ts}}, + participant_id=sender, ) - except Exception as e: - logger.debug(f"message handling error: {e} - {message}") + ) + except Exception as e: + logger.debug(f"message handling error: {e} - {message}") - @transport.event_handler("on_client_disconnected") - async def on_client_disconnected(transport, client): - logger.info(f"Client disconnected") + @transport.event_handler("on_client_disconnected") + async def on_client_disconnected(transport, client): + logger.info(f"Client disconnected") - @transport.event_handler("on_client_closed") - async def on_client_closed(transport, client): - logger.info(f"Client closed connection") - await task.cancel() + @transport.event_handler("on_client_closed") + async def on_client_closed(transport, client): + logger.info(f"Client closed connection") + await task.cancel() - runner = PipelineRunner(handle_sigint=False) + runner = PipelineRunner(handle_sigint=False) - await runner.run(task) + await runner.run(task) if __name__ == "__main__": diff --git a/examples/foundational/17-detect-user-idle.py b/examples/foundational/17-detect-user-idle.py index d7630c798..2f1652a0f 100644 --- a/examples/foundational/17-detect-user-idle.py +++ b/examples/foundational/17-detect-user-idle.py @@ -47,7 +47,7 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection): voice_id="71a7ad14-091c-4e8e-a314-022ece01c121", # British Reading Lady ) - llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o") + llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) messages = [ { diff --git a/examples/foundational/20a-persistent-context-openai.py b/examples/foundational/20a-persistent-context-openai.py index 22950e0fa..bba3fc00e 100644 --- a/examples/foundational/20a-persistent-context-openai.py +++ b/examples/foundational/20a-persistent-context-openai.py @@ -185,7 +185,7 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection): voice_id="71a7ad14-091c-4e8e-a314-022ece01c121", # British Reading Lady ) - llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o") + llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) # you can either register a single function for all function calls, or specific functions # llm.register_function(None, fetch_weather_from_api) diff --git a/examples/foundational/22-natural-conversation.py b/examples/foundational/22-natural-conversation.py index e2047c7aa..0878d4831 100644 --- a/examples/foundational/22-natural-conversation.py +++ b/examples/foundational/22-natural-conversation.py @@ -56,7 +56,7 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection): # statement. This doesn't really need to be an LLM, we could use NLP # libraries for that, but it was easier as an example because we # leverage the context aggregators. - statement_llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o") + statement_llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) statement_messages = [ { @@ -69,7 +69,7 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection): statement_context_aggregator = statement_llm.create_context_aggregator(statement_context) # This is the regular LLM. - llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o") + llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) messages = [ { diff --git a/examples/foundational/22b-natural-conversation-proposal.py b/examples/foundational/22b-natural-conversation-proposal.py index 05538bec0..b6e89556f 100644 --- a/examples/foundational/22b-natural-conversation-proposal.py +++ b/examples/foundational/22b-natural-conversation-proposal.py @@ -224,10 +224,10 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection): # This is the LLM that will be used to detect if the user has finished a # statement. This doesn't really need to be an LLM, we could use NLP # libraries for that, but we have the machinery to use an LLM, so we might as well! - statement_llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o") + statement_llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) # This is the regular LLM. - llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o") + llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) # You can also register a function_name of None to get all functions # sent to the same callback with an additional function_name parameter. llm.register_function("get_current_weather", fetch_weather_from_api) diff --git a/examples/foundational/22c-natural-conversation-mixed-llms.py b/examples/foundational/22c-natural-conversation-mixed-llms.py index 7d0d0a16b..e6b0d0df7 100644 --- a/examples/foundational/22c-natural-conversation-mixed-llms.py +++ b/examples/foundational/22c-natural-conversation-mixed-llms.py @@ -428,16 +428,10 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection): # This is the LLM that will be used to detect if the user has finished a # statement. This doesn't really need to be an LLM, we could use NLP # libraries for that, but we have the machinery to use an LLM, so we might as well! - statement_llm = AnthropicLLMService( - api_key=os.getenv("ANTHROPIC_API_KEY"), - model="claude-3-5-sonnet-20241022", - ) + statement_llm = AnthropicLLMService(api_key=os.getenv("ANTHROPIC_API_KEY")) # This is the regular LLM. - llm = OpenAILLMService( - api_key=os.getenv("OPENAI_API_KEY"), - model="gpt-4o", - ) + llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) # Register a function_name of None to get all functions # sent to the same callback with an additional function_name parameter. llm.register_function("get_current_weather", fetch_weather_from_api) diff --git a/examples/foundational/22d-natural-conversation-gemini-audio.py b/examples/foundational/22d-natural-conversation-gemini-audio.py index a0035e11c..40bc33e48 100644 --- a/examples/foundational/22d-natural-conversation-gemini-audio.py +++ b/examples/foundational/22d-natural-conversation-gemini-audio.py @@ -33,7 +33,10 @@ from pipecat.pipeline.parallel_pipeline import ParallelPipeline from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineParams, PipelineTask -from pipecat.processors.aggregators.llm_response import LLMAssistantResponseAggregator +from pipecat.processors.aggregators.llm_response import ( + LLMAssistantAggregatorParams, + LLMAssistantResponseAggregator, +) from pipecat.processors.aggregators.openai_llm_context import ( OpenAILLMContext, OpenAILLMContextFrame, @@ -478,7 +481,7 @@ class LLMAggregatorBuffer(LLMAssistantResponseAggregator): """Buffers the output of the transcription LLM. Used by the bot output gate.""" def __init__(self, **kwargs): - super().__init__(expect_stripped_words=False) + super().__init__(params=LLMAssistantAggregatorParams(expect_stripped_words=False)) self._transcription = "" async def process_frame(self, frame: Frame, direction: FrameDirection): diff --git a/examples/foundational/23-bot-background-sound-daily.py b/examples/foundational/23-bot-background-sound-daily.py index 0379b0a11..97947d768 100644 --- a/examples/foundational/23-bot-background-sound-daily.py +++ b/examples/foundational/23-bot-background-sound-daily.py @@ -62,7 +62,7 @@ async def main(): voice_id="71a7ad14-091c-4e8e-a314-022ece01c121", # British Reading Lady ) - llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o") + llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) messages = [ { diff --git a/examples/foundational/23-bot-background-sound-p2p.py b/examples/foundational/23-bot-background-sound-p2p.py index d854770c7..f18c58971 100644 --- a/examples/foundational/23-bot-background-sound-p2p.py +++ b/examples/foundational/23-bot-background-sound-p2p.py @@ -4,15 +4,13 @@ # SPDX-License-Identifier: BSD 2-Clause License # -""" -Usage +"""Usage ----- Set the path to your background audio file using the `INPUT_AUDIO_PATH` environment variable, then run the bot using: INPUT_AUDIO_PATH=path/to/your_audio.mp3 python 23-bot-background-sound.py Example: - INPUT_AUDIO_PATH=my_audio.mp3 python 23-bot-background-sound.py """ @@ -71,7 +69,7 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection): voice_id="71a7ad14-091c-4e8e-a314-022ece01c121", # British Reading Lady ) - llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o") + llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) messages = [ { diff --git a/examples/foundational/24-stt-mute-filter.py b/examples/foundational/24-stt-mute-filter.py index fb00c3114..98f054143 100644 --- a/examples/foundational/24-stt-mute-filter.py +++ b/examples/foundational/24-stt-mute-filter.py @@ -64,7 +64,7 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection): tts = DeepgramTTSService(api_key=os.getenv("DEEPGRAM_API_KEY"), voice="aura-helios-en") - llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o") + llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) llm.register_function("get_current_weather", fetch_weather_from_api) weather_function = FunctionSchema( diff --git a/examples/foundational/28-transcription-processor.py b/examples/foundational/28-transcription-processor.py index 069235a4e..42e31d5f6 100644 --- a/examples/foundational/28-transcription-processor.py +++ b/examples/foundational/28-transcription-processor.py @@ -109,10 +109,7 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection): voice_id="71a7ad14-091c-4e8e-a314-022ece01c121", # British Reading Lady ) - llm = OpenAILLMService( - api_key=os.getenv("OPENAI_API_KEY"), - model="gpt-4o", - ) + llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) messages = [ { diff --git a/examples/foundational/29-livekit-audio-chat.py b/examples/foundational/29-livekit-audio-chat.py index d761e5418..b8d44deb7 100644 --- a/examples/foundational/29-livekit-audio-chat.py +++ b/examples/foundational/29-livekit-audio-chat.py @@ -127,7 +127,7 @@ async def main(): ), ) - llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o") + llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) tts = CartesiaTTSService( api_key=os.getenv("CARTESIA_API_KEY"), diff --git a/examples/foundational/30-observer.py b/examples/foundational/30-observer.py index b53a0a026..ad11f2267 100644 --- a/examples/foundational/30-observer.py +++ b/examples/foundational/30-observer.py @@ -88,7 +88,7 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection): voice_id="71a7ad14-091c-4e8e-a314-022ece01c121", # British Reading Lady ) - llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o") + llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) messages = [ { diff --git a/examples/foundational/35-pattern-pair-voice-switching.py b/examples/foundational/35-pattern-pair-voice-switching.py index c12dfea72..aadbe0cc3 100644 --- a/examples/foundational/35-pattern-pair-voice-switching.py +++ b/examples/foundational/35-pattern-pair-voice-switching.py @@ -120,7 +120,7 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection): ) # Initialize LLM - llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o") + llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) # System prompt for storytelling with voice switching system_prompt = """You are an engaging storyteller that uses different voices to bring stories to life. diff --git a/examples/foundational/36-user-email-gathering.py b/examples/foundational/36-user-email-gathering.py index 8cbb826c8..39958ce79 100644 --- a/examples/foundational/36-user-email-gathering.py +++ b/examples/foundational/36-user-email-gathering.py @@ -63,7 +63,7 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection): # aiohttp_session=session, # ) - llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o") + llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) # You can aslo register a function_name of None to get all functions # sent to the same callback with an additional function_name parameter. llm.register_function("store_user_emails", store_user_emails) diff --git a/examples/foundational/37-mem0.py b/examples/foundational/37-mem0.py index 8fb8e41f0..e5fff8bca 100644 --- a/examples/foundational/37-mem0.py +++ b/examples/foundational/37-mem0.py @@ -210,10 +210,6 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection): @rtvi.event_handler("on_client_ready") async def on_client_ready(rtvi): await rtvi.set_bot_ready() - - @transport.event_handler("on_client_connected") - async def on_client_connected(transport, client): - logger.info(f"Client connected") # Get personalized greeting based on user memories. Can pass agent_id and run_id as per requirement of the application to manage short term memory or agent specific memory. greeting = await get_initial_greeting( memory_client=memory.memory_client, user_id=USER_ID, agent_id=None, run_id=None @@ -225,6 +221,10 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection): # Queue the context frame to start the conversation await task.queue_frames([context_aggregator.user().get_context_frame()]) + @transport.event_handler("on_client_connected") + async def on_client_connected(transport, client): + logger.info(f"Client connected") + @transport.event_handler("on_client_disconnected") async def on_client_disconnected(transport, client): logger.info(f"Client disconnected") diff --git a/examples/instant-voice/server/src/single_bot.py b/examples/instant-voice/server/src/single_bot.py index fbd3a5fc9..44d612fd1 100644 --- a/examples/instant-voice/server/src/single_bot.py +++ b/examples/instant-voice/server/src/single_bot.py @@ -98,14 +98,16 @@ async def main(): @rtvi.event_handler("on_client_ready") async def on_client_ready(rtvi): await rtvi.set_bot_ready() + # Kick off the conversation + await task.queue_frames([context_aggregator.user().get_context_frame()]) @daily_transport.event_handler("on_first_participant_joined") async def on_first_participant_joined(transport, participant): - await task.queue_frames([context_aggregator.user().get_context_frame()]) + logger.debug("First participant joined: {}", participant["id"]) @daily_transport.event_handler("on_participant_left") async def on_participant_left(transport, participant, reason): - print(f"Participant left: {participant}") + logger.debug(f"Participant left: {participant}") await task.cancel() runner = PipelineRunner(handle_sigint=False) diff --git a/examples/moondream-chatbot/bot.py b/examples/moondream-chatbot/bot.py index 4b2303959..87bc7e5fc 100644 --- a/examples/moondream-chatbot/bot.py +++ b/examples/moondream-chatbot/bot.py @@ -156,7 +156,7 @@ async def main(): voice_id="71a7ad14-091c-4e8e-a314-022ece01c121", # British Reading Lady ) - llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o") + llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) ta = TalkingAnimation() diff --git a/examples/news-chatbot/server/news_bot.py b/examples/news-chatbot/server/news_bot.py index a8f09ba43..9748f5c65 100644 --- a/examples/news-chatbot/server/news_bot.py +++ b/examples/news-chatbot/server/news_bot.py @@ -148,10 +148,13 @@ async def main(): @rtvi.event_handler("on_client_ready") async def on_client_ready(rtvi): await rtvi.set_bot_ready() + # Kick off the conversation + await task.queue_frames([context_aggregator.user().get_context_frame()]) @transport.event_handler("on_first_participant_joined") async def on_first_participant_joined(transport, participant): - await task.queue_frames([context_aggregator.user().get_context_frame()]) + logger.debug("First participant joined: {}", participant["id"]) + await transport.capture_participant_transcription(participant["id"]) @transport.event_handler("on_participant_left") async def on_participant_left(transport, participant, reason): diff --git a/examples/p2p-webrtc/daily-interop-bridge/README.md b/examples/p2p-webrtc/daily-interop-bridge/README.md new file mode 100644 index 000000000..9de3f379f --- /dev/null +++ b/examples/p2p-webrtc/daily-interop-bridge/README.md @@ -0,0 +1,61 @@ +# SmallWebRTC and Daily + +A Pipecat example demonstrating how to interoperate audio and video between `SmallWebRTCTransport` and `DailyTransport`. + +## 🚀 Quick Start + +### 1️⃣ Start the Bot Server + +#### 🔧 Set Up the Environment +1. Create and activate a virtual environment: + ```bash + python3 -m venv venv + source venv/bin/activate # On Windows: venv\Scripts\activate + ``` + +2. Install dependencies: + ```bash + pip install -r requirements.txt + ``` + +3. Configure environment variables: + - Copy `env.example` to `.env` + ```bash + cp env.example .env + ``` + - Add your API keys + +#### ▶️ Run the Server +```bash +python server.py +``` + +### 1️⃣ Connect the first client using Daily Prebuilt + +- Open your browser and navigate to the same URL that you configured inside your `.env` file: + - `DAILY_SAMPLE_ROOM_URL` + +### 2️⃣ Connect the second client using SmallWebRTC Prebuilt UI + +- Open your browser and navigate to: +👉 http://localhost:7860 + - (Or use your custom port, if configured) + +## ⚠️ Important Note +Ensure the bot server is running before using any client implementations. + +## 📌 Requirements + +- Python **3.10+** +- Node.js **16+** (for JavaScript components) +- Google API Key +- Modern web browser with WebRTC support + +--- + +### 💡 Notes +- Ensure all dependencies are installed before running the server. +- Check the `.env` file for missing configurations. +- WebRTC requires a secure environment (HTTPS) for full functionality in production. + +Happy coding! 🎉 \ No newline at end of file diff --git a/examples/p2p-webrtc/daily-interop-bridge/bot.py b/examples/p2p-webrtc/daily-interop-bridge/bot.py new file mode 100644 index 000000000..dc1d7a9ac --- /dev/null +++ b/examples/p2p-webrtc/daily-interop-bridge/bot.py @@ -0,0 +1,128 @@ +# +# Copyright (c) 2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# +import os +import sys + +from dotenv import load_dotenv +from loguru import logger + +from pipecat.frames.frames import ( + InputAudioRawFrame, + InputImageRawFrame, + OutputAudioRawFrame, + OutputImageRawFrame, +) +from pipecat.pipeline.parallel_pipeline import ParallelPipeline +from pipecat.pipeline.pipeline import Pipeline +from pipecat.pipeline.runner import PipelineRunner +from pipecat.pipeline.task import PipelineParams, PipelineTask +from pipecat.processors.frame_processor import Frame, FrameDirection, FrameProcessor +from pipecat.transports.base_transport import TransportParams +from pipecat.transports.network.small_webrtc import SmallWebRTCTransport +from pipecat.transports.services.daily import DailyParams, DailyTransport + +load_dotenv(override=True) + +logger.remove(0) +logger.add(sys.stderr, level="DEBUG") + + +class MirrorProcessor(FrameProcessor): + async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + + if isinstance(frame, InputAudioRawFrame): + await self.push_frame( + OutputAudioRawFrame( + audio=frame.audio, + sample_rate=frame.sample_rate, + num_channels=frame.num_channels, + ) + ) + elif isinstance(frame, InputImageRawFrame): + await self.push_frame( + OutputImageRawFrame(image=frame.image, size=frame.size, format=frame.format) + ) + else: + await self.push_frame(frame, direction) + + +async def run_bot(webrtc_connection): + pipecat_transport = SmallWebRTCTransport( + webrtc_connection=webrtc_connection, + params=TransportParams( + camera_in_enabled=True, + camera_out_enabled=True, + camera_out_is_live=True, + audio_in_enabled=True, + audio_out_enabled=True, + camera_out_width=1280, + camera_out_height=720, + vad_enabled=False, + ), + ) + + room_url = os.getenv("DAILY_SAMPLE_ROOM_URL", "") + daily_transport = DailyTransport( + room_url, + None, + "SmallWebRTC", + params=DailyParams( + camera_in_enabled=True, + camera_out_enabled=True, + camera_out_is_live=True, + audio_in_enabled=True, + audio_out_enabled=True, + camera_out_width=1280, + camera_out_height=720, + vad_enabled=False, + ), + ) + + pipeline = Pipeline( + [ + ParallelPipeline( + [ + daily_transport.input(), + MirrorProcessor(), + pipecat_transport.output(), + ], + [ + pipecat_transport.input(), + MirrorProcessor(), + daily_transport.output(), + ], + ) + ] + ) + + task = PipelineTask( + pipeline, + params=PipelineParams( + allow_interruptions=False, + ), + ) + + @daily_transport.event_handler("on_participant_joined") + async def on_participant_joined(transport, participant): + await transport.capture_participant_video(participant["id"]) + + @pipecat_transport.event_handler("on_client_connected") + async def on_client_connected(transport, client): + logger.info("Pipecat Client connected") + + @pipecat_transport.event_handler("on_client_disconnected") + async def on_client_disconnected(transport, client): + logger.info("Pipecat Client disconnected") + + @pipecat_transport.event_handler("on_client_closed") + async def on_client_closed(transport, client): + logger.info("Pipecat Client closed") + await task.cancel() + + runner = PipelineRunner(handle_sigint=False) + + await runner.run(task) diff --git a/examples/p2p-webrtc/daily-interop-bridge/env.example b/examples/p2p-webrtc/daily-interop-bridge/env.example new file mode 100644 index 000000000..8ea5961f2 --- /dev/null +++ b/examples/p2p-webrtc/daily-interop-bridge/env.example @@ -0,0 +1,2 @@ +DAILY_API_KEY= +DAILY_SAMPLE_ROOM_URL= \ No newline at end of file diff --git a/examples/p2p-webrtc/daily-interop-bridge/requirements.txt b/examples/p2p-webrtc/daily-interop-bridge/requirements.txt new file mode 100644 index 000000000..19d9ca501 --- /dev/null +++ b/examples/p2p-webrtc/daily-interop-bridge/requirements.txt @@ -0,0 +1,5 @@ +python-dotenv +fastapi[all] +uvicorn +aiortc +pipecat-ai[silero, webrtc, daily] \ No newline at end of file diff --git a/examples/p2p-webrtc/daily-interop-bridge/server.py b/examples/p2p-webrtc/daily-interop-bridge/server.py new file mode 100644 index 000000000..295fe4f44 --- /dev/null +++ b/examples/p2p-webrtc/daily-interop-bridge/server.py @@ -0,0 +1,89 @@ +import argparse +import asyncio +import logging +from contextlib import asynccontextmanager +from typing import Dict + +import uvicorn +from bot import run_bot +from dotenv import load_dotenv +from fastapi import BackgroundTasks, FastAPI +from fastapi.responses import RedirectResponse +from pipecat_ai_small_webrtc_prebuilt.frontend import SmallWebRTCPrebuiltUI + +from pipecat.transports.network.webrtc_connection import SmallWebRTCConnection + +# Load environment variables +load_dotenv(override=True) + +logger = logging.getLogger("pc") + +app = FastAPI() + +# Store connections by pc_id +pcs_map: Dict[str, SmallWebRTCConnection] = {} + +ice_servers = ["stun:stun.l.google.com:19302"] + +# Mount the frontend at / +app.mount("/prebuilt", SmallWebRTCPrebuiltUI) + + +@app.get("/", include_in_schema=False) +async def root_redirect(): + return RedirectResponse(url="/prebuilt/") + + +@app.post("/api/offer") +async def offer(request: dict, background_tasks: BackgroundTasks): + pc_id = request.get("pc_id") + + if pc_id and pc_id in pcs_map: + pipecat_connection = pcs_map[pc_id] + logger.info(f"Reusing existing connection for pc_id: {pc_id}") + await pipecat_connection.renegotiate( + sdp=request["sdp"], type=request["type"], restart_pc=request.get("restart_pc", False) + ) + else: + pipecat_connection = SmallWebRTCConnection(ice_servers) + await pipecat_connection.initialize(sdp=request["sdp"], type=request["type"]) + + @pipecat_connection.event_handler("closed") + async def handle_disconnected(webrtc_connection: SmallWebRTCConnection): + logger.info(f"Discarding peer connection for pc_id: {webrtc_connection.pc_id}") + pcs_map.pop(webrtc_connection.pc_id, None) + + background_tasks.add_task(run_bot, pipecat_connection) + + answer = pipecat_connection.get_answer() + # Updating the peer connection inside the map + pcs_map[answer["pc_id"]] = pipecat_connection + + return answer + + +@asynccontextmanager +async def lifespan(app: FastAPI): + yield # Run app + coros = [pc.close() for pc in pcs_map.values()] + await asyncio.gather(*coros) + pcs_map.clear() + + +if __name__ == "__main__": + parser = argparse.ArgumentParser(description="WebRTC demo") + parser.add_argument( + "--host", default="localhost", help="Host for HTTP server (default: localhost)" + ) + parser.add_argument( + "--port", type=int, default=7860, help="Port for HTTP server (default: 7860)" + ) + parser.add_argument("--verbose", "-v", action="count") + args = parser.parse_args() + + if args.verbose: + logging.basicConfig(level=logging.DEBUG) + else: + logging.basicConfig(level=logging.INFO) + + uvicorn.run(app, host=args.host, port=args.port) diff --git a/examples/p2p-webrtc/video-transform/server/bot.py b/examples/p2p-webrtc/video-transform/server/bot.py index 7d4a801b4..ea18f8b67 100644 --- a/examples/p2p-webrtc/video-transform/server/bot.py +++ b/examples/p2p-webrtc/video-transform/server/bot.py @@ -135,12 +135,12 @@ async def run_bot(webrtc_connection): async def on_client_ready(rtvi): logger.info("Pipecat client ready.") await rtvi.set_bot_ready() + # Kick off the conversation. + await task.queue_frames([context_aggregator.user().get_context_frame()]) @pipecat_transport.event_handler("on_client_connected") async def on_client_connected(transport, client): logger.info("Pipecat Client connected") - # Kick off the conversation. - await task.queue_frames([context_aggregator.user().get_context_frame()]) @pipecat_transport.event_handler("on_client_disconnected") async def on_client_disconnected(transport, client): diff --git a/examples/patient-intake/bot.py b/examples/patient-intake/bot.py index cf2d5ed92..64bb11743 100644 --- a/examples/patient-intake/bot.py +++ b/examples/patient-intake/bot.py @@ -324,7 +324,7 @@ async def main(): # voice_id="846d6cb0-2301-48b6-9683-48f5618ea2f6", # Spanish-speaking Lady # ) - llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o") + llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) messages = [] context = OpenAILLMContext(messages=messages) diff --git a/examples/phone-chatbot/bot_twilio.py b/examples/phone-chatbot/bot_twilio.py index 0ffbdb122..9f7740320 100644 --- a/examples/phone-chatbot/bot_twilio.py +++ b/examples/phone-chatbot/bot_twilio.py @@ -60,7 +60,7 @@ async def main(room_url: str, token: str, callId: str, sipUri: str): voice_id=os.getenv("ELEVENLABS_VOICE_ID", ""), ) - llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o") + llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) messages = [ { diff --git a/examples/phone-chatbot/call_transfer.py b/examples/phone-chatbot/call_transfer.py index aae460ec4..487391c67 100644 --- a/examples/phone-chatbot/call_transfer.py +++ b/examples/phone-chatbot/call_transfer.py @@ -305,7 +305,7 @@ async def main( tools = ToolsSchema(standard_tools=[terminate_call_function, dial_operator_function]) # Initialize LLM - llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o") + llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) # Register functions with the LLM llm.register_function( diff --git a/examples/phone-chatbot/simple_dialin.py b/examples/phone-chatbot/simple_dialin.py index cc06837dd..dee51f865 100644 --- a/examples/phone-chatbot/simple_dialin.py +++ b/examples/phone-chatbot/simple_dialin.py @@ -129,7 +129,7 @@ async def main( system_instruction = """You are Chatbot, a friendly, helpful robot. Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio so don't include special characters in your answers. Respond to what the user said in a creative and helpful way, but keep your responses brief. Start by introducing yourself. If the user ends the conversation, **IMMEDIATELY** call the `terminate_call` function. """ # Initialize LLM - llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o") + llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) # Register functions with the LLM llm.register_function("terminate_call", terminate_call) diff --git a/examples/phone-chatbot/simple_dialout.py b/examples/phone-chatbot/simple_dialout.py index 9686d935a..6b15d3af8 100644 --- a/examples/phone-chatbot/simple_dialout.py +++ b/examples/phone-chatbot/simple_dialout.py @@ -101,7 +101,7 @@ async def main( system_instruction = """You are Chatbot, a friendly, helpful robot. Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio so don't include special characters in your answers. Respond to what the user said in a creative and helpful way, but keep your responses brief. Start by introducing yourself. If the user ends the conversation, **IMMEDIATELY** call the `terminate_call` function. """ # Initialize LLM - llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o") + llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) # Register functions with the LLM llm.register_function("terminate_call", terminate_call) diff --git a/examples/sentry-metrics/bot.py b/examples/sentry-metrics/bot.py index 28e470955..853911f2b 100644 --- a/examples/sentry-metrics/bot.py +++ b/examples/sentry-metrics/bot.py @@ -63,7 +63,6 @@ async def main(): llm = OpenAILLMService( api_key=os.getenv("OPENAI_API_KEY"), - model="gpt-4o", metrics=SentryMetrics(), ) diff --git a/examples/simple-chatbot/server/README.md b/examples/simple-chatbot/server/README.md index 0de5a7d77..01ea1bff4 100644 --- a/examples/simple-chatbot/server/README.md +++ b/examples/simple-chatbot/server/README.md @@ -70,3 +70,17 @@ Run the server: ```bash python server.py ``` + +## Troubleshooting + +If you encounred this error: + +```bash +aiohttp.client_exceptions.ClientConnectorCertificateError: Cannot connect to host api.daily.co:443 ssl:True [SSLCertVerificationError: (1, '[SSL: CERTIFICATE_VERIFY_FAILED] certificate verify failed: unable to get local issuer certificate (_ssl.c:1000)')] +``` + +It's because Python cannot verify the SSL certificate from https://api.daily.co when making a POST request to create a room or token. + +This is a common issue when the system doesn't have the proper CA certificates. + +Install SSL Certificates (macOS): `/Applications/Python\ 3.12/Install\ Certificates.command` diff --git a/examples/simple-chatbot/server/bot-gemini.py b/examples/simple-chatbot/server/bot-gemini.py index 0bab332d6..db0ff3f2f 100644 --- a/examples/simple-chatbot/server/bot-gemini.py +++ b/examples/simple-chatbot/server/bot-gemini.py @@ -183,11 +183,12 @@ async def main(): @rtvi.event_handler("on_client_ready") async def on_client_ready(rtvi): await rtvi.set_bot_ready() + # Kick off the conversation + await task.queue_frames([context_aggregator.user().get_context_frame()]) @transport.event_handler("on_first_participant_joined") async def on_first_participant_joined(transport, participant): await transport.capture_participant_transcription(participant["id"]) - await task.queue_frames([context_aggregator.user().get_context_frame()]) @transport.event_handler("on_participant_left") async def on_participant_left(transport, participant, reason): diff --git a/examples/simple-chatbot/server/bot-openai.py b/examples/simple-chatbot/server/bot-openai.py index 3ad29a28d..5c4610c86 100644 --- a/examples/simple-chatbot/server/bot-openai.py +++ b/examples/simple-chatbot/server/bot-openai.py @@ -155,7 +155,7 @@ async def main(): ) # Initialize LLM service - llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o") + llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) messages = [ { @@ -210,11 +210,12 @@ async def main(): @rtvi.event_handler("on_client_ready") async def on_client_ready(rtvi): await rtvi.set_bot_ready() + # Kick off the conversation + await task.queue_frames([context_aggregator.user().get_context_frame()]) @transport.event_handler("on_first_participant_joined") async def on_first_participant_joined(transport, participant): await transport.capture_participant_transcription(participant["id"]) - await task.queue_frames([context_aggregator.user().get_context_frame()]) @transport.event_handler("on_participant_left") async def on_participant_left(transport, participant, reason): diff --git a/examples/telnyx-chatbot/bot.py b/examples/telnyx-chatbot/bot.py index 4d8584bfb..861aa1c7a 100644 --- a/examples/telnyx-chatbot/bot.py +++ b/examples/telnyx-chatbot/bot.py @@ -48,7 +48,7 @@ async def run_bot( ), ) - llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o") + llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY")) diff --git a/examples/translation-chatbot/bot.py b/examples/translation-chatbot/bot.py index 88c9c66b7..b7611e3a3 100644 --- a/examples/translation-chatbot/bot.py +++ b/examples/translation-chatbot/bot.py @@ -150,7 +150,7 @@ async def main(): in_language = "English" out_language = "Spanish" - llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o") + llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) context = OpenAILLMContext() context_aggregator = llm.create_context_aggregator(context) diff --git a/examples/twilio-chatbot/bot.py b/examples/twilio-chatbot/bot.py index 6821d8139..9b1501aaf 100644 --- a/examples/twilio-chatbot/bot.py +++ b/examples/twilio-chatbot/bot.py @@ -68,7 +68,7 @@ async def run_bot(websocket_client: WebSocket, stream_sid: str, testing: bool): ), ) - llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o") + llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"), audio_passthrough=True) diff --git a/examples/twilio-chatbot/client.py b/examples/twilio-chatbot/client.py index e5826fb19..97cca0cd0 100644 --- a/examples/twilio-chatbot/client.py +++ b/examples/twilio-chatbot/client.py @@ -98,7 +98,7 @@ async def run_client(client_name: str, server_url: str, duration_secs: int): ), ) - llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o") + llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) # We let the audio passthrough so we can record the conversation. stt = DeepgramSTTService( diff --git a/examples/websocket-server/bot.py b/examples/websocket-server/bot.py index 0e2f1d4b0..2f08f8b43 100644 --- a/examples/websocket-server/bot.py +++ b/examples/websocket-server/bot.py @@ -91,7 +91,7 @@ async def main(): ) ) - llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o") + llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY")) diff --git a/pyproject.toml b/pyproject.toml index fdabdb199..e10e6daf9 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -54,7 +54,7 @@ fal = [ "fal-client~=0.5.9" ] fireworks = [] fish = [ "ormsgpack~=1.7.0", "websockets~=13.1" ] gladia = [ "websockets~=13.1" ] -google = [ "google-cloud-speech~=2.31.1", "google-cloud-texttospeech~=2.25.1", "google-genai~=1.7.0", "google-generativeai~=0.8.4" ] +google = [ "google-cloud-speech~=2.31.1", "google-cloud-texttospeech~=2.25.1", "google-genai~=1.7.0", "google-generativeai~=0.8.4", "websockets~=13.1" ] grok = [] groq = [ "groq~=0.20.0" ] gstreamer = [ "pygobject~=3.50.0" ] diff --git a/src/pipecat/processors/aggregators/llm_response.py b/src/pipecat/processors/aggregators/llm_response.py index dccceea1f..1f7a33ed1 100644 --- a/src/pipecat/processors/aggregators/llm_response.py +++ b/src/pipecat/processors/aggregators/llm_response.py @@ -6,6 +6,7 @@ import asyncio from abc import abstractmethod +from dataclasses import dataclass from typing import Dict, List, Literal, Set from loguru import logger @@ -46,6 +47,16 @@ from pipecat.processors.frame_processor import FrameDirection, FrameProcessor from pipecat.utils.time import time_now_iso8601 +@dataclass +class LLMUserAggregatorParams: + aggregation_timeout: float = 1.0 + + +@dataclass +class LLMAssistantAggregatorParams: + expect_stripped_words: bool = True + + class LLMFullResponseAggregator(FrameProcessor): """This is an LLM aggregator that aggregates a full LLM completion. It aggregates LLM text frames (tokens) received between @@ -230,11 +241,23 @@ class LLMUserContextAggregator(LLMContextResponseAggregator): def __init__( self, context: OpenAILLMContext, - aggregation_timeout: float = 1.0, + *, + params: LLMUserAggregatorParams = LLMUserAggregatorParams(), **kwargs, ): super().__init__(context=context, role="user", **kwargs) - self._aggregation_timeout = aggregation_timeout + self._params = params + if "aggregation_timeout" in kwargs: + import warnings + + with warnings.catch_warnings(): + warnings.simplefilter("always") + warnings.warn( + "Parameter 'aggregation_timeout' is deprecated, use 'params' instead.", + DeprecationWarning, + ) + + self._params.aggregation_timeout = kwargs["aggregation_timeout"] self._seen_interim_results = False self._user_speaking = False @@ -357,7 +380,9 @@ class LLMUserContextAggregator(LLMContextResponseAggregator): async def _aggregation_task_handler(self): while True: try: - await asyncio.wait_for(self._aggregation_event.wait(), self._aggregation_timeout) + await asyncio.wait_for( + self._aggregation_event.wait(), self._params.aggregation_timeout + ) await self._maybe_push_bot_interruption() except asyncio.TimeoutError: if not self._user_speaking: @@ -394,9 +419,27 @@ class LLMAssistantContextAggregator(LLMContextResponseAggregator): """ - def __init__(self, context: OpenAILLMContext, *, expect_stripped_words: bool = True, **kwargs): + def __init__( + self, + context: OpenAILLMContext, + *, + params: LLMAssistantAggregatorParams = LLMAssistantAggregatorParams(), + **kwargs, + ): super().__init__(context=context, role="assistant", **kwargs) - self._expect_stripped_words = expect_stripped_words + self._params = params + + if "expect_stripped_words" in kwargs: + import warnings + + with warnings.catch_warnings(): + warnings.simplefilter("always") + warnings.warn( + "Parameter 'expect_stripped_words' is deprecated, use 'params' instead.", + DeprecationWarning, + ) + + self._params.expect_stripped_words = kwargs["expect_stripped_words"] self._started = 0 self._function_calls_in_progress: Dict[str, FunctionCallInProgressFrame] = {} @@ -558,7 +601,7 @@ class LLMAssistantContextAggregator(LLMContextResponseAggregator): if not self._started: return - if self._expect_stripped_words: + if self._params.expect_stripped_words: self._aggregation += f" {frame.text}" if self._aggregation else frame.text else: self._aggregation += frame.text @@ -572,8 +615,14 @@ class LLMAssistantContextAggregator(LLMContextResponseAggregator): class LLMUserResponseAggregator(LLMUserContextAggregator): - def __init__(self, messages: List[dict] = [], **kwargs): - super().__init__(context=OpenAILLMContext(messages), **kwargs) + def __init__( + self, + messages: List[dict] = [], + *, + params: LLMUserAggregatorParams = LLMUserAggregatorParams(), + **kwargs, + ): + super().__init__(context=OpenAILLMContext(messages), params=params, **kwargs) async def push_aggregation(self): if len(self._aggregation) > 0: @@ -588,8 +637,14 @@ class LLMUserResponseAggregator(LLMUserContextAggregator): class LLMAssistantResponseAggregator(LLMAssistantContextAggregator): - def __init__(self, messages: List[dict] = [], **kwargs): - super().__init__(context=OpenAILLMContext(messages), **kwargs) + def __init__( + self, + messages: List[dict] = [], + *, + params: LLMAssistantAggregatorParams = LLMAssistantAggregatorParams(), + **kwargs, + ): + super().__init__(context=OpenAILLMContext(messages), params=params, **kwargs) async def push_aggregation(self): if len(self._aggregation) > 0: diff --git a/src/pipecat/services/anthropic/llm.py b/src/pipecat/services/anthropic/llm.py index 9e75e198b..277e29f83 100644 --- a/src/pipecat/services/anthropic/llm.py +++ b/src/pipecat/services/anthropic/llm.py @@ -11,7 +11,7 @@ import io import json import re from dataclasses import dataclass -from typing import Any, Dict, List, Mapping, Optional, Union +from typing import Any, Dict, List, Optional, Union import httpx from loguru import logger @@ -35,7 +35,9 @@ from pipecat.frames.frames import ( ) from pipecat.metrics.metrics import LLMTokenUsage from pipecat.processors.aggregators.llm_response import ( + LLMAssistantAggregatorParams, LLMAssistantContextAggregator, + LLMUserAggregatorParams, LLMUserContextAggregator, ) from pipecat.processors.aggregators.openai_llm_context import ( @@ -49,10 +51,7 @@ try: from anthropic import NOT_GIVEN, AsyncAnthropic, NotGiven except ModuleNotFoundError as e: logger.error(f"Exception: {e}") - logger.error( - "In order to use Anthropic, you need to `pip install pipecat-ai[anthropic]`. " - + "Also, set `ANTHROPIC_API_KEY` environment variable." - ) + logger.error("In order to use Anthropic, you need to `pip install pipecat-ai[anthropic]`.") raise Exception(f"Missing module: {e}") @@ -120,8 +119,8 @@ class AnthropicLLMService(LLMService): self, context: OpenAILLMContext, *, - user_kwargs: Mapping[str, Any] = {}, - assistant_kwargs: Mapping[str, Any] = {}, + user_params: LLMUserAggregatorParams = LLMUserAggregatorParams(), + assistant_params: LLMAssistantAggregatorParams = LLMAssistantAggregatorParams(), ) -> AnthropicContextAggregatorPair: """Create an instance of AnthropicContextAggregatorPair from an OpenAILLMContext. Constructor keyword arguments for both the user and @@ -129,12 +128,10 @@ class AnthropicLLMService(LLMService): Args: context (OpenAILLMContext): The LLM context. - user_kwargs (Mapping[str, Any], optional): Additional keyword - arguments for the user context aggregator constructor. Defaults - to an empty mapping. - assistant_kwargs (Mapping[str, Any], optional): Additional keyword - arguments for the assistant context aggregator - constructor. Defaults to an empty mapping. + user_params (LLMUserAggregatorParams, optional): User aggregator + parameters. + assistant_params (LLMAssistantAggregatorParams, optional): User + aggregator parameters. Returns: AnthropicContextAggregatorPair: A pair of context aggregators, one @@ -146,8 +143,8 @@ class AnthropicLLMService(LLMService): if isinstance(context, OpenAILLMContext): context = AnthropicLLMContext.from_openai_context(context) - user = AnthropicUserContextAggregator(context, **user_kwargs) - assistant = AnthropicAssistantContextAggregator(context, **assistant_kwargs) + user = AnthropicUserContextAggregator(context, params=user_params) + assistant = AnthropicAssistantContextAggregator(context, params=assistant_params) return AnthropicContextAggregatorPair(_user=user, _assistant=assistant) async def _process_context(self, context: OpenAILLMContext): diff --git a/src/pipecat/services/aws/tts.py b/src/pipecat/services/aws/tts.py index cc9ce6457..db6e168ab 100644 --- a/src/pipecat/services/aws/tts.py +++ b/src/pipecat/services/aws/tts.py @@ -231,9 +231,9 @@ class PollyTTSService(TTSService): yield TTSStartedFrame() - chunk_size = 8192 - for i in range(0, len(audio_data), chunk_size): - chunk = audio_data[i : i + chunk_size] + CHUNK_SIZE = 1024 + for i in range(0, len(audio_data), CHUNK_SIZE): + chunk = audio_data[i : i + CHUNK_SIZE] if len(chunk) > 0: await self.stop_ttfb_metrics() frame = TTSAudioRawFrame(chunk, self.sample_rate, 1) diff --git a/src/pipecat/services/deepgram/stt.py b/src/pipecat/services/deepgram/stt.py index 088b77829..7b5209e0b 100644 --- a/src/pipecat/services/deepgram/stt.py +++ b/src/pipecat/services/deepgram/stt.py @@ -45,6 +45,7 @@ class DeepgramSTTService(STTService): *, api_key: str, url: str = "", + base_url: str = "", sample_rate: Optional[int] = None, live_options: Optional[LiveOptions] = None, addons: Optional[Dict] = None, @@ -53,6 +54,17 @@ class DeepgramSTTService(STTService): sample_rate = sample_rate or (live_options.sample_rate if live_options else None) super().__init__(sample_rate=sample_rate, **kwargs) + if url: + import warnings + + with warnings.catch_warnings(): + warnings.simplefilter("always") + warnings.warn( + "Parameter 'url' is deprecated, use 'base_url' instead.", + DeprecationWarning, + ) + base_url = url + default_options = LiveOptions( encoding="linear16", language=Language.EN, @@ -81,7 +93,7 @@ class DeepgramSTTService(STTService): self._client = DeepgramClient( api_key, config=DeepgramClientOptions( - url=url, + url=base_url, options={"keepalive": "true"}, # verbose=logging.DEBUG ), ) diff --git a/src/pipecat/services/deepgram/tts.py b/src/pipecat/services/deepgram/tts.py index 95e08e7af..ec8a755a0 100644 --- a/src/pipecat/services/deepgram/tts.py +++ b/src/pipecat/services/deepgram/tts.py @@ -4,7 +4,6 @@ # SPDX-License-Identifier: BSD 2-Clause License # -import asyncio from typing import AsyncGenerator, Optional from loguru import logger @@ -19,7 +18,7 @@ from pipecat.frames.frames import ( from pipecat.services.tts_service import TTSService try: - from deepgram import DeepgramClient, SpeakOptions + from deepgram import DeepgramClient, DeepgramClientOptions, SpeakOptions except ModuleNotFoundError as e: logger.error(f"Exception: {e}") logger.error("In order to use Deepgram, you need to `pip install pipecat-ai[deepgram]`.") @@ -32,6 +31,7 @@ class DeepgramTTSService(TTSService): *, api_key: str, voice: str = "aura-helios-en", + base_url: str = "", sample_rate: Optional[int] = None, encoding: str = "linear16", **kwargs, @@ -42,7 +42,9 @@ class DeepgramTTSService(TTSService): "encoding": encoding, } self.set_voice(voice) - self._deepgram_client = DeepgramClient(api_key=api_key) + + client_options = DeepgramClientOptions(url=base_url) + self._deepgram_client = DeepgramClient(api_key, config=client_options) def can_generate_metrics(self) -> bool: return True @@ -60,8 +62,8 @@ class DeepgramTTSService(TTSService): try: await self.start_ttfb_metrics() - response = await asyncio.to_thread( - self._deepgram_client.speak.v("1").stream, {"text": text}, options + response = await self._deepgram_client.speak.asyncrest.v("1").stream_memory( + {"text": text}, options ) await self.start_tts_usage_metrics(text) diff --git a/src/pipecat/services/elevenlabs/tts.py b/src/pipecat/services/elevenlabs/tts.py index d3a066882..acb4a7b12 100644 --- a/src/pipecat/services/elevenlabs/tts.py +++ b/src/pipecat/services/elevenlabs/tts.py @@ -18,6 +18,7 @@ from pipecat.frames.frames import ( EndFrame, ErrorFrame, Frame, + LLMFullResponseEndFrame, StartFrame, StartInterruptionFrame, TTSAudioRawFrame, @@ -25,7 +26,7 @@ from pipecat.frames.frames import ( TTSStoppedFrame, ) from pipecat.processors.frame_processor import FrameDirection -from pipecat.services.tts_service import InterruptibleWordTTSService, TTSService +from pipecat.services.tts_service import InterruptibleWordTTSService, WordTTSService from pipecat.transcriptions.language import Language # See .env.example for ElevenLabs configuration needed @@ -441,8 +442,8 @@ class ElevenLabsTTSService(InterruptibleWordTTSService): logger.error(f"{self} exception: {e}") -class ElevenLabsHttpTTSService(TTSService): - """ElevenLabs Text-to-Speech service using HTTP streaming. +class ElevenLabsHttpTTSService(WordTTSService): + """ElevenLabs Text-to-Speech service using HTTP streaming with word timestamps. Args: api_key: ElevenLabs API key @@ -475,7 +476,13 @@ class ElevenLabsHttpTTSService(TTSService): params: InputParams = InputParams(), **kwargs, ): - super().__init__(sample_rate=sample_rate, **kwargs) + super().__init__( + aggregate_sentences=True, + push_text_frames=False, + push_stop_frames=True, + sample_rate=sample_rate, + **kwargs, + ) self._api_key = api_key self._base_url = base_url @@ -498,34 +505,136 @@ class ElevenLabsHttpTTSService(TTSService): self._output_format = "" # initialized in start() self._voice_settings = self._set_voice_settings() + # Track cumulative time to properly sequence word timestamps across utterances + self._cumulative_time = 0 + self._started = False + + # Store previous text for context within a turn + self._previous_text = "" + + def language_to_service_language(self, language: Language) -> Optional[str]: + """Convert pipecat Language to ElevenLabs language code.""" + return language_to_elevenlabs_language(language) + def can_generate_metrics(self) -> bool: + """Indicate that this service can generate usage metrics.""" return True def _set_voice_settings(self): return build_elevenlabs_voice_settings(self._settings) + def _reset_state(self): + """Reset internal state variables.""" + self._cumulative_time = 0 + self._started = False + self._previous_text = "" + logger.debug(f"{self}: Reset internal state") + async def start(self, frame: StartFrame): + """Initialize the service upon receiving a StartFrame.""" await super().start(frame) self._output_format = output_format_from_sample_rate(self.sample_rate) + self._reset_state() - async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: - """Generate speech from text using ElevenLabs streaming API. + async def push_frame(self, frame: Frame, direction: FrameDirection = FrameDirection.DOWNSTREAM): + await super().push_frame(frame, direction) + if isinstance(frame, (StartInterruptionFrame, TTSStoppedFrame)): + # Reset timing on interruption or stop + self._reset_state() + + if isinstance(frame, TTSStoppedFrame): + await self.add_word_timestamps([("LLMFullResponseEndFrame", 0), ("Reset", 0)]) + + elif isinstance(frame, LLMFullResponseEndFrame): + # End of turn - reset previous text + self._previous_text = "" + + def calculate_word_times(self, alignment_info: Mapping[str, Any]) -> List[Tuple[str, float]]: + """Calculate word timing from character alignment data. + + Example input data: + { + "characters": [" ", "H", "e", "l", "l", "o", " ", "w", "o", "r", "l", "d"], + "character_start_times_seconds": [0.0, 0.1, 0.15, 0.2, 0.25, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9], + "character_end_times_seconds": [0.1, 0.15, 0.2, 0.25, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9, 1.0] + } + + Would produce word times (with cumulative_time=0): + [("Hello", 0.1), ("world", 0.5)] Args: - text: The text to convert to speech + alignment_info: Character timing data from ElevenLabs + + Returns: + List of (word, timestamp) pairs + """ + chars = alignment_info.get("characters", []) + char_start_times = alignment_info.get("character_start_times_seconds", []) + + if not chars or not char_start_times or len(chars) != len(char_start_times): + logger.warning( + f"Invalid alignment data: chars={len(chars)}, times={len(char_start_times)}" + ) + return [] + + # Build the words and find their start times + words = [] + word_start_times = [] + current_word = "" + first_char_idx = -1 + + for i, char in enumerate(chars): + if char == " ": + if current_word: # Only add non-empty words + words.append(current_word) + # Use time of the first character of the word, offset by cumulative time + word_start_times.append( + self._cumulative_time + char_start_times[first_char_idx] + ) + current_word = "" + first_char_idx = -1 + else: + if not current_word: # This is the first character of a new word + first_char_idx = i + current_word += char + + # Don't forget the last word if there's no trailing space + if current_word and first_char_idx >= 0: + words.append(current_word) + word_start_times.append(self._cumulative_time + char_start_times[first_char_idx]) + + # Create word-time pairs + word_times = list(zip(words, word_start_times)) + + return word_times + + async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: + """Generate speech from text using ElevenLabs streaming API with timestamps. + + Makes a request to the ElevenLabs API to generate audio and timing data. + Tracks the duration of each utterance to ensure correct sequencing. + Includes previous text as context for better prosody continuity. + + Args: + text: Text to convert to speech Yields: - Frames containing audio data and status information + Audio and control frames """ logger.debug(f"{self}: Generating TTS [{text}]") - url = f"{self._base_url}/v1/text-to-speech/{self._voice_id}/stream" + # Use the with-timestamps endpoint + url = f"{self._base_url}/v1/text-to-speech/{self._voice_id}/stream/with-timestamps" payload: Dict[str, Union[str, Dict[str, Union[float, bool]]]] = { "text": text, "model_id": self._model_name, } + # Include previous text as context if available + if self._previous_text: + payload["previous_text"] = self._previous_text + if self._voice_settings: payload["voice_settings"] = self._voice_settings @@ -550,8 +659,6 @@ class ElevenLabsHttpTTSService(TTSService): if self._settings["optimize_streaming_latency"] is not None: params["optimize_streaming_latency"] = self._settings["optimize_streaming_latency"] - logger.debug(f"ElevenLabs request - payload: {payload}, params: {params}") - try: await self.start_ttfb_metrics() @@ -566,17 +673,66 @@ class ElevenLabsHttpTTSService(TTSService): await self.start_tts_usage_metrics(text) - # Process the streaming response - CHUNK_SIZE = 1024 + # Start TTS sequence if not already started + if not self._started: + self.start_word_timestamps() + yield TTSStartedFrame() + self._started = True + + # Track the duration of this utterance based on the last character's end time + utterance_duration = 0 + async for line in response.content: + line_str = line.decode("utf-8").strip() + if not line_str: + continue + + try: + # Parse the JSON object + data = json.loads(line_str) + + # Process audio if present + if data and "audio_base64" in data: + await self.stop_ttfb_metrics() + audio = base64.b64decode(data["audio_base64"]) + yield TTSAudioRawFrame(audio, self.sample_rate, 1) + + # Process alignment if present + if data and "alignment" in data: + alignment = data["alignment"] + if alignment: # Ensure alignment is not None + # Get end time of the last character in this chunk + char_end_times = alignment.get("character_end_times_seconds", []) + if char_end_times: + chunk_end_time = char_end_times[-1] + # Update to the longest end time seen so far + utterance_duration = max(utterance_duration, chunk_end_time) + + # Calculate word timestamps + word_times = self.calculate_word_times(alignment) + if word_times: + await self.add_word_timestamps(word_times) + except json.JSONDecodeError as e: + logger.warning(f"Failed to parse JSON from stream: {e}") + continue + except Exception as e: + logger.error(f"Error processing response: {e}", exc_info=True) + continue + + # After processing all chunks, add the total utterance duration + # to the cumulative time to ensure next utterance starts after this one + if utterance_duration > 0: + self._cumulative_time += utterance_duration + + # Append the current text to previous_text for context continuity + # Only add a space if there's already text + if self._previous_text: + self._previous_text += " " + text + else: + self._previous_text = text - yield TTSStartedFrame() - async for chunk in response.content.iter_chunked(CHUNK_SIZE): - if len(chunk) > 0: - await self.stop_ttfb_metrics() - yield TTSAudioRawFrame(chunk, self.sample_rate, 1) except Exception as e: logger.error(f"Error in run_tts: {e}") yield ErrorFrame(error=str(e)) finally: await self.stop_ttfb_metrics() - yield TTSStoppedFrame() + # Let the parent class handle TTSStoppedFrame diff --git a/src/pipecat/services/gemini_multimodal_live/gemini.py b/src/pipecat/services/gemini_multimodal_live/gemini.py index 1051d2f60..d953e5065 100644 --- a/src/pipecat/services/gemini_multimodal_live/gemini.py +++ b/src/pipecat/services/gemini_multimodal_live/gemini.py @@ -10,9 +10,8 @@ import json import time from dataclasses import dataclass from enum import Enum -from typing import Any, Dict, List, Mapping, Optional, Union +from typing import Any, Dict, List, Optional, Union -import websockets from loguru import logger from pydantic import BaseModel, Field @@ -45,6 +44,10 @@ from pipecat.frames.frames import ( UserStoppedSpeakingFrame, ) from pipecat.metrics.metrics import LLMTokenUsage +from pipecat.processors.aggregators.llm_response import ( + LLMAssistantAggregatorParams, + LLMUserAggregatorParams, +) from pipecat.processors.aggregators.openai_llm_context import ( OpenAILLMContext, OpenAILLMContextFrame, @@ -61,6 +64,13 @@ from pipecat.utils.time import time_now_iso8601 from . import events from .audio_transcriber import AudioTranscriber +try: + import websockets +except ModuleNotFoundError as e: + logger.error(f"Exception: {e}") + logger.error("In order to use Google AI, you need to `pip install pipecat-ai[google]`.") + raise Exception(f"Missing module: {e}") + def language_to_gemini_language(language: Language) -> Optional[str]: """Maps a Language enum value to a Gemini Live supported language code. @@ -871,8 +881,8 @@ class GeminiMultimodalLiveLLMService(LLMService): self, context: OpenAILLMContext, *, - user_kwargs: Mapping[str, Any] = {}, - assistant_kwargs: Mapping[str, Any] = {}, + user_params: LLMUserAggregatorParams = LLMUserAggregatorParams(), + assistant_params: LLMAssistantAggregatorParams = LLMAssistantAggregatorParams(), ) -> GeminiMultimodalLiveContextAggregatorPair: """Create an instance of GeminiMultimodalLiveContextAggregatorPair from an OpenAILLMContext. Constructor keyword arguments for both the user and @@ -880,12 +890,10 @@ class GeminiMultimodalLiveLLMService(LLMService): Args: context (OpenAILLMContext): The LLM context. - user_kwargs (Mapping[str, Any], optional): Additional keyword - arguments for the user context aggregator constructor. Defaults - to an empty mapping. - assistant_kwargs (Mapping[str, Any], optional): Additional keyword - arguments for the assistant context aggregator - constructor. Defaults to an empty mapping. + user_params (LLMUserAggregatorParams, optional): User aggregator + parameters. + assistant_params (LLMAssistantAggregatorParams, optional): User + aggregator parameters. Returns: GeminiMultimodalLiveContextAggregatorPair: A pair of context @@ -896,11 +904,8 @@ class GeminiMultimodalLiveLLMService(LLMService): context.set_llm_adapter(self.get_llm_adapter()) GeminiMultimodalLiveContext.upgrade(context) - user = GeminiMultimodalLiveUserContextAggregator(context, **user_kwargs) + user = GeminiMultimodalLiveUserContextAggregator(context, params=user_params) - default_assistant_kwargs = {"expect_stripped_words": True} - default_assistant_kwargs.update(assistant_kwargs) - assistant = GeminiMultimodalLiveAssistantContextAggregator( - context, **default_assistant_kwargs - ) + assistant_params.expect_stripped_words = True + assistant = GeminiMultimodalLiveAssistantContextAggregator(context, params=assistant_params) return GeminiMultimodalLiveContextAggregatorPair(_user=user, _assistant=assistant) diff --git a/src/pipecat/services/google/llm.py b/src/pipecat/services/google/llm.py index a9dd0cb3a..bf9714817 100644 --- a/src/pipecat/services/google/llm.py +++ b/src/pipecat/services/google/llm.py @@ -9,21 +9,14 @@ import io import json import os import uuid - -from google.api_core.exceptions import DeadlineExceeded - -from pipecat.adapters.services.gemini_adapter import GeminiLLMAdapter - -# Suppress gRPC fork warnings -os.environ["GRPC_ENABLE_FORK_SUPPORT"] = "false" - from dataclasses import dataclass -from typing import Any, Dict, List, Mapping, Optional, Union +from typing import Any, Dict, List, Optional from loguru import logger from PIL import Image from pydantic import BaseModel, Field +from pipecat.adapters.services.gemini_adapter import GeminiLLMAdapter from pipecat.frames.frames import ( AudioRawFrame, Frame, @@ -39,6 +32,10 @@ from pipecat.frames.frames import ( VisionImageRawFrame, ) from pipecat.metrics.metrics import LLMTokenUsage +from pipecat.processors.aggregators.llm_response import ( + LLMAssistantAggregatorParams, + LLMUserAggregatorParams, +) from pipecat.processors.aggregators.openai_llm_context import ( OpenAILLMContext, OpenAILLMContextFrame, @@ -51,11 +48,14 @@ from pipecat.services.openai.llm import ( OpenAIUserContextAggregator, ) +# Suppress gRPC fork warnings +os.environ["GRPC_ENABLE_FORK_SUPPORT"] = "false" + try: import google.ai.generativelanguage as glm import google.generativeai as gai + from google.api_core.exceptions import DeadlineExceeded from google.generativeai.types import GenerationConfig - except ModuleNotFoundError as e: logger.error(f"Exception: {e}") logger.error("In order to use Google AI, you need to `pip install pipecat-ai[google]`.") @@ -686,8 +686,8 @@ class GoogleLLMService(LLMService): self, context: OpenAILLMContext, *, - user_kwargs: Mapping[str, Any] = {}, - assistant_kwargs: Mapping[str, Any] = {}, + user_params: LLMUserAggregatorParams = LLMUserAggregatorParams(), + assistant_params: LLMAssistantAggregatorParams = LLMAssistantAggregatorParams(), ) -> GoogleContextAggregatorPair: """Create an instance of GoogleContextAggregatorPair from an OpenAILLMContext. Constructor keyword arguments for both the user and @@ -695,12 +695,10 @@ class GoogleLLMService(LLMService): Args: context (OpenAILLMContext): The LLM context. - user_kwargs (Mapping[str, Any], optional): Additional keyword - arguments for the user context aggregator constructor. Defaults - to an empty mapping. - assistant_kwargs (Mapping[str, Any], optional): Additional keyword - arguments for the assistant context aggregator - constructor. Defaults to an empty mapping. + user_params (LLMUserAggregatorParams, optional): User aggregator + parameters. + assistant_params (LLMAssistantAggregatorParams, optional): User + aggregator parameters. Returns: GoogleContextAggregatorPair: A pair of context aggregators, one for @@ -712,6 +710,6 @@ class GoogleLLMService(LLMService): if isinstance(context, OpenAILLMContext): context = GoogleLLMContext.upgrade_to_google(context) - user = GoogleUserContextAggregator(context, **user_kwargs) - assistant = GoogleAssistantContextAggregator(context, **assistant_kwargs) + user = GoogleUserContextAggregator(context, params=user_params) + assistant = GoogleAssistantContextAggregator(context, params=assistant_params) return GoogleContextAggregatorPair(_user=user, _assistant=assistant) diff --git a/src/pipecat/services/google/llm_vertex.py b/src/pipecat/services/google/llm_vertex.py index 1a23fe7d5..165146c04 100644 --- a/src/pipecat/services/google/llm_vertex.py +++ b/src/pipecat/services/google/llm_vertex.py @@ -65,7 +65,9 @@ class GoogleVertexLLMService(OpenAILLMService): base_url = self._get_base_url(params) self._api_key = self._get_api_token(credentials, credentials_path) - super().__init__(api_key=self._api_key, base_url=base_url, model=model, **kwargs) + super().__init__( + api_key=self._api_key, base_url=base_url, model=model, params=params, **kwargs + ) @staticmethod def _get_base_url(params: InputParams) -> str: diff --git a/src/pipecat/services/google/tts.py b/src/pipecat/services/google/tts.py index ef9023a8c..5bfdada21 100644 --- a/src/pipecat/services/google/tts.py +++ b/src/pipecat/services/google/tts.py @@ -346,9 +346,9 @@ class GoogleTTSService(TTSService): audio_content = response.audio_content[44:] # Read and yield audio data in chunks - chunk_size = 8192 - for i in range(0, len(audio_content), chunk_size): - chunk = audio_content[i : i + chunk_size] + CHUNK_SIZE = 1024 + for i in range(0, len(audio_content), CHUNK_SIZE): + chunk = audio_content[i : i + CHUNK_SIZE] if not chunk: break await self.stop_ttfb_metrics() diff --git a/src/pipecat/services/grok/llm.py b/src/pipecat/services/grok/llm.py index 57517eb3e..90c8df14f 100644 --- a/src/pipecat/services/grok/llm.py +++ b/src/pipecat/services/grok/llm.py @@ -5,11 +5,14 @@ # from dataclasses import dataclass -from typing import Any, Mapping from loguru import logger from pipecat.metrics.metrics import LLMTokenUsage +from pipecat.processors.aggregators.llm_response import ( + LLMAssistantAggregatorParams, + LLMUserAggregatorParams, +) from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext from pipecat.services.openai.llm import ( OpenAIAssistantContextAggregator, @@ -124,8 +127,8 @@ class GrokLLMService(OpenAILLMService): self, context: OpenAILLMContext, *, - user_kwargs: Mapping[str, Any] = {}, - assistant_kwargs: Mapping[str, Any] = {}, + user_params: LLMUserAggregatorParams = LLMUserAggregatorParams(), + assistant_params: LLMAssistantAggregatorParams = LLMAssistantAggregatorParams(), ) -> GrokContextAggregatorPair: """Create an instance of GrokContextAggregatorPair from an OpenAILLMContext. Constructor keyword arguments for both the user and @@ -133,12 +136,10 @@ class GrokLLMService(OpenAILLMService): Args: context (OpenAILLMContext): The LLM context. - user_kwargs (Mapping[str, Any], optional): Additional keyword - arguments for the user context aggregator constructor. Defaults - to an empty mapping. - assistant_kwargs (Mapping[str, Any], optional): Additional keyword - arguments for the assistant context aggregator - constructor. Defaults to an empty mapping. + user_params (LLMUserAggregatorParams, optional): User aggregator + parameters. + assistant_params (LLMAssistantAggregatorParams, optional): User + aggregator parameters. Returns: GrokContextAggregatorPair: A pair of context aggregators, one for @@ -148,6 +149,6 @@ class GrokLLMService(OpenAILLMService): """ context.set_llm_adapter(self.get_llm_adapter()) - user = OpenAIUserContextAggregator(context, **user_kwargs) - assistant = OpenAIAssistantContextAggregator(context, **assistant_kwargs) + user = OpenAIUserContextAggregator(context, params=user_params) + assistant = OpenAIAssistantContextAggregator(context, params=assistant_params) return GrokContextAggregatorPair(_user=user, _assistant=assistant) diff --git a/src/pipecat/services/llm_service.py b/src/pipecat/services/llm_service.py index 7f7238b47..6ac841f25 100644 --- a/src/pipecat/services/llm_service.py +++ b/src/pipecat/services/llm_service.py @@ -6,7 +6,7 @@ import asyncio from dataclasses import dataclass -from typing import Any, Mapping, Optional, Set, Tuple, Type +from typing import Any, Optional, Set, Tuple, Type from loguru import logger @@ -20,6 +20,10 @@ from pipecat.frames.frames import ( StartInterruptionFrame, UserImageRequestFrame, ) +from pipecat.processors.aggregators.llm_response import ( + LLMAssistantAggregatorParams, + LLMUserAggregatorParams, +) from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext from pipecat.processors.frame_processor import FrameDirection from pipecat.services.ai_service import AIService @@ -55,8 +59,8 @@ class LLMService(AIService): self, context: OpenAILLMContext, *, - user_kwargs: Mapping[str, Any] = {}, - assistant_kwargs: Mapping[str, Any] = {}, + user_params: LLMUserAggregatorParams = LLMUserAggregatorParams(), + assistant_params: LLMAssistantAggregatorParams = LLMAssistantAggregatorParams(), ) -> Any: pass diff --git a/src/pipecat/services/openai/llm.py b/src/pipecat/services/openai/llm.py index 5ca51a479..07b564fe1 100644 --- a/src/pipecat/services/openai/llm.py +++ b/src/pipecat/services/openai/llm.py @@ -6,7 +6,7 @@ import json from dataclasses import dataclass -from typing import Any, Mapping +from typing import Any from pipecat.frames.frames import ( FunctionCallCancelFrame, @@ -15,7 +15,9 @@ from pipecat.frames.frames import ( UserImageRawFrame, ) from pipecat.processors.aggregators.llm_response import ( + LLMAssistantAggregatorParams, LLMAssistantContextAggregator, + LLMUserAggregatorParams, LLMUserContextAggregator, ) from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext @@ -38,7 +40,7 @@ class OpenAILLMService(BaseOpenAILLMService): def __init__( self, *, - model: str = "gpt-4o", + model: str = "gpt-4.1", params: BaseOpenAILLMService.InputParams = BaseOpenAILLMService.InputParams(), **kwargs, ): @@ -48,8 +50,8 @@ class OpenAILLMService(BaseOpenAILLMService): self, context: OpenAILLMContext, *, - user_kwargs: Mapping[str, Any] = {}, - assistant_kwargs: Mapping[str, Any] = {}, + user_params: LLMUserAggregatorParams = LLMUserAggregatorParams(), + assistant_params: LLMAssistantAggregatorParams = LLMAssistantAggregatorParams(), ) -> OpenAIContextAggregatorPair: """Create an instance of OpenAIContextAggregatorPair from an OpenAILLMContext. Constructor keyword arguments for both the user and @@ -57,12 +59,8 @@ class OpenAILLMService(BaseOpenAILLMService): Args: context (OpenAILLMContext): The LLM context. - user_kwargs (Mapping[str, Any], optional): Additional keyword - arguments for the user context aggregator constructor. Defaults - to an empty mapping. - assistant_kwargs (Mapping[str, Any], optional): Additional keyword - arguments for the assistant context aggregator - constructor. Defaults to an empty mapping. + user_params (LLMUserAggregatorParams, optional): User aggregator parameters. + assistant_params (LLMAssistantAggregatorParams, optional): User aggregator parameters. Returns: OpenAIContextAggregatorPair: A pair of context aggregators, one for @@ -71,8 +69,8 @@ class OpenAILLMService(BaseOpenAILLMService): """ context.set_llm_adapter(self.get_llm_adapter()) - user = OpenAIUserContextAggregator(context, **user_kwargs) - assistant = OpenAIAssistantContextAggregator(context, **assistant_kwargs) + user = OpenAIUserContextAggregator(context, params=user_params) + assistant = OpenAIAssistantContextAggregator(context, params=assistant_params) return OpenAIContextAggregatorPair(_user=user, _assistant=assistant) diff --git a/src/pipecat/services/openai_realtime_beta/openai.py b/src/pipecat/services/openai_realtime_beta/openai.py index 6da1a3109..94d848b77 100644 --- a/src/pipecat/services/openai_realtime_beta/openai.py +++ b/src/pipecat/services/openai_realtime_beta/openai.py @@ -8,19 +8,9 @@ import base64 import json import time from dataclasses import dataclass -from typing import Any, Mapping from loguru import logger -try: - import websockets -except ModuleNotFoundError as e: - logger.error(f"Exception: {e}") - logger.error( - "In order to use OpenAI, you need to `pip install pipecat-ai[openai]`. Also, set `OPENAI_API_KEY` environment variable." - ) - raise Exception(f"Missing module: {e}") - from pipecat.adapters.services.open_ai_realtime_adapter import OpenAIRealtimeLLMAdapter from pipecat.frames.frames import ( BotStoppedSpeakingFrame, @@ -48,6 +38,10 @@ from pipecat.frames.frames import ( UserStoppedSpeakingFrame, ) from pipecat.metrics.metrics import LLMTokenUsage +from pipecat.processors.aggregators.llm_response import ( + LLMAssistantAggregatorParams, + LLMUserAggregatorParams, +) from pipecat.processors.aggregators.openai_llm_context import ( OpenAILLMContext, OpenAILLMContextFrame, @@ -65,6 +59,13 @@ from .context import ( ) from .frames import RealtimeFunctionCallResultFrame, RealtimeMessagesUpdateFrame +try: + import websockets +except ModuleNotFoundError as e: + logger.error(f"Exception: {e}") + logger.error("In order to use OpenAI, you need to `pip install pipecat-ai[openai]`.") + raise Exception(f"Missing module: {e}") + @dataclass class CurrentAudioResponse: @@ -650,8 +651,8 @@ class OpenAIRealtimeBetaLLMService(LLMService): self, context: OpenAILLMContext, *, - user_kwargs: Mapping[str, Any] = {}, - assistant_kwargs: Mapping[str, Any] = {}, + user_params: LLMUserAggregatorParams = LLMUserAggregatorParams(), + assistant_params: LLMAssistantAggregatorParams = LLMAssistantAggregatorParams(), ) -> OpenAIContextAggregatorPair: """Create an instance of OpenAIContextAggregatorPair from an OpenAILLMContext. Constructor keyword arguments for both the user and @@ -659,12 +660,10 @@ class OpenAIRealtimeBetaLLMService(LLMService): Args: context (OpenAILLMContext): The LLM context. - user_kwargs (Mapping[str, Any], optional): Additional keyword - arguments for the user context aggregator constructor. Defaults - to an empty mapping. - assistant_kwargs (Mapping[str, Any], optional): Additional keyword - arguments for the assistant context aggregator - constructor. Defaults to an empty mapping. + user_params (LLMUserAggregatorParams, optional): User aggregator + parameters. + assistant_params (LLMAssistantAggregatorParams, optional): User + aggregator parameters. Returns: OpenAIContextAggregatorPair: A pair of context aggregators, one for @@ -675,9 +674,8 @@ class OpenAIRealtimeBetaLLMService(LLMService): context.set_llm_adapter(self.get_llm_adapter()) OpenAIRealtimeLLMContext.upgrade_to_realtime(context) - user = OpenAIRealtimeUserContextAggregator(context, **user_kwargs) + user = OpenAIRealtimeUserContextAggregator(context, params=user_params) - default_assistant_kwargs = {"expect_stripped_words": False} - default_assistant_kwargs.update(assistant_kwargs) - assistant = OpenAIRealtimeAssistantContextAggregator(context, **default_assistant_kwargs) + assistant_params.expect_stripped_words = False + assistant = OpenAIRealtimeAssistantContextAggregator(context, params=assistant_params) return OpenAIContextAggregatorPair(_user=user, _assistant=assistant) diff --git a/src/pipecat/services/openpipe/llm.py b/src/pipecat/services/openpipe/llm.py index 63e499714..2a2dd1d26 100644 --- a/src/pipecat/services/openpipe/llm.py +++ b/src/pipecat/services/openpipe/llm.py @@ -25,7 +25,7 @@ class OpenPipeLLMService(OpenAILLMService): def __init__( self, *, - model: str = "gpt-4o", + model: str = "gpt-4.1", api_key: Optional[str] = None, base_url: Optional[str] = None, openpipe_api_key: Optional[str] = None, diff --git a/src/pipecat/services/tavus/video.py b/src/pipecat/services/tavus/video.py index cbc31a2ba..3699ba512 100644 --- a/src/pipecat/services/tavus/video.py +++ b/src/pipecat/services/tavus/video.py @@ -6,7 +6,9 @@ """This module implements Tavus as a sink transport layer""" +import asyncio import base64 +from typing import Optional import aiohttp from loguru import logger @@ -16,6 +18,7 @@ from pipecat.frames.frames import ( CancelFrame, EndFrame, Frame, + StartFrame, StartInterruptionFrame, TransportMessageUrgentFrame, TTSAudioRawFrame, @@ -50,6 +53,10 @@ class TavusVideoService(AIService): self._resampler = create_default_resampler() + self._audio_buffer = bytearray() + self._queue = asyncio.Queue() + self._send_task: Optional[asyncio.Task] = None + async def initialize(self) -> str: url = "https://tavusapi.com/v2/conversations" headers = {"Content-Type": "application/json", "x-api-key": self._api_key} @@ -78,45 +85,98 @@ class TavusVideoService(AIService): logger.debug(f"TavusVideoService persona grabbed {response_json}") return response_json["persona_name"] + async def start(self, frame: StartFrame): + await super().start(frame) + await self._create_send_task() + async def stop(self, frame: EndFrame): await super().stop(frame) await self._end_conversation() + await self._cancel_send_task() async def cancel(self, frame: CancelFrame): await super().cancel(frame) await self._end_conversation() + await self._cancel_send_task() - async def _end_conversation(self) -> None: + async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + + if isinstance(frame, StartInterruptionFrame): + await self._handle_interruptions() + await self.push_frame(frame, direction) + elif isinstance(frame, TTSStartedFrame): + await self.start_processing_metrics() + await self.start_ttfb_metrics() + self._current_idx_str = str(frame.id) + elif isinstance(frame, TTSAudioRawFrame): + await self._queue_audio(frame.audio, frame.sample_rate, done=False) + elif isinstance(frame, TTSStoppedFrame): + await self._queue_audio(b"\x00\x00", self._sample_rate, done=True) + await self.stop_ttfb_metrics() + await self.stop_processing_metrics() + else: + await self.push_frame(frame, direction) + + async def _handle_interruptions(self): + await self._cancel_send_task() + await self._create_send_task() + await self._send_interrupt_message() + + async def _end_conversation(self): url = f"https://tavusapi.com/v2/conversations/{self._conversation_id}/end" headers = {"Content-Type": "application/json", "x-api-key": self._api_key} async with self._session.post(url, headers=headers) as r: r.raise_for_status() - async def _encode_audio_and_send(self, audio: bytes, in_rate: int, done: bool) -> None: + async def _queue_audio(self, audio: bytes, in_rate: int, done: bool): + await self._queue.put((audio, in_rate, done)) + + async def _create_send_task(self): + if not self._send_task: + self._queue = asyncio.Queue() + self._send_task = self.create_task(self._send_task_handler()) + + async def _cancel_send_task(self): + if self._send_task: + await self.cancel_task(self._send_task) + self._send_task = None + + async def _send_task_handler(self): + # Daily app-messages have a 4kb limit and also a rate limit of 20 + # messages per second. Below, we only consider the rate limit because 1 + # second of a 24000 sample rate would be 48000 bytes (16-bit samples and + # 1 channel). So, that is 48000 / 20 = 2400, which is below the 4kb + # limit (even including base64 encoding). For a sample rate of 16000, + # that would be 32000 / 20 = 1600. + MAX_CHUNK_SIZE = int((self._sample_rate * 2) / 20) + SLEEP_TIME = 1 / 20 + + audio_buffer = bytearray() + while True: + (audio, in_rate, done) = await self._queue.get() + + if done: + # Send any remaining audio. + if len(audio_buffer) > 0: + await self._encode_audio_and_send(bytes(audio_buffer), done) + await self._encode_audio_and_send(audio, done) + audio_buffer.clear() + else: + audio = await self._resampler.resample(audio, in_rate, self._sample_rate) + audio_buffer.extend(audio) + while len(audio_buffer) >= MAX_CHUNK_SIZE: + chunk = audio_buffer[:MAX_CHUNK_SIZE] + audio_buffer = audio_buffer[MAX_CHUNK_SIZE:] + await self._encode_audio_and_send(bytes(chunk), done) + await asyncio.sleep(SLEEP_TIME) + + async def _encode_audio_and_send(self, audio: bytes, done: bool): """Encodes audio to base64 and sends it to Tavus""" - if not done: - audio = await self._resampler.resample(audio, in_rate, self._sample_rate) audio_base64 = base64.b64encode(audio).decode("utf-8") logger.trace(f"{self}: sending {len(audio)} bytes") await self._send_audio_message(audio_base64, done=done) - async def process_frame(self, frame: Frame, direction: FrameDirection): - await super().process_frame(frame, direction) - if isinstance(frame, TTSStartedFrame): - await self.start_processing_metrics() - await self.start_ttfb_metrics() - self._current_idx_str = str(frame.id) - elif isinstance(frame, TTSAudioRawFrame): - await self._encode_audio_and_send(frame.audio, frame.sample_rate, done=False) - elif isinstance(frame, TTSStoppedFrame): - await self._encode_audio_and_send(b"\x00", self._sample_rate, done=True) - await self.stop_ttfb_metrics() - await self.stop_processing_metrics() - elif isinstance(frame, StartInterruptionFrame): - await self._send_interrupt_message() - else: - await self.push_frame(frame, direction) - async def _send_interrupt_message(self) -> None: transport_frame = TransportMessageUrgentFrame( message={ @@ -127,7 +187,7 @@ class TavusVideoService(AIService): ) await self.push_frame(transport_frame) - async def _send_audio_message(self, audio_base64: str, done: bool) -> None: + async def _send_audio_message(self, audio_base64: str, done: bool): transport_frame = TransportMessageUrgentFrame( message={ "message_type": "conversation", diff --git a/src/pipecat/transports/base_output.py b/src/pipecat/transports/base_output.py index d1a26a173..797f2ea16 100644 --- a/src/pipecat/transports/base_output.py +++ b/src/pipecat/transports/base_output.py @@ -386,10 +386,13 @@ class BaseOutputTransport(FrameProcessor): async def _draw_image(self, frame: OutputImageRawFrame): desired_size = (self._params.camera_out_width, self._params.camera_out_height) + # TODO: we should refactor in the future to support dynamic resolutions + # which is kind of what happens in P2P connections. + # We need to add support for that inside the DailyTransport if frame.size != desired_size: image = Image.frombytes(frame.format, frame.size, frame.image) resized_image = image.resize(desired_size) - logger.warning(f"{frame} does not have the expected size {desired_size}, resizing") + # logger.warning(f"{frame} does not have the expected size {desired_size}, resizing") frame = OutputImageRawFrame( resized_image.tobytes(), resized_image.size, resized_image.format ) diff --git a/src/pipecat/transports/services/helpers/daily_rest.py b/src/pipecat/transports/services/helpers/daily_rest.py index b64aba2d6..8bc6cf3ce 100644 --- a/src/pipecat/transports/services/helpers/daily_rest.py +++ b/src/pipecat/transports/services/helpers/daily_rest.py @@ -68,9 +68,9 @@ class DailyRoomProperties(BaseModel, extra="allow"): exp: Optional[float] = None enable_chat: bool = False - enable_prejoin_ui: bool = True + enable_prejoin_ui: bool = False enable_emoji_reactions: bool = False - eject_at_room_exp: bool = True + eject_at_room_exp: bool = False enable_dialout: Optional[bool] = None enable_recording: Optional[Literal["cloud", "local", "raw-tracks"]] = None geo: Optional[str] = None @@ -291,6 +291,7 @@ class DailyRESTHelper: self, room_url: str, expiry_time: float = 60 * 60, + eject_at_token_exp: bool = False, owner: bool = True, params: Optional[DailyMeetingTokenParams] = None, ) -> str: @@ -324,12 +325,16 @@ class DailyRESTHelper: if params is None: params = DailyMeetingTokenParams( properties=DailyMeetingTokenProperties( - room_name=room_name, is_owner=owner, exp=expiration + room_name=room_name, + is_owner=owner, + exp=expiration, + eject_at_token_exp=eject_at_token_exp, ) ) else: params.properties.room_name = room_name params.properties.exp = expiration + params.properties.eject_at_token_exp = eject_at_token_exp params.properties.is_owner = owner json = params.model_dump(exclude_none=True) diff --git a/tests/integration/test_integration_unified_function_calling.py b/tests/integration/test_integration_unified_function_calling.py index 26dcb0d73..09611fd3a 100644 --- a/tests/integration/test_integration_unified_function_calling.py +++ b/tests/integration/test_integration_unified_function_calling.py @@ -80,7 +80,7 @@ async def _test_llm_function_calling(llm: LLMService): @pytest.mark.skipif(os.getenv("OPENAI_API_KEY") is None, reason="OPENAI_API_KEY is not set") @pytest.mark.asyncio async def test_unified_function_calling_openai(): - llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o") + llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) # This will fail if an exception is raised await _test_llm_function_calling(llm) diff --git a/tests/test_context_aggregators.py b/tests/test_context_aggregators.py index 05734a64e..dfe210e07 100644 --- a/tests/test_context_aggregators.py +++ b/tests/test_context_aggregators.py @@ -26,7 +26,11 @@ from pipecat.frames.frames import ( UserStartedSpeakingFrame, UserStoppedSpeakingFrame, ) -from pipecat.processors.aggregators.llm_response import LLMUserContextAggregator +from pipecat.processors.aggregators.llm_response import ( + LLMAssistantAggregatorParams, + LLMUserAggregatorParams, + LLMUserContextAggregator, +) from pipecat.processors.aggregators.openai_llm_context import ( OpenAILLMContext, OpenAILLMContextFrame, @@ -163,7 +167,9 @@ class BaseTestUserContextAggregator: assert self.AGGREGATOR_CLASS is not None, "AGGREGATOR_CLASS must be set in a subclass" context = self.CONTEXT_CLASS() - aggregator = self.AGGREGATOR_CLASS(context, aggregation_timeout=AGGREGATION_TIMEOUT) + aggregator = self.AGGREGATOR_CLASS( + context, params=LLMUserAggregatorParams(aggregation_timeout=AGGREGATION_TIMEOUT) + ) frames_to_send = [ UserStartedSpeakingFrame(), InterimTranscriptionFrame(text="How ", user_id="cat", timestamp=""), @@ -189,7 +195,9 @@ class BaseTestUserContextAggregator: assert self.AGGREGATOR_CLASS is not None, "AGGREGATOR_CLASS must be set in a subclass" context = self.CONTEXT_CLASS() - aggregator = self.AGGREGATOR_CLASS(context, aggregation_timeout=AGGREGATION_TIMEOUT) + aggregator = self.AGGREGATOR_CLASS( + context, params=LLMUserAggregatorParams(aggregation_timeout=AGGREGATION_TIMEOUT) + ) frames_to_send = [ UserStartedSpeakingFrame(), InterimTranscriptionFrame(text="How ", user_id="cat", timestamp=""), @@ -216,7 +224,9 @@ class BaseTestUserContextAggregator: assert self.AGGREGATOR_CLASS is not None, "AGGREGATOR_CLASS must be set in a subclass" context = self.CONTEXT_CLASS() - aggregator = self.AGGREGATOR_CLASS(context, aggregation_timeout=AGGREGATION_TIMEOUT) + aggregator = self.AGGREGATOR_CLASS( + context, params=LLMUserAggregatorParams(aggregation_timeout=AGGREGATION_TIMEOUT) + ) frames_to_send = [ UserStartedSpeakingFrame(), UserStoppedSpeakingFrame(), @@ -240,7 +250,9 @@ class BaseTestUserContextAggregator: assert self.AGGREGATOR_CLASS is not None, "AGGREGATOR_CLASS must be set in a subclass" context = self.CONTEXT_CLASS() - aggregator = self.AGGREGATOR_CLASS(context, aggregation_timeout=AGGREGATION_TIMEOUT) + aggregator = self.AGGREGATOR_CLASS( + context, params=LLMUserAggregatorParams(aggregation_timeout=AGGREGATION_TIMEOUT) + ) frames_to_send = [ UserStartedSpeakingFrame(), UserStoppedSpeakingFrame(), @@ -265,7 +277,9 @@ class BaseTestUserContextAggregator: assert self.AGGREGATOR_CLASS is not None, "AGGREGATOR_CLASS must be set in a subclass" context = self.CONTEXT_CLASS() - aggregator = self.AGGREGATOR_CLASS(context, aggregation_timeout=AGGREGATION_TIMEOUT) + aggregator = self.AGGREGATOR_CLASS( + context, params=LLMUserAggregatorParams(aggregation_timeout=AGGREGATION_TIMEOUT) + ) frames_to_send = [ UserStartedSpeakingFrame(), TranscriptionFrame(text="Hello Pipecat!", user_id="cat", timestamp=""), @@ -293,7 +307,9 @@ class BaseTestUserContextAggregator: assert self.AGGREGATOR_CLASS is not None, "AGGREGATOR_CLASS must be set in a subclass" context = self.CONTEXT_CLASS() - aggregator = self.AGGREGATOR_CLASS(context, aggregation_timeout=AGGREGATION_TIMEOUT) + aggregator = self.AGGREGATOR_CLASS( + context, params=LLMUserAggregatorParams(aggregation_timeout=AGGREGATION_TIMEOUT) + ) frames_to_send = [ UserStartedSpeakingFrame(), UserStoppedSpeakingFrame(), @@ -318,7 +334,9 @@ class BaseTestUserContextAggregator: assert self.AGGREGATOR_CLASS is not None, "AGGREGATOR_CLASS must be set in a subclass" context = self.CONTEXT_CLASS() - aggregator = self.AGGREGATOR_CLASS(context, aggregation_timeout=AGGREGATION_TIMEOUT) + aggregator = self.AGGREGATOR_CLASS( + context, params=LLMUserAggregatorParams(aggregation_timeout=AGGREGATION_TIMEOUT) + ) frames_to_send = [ UserStartedSpeakingFrame(), InterimTranscriptionFrame(text="Hello ", user_id="cat", timestamp=""), @@ -346,7 +364,9 @@ class BaseTestUserContextAggregator: assert self.AGGREGATOR_CLASS is not None, "AGGREGATOR_CLASS must be set in a subclass" context = self.CONTEXT_CLASS() - aggregator = self.AGGREGATOR_CLASS(context, aggregation_timeout=AGGREGATION_TIMEOUT) + aggregator = self.AGGREGATOR_CLASS( + context, params=LLMUserAggregatorParams(aggregation_timeout=AGGREGATION_TIMEOUT) + ) frames_to_send = [ TranscriptionFrame(text="Hello!", user_id="cat", timestamp=""), SleepFrame(sleep=AGGREGATION_SLEEP), @@ -366,7 +386,9 @@ class BaseTestUserContextAggregator: assert self.AGGREGATOR_CLASS is not None, "AGGREGATOR_CLASS must be set in a subclass" context = self.CONTEXT_CLASS() - aggregator = self.AGGREGATOR_CLASS(context, aggregation_timeout=AGGREGATION_TIMEOUT) + aggregator = self.AGGREGATOR_CLASS( + context, params=LLMUserAggregatorParams(aggregation_timeout=AGGREGATION_TIMEOUT) + ) frames_to_send = [ InterimTranscriptionFrame(text="Hello ", user_id="cat", timestamp=""), SleepFrame(), @@ -389,8 +411,7 @@ class BaseTestUserContextAggregator: context = self.CONTEXT_CLASS() aggregator = self.AGGREGATOR_CLASS( - context, - aggregation_timeout=AGGREGATION_TIMEOUT, + context, params=LLMUserAggregatorParams(aggregation_timeout=AGGREGATION_TIMEOUT) ) frames_to_send = [ UserStartedSpeakingFrame(), @@ -469,7 +490,9 @@ class BaseTestAssistantContextAggreagator: assert self.AGGREGATOR_CLASS is not None, "AGGREGATOR_CLASS must be set in a subclass" context = self.CONTEXT_CLASS() - aggregator = self.AGGREGATOR_CLASS(context, expect_stripped_words=False) + aggregator = self.AGGREGATOR_CLASS( + context, params=LLMAssistantAggregatorParams(expect_stripped_words=False) + ) frames_to_send = [ LLMFullResponseStartFrame(), TextFrame(text="Hello "), @@ -513,7 +536,9 @@ class BaseTestAssistantContextAggreagator: assert self.AGGREGATOR_CLASS is not None, "AGGREGATOR_CLASS must be set in a subclass" context = self.CONTEXT_CLASS() - aggregator = self.AGGREGATOR_CLASS(context, expect_stripped_words=False) + aggregator = self.AGGREGATOR_CLASS( + context, params=LLMAssistantAggregatorParams(expect_stripped_words=False) + ) frames_to_send = [ LLMFullResponseStartFrame(), TextFrame(text="Hello "), @@ -538,7 +563,9 @@ class BaseTestAssistantContextAggreagator: assert self.AGGREGATOR_CLASS is not None, "AGGREGATOR_CLASS must be set in a subclass" context = self.CONTEXT_CLASS() - aggregator = self.AGGREGATOR_CLASS(context, expect_stripped_words=False) + aggregator = self.AGGREGATOR_CLASS( + context, params=LLMAssistantAggregatorParams(expect_stripped_words=False) + ) frames_to_send = [ LLMFullResponseStartFrame(), TextFrame(text="Hello "), diff --git a/tests/test_langchain.py b/tests/test_langchain.py index 6534f6bb0..3d907e084 100644 --- a/tests/test_langchain.py +++ b/tests/test_langchain.py @@ -20,6 +20,7 @@ from pipecat.frames.frames import ( ) from pipecat.pipeline.pipeline import Pipeline from pipecat.processors.aggregators.llm_response import ( + LLMAssistantAggregatorParams, LLMAssistantResponseAggregator, LLMUserResponseAggregator, ) @@ -63,7 +64,9 @@ class TestLangchain(unittest.IsolatedAsyncioTestCase): self.mock_proc = self.MockProcessor("token_collector") tma_in = LLMUserResponseAggregator(messages) - tma_out = LLMAssistantResponseAggregator(messages, expect_stripped_words=False) + tma_out = LLMAssistantResponseAggregator( + messages, params=LLMAssistantAggregatorParams(expect_stripped_words=False) + ) pipeline = Pipeline([tma_in, proc, self.mock_proc, tma_out])