hacking classic-pipeline.py to be runnable by bot_runner

examples/fast-bot-metrics/bot.py

@@ -0,0 +1 @@
+classic-pipeline.py

examples/fast-bot-metrics/bot_runner.py

@@ -37,9 +37,10 @@ daily_rest_helper = DailyRESTHelper(
 
 
 class RunnerSettings(BaseModel):
-    prompt: Optional[str] = None
+    prompt: Optional[str] = "You are a helpful assistant."
-    deepgram_voice: Optional[str] = None
+    deepgram_voice: Optional[str] = os.getenv("DEEPGRAM_VOICE")
-    openai_model: Optional[str] = "meta-llama/Meta-Llama-3-70B-Instruct"
+    openai_model: Optional[str] = os.getenv("OPENAI_MODEL")
+    openai_api_key: Optional[str] = os.getenv("OPENAI_API_KEY")
     test: Optional[bool] = None
 
 # ----------------- API ----------------- #
@@ -114,10 +115,10 @@ async def start_bot(request: Request) -> JSONResponse:
             prompt=runner_settings.prompt,
             deepgram_voice=runner_settings.deepgram_voice,
             deepgram_api_key=os.getenv("DEEPGRAM_API_KEY"),
-            deepgram_base_url="http://0.0.0.0:8080/v1/speak",
+            # deepgram_base_url="http://0.0.0.0:8080/v1/speak",
             openai_model=runner_settings.openai_model,
-            openai_api_key="ollama",
+            openai_api_key=runner_settings.openai_api_key,
-            openai_base_url="http://0.0.0.0:8000/v1",
+            # openai_base_url="http://0.0.0.0:8000/v1",
         )
         bot_settings_str = bot_settings.model_dump_json(exclude_none=True)
 

examples/fast-bot-metrics/classic-pipeline.py

@@ -6,12 +6,14 @@
 
 from loguru import logger
 from runner import configure
+import argparse
 import asyncio
 import aiohttp
 import os
 import sys
-from typing import List
+from typing import List, Optional
 
+from pydantic import BaseModel, ValidationError
 
 from pipecat.vad.vad_analyzer import VADParams
 from pipecat.vad.silero import SileroVADAnalyzer
@@ -41,17 +43,29 @@ from fastbothelpers import (
 from dotenv import load_dotenv
 load_dotenv(override=True)
 
-
 logger.remove(0)
 logger.add(sys.stderr, level="DEBUG")
 
 
-async def main(room_url: str, token):
+class BotSettings(BaseModel):
+    room_url: str
+    room_token: str
+    bot_name: str = "Pipecat"
+    prompt: Optional[str] = "You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Respond to what the user said in a creative and helpful way in a few short sentences."
+    deepgram_api_key: Optional[str] = None
+    deepgram_voice: Optional[str] = "aura-asteria-en"
+    deepgram_base_url: Optional[str] = "https://api.deepgram.com/v1/speak"
+    openai_api_key: Optional[str] = None
+    openai_model: Optional[str] = "gpt-4o"
+    openai_base_url: Optional[str] = None
+
+
+async def main(settings: BotSettings):
     async with aiohttp.ClientSession() as session:
         transport = DailyTransport(
-            room_url,
+            settings.room_url,
-            token,
+            settings.room_token,
-            "Respond bot",
+            settings.bot_name,
             DailyParams(
                 audio_out_enabled=True,
                 transcription_enabled=False,
@@ -163,6 +177,18 @@ Respond to what the user said in a creative and helpful way. Be concise in your
         await runner.run(task)
 
 
+# if __name__ == "__main__":
+#     (url, token) = configure()
+#     asyncio.run(main(url, token))
+
 if __name__ == "__main__":
-    (url, token) = configure()
+    parser = argparse.ArgumentParser(description="Pipecat Bot")
-    asyncio.run(main(url, token))
+    parser.add_argument("-s", "--settings", type=str, required=True, help="Pipecat bot settings")
+
+    args, unknown = parser.parse_known_args()
+
+    try:
+        settings = BotSettings.model_validate_json(args.settings)
+        asyncio.run(main(settings))
+    except ValidationError as e:
+        print(e)

examples/fast-bot-metrics/example-bot-runner-bot.py

@@ -0,0 +1,158 @@
+#
+# Copyright (c) 2024, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import argparse
+import asyncio
+import aiohttp
+import sys
+
+from pydantic import BaseModel, ValidationError
+from typing import Optional
+
+from pipecat.frames.frames import EndFrame, LLMMessagesFrame
+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, LLMUserResponseAggregator)
+from pipecat.services.deepgram import DeepgramTTSService
+from pipecat.services.openai import OpenAILLMService
+from pipecat.transports.services.daily import DailyParams, DailyTransport, DailyTransportMessageFrame
+from pipecat.vad.silero import SileroVADAnalyzer
+
+from loguru import logger
+
+logger.remove(0)
+logger.add(sys.stderr, level="DEBUG")
+
+
+class BotSettings(BaseModel):
+    room_url: str
+    room_token: str
+    bot_name: str = "Pipecat"
+    prompt: Optional[str] = "You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Respond to what the user said in a creative and helpful way in a few short sentences."
+    deepgram_api_key: Optional[str] = None
+    deepgram_voice: Optional[str] = "aura-asteria-en"
+    deepgram_base_url: Optional[str] = "https://api.deepgram.com/v1/speak"
+    openai_api_key: Optional[str] = None
+    openai_model: Optional[str] = "gpt-4o"
+    openai_base_url: Optional[str] = None
+
+
+async def main(settings: BotSettings):
+    async with aiohttp.ClientSession() as session:
+        transport = DailyTransport(
+            settings.room_url,
+            settings.room_token,
+            settings.bot_name,
+            DailyParams(
+                audio_out_enabled=True,
+                transcription_enabled=True,
+                vad_enabled=True,
+                vad_analyzer=SileroVADAnalyzer()
+            )
+        )
+
+        tts = DeepgramTTSService(
+            aiohttp_session=session,
+            api_key=settings.deepgram_api_key,
+            voice=settings.deepgram_voice,
+            base_url=settings.deepgram_base_url
+        )
+
+        llm = OpenAILLMService(
+            api_key=settings.openai_api_key,
+            model=settings.openai_model,
+            base_url=settings.openai_base_url
+        )
+
+        messages = [
+            {
+                "role": "system",
+                "content": settings.prompt,
+            },
+        ]
+
+        tma_in = LLMUserResponseAggregator(messages)
+        tma_out = LLMAssistantResponseAggregator(messages)
+
+        pipeline = Pipeline([
+            transport.input(),   # Transport user input
+            tma_in,              # User responses
+            llm,                 # LLM
+            tts,                 # TTS
+            transport.output(),  # Transport bot output
+            tma_out,             # Assistant spoken responses
+        ])
+
+        task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True, enable_metrics=True))
+
+        # When the first participant joins, the bot should introduce itself.
+        @transport.event_handler("on_first_participant_joined")
+        async def on_first_participant_joined(transport, participant):
+            # Kick off the conversation.
+            messages.append(
+                {"role": "system", "content": "Please introduce yourself to the user."})
+            await task.queue_frame(LLMMessagesFrame(messages))
+
+        # When a participant joins, start transcription for that participant so the
+        # bot can "hear" and respond to them.
+        @transport.event_handler("on_participant_joined")
+        async def on_participant_joined(transport, participant):
+            transport.capture_participant_transcription(participant["id"])
+
+        # When the participant leaves, we exit the bot.
+        @transport.event_handler("on_participant_left")
+        async def on_participant_left(transport, participant, reason):
+            await task.queue_frame(EndFrame())
+
+        # If the call is ended make sure we quit as well.
+        @transport.event_handler("on_call_state_updated")
+        async def on_call_state_updated(transport, state):
+            if state == "left":
+                await task.queue_frame(EndFrame())
+
+        # Handle "latency-ping" messages. The client will send app messages that look like
+        # this:
+        #   { "latency-ping": { ts: <client-side timestamp> }}
+        #
+        # 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 tma_in.push_frame(
+                        DailyTransportMessageFrame(
+                            message={"latency-pong-pipeline-delivery": {"ts": ts}},
+                            participant_id=sender))
+            except Exception as e:
+                logger.debug(f"message handling error: {e} - {message}")
+
+        runner = PipelineRunner()
+
+        await runner.run(task)
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(description="Pipecat Bot")
+    parser.add_argument("-s", "--settings", type=str, required=True, help="Pipecat bot settings")
+
+    args, unknown = parser.parse_known_args()
+
+    try:
+        settings = BotSettings.model_validate_json(args.settings)
+        asyncio.run(main(settings))
+    except ValidationError as e:
+        print(e)