Boson Higgs Realtime

uv add pipecat-boson

export BOSON_API_KEY=bai-xxxx
export BOSON_REALTIME_URL=wss://api.boson.ai/v1/realtime/
export BOSON_REALTIME_MODEL=higgs-realtime

import os

from pipecat_boson.realtime import BosonRealtimeLLMService


llm = BosonRealtimeLLMService(
    url=os.environ["BOSON_REALTIME_URL"],
    api_key=os.environ["BOSON_API_KEY"],
    model=os.getenv("BOSON_REALTIME_MODEL", "higgs-realtime"),
    voice="default",
    instructions="You are a concise and helpful voice assistant.",
)

from pipecat.pipeline.pipeline import Pipeline
from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.processors.aggregators.llm_response_universal import (
    LLMContextAggregatorPair,
)


context = LLMContext()
user_aggregator, assistant_aggregator = LLMContextAggregatorPair(
    context,
    realtime_service_mode=True,
)

pipeline = Pipeline(
    [
        transport.input(),
        user_aggregator,
        llm,
        transport.output(),
        assistant_aggregator,
    ]
)