simple_dialin.py

#
# Copyright (c) 2024–2025, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import argparse
import asyncio
import os
import sys
from loguru import logger

from call_connection_manager import CallConfigManager
from pipecat.audio.vad.silero import SileroVADAnalyzer
from pipecat.frames.frames import EndTaskFrame, LLMMessagesFrame
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineParams, PipelineTask
from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
from pipecat.services.cartesia.tts import CartesiaTTSService
from pipecat.services.openai.llm import OpenAILLMService
from pipecat.transports.services.daily import DailyDialinSettings, DailyParams, DailyTransport

logger.remove(0)
logger.add(sys.stderr, level="DEBUG")

async def main(room_url: str, token: str, body: dict):
    call_config_manager = CallConfigManager.from_json_string(body) if body else CallConfigManager()
    caller_info = call_config_manager.get_caller_info()
    caller_number = caller_info["caller_number"]
    dialed_number = caller_info["dialed_number"]
    test_mode = call_config_manager.is_test_mode()
    dialin_settings = call_config_manager.get_dialin_settings()

    logger.info(f"Caller number: {caller_number}")
    logger.info(f"Dialed number: {dialed_number}")

    # ------------ TRANSPORT SETUP ------------
    if test_mode:
        logger.info("Running in test mode")
        transport_params = DailyParams(
            api_url=os.environ.get("DAILY_API_URL", "https://api.daily.co/v1"),
            api_key=os.environ.get("HF_DAILY_API_KEY", ""),
            audio_in_enabled=True,
            audio_out_enabled=True,
            video_out_enabled=False,
            vad_analyzer=SileroVADAnalyzer(),
            transcription_enabled=True,
        )
    else:
        daily_dialin_settings = DailyDialinSettings(
            call_id=dialin_settings.get("call_id"), call_domain=dialin_settings.get("call_domain")
        )
        transport_params = DailyParams(
            api_url=os.environ.get("DAILY_API_URL", "https://api.daily.co/v1"),
            api_key=os.environ.get("HF_DAILY_API_KEY", ""),
            dialin_settings=daily_dialin_settings,
            audio_in_enabled=True,
            audio_out_enabled=True,
            video_out_enabled=False,
            vad_analyzer=SileroVADAnalyzer(),
            transcription_enabled=True,
        )

    transport = DailyTransport(room_url, token, "Simple Dialin Bot", transport_params)
    tts = CartesiaTTSService(
        api_key=os.environ.get("HF_CARTESIA_API_KEY", ""),
        voice_id="b7d50908-b17c-442d-ad8d-810c63997ed9",
    )
    llm = OpenAILLMService(api_key=os.environ.get("HF_OPENAI_API_KEY"))

    # ------------ LLM AND CONTEXT SETUP ------------
    system_instruction = """You are a friendly, helpful robot. Greet the user and ask how you can assist them."""
    messages = [call_config_manager.create_system_message(system_instruction)]
    context = OpenAILLMContext(messages)
    context_aggregator = llm.create_context_aggregator(context)

    # ------------ PIPELINE SETUP ------------
    pipeline = Pipeline([
        transport.input(),
        context_aggregator.user(),
        llm,
        tts,
        transport.output(),
        context_aggregator.assistant(),
    ])

    task = PipelineTask(pipeline, params=PipelineParams(allow_interruptions=True))

    @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):
        logger.debug(f"Participant left: {participant}, reason: {reason}")
        await task.cancel()

    # ------------ RUN PIPELINE ------------
    runner = PipelineRunner()
    await runner.run(task)

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Pipecat Simple Dialin Bot")
    parser.add_argument("-u", "--url", type=str, help="Room URL")
    parser.add_argument("-t", "--token", type=str, help="Room Token")
    parser.add_argument("-b", "--body", type=str, help="JSON configuration string")
    args = parser.parse_args()
    logger.info(f"Room URL: {args.url}")
    logger.info(f"Token: {args.token}")
    logger.info(f"Body provided: {bool(args.body)}")
    asyncio.run(main(args.url, args.token, args.body))