OpenAI GPT-Live

uv add "pipecat-ai[openai]"

import os
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.processors.aggregators.llm_response_universal import LLMContextAggregatorPair
from pipecat.services.llm_service import FunctionCallParams
from pipecat.services.openai.live.llm import OpenAILiveLLMService
from pipecat.services.openai.responses.llm import OpenAIResponsesLLMService
from pipecat.workers.runner import WorkerRunner


async def get_current_weather(params: FunctionCallParams, location: str, format: str):
    """Get the current weather.

    Args:
        location: The city and state, e.g. "San Francisco, CA".
        format: The temperature unit, "celsius" or "fahrenheit".
    """
    await params.result_callback({"conditions": "nice", "temperature": 75, "format": format})


llm = OpenAILiveLLMService(
    api_key=os.getenv("OPENAI_API_KEY"),
    settings=OpenAILiveLLMService.Settings(
        system_instruction="You are a friendly, concise voice assistant. Delegate when the user asks for current information.",
    ),
    delegation=OpenAILiveLLMService.ResponsesDelegation(
        settings=OpenAIResponsesLLMService.Settings(
            model="gpt-5.6-terra",
            system_instruction="Use the available tools to answer questions about the weather.",
            reasoning=OpenAIResponsesLLMService.ReasoningConfig(effort="low"),
        ),
    ),
)

# The context's tools are the backend model's tools; their handlers run here.
# The trailing developer message asks the model to open the conversation.
context = LLMContext(
    [{"role": "developer", "content": "Greet the user and ask how you can help."}],
    [get_current_weather],
)
user_aggregator, assistant_aggregator = LLMContextAggregatorPair(context)

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

worker = PipelineWorker(pipeline)
runner = WorkerRunner()
await runner.add_workers(worker)


@transport.event_handler("on_client_connected")
async def on_client_connected(transport, client):
    # Start the GPT-Live session from the context.
    await worker.queue_frames([LLMRunFrame()])


await runner.run()

from pipecat.services.anthropic.llm import AnthropicLLMService
from pipecat.workers.llm import BackendLLMWorker

backend = BackendLLMWorker(
    llm=AnthropicLLMService(
        api_key=os.getenv("ANTHROPIC_API_KEY"),
        settings=AnthropicLLMService.Settings(
            system_instruction="You are the backend of a voice assistant. Each message is the recent conversation as a transcript; work out what is being asked and answer it.",
        ),
    ),
    context=LLMContext(tools=[get_current_weather]),
)

llm = OpenAILiveLLMService(
    api_key=os.getenv("OPENAI_API_KEY"),
    settings=OpenAILiveLLMService.Settings(system_instruction=FRONTEND_INSTRUCTIONS),
    delegation=OpenAILiveLLMService.ClientDelegation(backend=backend),
)

# The GPT-Live model's own context: no tools here, they belong to the backend.
context = LLMContext(
    [{"role": "developer", "content": "Greet the user and ask how you can help."}],
)

from pipecat.services.llm_service import FunctionCallParams
from pipecat.workers.llm import BackendLLMWorker, BackendOutput


async def rebook_flight(params: FunctionCallParams, flight_number: str):
    booking = await booking_system.rebook(flight_number)
    told = f"You're booked on {booking.flight}. Your confirmation code is {booking.code}."
    backend = params.pipeline_worker
    assert isinstance(backend, BackendLLMWorker)
    await backend.send_output(BackendOutput(text=told))
    await params.result_callback({"status": "confirmed", "user_told": told})

async def load_conversation(params: FunctionCallParams, filename: str):
    with open(filename) as file:
        params.context.set_messages(json.load(file))
    params.context.add_message(
        {
            "role": "developer",
            "content": "The saved conversation above has just been restored. Briefly tell the user it's loaded.",
        }
    )
    await params.result_callback({"status": "loaded"})
    await params.llm.reset_conversation()

async def reset_conversation()

@llm.event_handler("on_session_started")
async def on_session_started(service, session):
    print(f"Session started: {session.id}")

@llm.event_handler("on_delegation_created")
async def on_delegation_created(service, delegation):
    print(f"Delegated to the {delegation.target} backend: {delegation.id}")