Realtime API

# /// script
# requires-python = ">=3.11"
# dependencies = [
#     "pipecat-ai[local,openai]==1.8.1",
# ]
# ///
"""Talk to a Mirai Realtime session from your microphone, with stock Pipecat.

This is Pipecat's own `OpenAIRealtimeLLMService`, unchanged. The only
Mirai-specific lines are the API key and `base_url`: speech recognition,
turn detection, the language model and speech synthesis all run on Mirai's
side, next to each other, and your process only moves audio.

    export MIRA_API_KEY=sk_live_…
    uv run pipecat_realtime.py

Optional:
    MIRA_AGENT_ID      start from one of your agents (prompt, tools, voice)
    MIRA_VOICE         a Mirai voice: neha (default), ashu, shruti, sameer
    MIRA_LANGUAGE      the caller's language, e.g. hi, en (default: auto)
    MIRA_REALTIME_URL  another endpoint (default: production); a sandbox key
                       needs wss://sandbox.voice.miraiminds.co/v2/realtime

Use headphones: the local audio transport has no echo cancellation, so on
speakers the agent hears itself and answers its own voice.
"""

import asyncio
import os

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
from pipecat.services.openai.realtime.events import (
    AudioConfiguration,
    AudioInput,
    AudioOutput,
    InputAudioTranscription,
    SessionProperties,
    TurnDetection,
)
from pipecat.services.openai.realtime.llm import OpenAIRealtimeLLMService
from pipecat.transports.local.audio import LocalAudioTransport, LocalAudioTransportParams

BASE_URL = os.getenv("MIRA_REALTIME_URL", "wss://prod.voice.miraiminds.co/v2/realtime")
AGENT_ID = os.getenv("MIRA_AGENT_ID", "")
if AGENT_ID:
    # Pipecat appends "?model=…" to whatever base_url it is given. Mirai reads
    # agent_id correctly either way and ignores the model name.
    BASE_URL += f"?agent_id={AGENT_ID}"

INSTRUCTIONS = (
    "You are Mira, a friendly voice assistant. Keep every answer to one or two "
    "short sentences, because it will be spoken. No markdown, lists or emoji. "
    "Reply in the language the caller uses."
)

# Pipecat's Realtime service sends and expects 16-bit PCM at 24 kHz, the
# protocol's default format.
SAMPLE_RATE = 24000


async def main():
    transport = LocalAudioTransport(
        LocalAudioTransportParams(audio_in_enabled=True, audio_out_enabled=True)
    )

    session = SessionProperties(
        instructions=None if AGENT_ID else INSTRUCTIONS,
        audio=AudioConfiguration(
            input=AudioInput(
                transcription=InputAudioTranscription(language=os.getenv("MIRA_LANGUAGE")),
                # Mirai's own voice activity detection and turn model decide
                # when you have finished speaking.
                turn_detection=TurnDetection(),
            ),
            output=AudioOutput(voice=os.getenv("MIRA_VOICE", "neha")),
        ),
    )

    llm = OpenAIRealtimeLLMService(
        api_key=os.environ["MIRA_API_KEY"],
        base_url=BASE_URL,
        settings=OpenAIRealtimeLLMService.Settings(session_properties=session),
    )

    # The first context runs once the session is configured: the agent speaks
    # first. Mirai never greets on its own in a Realtime session.
    context = LLMContext([{"role": "user", "content": "Say hello in one short sentence."}])
    aggregators = LLMContextAggregatorPair(context)

    pipeline = Pipeline(
        [
            transport.input(),
            aggregators.user(),
            llm,
            transport.output(),
            aggregators.assistant(),
        ]
    )
    task = PipelineTask(
        pipeline,
        params=PipelineParams(
            audio_in_sample_rate=SAMPLE_RATE,
            audio_out_sample_rate=SAMPLE_RATE,
            enable_metrics=True,
        ),
    )
    await task.queue_frames([LLMRunFrame()])
    await PipelineRunner(handle_sigint=True).run(task)


if __name__ == "__main__":
    asyncio.run(main())