Pipecat

Phone ⇄ Vonyx ⇄ (SIP) ⇄ LiveKit SIP ⇄ LiveKit room ⇄ Pipecat bot (LiveKitTransport)

pip install "pipecat-ai[livekit,deepgram,openai,cartesia,silero]" fastapi uvicorn python-dotenv

LIVEKIT_URL=wss://your-project.livekit.cloud
LIVEKIT_API_KEY=your_livekit_api_key
LIVEKIT_API_SECRET=your_livekit_api_secret
LIVEKIT_SIP_TRUNK_ID=ST_xxxxxxxx
DEEPGRAM_API_KEY=...
OPENAI_API_KEY=...
CARTESIA_API_KEY=...

import asyncio
import os
 
from dotenv import load_dotenv
from livekit import api
from loguru import logger
 
from pipecat.audio.vad.silero import SileroVADAnalyzer
from pipecat.frames.frames import TTSSpeakFrame
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.worker import PipelineWorker
from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.processors.aggregators.llm_response_universal import (
    LLMContextAggregatorPair,
    LLMUserAggregatorParams,
)
from pipecat.runner.livekit import generate_token_with_agent
from pipecat.services.cartesia.tts import CartesiaTTSService
from pipecat.services.deepgram.stt import DeepgramSTTService
from pipecat.services.openai.llm import OpenAILLMService
from pipecat.transports.livekit.transport import LiveKitParams, LiveKitTransport
from pipecat.workers.runner import WorkerRunner
 
load_dotenv(override=True)
 
GREETING = "Hi, this is the assistant. How can I help?"
 
 
async def run_bot(room_name: str, dial_to: str | None = None):
    token = generate_token_with_agent(
        room_name, "Pipecat Bot", os.environ["LIVEKIT_API_KEY"], os.environ["LIVEKIT_API_SECRET"]
    )
    transport = LiveKitTransport(
        url=os.environ["LIVEKIT_URL"],
        token=token,
        room_name=room_name,
        params=LiveKitParams(audio_in_enabled=True, audio_out_enabled=True),
    )
 
    stt = DeepgramSTTService(api_key=os.environ["DEEPGRAM_API_KEY"])
    llm = OpenAILLMService(
        api_key=os.environ["OPENAI_API_KEY"],
        settings=OpenAILLMService.Settings(
            system_instruction="You are a helpful phone assistant. Keep replies short and spoken.",
        ),
    )
    tts = CartesiaTTSService(
        api_key=os.environ["CARTESIA_API_KEY"],
        settings=CartesiaTTSService.Settings(voice="YOUR_CARTESIA_VOICE_ID"),
    )
 
    context = LLMContext()
    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,
    ])
    worker = PipelineWorker(pipeline)
 
    async def dial():
        lkapi = api.LiveKitAPI()
        try:
            await lkapi.sip.create_sip_participant(api.CreateSIPParticipantRequest(
                sip_trunk_id=os.environ["LIVEKIT_SIP_TRUNK_ID"],
                sip_call_to=dial_to,
                room_name=room_name,
                participant_identity="callee",
                wait_until_answered=True,
            ))
        except api.SipCallError as e:
            logger.error(f"Call failed: {e.sip_status_code} {e.sip_status}")
            await worker.cancel()
            return
        finally:
            await lkapi.aclose()
        await worker.queue_frame(TTSSpeakFrame(GREETING))
 
    @transport.event_handler("on_connected")
    async def on_connected(transport):
        if dial_to:
            asyncio.create_task(dial())
 
    @transport.event_handler("on_first_participant_joined")
    async def on_first_participant_joined(transport, participant_id):
        if not dial_to:  # inbound: the caller is already on the line
            await worker.queue_frame(TTSSpeakFrame(GREETING))
 
    @transport.event_handler("on_participant_left")
    async def on_participant_left(transport, participant_id, reason):
        await worker.cancel()  # the phone side hung up
 
    runner = WorkerRunner()
    await runner.add_workers(worker)
    await runner.run()

lkapi = api.LiveKitAPI()
await lkapi.room.delete_room(api.DeleteRoomRequest(room=room_name))
await lkapi.aclose()