OpenAI Responses LLM

​

Overview

• OpenAIResponsesLLMService (WebSocket-based, recommended): Maintains a persistent WebSocket connection for lower-latency inference and automatically uses previous_response_id to send only incremental context when possible.
• OpenAIResponsesHttpLLMService (HTTP-based): Uses server-sent events (SSE) via HTTP streaming. Each request opens a new connection. Use this when WebSocket is not available or preferred.

​

WebSocket vs HTTP

• You need the lowest possible latency for real-time conversations
• Your workflow involves frequent tool/function calls
• You want automatic incremental context optimization without server-side storage

• WebSocket connections are blocked by your infrastructure
• You prefer stateless request/response patterns
• You don’t need the incremental context optimization

​

Installation

uv add "pipecat-ai[openai]"

​

Prerequisites

​

OpenAI Account Setup

1. OpenAI Account: Sign up at OpenAI Platform
2. API Key: Generate an API key from your account dashboard
3. Model Selection: Choose from available models (GPT-4.1, GPT-4o, GPT-4o-mini, etc.)
4. Usage Limits: Set up billing and usage limits as needed

​

Required Environment Variables

• OPENAI_API_KEY: Your OpenAI API key for authentication

​

Configuration

​

Common Parameters

​

WebSocket-Specific Parameters

​

Settings

​

Reasoning Configuration

• "none": Disables reasoning for latency-sensitive use
• "minimal", "low", "medium", "high", "xhigh": Progressive levels of reasoning effort
• "max": Applies the model’s deepest reasoning

from pipecat.services.openai.responses.llm import (
    OpenAIResponsesLLMService,
    OpenAIResponsesReasoningConfig,
)

llm = OpenAIResponsesLLMService(
    settings=OpenAIResponsesLLMService.Settings(
        model="gpt-5.4",  # Reasoning-capable model
        reasoning=OpenAIResponsesReasoningConfig(
            effort="high",
            summary="concise",
        ),
    ),
)

​

Usage

​

Basic Setup

from pipecat.services.openai.responses.llm import OpenAIResponsesLLMService

llm = OpenAIResponsesLLMService(
    api_key=os.getenv("OPENAI_API_KEY"),
    settings=OpenAIResponsesLLMService.Settings(
        model="gpt-4.1",
        system_instruction="You are a helpful assistant.",
    ),
)

from pipecat.services.openai.responses.llm import OpenAIResponsesHttpLLMService

llm = OpenAIResponsesHttpLLMService(
    api_key=os.getenv("OPENAI_API_KEY"),
    settings=OpenAIResponsesHttpLLMService.Settings(
        model="gpt-4.1",
        system_instruction="You are a helpful assistant.",
    ),
)

​

With Custom Settings

from pipecat.services.openai.responses.llm import OpenAIResponsesLLMService

llm = OpenAIResponsesLLMService(
    api_key=os.getenv("OPENAI_API_KEY"),
    settings=OpenAIResponsesLLMService.Settings(
        model="gpt-4.1",
        temperature=0.7,
        max_completion_tokens=1000,
    ),
)

​

Updating Settings at Runtime

from pipecat.frames.frames import LLMUpdateSettingsFrame

await worker.queue_frame(
    LLMUpdateSettingsFrame(
        delta=OpenAIResponsesLLMService.Settings(
            temperature=0.3,
            max_completion_tokens=500,
        )
    )
)

​

Out-of-Band Inference

from pipecat.processors.aggregators.llm_context import LLMContext

context = LLMContext()
context.add_message({"role": "user", "content": "What is the capital of France?"})

response = await llm.run_inference(
    context=context,
    max_tokens=100,
    system_instruction="You are a helpful geography assistant.",
)
print(response)  # "The capital of France is Paris."

​

Structured Output with Response Schema

schema = {
    "type": "object",
    "properties": {"answer": {"type": "string", "enum": ["yes", "no"]}},
    "required": ["answer"],
    "additionalProperties": False,
}

response = await llm.run_inference(
    context=context,
    response_schema=schema,
)
# Returns JSON text matching the schema, e.g., '{"answer": "yes"}'

​

Notes

• Persistent WebSocket connection: OpenAIResponsesLLMService keeps a WebSocket connection open and reconnects automatically when it drops, including at OpenAI’s 60-minute connection limit. For HTTP streaming, use OpenAIResponsesHttpLLMService.
• Incremental context optimization: The WebSocket variant uses previous_response_id to send only incremental context when the conversation prefix hasn’t changed, reducing latency and costs. This works with store=False (no server-side storage) via a connection-local cache.
• Responses API vs Chat Completions API: The Responses API has a different request/response structure compared to the Chat Completions API. Use OpenAILLMService for the Chat Completions API and OpenAIResponsesLLMService or OpenAIResponsesHttpLLMService for the Responses API.
• Universal LLM Context: Both services work with the universal LLMContext and LLMContextAggregatorPair, making it easy to switch between different LLM providers.
• Function calling: Supports OpenAI’s tool/function calling format. Register function handlers with llm.register_function() to handle tool calls automatically.
• Usage metrics: Automatically tracks token usage, including cached tokens and reasoning tokens.
• Service tiers: Supports OpenAI’s service tier system for prioritizing requests.
• Empty responses: If a response completes with no text and no function call (for example, only reasoning items or a refusal), the service logs a warning listing the response’s output items and pushes a non-fatal ErrorFrame. Nothing reaches the TTS, so handle it in on_pipeline_error, for example by running the LLM again.

​

Event Handlers

# OpenAIResponsesHttpLLMService
@llm.event_handler("on_completion_timeout")
async def on_completion_timeout(service):
    print("LLM completion timed out")

# OpenAIResponsesLLMService
@llm.event_handler("on_connection_error")
async def on_connection_error(service, error):
    print(f"LLM connection error: {error}")