Pipecat voice agent routing

mkdir speko-pipecat && cd speko-pipecat
uv init --bare --python 3.11
uv add "pipecat-ai[openai,silero,webrtc,runner]~=1.6.0" "python-dotenv>=1.0,<2"
export SPEKO_API_KEY=sk_live_...

curl -s https://api.speko.ai/v1/models

"""Minimal Pipecat WebRTC bot using Speko's OpenAI-compatible API."""

from __future__ import annotations

import os

from dotenv import load_dotenv
from pipecat.audio.vad.silero import SileroVADAnalyzer
from pipecat.frames.frames import LLMRunFrame
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.worker import PipelineParams, PipelineWorker
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.openai.llm import OpenAILLMService
from pipecat.services.openai.stt import OpenAISTTService
from pipecat.services.openai.tts import OpenAITTSService
from pipecat.transcriptions.language import Language
from pipecat.transports.base_transport import BaseTransport, TransportParams
from pipecat.workers.runner import WorkerRunner


load_dotenv()

SPEKO_API_KEY = os.getenv("SPEKO_API_KEY")
SPEKO_BASE_URL = "https://api.speko.ai/v1"
SPEKO_OBJECTIVE = os.getenv("SPEKO_OBJECTIVE", "balanced")
SPEKO_LANGUAGE = os.getenv("SPEKO_LANGUAGE", "en")

if not SPEKO_API_KEY:
    raise RuntimeError("SPEKO_API_KEY is required")


transport_params = {
    "webrtc": lambda: TransportParams(
        audio_in_enabled=True,
        audio_out_enabled=True,
    ),
}


def language_from_env() -> Language | str:
    """Preserve regional tags within the router's enabled primary languages."""

    primary = SPEKO_LANGUAGE.split("-", 1)[0].lower()
    if primary not in {"en", "es"}:
        raise ValueError("SPEKO_LANGUAGE must use the enabled en or es primary tag")

    try:
        return Language(SPEKO_LANGUAGE)
    except ValueError:
        return SPEKO_LANGUAGE


async def run_bot(
    transport: BaseTransport,
    runner_args: RunnerArguments,
) -> None:
    stt = OpenAISTTService(
        api_key=SPEKO_API_KEY,
        base_url=SPEKO_BASE_URL,
        settings=OpenAISTTService.Settings(
            model="auto",
            language=language_from_env(),
        ),
    )
    llm = OpenAILLMService(
        api_key=SPEKO_API_KEY,
        base_url=SPEKO_BASE_URL,
        default_headers={"X-Speko-Objective": SPEKO_OBJECTIVE},
        settings=OpenAILLMService.Settings(
            model="auto",
            system_instruction="You are a helpful assistant in a voice conversation.",
        ),
    )
    tts = OpenAITTSService(
        api_key=SPEKO_API_KEY,
        base_url=SPEKO_BASE_URL,
        # Speko returns 24 kHz mono PCM for routed TTS.
        sample_rate=24000,
        settings=OpenAITTSService.Settings(
            model="auto",
            voice="alloy",
        ),
    )

    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,
        params=PipelineParams(enable_metrics=True),
        idle_timeout_secs=runner_args.pipeline_idle_timeout_secs,
    )

    @transport.event_handler("on_client_connected")
    async def on_client_connected(transport: BaseTransport, client: object) -> None:
        del transport, client
        context.add_message(
            {
                "role": "developer",
                "content": "Introduce yourself in one short sentence.",
            }
        )
        await worker.queue_frames([LLMRunFrame()])

    @transport.event_handler("on_client_disconnected")
    async def on_client_disconnected(transport: BaseTransport, client: object) -> None:
        del transport, client
        await worker.cancel()

    runner = WorkerRunner(handle_sigint=runner_args.handle_sigint)
    await runner.add_workers(worker)
    await runner.run()


async def bot(runner_args: RunnerArguments) -> None:
    transport = await create_transport(runner_args, transport_params)
    await run_bot(transport, runner_args)


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

    main()

uv run python bot.py -t webrtc
# Then open http://localhost:7860, click Connect, and start talking.

import os

from pipecat.services.openai.llm import OpenAILLMService

llm = OpenAILLMService(
    api_key=os.environ["SPEKO_API_KEY"],
    base_url="https://api.speko.ai/v1",
    default_headers={"X-Speko-Objective": "latency"},
    settings=OpenAILLMService.Settings(model="auto"),
)

import os

from pipecat.services.openai.tts import OpenAITTSService

tts = OpenAITTSService(
    api_key=os.environ["SPEKO_API_KEY"],
    base_url="https://api.speko.ai/v1",
    sample_rate=24000,
    # The id pins the provider. The voice stays a preset name, and the router
    # swaps in that provider's own default voice.
    settings=OpenAITTSService.Settings(model="cartesia:sonic-3.5", voice="alloy"),
)