Sending Search Response to RTVI

examples/news-chatbot/server/README.md

@@ -0,0 +1,52 @@
+# News Chatbot Server
+
+A FastAPI server that manages bot instances and provide endpoint for Pipecat client connections.
+
+## Endpoints
+
+- `POST /connect` - Pipecat client connection endpoint
+
+## Environment Variables
+
+Copy `env.example` to `.env` and configure:
+
+```ini
+# Required API Keys
+DAILY_API_KEY=           # Your Daily API key
+DEEPGRAM_API_KEY=        # Your Deepgram API key
+GOOGLE_API_KEY=          # Your Google/Gemini API key
+CARTESIA_API_KEY=        # Your Cartesia API key
+
+# Optional Configuration
+DAILY_API_URL=           # Optional: Daily API URL (defaults to https://api.daily.co/v1)
+DAILY_SAMPLE_ROOM_URL=   # Optional: Fixed room URL for development
+HOST=                    # Optional: Host address (defaults to 0.0.0.0)
+FAST_API_PORT=           # Optional: Port number (defaults to 7860)
+```
+
+## Running the Server
+
+Set up and activate your virtual environment:
+
+```bash
+python3 -m venv venv
+source venv/bin/activate  # On Windows: venv\Scripts\activate
+```
+
+Install dependencies:
+
+```bash
+pip install -r requirements.txt
+```
+
+If you want to use the local version of `pipecat` in this repo rather than the last published version, also run:
+
+```bash
+pip install --editable "../../../[daily,deepgram,google,cartesia,openai,silero]"
+```
+
+Run the server:
+
+```bash
+python server.py
+```

examples/news-chatbot/server/env.example

@@ -0,0 +1,5 @@
+DAILY_SAMPLE_ROOM_URL=https://yourdomain.daily.co/yourroom # (for joining the bot to the same room repeatedly for local dev)
+DAILY_API_KEY=
+CARTESIA_API_KEY=
+DEEPGRAM_API_KEY=
+GOOGLE_API_KEY=

examples/news-chatbot/server/news_bot.py

@@ -0,0 +1,166 @@
+#
+# Copyright (c) 2024, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import asyncio
+import os
+import sys
+from pathlib import Path
+
+import aiohttp
+from dotenv import load_dotenv
+from loguru import logger
+
+from pipecat.audio.vad.silero import SileroVADAnalyzer
+from pipecat.frames.frames import EndFrame, Frame
+from pipecat.pipeline.pipeline import Pipeline
+from pipecat.pipeline.runner import PipelineRunner
+from pipecat.pipeline.task import PipelineParams, PipelineTask
+from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
+from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
+from pipecat.processors.frameworks.rtvi import RTVIConfig, RTVIProcessor
+from pipecat.services.cartesia import CartesiaTTSService
+from pipecat.services.deepgram import DeepgramSTTService
+from pipecat.services.google import GoogleLLMService, LLMSearchResponseFrame
+from pipecat.transports.services.daily import DailyParams, DailyTransport
+from pipecat.utils.text.markdown_text_filter import MarkdownTextFilter
+
+sys.path.append(str(Path(__file__).parent.parent))
+from runner import configure
+
+load_dotenv(override=True)
+
+logger.remove(0)
+logger.add(sys.stderr, level="DEBUG")
+
+
+# Function handlers for the LLM
+# https://ai.google.dev/gemini-api/docs/grounding?lang=python#dynamic-retrieval
+# Some queries are likely to benefit more from Grounding with Google Search than others.
+# The dynamic retrieval feature gives you additional control over when to use Grounding with Google Search.
+# If the dynamic retrieval mode is unspecified, Grounding with Google Search is always triggered.
+# If the mode is set to dynamic, the model decides when to use grounding based on a threshold that you can configure.
+# The threshold is a floating-point value in the range [0,1] and defaults to 0.3.
+# If the threshold value is 0, the response is always grounded with Google Search; if it's 1, it never is.
+search_tool = {
+    "google_search_retrieval": {
+        "dynamic_retrieval_config": {
+            "mode": "MODE_DYNAMIC",
+            "dynamic_threshold": 0,
+        }  # always grounding
+    }
+}
+tools = [search_tool]
+
+system_instruction = """
+You are an expert at providing the most recent news from any place. Your responses will be converted to audio, so ensure they are formatted in plain text without special characters (e.g., *, _, -) or overly complex formatting.
+
+Guidelines:
+- Use the Google search API to retrieve the current date and provide the latest news.
+- Always deliver accurate and concise responses.
+- Ensure all responses are clear, using plain text only. Avoid any special characters or symbols.
+
+Start every interaction by asking how you can assist the user.
+"""
+
+
+class LLMSearchLoggerProcessor(FrameProcessor):
+    async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+
+        if isinstance(frame, LLMSearchResponseFrame):
+            print(f"LLMSearchLoggerProcessor: {frame}")
+
+        await self.push_frame(frame)
+
+
+async def main():
+    async with aiohttp.ClientSession() as session:
+        (room_url, token) = await configure(session)
+
+        transport = DailyTransport(
+            room_url,
+            token,
+            "Latest news!",
+            DailyParams(
+                audio_out_enabled=True,
+                vad_enabled=True,
+                vad_analyzer=SileroVADAnalyzer(),
+                vad_audio_passthrough=True,
+            ),
+        )
+
+        stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))
+
+        tts = CartesiaTTSService(
+            api_key=os.getenv("CARTESIA_API_KEY"),
+            voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22",  # British Lady
+            text_filter=MarkdownTextFilter(),
+        )
+
+        llm = GoogleLLMService(
+            api_key=os.getenv("GOOGLE_API_KEY"),
+            system_instruction=system_instruction,
+            tools=tools,
+        )
+
+        context = OpenAILLMContext(
+            [
+                {
+                    "role": "user",
+                    "content": "Start by greeting the user warmly, introducing yourself, and mentioning the current day. Be friendly and engaging to set a positive tone for the interaction.",
+                }
+            ],
+        )
+        context_aggregator = llm.create_context_aggregator(context)
+
+        llm_search_logger = LLMSearchLoggerProcessor()
+
+        #
+        # RTVI events for Pipecat client UI
+        #
+        rtvi = RTVIProcessor(config=RTVIConfig(config=[]))
+
+        pipeline = Pipeline(
+            [
+                transport.input(),
+                stt,
+                rtvi,
+                context_aggregator.user(),
+                llm,
+                llm_search_logger,
+                tts,
+                transport.output(),
+                context_aggregator.assistant(),
+            ]
+        )
+
+        task = PipelineTask(
+            pipeline,
+            PipelineParams(
+                allow_interruptions=True,
+                observers=[rtvi.observer()],
+            ),
+        )
+
+        @rtvi.event_handler("on_client_ready")
+        async def on_client_ready(rtvi):
+            await rtvi.set_bot_ready()
+
+        @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()])
+
+        @transport.event_handler("on_participant_left")
+        async def on_participant_left(transport, participant, reason):
+            print(f"Participant left: {participant}")
+            await task.queue_frame(EndFrame())
+
+        runner = PipelineRunner()
+        await runner.run(task)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

examples/news-chatbot/server/requirements.txt

@@ -0,0 +1,4 @@
+python-dotenv
+fastapi[all]
+uvicorn
+pipecat-ai[daily,google,deepgram,cartesia,silero,openai]

examples/news-chatbot/server/runner.py

@@ -0,0 +1,63 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import argparse
+import os
+
+import aiohttp
+
+from pipecat.transports.services.helpers.daily_rest import DailyRESTHelper
+
+
+async def configure(aiohttp_session: aiohttp.ClientSession):
+    (url, token, _) = await configure_with_args(aiohttp_session)
+    return (url, token)
+
+
+async def configure_with_args(
+    aiohttp_session: aiohttp.ClientSession, parser: argparse.ArgumentParser | None = None
+):
+    if not parser:
+        parser = argparse.ArgumentParser(description="Daily AI SDK Bot Sample")
+    parser.add_argument(
+        "-u", "--url", type=str, required=False, help="URL of the Daily room to join"
+    )
+    parser.add_argument(
+        "-k",
+        "--apikey",
+        type=str,
+        required=False,
+        help="Daily API Key (needed to create an owner token for the room)",
+    )
+
+    args, unknown = parser.parse_known_args()
+
+    url = args.url or os.getenv("DAILY_SAMPLE_ROOM_URL")
+    key = args.apikey or os.getenv("DAILY_API_KEY")
+
+    if not url:
+        raise Exception(
+            "No Daily room specified. use the -u/--url option from the command line, or set DAILY_SAMPLE_ROOM_URL in your environment to specify a Daily room URL."
+        )
+
+    if not key:
+        raise Exception(
+            "No Daily API key specified. use the -k/--apikey option from the command line, or set DAILY_API_KEY in your environment to specify a Daily API key, available from https://dashboard.daily.co/developers."
+        )
+
+    daily_rest_helper = DailyRESTHelper(
+        daily_api_key=key,
+        daily_api_url=os.getenv("DAILY_API_URL", "https://api.daily.co/v1"),
+        aiohttp_session=aiohttp_session,
+    )
+
+    # Create a meeting token for the given room with an expiration 1 hour in
+    # the future.
+    expiry_time: float = 60 * 60
+
+    token = await daily_rest_helper.get_token(url, expiry_time)
+
+    return (url, token, args)

examples/news-chatbot/server/server.py

@@ -0,0 +1,147 @@
+#
+# Copyright (c) 2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import argparse
+import os
+import subprocess
+from contextlib import asynccontextmanager
+from typing import Any, Dict
+
+import aiohttp
+from dotenv import load_dotenv
+from fastapi import FastAPI, HTTPException, Request
+from fastapi.middleware.cors import CORSMiddleware
+from fastapi.responses import JSONResponse
+
+from pipecat.transports.services.helpers.daily_rest import DailyRESTHelper, DailyRoomParams
+
+# Load environment variables from .env file
+load_dotenv(override=True)
+
+# Dictionary to track bot processes: {pid: (process, room_url)}
+bot_procs = {}
+
+# Store Daily API helpers
+daily_helpers = {}
+
+
+def cleanup():
+    """Cleanup function to terminate all bot processes.
+
+    Called during server shutdown.
+    """
+    for entry in bot_procs.values():
+        proc = entry[0]
+        proc.terminate()
+        proc.wait()
+
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    """FastAPI lifespan manager that handles startup and shutdown tasks.
+
+    - Creates aiohttp session
+    - Initializes Daily API helper
+    - Cleans up resources on shutdown
+    """
+    aiohttp_session = aiohttp.ClientSession()
+    daily_helpers["rest"] = DailyRESTHelper(
+        daily_api_key=os.getenv("DAILY_API_KEY", ""),
+        daily_api_url=os.getenv("DAILY_API_URL", "https://api.daily.co/v1"),
+        aiohttp_session=aiohttp_session,
+    )
+    yield
+    await aiohttp_session.close()
+    cleanup()
+
+
+# Initialize FastAPI app with lifespan manager
+app = FastAPI(lifespan=lifespan)
+
+# Configure CORS to allow requests from any origin
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+
+async def create_room_and_token() -> tuple[str, str]:
+    """Helper function to create a Daily room and generate an access token.
+
+    Returns:
+        tuple[str, str]: A tuple containing (room_url, token)
+
+    Raises:
+        HTTPException: If room creation or token generation fails
+    """
+    room = await daily_helpers["rest"].create_room(DailyRoomParams())
+    if not room.url:
+        raise HTTPException(status_code=500, detail="Failed to create room")
+
+    token = await daily_helpers["rest"].get_token(room.url)
+    if not token:
+        raise HTTPException(status_code=500, detail=f"Failed to get token for room: {room.url}")
+
+    return room.url, token
+
+
+@app.post("/connect")
+async def bot_connect(request: Request) -> Dict[Any, Any]:
+    """Connect endpoint that creates a room and returns connection credentials.
+
+    This endpoint is called by client to establish a connection.
+
+    Returns:
+        Dict[Any, Any]: Authentication bundle containing room_url and token
+
+    Raises:
+        HTTPException: If room creation, token generation, or bot startup fails
+    """
+    print("Creating room for RTVI connection")
+    room_url, token = await create_room_and_token()
+    print(f"Room URL: {room_url}")
+
+    # Start the bot process
+    try:
+        bot_file = "news_bot"
+        proc = subprocess.Popen(
+            [f"python3 -m {bot_file} -u {room_url} -t {token}"],
+            shell=True,
+            bufsize=1,
+            cwd=os.path.dirname(os.path.abspath(__file__)),
+        )
+        bot_procs[proc.pid] = (proc, room_url)
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"Failed to start subprocess: {e}")
+
+    # Return the authentication bundle in format expected by DailyTransport
+    return {"room_url": room_url, "token": token}
+
+
+if __name__ == "__main__":
+    import uvicorn
+
+    # Parse command line arguments for server configuration
+    default_host = os.getenv("HOST", "0.0.0.0")
+    default_port = int(os.getenv("FAST_API_PORT", "7860"))
+
+    parser = argparse.ArgumentParser(description="Daily Travel Companion FastAPI server")
+    parser.add_argument("--host", type=str, default=default_host, help="Host address")
+    parser.add_argument("--port", type=int, default=default_port, help="Port number")
+    parser.add_argument("--reload", action="store_true", help="Reload code on change")
+
+    config = parser.parse_args()
+
+    # Start the FastAPI server
+    uvicorn.run(
+        "server:app",
+        host=config.host,
+        port=config.port,
+        reload=config.reload,
+    )