Split features/ into audio/, observability/, and rag/ subfolders

Extract focused example groups from the catch-all features/ folder: - audio/: audio recording, background sound, sound effects - observability/: observer, heartbeats, sentry metrics - rag/: mem0, gemini-rag, gemini grounding metadata Update README to document the new folders.
2026-03-31 13:15:06 -04:00
parent e719cbbe6d
commit 1d85aedcae
10 changed files with 13 additions and 1 deletions
--- a/examples/rag/gemini-grounding-metadata.py
+++ b/examples/rag/gemini-grounding-metadata.py
@@ -0,0 +1,175 @@
+#
+# Copyright (c) 2024-2026, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+
+import os
+import sys
+from pathlib import Path
+
+from dotenv import load_dotenv
+from loguru import logger
+
+from pipecat.audio.vad.silero import SileroVADAnalyzer
+from pipecat.frames.frames import LLMRunFrame
+from pipecat.observers.base_observer import BaseObserver, FramePushed
+from pipecat.pipeline.pipeline import Pipeline
+from pipecat.pipeline.runner import PipelineRunner
+from pipecat.pipeline.task import PipelineParams, PipelineTask
+from pipecat.processors.aggregators.llm_context import LLMContext
+from pipecat.processors.aggregators.llm_response_universal import (
+    LLMContextAggregatorPair,
+    LLMUserAggregatorParams,
+)
+from pipecat.runner.types import RunnerArguments
+from pipecat.runner.utils import create_transport
+from pipecat.services.cartesia.tts import CartesiaTTSService
+from pipecat.services.deepgram.stt import DeepgramSTTService
+from pipecat.services.google.llm import GoogleLLMService, LLMSearchResponseFrame
+from pipecat.services.llm_service import LLMService
+from pipecat.transports.base_transport import BaseTransport, TransportParams
+from pipecat.transports.daily.transport import DailyParams
+from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams
+
+sys.path.append(str(Path(__file__).parent.parent))
+
+load_dotenv(override=True)
+
+
+# Function handlers for the LLM
+search_tool = {"google_search": {}}
+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 avoid using special characters or overly complex formatting.
+
+Always use the google search API to retrieve the latest news. You must also use it to check which day is today.
+
+You can:
+- Use the Google search API to check the current date.
+- Provide the most recent and relevant news from any place by using the google search API.
+- Answer any questions the user may have, ensuring your responses are accurate and concise.
+
+Start each interaction by asking the user about which place they would like to know the information.
+"""
+
+
+class LLMSearchLoggerObserver(BaseObserver):
+    async def on_push_frame(self, data: FramePushed):
+        src = data.source
+        dst = data.destination
+        frame = data.frame
+        timestamp = data.timestamp
+
+        if not isinstance(src, LLMService) and not isinstance(dst, LLMService):
+            return
+
+        time_sec = timestamp / 1_000_000_000
+
+        arrow = "→"
+
+        if isinstance(frame, LLMSearchResponseFrame):
+            logger.debug(f"🧠 {arrow} {dst} LLM SEARCH RESPONSE FRAME: {frame} at {time_sec:.2f}s")
+
+
+# We use lambdas to defer transport parameter creation until the transport
+# type is selected at runtime.
+transport_params = {
+    "daily": lambda: DailyParams(
+        audio_in_enabled=True,
+        audio_out_enabled=True,
+    ),
+    "twilio": lambda: FastAPIWebsocketParams(
+        audio_in_enabled=True,
+        audio_out_enabled=True,
+    ),
+    "webrtc": lambda: TransportParams(
+        audio_in_enabled=True,
+        audio_out_enabled=True,
+    ),
+}
+
+
+async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
+    logger.info(f"Starting bot")
+
+    stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))
+
+    tts = CartesiaTTSService(
+        api_key=os.getenv("CARTESIA_API_KEY"),
+        settings=CartesiaTTSService.Settings(
+            voice="71a7ad14-091c-4e8e-a314-022ece01c121",  # British Reading Lady
+        ),
+    )
+
+    # Initialize the Gemini Multimodal Live model
+    llm = GoogleLLMService(
+        api_key=os.getenv("GOOGLE_API_KEY"),
+        settings=GoogleLLMService.Settings(
+            system_instruction=system_instruction,
+        ),
+        tools=tools,
+    )
+
+    context = LLMContext(
+        [
+            {
+                "role": "developer",
+                "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.",
+            }
+        ],
+    )
+    user_aggregator, assistant_aggregator = LLMContextAggregatorPair(
+        context,
+        user_params=LLMUserAggregatorParams(vad_analyzer=SileroVADAnalyzer()),
+    )
+
+    pipeline = Pipeline(
+        [
+            transport.input(),
+            stt,
+            user_aggregator,
+            llm,
+            tts,
+            transport.output(),
+            assistant_aggregator,
+        ]
+    )
+
+    task = PipelineTask(
+        pipeline,
+        params=PipelineParams(
+            enable_metrics=True,
+            enable_usage_metrics=True,
+        ),
+        idle_timeout_secs=runner_args.pipeline_idle_timeout_secs,
+        observers=[LLMSearchLoggerObserver()],
+    )
+
+    @transport.event_handler("on_client_connected")
+    async def on_client_connected(transport, client):
+        logger.info(f"Client connected")
+        # Start conversation - empty prompt to let LLM follow system instructions
+        await task.queue_frames([LLMRunFrame()])
+
+    @transport.event_handler("on_client_disconnected")
+    async def on_client_disconnected(transport, client):
+        logger.info(f"Client disconnected")
+        await task.cancel()
+
+    runner = PipelineRunner(handle_sigint=runner_args.handle_sigint)
+    await runner.run(task)
+
+
+async def bot(runner_args: RunnerArguments):
+    """Main bot entry point compatible with Pipecat Cloud."""
+    transport = await create_transport(runner_args, transport_params)
+    await run_bot(transport, runner_args)
+
+
+if __name__ == "__main__":
+    from pipecat.runner.run import main
+
+    main()
--- a/examples/rag/gemini-rag.py
+++ b/examples/rag/gemini-rag.py
@@ -0,0 +1,274 @@
+#
+# Copyright (c) 2024-2026, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+"""CrossFit Games 2025 Rulebook RAG Demo.
+
+This example demonstrates a Model-Assisted Generation (MAG) chatbot using Google's Gemini model.
+This example uses 2 Gemini models:
+- Gemini 2.0 Flash: This is the voice model that is used to generate the response.
+- Gemini 2.0 Flash Lite: This is the model that is used to answer questions about the CrossFit Games 2025 rulebook - information that isn't yet publicly
+indexed by Gemini (or any other LLM).
+
+How it works:
+- The voice model (Gemini 2.0 Flash) is configured to call a function whenever the user asks a question.
+- The function call is a tool call to the MAG model (Gemini 2.0 Flash Lite).
+- The MAG model generates a response based on the question. The MAG model has the entire contents of the CrossFit Games 2025 rulebook in it's context window.
+- The response is returned to the voice model (Gemini 2.0 Flash), which then generates the response to the user.
+
+Why this works:
+- Gemini 2.0 Flash is fast
+- Gemini 2.0 Flash Lite is faster
+- Gemini 2.0 Flash Lite has a large (1 million tokens) context window
+- IMPORTANT: The generated response from Gemini 2.0 Flash Lite is limited to 50 words or less and 64 tokens.
+You can see this in the RAG_PROMPT variable and the generation_config in the query_knowledge_base function.
+Long generations are slower and more expensive, in the world of Voice AI, we don't need long generations.
+
+Example questions to ask and compare to other RAG solutions:
+- What lenses are not allowed?
+- How many people can be on a team?
+- What do winning gyms get?
+- What happens if I skip a workout?
+- Can I switch my team members for the Games?
+- What happens if I start too early?
+
+Notes:
+- The RAG model is Gemini 2.0 Flash Lite.
+- The voice model is Gemini 2.0 Flash.
+- The RAG content is stored in the assets/rag-content.txt file.
+- The model for voice is Gemini 2.0 Flash, but can be easily switched to any other model.
+
+Customization options:
+- update assets/rag-content.txt with your own knowledge base
+- increase/decrease the RAG_MODEL's generation length
+- use a different voice model
+- play with the RAG_PROMPT
+- change the function calling logic
+"""
+
+import json
+import os
+import time
+
+from dotenv import load_dotenv
+from google import genai
+from loguru import logger
+
+from pipecat.adapters.schemas.function_schema import FunctionSchema
+from pipecat.adapters.schemas.tools_schema import ToolsSchema
+from pipecat.audio.vad.silero import SileroVADAnalyzer
+from pipecat.frames.frames import LLMRunFrame
+from pipecat.pipeline.pipeline import Pipeline
+from pipecat.pipeline.runner import PipelineRunner
+from pipecat.pipeline.task import PipelineParams, PipelineTask
+from pipecat.processors.aggregators.llm_context import LLMContext
+from pipecat.processors.aggregators.llm_response_universal import (
+    LLMContextAggregatorPair,
+    LLMUserAggregatorParams,
+)
+from pipecat.runner.types import RunnerArguments
+from pipecat.runner.utils import create_transport
+from pipecat.services.cartesia.tts import CartesiaTTSService
+from pipecat.services.deepgram.stt import DeepgramSTTService
+from pipecat.services.google.llm import GoogleLLMService
+from pipecat.services.llm_service import FunctionCallParams
+from pipecat.transports.base_transport import BaseTransport, TransportParams
+from pipecat.transports.daily.transport import DailyParams
+from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams
+
+load_dotenv(override=True)
+
+# Initialize the client globally
+client = genai.Client(api_key=os.environ["GOOGLE_API_KEY"])
+
+
+def get_rag_content():
+    """Get the RAG content from the file."""
+    script_dir = os.path.dirname(os.path.abspath(__file__))
+    rag_content_path = os.path.join(script_dir, "assets", "rag-content.txt")
+    with open(rag_content_path, "r") as f:
+        return f.read()
+
+
+RAG_MODEL = "gemini-2.0-flash-lite-preview-02-05"
+VOICE_MODEL = "gemini-2.0-flash"
+RAG_CONTENT = get_rag_content()
+RAG_PROMPT = f"""
+You are a helpful assistant designed to answer user questions based solely on the provided knowledge base.
+
+**Instructions:**
+
+1.  **Knowledge Base Only:** Answer questions *exclusively* using the information in the "Knowledge Base" section below. Do not use any outside information.
+2.  **Conversation History:** Use the "Conversation History" (ordered oldest to newest) to understand the context of the current question.
+3.  **Concise Response:**  Respond in 50 words or fewer.  The response will be spoken, so avoid symbols, abbreviations, or complex formatting. Use plain, natural language.
+4.  **Unknown Answer:** If the answer is not found within the "Knowledge Base," respond with "I don't know." Do not guess or make up an answer.
+5. Do not introduce your response. Just provide the answer.
+6. You must follow all instructions.
+
+**Input Format:**
+
+Each request will include:
+
+*   **Conversation History:**  (A list of previous user and assistant messages, if any)
+
+**Knowledge Base:**
+Here is the knowledge base you have access to:
+{RAG_CONTENT}
+"""
+
+
+async def query_knowledge_base(params: FunctionCallParams):
+    """Query the knowledge base for the answer to the question."""
+    logger.info(f"Querying knowledge base for question: {params.arguments['question']}")
+
+    # for our case, the first two messages are the instructions and the user message
+    # so we remove them.
+    conversation_turns = params.context.get_messages()[2:]
+
+    def _is_tool_call(turn):
+        if turn.get("role", None) == "tool":
+            return True
+        if turn.get("tool_calls", None):
+            return True
+        return False
+
+    # filter out tool calls
+    messages = [turn for turn in conversation_turns if not _is_tool_call(turn)]
+    # use the last 3 turns as the conversation history/context
+    messages = messages[-3:]
+    messages_json = json.dumps(messages, ensure_ascii=False, indent=2)
+
+    logger.info(f"Conversation turns: {messages_json}")
+
+    start = time.perf_counter()
+    full_prompt = f"System: {RAG_PROMPT}\n\nConversation History: {messages_json}"
+
+    response = await client.aio.models.generate_content(
+        model=RAG_MODEL,
+        contents=[full_prompt],
+        config={
+            "temperature": 0.1,
+            "max_output_tokens": 64,
+        },
+    )
+    end = time.perf_counter()
+    logger.info(f"Time taken: {end - start:.2f} seconds")
+    logger.info(response.text)
+    await params.result_callback(response.text)
+
+
+# We use lambdas to defer transport parameter creation until the transport
+# type is selected at runtime.
+transport_params = {
+    "daily": lambda: DailyParams(
+        audio_in_enabled=True,
+        audio_out_enabled=True,
+    ),
+    "twilio": lambda: FastAPIWebsocketParams(
+        audio_in_enabled=True,
+        audio_out_enabled=True,
+    ),
+    "webrtc": lambda: TransportParams(
+        audio_in_enabled=True,
+        audio_out_enabled=True,
+    ),
+}
+
+
+async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
+    logger.info(f"Starting bot")
+
+    stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))
+
+    tts = CartesiaTTSService(
+        api_key=os.getenv("CARTESIA_API_KEY"),
+        settings=CartesiaTTSService.Settings(
+            voice="f9836c6e-a0bd-460e-9d3c-f7299fa60f94",  # Southern Lady
+        ),
+    )
+
+    system_prompt = """\
+You are a helpful assistant who converses with a user and answers questions.
+
+You have access to the tool, query_knowledge_base, that allows you to query the knowledge base for the answer to the user's question.
+
+Your response will be turned into speech so use only simple words and punctuation.
+"""
+
+    llm = GoogleLLMService(
+        api_key=os.getenv("GOOGLE_API_KEY"),
+        settings=GoogleLLMService.Settings(
+            model=VOICE_MODEL,
+            system_instruction=system_prompt,
+        ),
+    )
+    llm.register_function("query_knowledge_base", query_knowledge_base)
+
+    query_function = FunctionSchema(
+        name="query_knowledge_base",
+        description="Query the knowledge base for the answer to the question.",
+        properties={
+            "question": {
+                "type": "string",
+                "description": "The question to query the knowledge base with.",
+            },
+        },
+        required=["question"],
+    )
+    tools = ToolsSchema(standard_tools=[query_function])
+
+    context = LLMContext(tools=tools)
+    user_aggregator, assistant_aggregator = LLMContextAggregatorPair(
+        context,
+        user_params=LLMUserAggregatorParams(vad_analyzer=SileroVADAnalyzer()),
+    )
+
+    pipeline = Pipeline(
+        [
+            transport.input(),
+            stt,
+            user_aggregator,
+            llm,
+            tts,
+            transport.output(),
+            assistant_aggregator,
+        ]
+    )
+    task = PipelineTask(
+        pipeline,
+        params=PipelineParams(
+            enable_metrics=True,
+            enable_usage_metrics=True,
+        ),
+        idle_timeout_secs=runner_args.pipeline_idle_timeout_secs,
+    )
+
+    @transport.event_handler("on_client_connected")
+    async def on_client_connected(transport, client):
+        logger.info(f"Client connected")
+        # Start conversation - empty prompt to let LLM follow system instructions
+        context.add_message(
+            {"role": "developer", "content": "Please introduce yourself to the user."}
+        )
+        await task.queue_frames([LLMRunFrame()])
+
+    @transport.event_handler("on_client_disconnected")
+    async def on_client_disconnected(transport, client):
+        logger.info(f"Client disconnected")
+        await task.cancel()
+
+    runner = PipelineRunner(handle_sigint=runner_args.handle_sigint)
+    await runner.run(task)
+
+
+async def bot(runner_args: RunnerArguments):
+    """Main bot entry point compatible with Pipecat Cloud."""
+    transport = await create_transport(runner_args, transport_params)
+    await run_bot(transport, runner_args)
+
+
+if __name__ == "__main__":
+    from pipecat.runner.run import main
+
+    main()
--- a/examples/rag/mem0.py
+++ b/examples/rag/mem0.py
@@ -0,0 +1,270 @@
+#
+# Copyright (c) 2024-2026, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+"""Mem0 Personalized Voice Agent Example with Pipecat.
+
+This example demonstrates how to create a conversational AI assistant with memory capabilities
+using Mem0 integration. It shows how to build an agent that remembers previous interactions
+and personalizes responses based on conversation history.
+
+The example:
+    1. Sets up a video/audio conversation between a user and an AI assistant
+    2. Uses Mem0 to store and retrieve memories from conversations
+    3. Creates personalized greetings based on previous interactions
+    4. Handles multi-modal interaction through audio
+    5. Demonstrates two approaches for memory management:
+       - Using Mem0 API (cloud-based memory storage)
+       - Using local configuration with custom LLM (self-hosted memory)
+
+Requirements:
+    - OpenAI API key
+    - ElevenLabs API key (for text-to-speech)
+    - Daily API key (for video/audio transport)
+    - Mem0 API key (for cloud-based memory storage)
+    - [Optional] Anthropic API key (if using Claude with local config)
+
+    Environment variables (set in .env or in your terminal using `export`):
+        DAILY_ROOM_URL=daily_room_url
+        DAILY_API_KEY=daily_api_key
+        OPENAI_API_KEY=openai_api_key
+        ELEVENLABS_API_KEY=elevenlabs_api_key
+        MEM0_API_KEY=mem0_api_key
+        ANTHROPIC_API_KEY=anthropic_api_key (if using Claude with local config)
+
+The bot runs as part of a pipeline that processes audio frames and manages the conversation flow.
+"""
+
+import os
+
+from dotenv import load_dotenv
+from loguru import logger
+
+from pipecat.audio.vad.silero import SileroVADAnalyzer
+from pipecat.frames.frames import LLMRunFrame
+from pipecat.pipeline.pipeline import Pipeline
+from pipecat.pipeline.runner import PipelineRunner
+from pipecat.pipeline.task import PipelineParams, PipelineTask
+from pipecat.processors.aggregators.llm_context import LLMContext
+from pipecat.processors.aggregators.llm_response_universal import (
+    LLMContextAggregatorPair,
+    LLMUserAggregatorParams,
+)
+from pipecat.runner.types import RunnerArguments
+from pipecat.runner.utils import create_transport
+from pipecat.services.deepgram.stt import DeepgramSTTService
+from pipecat.services.elevenlabs.tts import ElevenLabsTTSService
+from pipecat.services.mem0.memory import Mem0MemoryService
+from pipecat.services.openai.llm import OpenAILLMService
+from pipecat.transports.base_transport import BaseTransport, TransportParams
+from pipecat.transports.daily.transport import DailyParams
+from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams
+
+load_dotenv(override=True)
+
+
+async def get_initial_greeting(memory_service: Mem0MemoryService) -> str:
+    """Fetch all memories for the user and create a personalized greeting.
+
+    Args:
+        memory_service: The Mem0 memory service instance.
+
+    Returns:
+        A personalized greeting based on user memories.
+    """
+    try:
+        results = await memory_service.get_memories()
+        if not results:
+            logger.debug("No memories found for this user.")
+            return "Hello! It's nice to meet you. How can I help you today?"
+
+        # Create a personalized greeting based on memories
+        greeting = "Hello! It's great to see you again. "
+        greeting += "Based on our previous conversations, I remember: "
+        for i, memory in enumerate(results[:3], 1):
+            memory_content = memory.get("memory", "")
+            # Keep memory references brief
+            if len(memory_content) > 100:
+                memory_content = memory_content[:97] + "..."
+            greeting += f"{memory_content} "
+
+        greeting += "How can I help you today?"
+
+        logger.debug(f"Created personalized greeting from {len(results)} memories")
+        return greeting
+
+    except Exception as e:
+        logger.error(f"Error retrieving initial memories from Mem0: {e}")
+        return "Hello! How can I help you today?"
+
+
+# We use lambdas to defer transport parameter creation until the transport
+# type is selected at runtime.
+transport_params = {
+    "daily": lambda: DailyParams(
+        audio_in_enabled=True,
+        audio_out_enabled=True,
+    ),
+    "twilio": lambda: FastAPIWebsocketParams(
+        audio_in_enabled=True,
+        audio_out_enabled=True,
+    ),
+    "webrtc": lambda: TransportParams(
+        audio_in_enabled=True,
+        audio_out_enabled=True,
+    ),
+}
+
+
+async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
+    """Main bot execution function.
+
+    Sets up and runs the bot pipeline including:
+    - Daily video transport
+    - Speech-to-text and text-to-speech services
+    - Language model integration
+    - Mem0 memory service (using either API or local configuration)
+    - RTVI event handling
+    """
+    # Note: You can pass the user_id as a parameter in API call
+    USER_ID = "pipecat-demo-user"
+
+    logger.info(f"Starting bot")
+
+    stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))
+
+    # Initialize text-to-speech service
+    tts = ElevenLabsTTSService(
+        api_key=os.getenv("ELEVENLABS_API_KEY"),
+        settings=ElevenLabsTTSService.Settings(
+            voice="pNInz6obpgDQGcFmaJgB",
+        ),
+    )
+
+    # =====================================================================
+    # OPTION 1: Using Mem0 API (cloud-based approach)
+    # This approach uses Mem0's cloud service for memory management
+    # Requires: MEM0_API_KEY set in your environment
+    # =====================================================================
+    memory = Mem0MemoryService(
+        api_key=os.getenv("MEM0_API_KEY"),  # Your Mem0 API key
+        user_id=USER_ID,  # Unique identifier for the user
+        agent_id="agent1",  # Optional identifier for the agent
+        run_id="session1",  # Optional identifier for the run
+        params=Mem0MemoryService.InputParams(
+            search_limit=10,
+            search_threshold=0.3,
+            api_version="v2",
+            system_prompt="Based on previous conversations, I recall: \n\n",
+            add_as_system_message=True,
+            position=1,
+        ),
+    )
+
+    # =====================================================================
+    # OPTION 2: Using Mem0 with local configuration (self-hosted approach)
+    # This approach uses a local LLM configuration for memory management
+    # Requires: Anthropic API key if using Claude model
+    # =====================================================================
+    # Uncomment the following code and comment out the previous memory initialization to use local config
+
+    # local_config = {
+    #     "llm": {
+    #         "provider": "anthropic",
+    #         "config": {
+    #             "model": "claude-3-5-sonnet-20240620",
+    #             "api_key": os.getenv("ANTHROPIC_API_KEY"),  # Make sure to set this in your .env
+    #         }
+    #     },
+    #     "embedder": {
+    #         "provider": "openai",
+    #         "config": {
+    #             "model": "text-embedding-3-large"
+    #         }
+    #     }
+    # }
+
+    # # Initialize Mem0 memory service with local configuration
+    # memory = Mem0MemoryService(
+    #     local_config=local_config,  # Use local LLM for memory processing
+    #     user_id=USER_ID,            # Unique identifier for the user
+    #     # agent_id="agent1",        # Optional identifier for the agent
+    #     # run_id="session1",        # Optional identifier for the run
+    # )
+
+    # Initialize LLM service
+    llm = OpenAILLMService(
+        api_key=os.getenv("OPENAI_API_KEY"),
+        settings=OpenAILLMService.Settings(
+            system_instruction="""You are a personal assistant. You can remember things about the person you are talking to.
+                        Some Guidelines:
+                        - Make sure your responses are friendly yet short and concise.
+                        - If the user asks you to remember something, make sure to remember it.
+                        - Greet the user by their name if you know about it.
+                    """,
+        ),
+    )
+
+    # Set up conversation context and management
+    # The context_aggregator will automatically collect conversation context
+    context = LLMContext()
+    user_aggregator, assistant_aggregator = LLMContextAggregatorPair(
+        context,
+        user_params=LLMUserAggregatorParams(vad_analyzer=SileroVADAnalyzer()),
+    )
+
+    pipeline = Pipeline(
+        [
+            transport.input(),
+            stt,
+            user_aggregator,
+            memory,
+            llm,
+            tts,
+            transport.output(),
+            assistant_aggregator,
+        ]
+    )
+
+    task = PipelineTask(
+        pipeline,
+        params=PipelineParams(
+            enable_metrics=True,
+            enable_usage_metrics=True,
+        ),
+        idle_timeout_secs=runner_args.pipeline_idle_timeout_secs,
+    )
+
+    @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
+        greeting = await get_initial_greeting(memory)
+
+        # Add the greeting as an assistant message to start the conversation
+        context.add_message({"role": "developer", "content": greeting})
+
+        # Queue the context frame to start the conversation
+        await task.queue_frames([LLMRunFrame()])
+
+    @transport.event_handler("on_client_disconnected")
+    async def on_client_disconnected(transport, client):
+        logger.info(f"Client disconnected")
+        await task.cancel()
+
+    runner = PipelineRunner(handle_sigint=runner_args.handle_sigint)
+    await runner.run(task)
+
+
+async def bot(runner_args: RunnerArguments):
+    """Main bot entry point compatible with Pipecat Cloud."""
+    transport = await create_transport(runner_args, transport_params)
+    await run_bot(transport, runner_args)
+
+
+if __name__ == "__main__":
+    from pipecat.runner.run import main
+
+    main()