OpenAI Realtime Beta

pip install pipecat-ai[openai]

class SessionProperties(BaseModel):
    modalities: Optional[List[Literal["text", "audio"]]]
    instructions: Optional[str]
    voice: Optional[str]
    input_audio_format: Optional[Literal["pcm16", "g711_ulaw", "g711_alaw"]]
    output_audio_format: Optional[Literal["pcm16", "g711_ulaw", "g711_alaw"]]
    input_audio_transcription: Optional[InputAudioTranscription]
    turn_detection: Optional[Union[TurnDetection, bool]]
    tools: Optional[List[Dict]]
    tool_choice: Optional[Literal["auto", "none", "required"]]
    temperature: Optional[float]
    max_response_output_tokens: Optional[Union[int, Literal["inf"]]]

from pipecat.services.openai import OpenAIRealtimeBetaLLMService
from pipecat.services.openai.events import SessionProperties, TurnDetection

# Configure service
service = OpenAIRealtimeBetaLLMService(
    api_key="your-api-key",
    session_properties=SessionProperties(
        modalities=["audio", "text"],
        voice="alloy",
        turn_detection=TurnDetection(
            threshold=0.5,
            silence_duration_ms=800
        ),
        temperature=0.7
    )
)

# Use in pipeline
pipeline = Pipeline([
    audio_input,       # Produces InputAudioRawFrame
    service,           # Processes speech/generates responses
    audio_output       # Handles TTSAudioRawFrame
])

# Define tools
tools = [{
    "type": "function",
    "function": {
        "name": "get_weather",
        "description": "Get weather information",
        "parameters": {
            "type": "object",
            "properties": {
                "location": {"type": "string"}
            }
        }
    }
}]

# Configure service with tools
service = OpenAIRealtimeBetaLLMService(
    api_key="your-api-key",
    session_properties=SessionProperties(
        tools=tools,
        tool_choice="auto"
    )
)

# Register function handler
@service.function("get_weather")
async def handle_weather(location: str):
    # Implementation
    return {"temperature": 72, "condition": "sunny"}

# Server-side VAD
turn_detection = TurnDetection(
    type="server_vad",
    threshold=0.5,
    prefix_padding_ms=300,
    silence_duration_ms=800
)

# Disable turn detection
turn_detection = False

# Create context
context = OpenAIRealtimeLLMContext(
    messages=[],
    tools=[],
    system="You are a helpful assistant"
)

# Create aggregators
aggregators = service.create_context_aggregator(context)