Build a Pipecat voice agent with Prism

.env.local

RIME_API_KEY=YOUR_RIME_API_KEY

from dotenv import load_dotenv
from loguru import logger

from pipecat.frames.frames import LLMRunFrame
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.worker import PipelineWorker
from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.runner.types import RunnerArguments
from pipecat.runner.utils import create_transport
from pipecat.services.llm_service import FunctionCallParams
from pipecat.services.rime.realtime.llm import RimeRealtimeLLMService
from pipecat.transports.base_transport import TransportParams
from pipecat.workers.runner import WorkerRunner

load_dotenv(".env.local")


async def bot(runner_args: RunnerArguments):
    transport = await create_transport(
        runner_args,
        {"webrtc": lambda: TransportParams(
            audio_in_enabled=True, audio_out_enabled=True
        )},
    )

    async def get_opening_hours(params: FunctionCallParams):
        """Look up Juniper Cafe's opening hours for today."""
        logger.info("get_opening_hours called")
        await params.result_callback(
            "Juniper Cafe is open from 8 a.m. to 4 p.m. today."
        )

    llm = RimeRealtimeLLMService(
        settings=RimeRealtimeLLMService.Settings(
            system_instruction=(
                "You answer the phone for Juniper Cafe. Answer in one or two "
                "short sentences. For opening hours, use get_opening_hours."
            ),
            voice="astra",
            tools=[get_opening_hours],
        ),
    )
    context = LLMContext([
        {"role": "system", "content": "Greet the caller and ask how you can help."}
    ])
    user, assistant = llm.create_context_aggregator(context)
    pipeline = Pipeline([
        transport.input(), user, llm, transport.output(), assistant
    ])
    worker = PipelineWorker(
        pipeline, idle_timeout_secs=runner_args.pipeline_idle_timeout_secs
    )
    runner = WorkerRunner(handle_sigint=runner_args.handle_sigint)
    await runner.add_workers(worker)

    @worker.rtvi.event_handler("on_client_ready")
    async def on_client_ready(rtvi):
        await worker.queue_frame(LLMRunFrame())

    @transport.event_handler("on_client_disconnected")
    async def on_disconnected(transport, client):
        await runner.cancel()

    await runner.run()


if __name__ == "__main__":
    from pipecat.runner.run import main

    main()