wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #4141 from pipecat-ai/pk/openai-responses-websocket-service

Browse Source 
feat: add WebSocket-based OpenAI Responses LLM service
This commit is contained in:
kompfner
2026-03-30 15:25:32 -04:00
committed by GitHub
parent 87b8f38a48 0efef19d60
commit 0c11eb6fd0
13 changed files with 2788 additions and 120 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
changelog/4141.added.md Normal file
View File

@@ -0,0 +1 @@
- ⚠️ Added WebSocket-based `OpenAIResponsesLLMService` as the new default for the OpenAI Responses API. It maintains a persistent connection to `wss://api.openai.com/v1/responses` and automatically uses `previous_response_id` to send only incremental context, falling back to full context on reconnection or cache miss. The previous HTTP-based implementation is now available as `OpenAIResponsesHttpLLMService`.

125
examples/foundational/07-interruptible-openai-responses-http.py Normal file
View File

@@ -0,0 +1,125 @@
#
# Copyright (c) 2024-2026, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import os
from dotenv import load_dotenv
from loguru import logger
from pipecat.audio.vad.silero import SileroVADAnalyzer
from pipecat.frames.frames import LLMRunFrame
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineParams, PipelineTask
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.cartesia.tts import CartesiaTTSService
from pipecat.services.deepgram.stt import DeepgramSTTService
from pipecat.services.openai.responses.llm import OpenAIResponsesHttpLLMService
from pipecat.transports.base_transport import BaseTransport, TransportParams
from pipecat.transports.daily.transport import DailyParams
from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams
load_dotenv(override=True)
# We use lambdas to defer transport parameter creation until the transport
# type is selected at runtime.
transport_params = {
"daily": lambda: DailyParams(
audio_in_enabled=True,
audio_out_enabled=True,
),
"twilio": lambda: FastAPIWebsocketParams(
audio_in_enabled=True,
audio_out_enabled=True,
),
"webrtc": lambda: TransportParams(
audio_in_enabled=True,
audio_out_enabled=True,
),
}
async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
logger.info(f"Starting bot")
stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))
tts = CartesiaTTSService(
api_key=os.getenv("CARTESIA_API_KEY"),
settings=CartesiaTTSService.Settings(
voice="71a7ad14-091c-4e8e-a314-022ece01c121", # British Reading Lady
),
)
llm = OpenAIResponsesHttpLLMService(
api_key=os.getenv("OPENAI_API_KEY"),
settings=OpenAIResponsesHttpLLMService.Settings(
system_instruction="You are a helpful assistant in a voice conversation. Your responses will be spoken aloud, so avoid emojis, bullet points, or other formatting that can't be spoken. Respond to what the user said in a creative, helpful, and brief way.",
),
)
context = LLMContext()
user_aggregator, assistant_aggregator = LLMContextAggregatorPair(
context,
user_params=LLMUserAggregatorParams(vad_analyzer=SileroVADAnalyzer()),
)
pipeline = Pipeline(
[
transport.input(), # Transport user input
stt,
user_aggregator, # User responses
llm, # LLM
tts, # TTS
transport.output(), # Transport bot output
assistant_aggregator, # Assistant spoken responses
]
)
task = PipelineTask(
pipeline,
params=PipelineParams(
enable_metrics=True,
enable_usage_metrics=True,
),
idle_timeout_secs=runner_args.pipeline_idle_timeout_secs,
)
@transport.event_handler("on_client_connected")
async def on_client_connected(transport, client):
logger.info(f"Client connected")
# Kick off the conversation.
context.add_message(
{"role": "developer", "content": "Please introduce yourself to the user."}
)
await task.queue_frames([LLMRunFrame()])
@transport.event_handler("on_client_disconnected")
async def on_client_disconnected(transport, client):
logger.info(f"Client disconnected")
await task.cancel()
runner = PipelineRunner(handle_sigint=runner_args.handle_sigint)
await runner.run(task)
async def bot(runner_args: RunnerArguments):
"""Main bot entry point compatible with Pipecat Cloud."""
transport = await create_transport(runner_args, transport_params)
await run_bot(transport, runner_args)
if __name__ == "__main__":
from pipecat.runner.run import main
main()

139
examples/foundational/12-describe-image-openai-responses-http.py Normal file
View File

@@ -0,0 +1,139 @@
#
# Copyright (c) 2024-2026, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import os
from dotenv import load_dotenv
from loguru import logger
from PIL import Image
from pipecat.audio.vad.silero import SileroVADAnalyzer
from pipecat.frames.frames import LLMRunFrame
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineParams, PipelineTask
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.cartesia.tts import CartesiaTTSService
from pipecat.services.deepgram.stt import DeepgramSTTService
from pipecat.services.openai.responses.llm import OpenAIResponsesHttpLLMService
from pipecat.transports.base_transport import BaseTransport, TransportParams
from pipecat.transports.daily.transport import DailyParams
load_dotenv(override=True)
# We use lambdas to defer transport parameter creation until the transport
# type is selected at runtime.
transport_params = {
"daily": lambda: DailyParams(
audio_in_enabled=True,
audio_out_enabled=True,
),
"webrtc": lambda: TransportParams(
audio_in_enabled=True,
audio_out_enabled=True,
),
}
async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
logger.info(f"Starting bot")
stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))
tts = CartesiaTTSService(
api_key=os.getenv("CARTESIA_API_KEY"),
settings=CartesiaTTSService.Settings(
voice="71a7ad14-091c-4e8e-a314-022ece01c121", # British Reading Lady
),
)
llm = OpenAIResponsesHttpLLMService(
api_key=os.getenv("OPENAI_API_KEY"),
settings=OpenAIResponsesHttpLLMService.Settings(
system_instruction="You are a helpful assistant in a voice conversation. Your responses will be spoken aloud, so avoid emojis, bullet points, or other formatting that can't be spoken. Respond to what the user said in a creative, helpful, and brief way. You are also able to describe images.",
),
)
context = LLMContext()
user_aggregator, assistant_aggregator = LLMContextAggregatorPair(
context,
user_params=LLMUserAggregatorParams(vad_analyzer=SileroVADAnalyzer()),
)
pipeline = Pipeline(
[
transport.input(), # Transport user input
stt, # STT
user_aggregator, # User responses
llm, # LLM
tts, # TTS
transport.output(), # Transport bot output
assistant_aggregator, # Assistant spoken responses
]
)
task = PipelineTask(
pipeline,
params=PipelineParams(
enable_metrics=True,
enable_usage_metrics=True,
),
idle_timeout_secs=runner_args.pipeline_idle_timeout_secs,
)
@transport.event_handler("on_client_connected")
async def on_client_connected(transport, client):
logger.info(f"Client connected")
if not runner_args.body:
script_dir = os.path.dirname(__file__)
runner_args.body = {
"image_path": os.path.join(script_dir, "assets", "cat.jpg"),
"question": "Describe this image",
}
image_path = runner_args.body["image_path"]
question = runner_args.body["question"]
# Kick off the conversation.
image = Image.open(image_path)
message = await LLMContext.create_image_message(
image=image.tobytes(),
format="RGB",
size=image.size,
text=question,
)
context.add_message(message)
await task.queue_frames([LLMRunFrame()])
@transport.event_handler("on_client_disconnected")
async def on_client_disconnected(transport, client):
logger.info(f"Client disconnected")
await task.cancel()
runner = PipelineRunner(handle_sigint=runner_args.handle_sigint)
await runner.run(task)
async def bot(runner_args: RunnerArguments):
"""Main bot entry point compatible with Pipecat Cloud."""
transport = await create_transport(runner_args, transport_params)
await run_bot(transport, runner_args)
if __name__ == "__main__":
from pipecat.runner.run import main
main()

175
examples/foundational/14-function-calling-openai-responses-http.py Normal file
View File

@@ -0,0 +1,175 @@
#
# Copyright (c) 2024-2026, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import os
from dotenv import load_dotenv
from loguru import logger
from pipecat.adapters.schemas.function_schema import FunctionSchema
from pipecat.adapters.schemas.tools_schema import ToolsSchema
from pipecat.audio.vad.silero import SileroVADAnalyzer
from pipecat.frames.frames import LLMRunFrame, TTSSpeakFrame
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineParams, PipelineTask
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.cartesia.tts import CartesiaTTSService
from pipecat.services.deepgram.stt import DeepgramSTTService
from pipecat.services.llm_service import FunctionCallParams
from pipecat.services.openai.responses.llm import OpenAIResponsesHttpLLMService
from pipecat.transports.base_transport import BaseTransport, TransportParams
from pipecat.transports.daily.transport import DailyParams
from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams
load_dotenv(override=True)
async def fetch_weather_from_api(params: FunctionCallParams):
await params.result_callback({"conditions": "nice", "temperature": "75"})
async def fetch_restaurant_recommendation(params: FunctionCallParams):
await params.result_callback({"name": "The Golden Dragon"})
# We use lambdas to defer transport parameter creation until the transport
# type is selected at runtime.
transport_params = {
"daily": lambda: DailyParams(
audio_in_enabled=True,
audio_out_enabled=True,
),
"twilio": lambda: FastAPIWebsocketParams(
audio_in_enabled=True,
audio_out_enabled=True,
),
"webrtc": lambda: TransportParams(
audio_in_enabled=True,
audio_out_enabled=True,
),
}
async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
logger.info(f"Starting bot")
stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))
tts = CartesiaTTSService(
api_key=os.getenv("CARTESIA_API_KEY"),
settings=CartesiaTTSService.Settings(
voice="71a7ad14-091c-4e8e-a314-022ece01c121", # British Reading Lady
),
)
llm = OpenAIResponsesHttpLLMService(
api_key=os.getenv("OPENAI_API_KEY"),
settings=OpenAIResponsesHttpLLMService.Settings(
system_instruction="You are a helpful assistant in a voice conversation. Your responses will be spoken aloud, so avoid emojis, bullet points, or other formatting that can't be spoken. Respond to what the user said in a creative, helpful, and brief way.",
),
)
# You can also register a function_name of None to get all functions
# sent to the same callback with an additional function_name parameter.
llm.register_function("get_current_weather", fetch_weather_from_api)
llm.register_function("get_restaurant_recommendation", fetch_restaurant_recommendation)
@llm.event_handler("on_function_calls_started")
async def on_function_calls_started(service, function_calls):
await tts.queue_frame(TTSSpeakFrame("Let me check on that."))
weather_function = FunctionSchema(
name="get_current_weather",
description="Get the current weather",
properties={
"location": {
"type": "string",
"description": "The city and state, e.g. San Francisco, CA",
},
"format": {
"type": "string",
"enum": ["celsius", "fahrenheit"],
"description": "The temperature unit to use. Infer this from the user's location.",
},
},
required=["location", "format"],
)
restaurant_function = FunctionSchema(
name="get_restaurant_recommendation",
description="Get a restaurant recommendation",
properties={
"location": {
"type": "string",
"description": "The city and state, e.g. San Francisco, CA",
},
},
required=["location"],
)
tools = ToolsSchema(standard_tools=[weather_function, restaurant_function])
context = LLMContext(tools=tools)
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,
]
)
task = PipelineTask(
pipeline,
params=PipelineParams(
enable_metrics=True,
enable_usage_metrics=True,
),
idle_timeout_secs=runner_args.pipeline_idle_timeout_secs,
)
@transport.event_handler("on_client_connected")
async def on_client_connected(transport, client):
logger.info(f"Client connected")
# Kick off the conversation.
context.add_message(
{"role": "developer", "content": "Please introduce yourself to the user."}
)
await task.queue_frames([LLMRunFrame()])
@transport.event_handler("on_client_disconnected")
async def on_client_disconnected(transport, client):
logger.info(f"Client disconnected")
await task.cancel()
runner = PipelineRunner(handle_sigint=runner_args.handle_sigint)
await runner.run(task)
async def bot(runner_args: RunnerArguments):
"""Main bot entry point compatible with Pipecat Cloud."""
transport = await create_transport(runner_args, transport_params)
await run_bot(transport, runner_args)
if __name__ == "__main__":
from pipecat.runner.run import main
main()

6
examples/foundational/14-function-calling-openai-responses.py
View File

@@ -86,7 +86,11 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
@llm.event_handler("on_function_calls_started")
async def on_function_calls_started(service, function_calls):
await tts.queue_frame(TTSSpeakFrame("Let me check on that."))
# Avoid appending this filler message to the LLM context — it would
# alter the conversation history and prevent
# OpenAIResponsesLLMService's previous_response_id optimization from
# matching, forcing a full context resend.
await tts.queue_frame(TTSSpeakFrame("Let me check on that.", append_to_context=False))
weather_function = FunctionSchema(
name="get_current_weather",

195
examples/foundational/14d-function-calling-openai-responses-video-http.py Normal file
View File

@@ -0,0 +1,195 @@
#
# Copyright (c) 2024-2026, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import os
from dotenv import load_dotenv
from loguru import logger
from pipecat.adapters.schemas.function_schema import FunctionSchema
from pipecat.adapters.schemas.tools_schema import ToolsSchema
from pipecat.audio.vad.silero import SileroVADAnalyzer
from pipecat.frames.frames import LLMRunFrame, TTSSpeakFrame, UserImageRequestFrame
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineParams, PipelineTask
from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.processors.aggregators.llm_response_universal import (
LLMContextAggregatorPair,
LLMUserAggregatorParams,
)
from pipecat.processors.frame_processor import FrameDirection
from pipecat.runner.types import RunnerArguments
from pipecat.runner.utils import (
create_transport,
get_transport_client_id,
maybe_capture_participant_camera,
)
from pipecat.services.cartesia.tts import CartesiaTTSService
from pipecat.services.deepgram.stt import DeepgramSTTService
from pipecat.services.llm_service import FunctionCallParams
from pipecat.services.openai.responses.llm import OpenAIResponsesHttpLLMService
from pipecat.transports.base_transport import BaseTransport, TransportParams
from pipecat.transports.daily.transport import DailyParams
load_dotenv(override=True)
async def fetch_user_image(params: FunctionCallParams):
"""Fetch the user image and push it to the LLM.
When called, this function pushes a UserImageRequestFrame upstream to the
transport. As a result, the transport will request the user image and push a
UserImageRawFrame downstream which will be added to the context by the LLM
assistant aggregator. The result_callback will be invoked once the image is
retrieved and processed.
"""
user_id = params.arguments["user_id"]
question = params.arguments["question"]
logger.debug(f"Requesting image with user_id={user_id}, question={question}")
# Request a user image frame and indicate that it should be added to the
# context. Also associate it to the function call. Pass the result_callback
# so it can be invoked when the image is actually retrieved.
await params.llm.push_frame(
UserImageRequestFrame(
user_id=user_id,
text=question,
append_to_context=True,
function_name=params.function_name,
tool_call_id=params.tool_call_id,
result_callback=params.result_callback,
),
FrameDirection.UPSTREAM,
)
# We use lambdas to defer transport parameter creation until the transport
# type is selected at runtime.
transport_params = {
"daily": lambda: DailyParams(
audio_in_enabled=True,
audio_out_enabled=True,
video_in_enabled=True,
),
"webrtc": lambda: TransportParams(
audio_in_enabled=True,
audio_out_enabled=True,
video_in_enabled=True,
),
}
async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
logger.info(f"Starting bot")
stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))
tts = CartesiaTTSService(
api_key=os.getenv("CARTESIA_API_KEY"),
settings=CartesiaTTSService.Settings(
voice="71a7ad14-091c-4e8e-a314-022ece01c121", # British Reading Lady
),
)
llm = OpenAIResponsesHttpLLMService(
api_key=os.getenv("OPENAI_API_KEY"),
settings=OpenAIResponsesHttpLLMService.Settings(
system_instruction="You are a helpful assistant in a voice conversation. Your responses will be spoken aloud, so avoid emojis, bullet points, or other formatting that can't be spoken. Respond to what the user said in a creative, helpful, and brief way. You are able to describe images from the user camera.",
),
)
llm.register_function("fetch_user_image", fetch_user_image)
@llm.event_handler("on_function_calls_started")
async def on_function_calls_started(service, function_calls):
await tts.queue_frame(TTSSpeakFrame("Let me check on that.", append_to_context=False))
fetch_image_function = FunctionSchema(
name="fetch_user_image",
description="Called when the user requests a description of their camera feed",
properties={
"user_id": {
"type": "string",
"description": "The ID of the user to grab the image from",
},
"question": {
"type": "string",
"description": "The question that the user is asking about the image",
},
},
required=["user_id", "question"],
)
tools = ToolsSchema(standard_tools=[fetch_image_function])
context = LLMContext(tools=tools)
user_aggregator, assistant_aggregator = LLMContextAggregatorPair(
context,
user_params=LLMUserAggregatorParams(vad_analyzer=SileroVADAnalyzer()),
)
pipeline = Pipeline(
[
transport.input(), # Transport user input
stt, # STT
user_aggregator, # User responses
llm, # LLM
tts, # TTS
transport.output(), # Transport bot output
assistant_aggregator, # Assistant spoken responses
]
)
task = PipelineTask(
pipeline,
params=PipelineParams(
enable_metrics=True,
enable_usage_metrics=True,
),
idle_timeout_secs=runner_args.pipeline_idle_timeout_secs,
)
@transport.event_handler("on_client_connected")
async def on_client_connected(transport, client):
logger.info(f"Client connected")
await maybe_capture_participant_camera(transport, client)
client_id = get_transport_client_id(transport, client)
# Kick off the conversation.
context.add_message(
{
"role": "developer",
"content": f"Please introduce yourself to the user. Use '{client_id}' as the user ID during function calls.",
}
)
await task.queue_frames([LLMRunFrame()])
@transport.event_handler("on_client_disconnected")
async def on_client_disconnected(transport, client):
logger.info(f"Client disconnected")
await task.cancel()
@tts.event_handler("on_tts_request")
async def on_tts_request(tts, context_id: str, text: str):
logger.debug(f"On TTS request: {context_id}: {text}")
runner = PipelineRunner(handle_sigint=runner_args.handle_sigint)
await runner.run(task)
async def bot(runner_args: RunnerArguments):
"""Main bot entry point compatible with Pipecat Cloud."""
transport = await create_transport(runner_args, transport_params)
await run_bot(transport, runner_args)
if __name__ == "__main__":
from pipecat.runner.run import main
main()

249
examples/foundational/20a-persistent-context-openai-responses-http.py Normal file
View File

@@ -0,0 +1,249 @@
#
# Copyright (c) 2024-2026, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import glob
import json
import os
from datetime import datetime
from dotenv import load_dotenv
from loguru import logger
from pipecat.adapters.schemas.function_schema import FunctionSchema
from pipecat.adapters.schemas.tools_schema import ToolsSchema
from pipecat.audio.vad.silero import SileroVADAnalyzer
from pipecat.frames.frames import LLMRunFrame, TTSSpeakFrame
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineParams, PipelineTask
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.cartesia.tts import CartesiaTTSService
from pipecat.services.deepgram.stt import DeepgramSTTService
from pipecat.services.llm_service import FunctionCallParams
from pipecat.services.openai.responses.llm import OpenAIResponsesHttpLLMService
from pipecat.transports.base_transport import BaseTransport, TransportParams
from pipecat.transports.daily.transport import DailyParams
from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams
load_dotenv(override=True)
BASE_FILENAME = "/tmp/pipecat_conversation_"
async def fetch_weather_from_api(params: FunctionCallParams):
temperature = 75 if params.arguments["format"] == "fahrenheit" else 24
await params.result_callback(
{
"conditions": "nice",
"temperature": temperature,
"format": params.arguments["format"],
"timestamp": datetime.now().strftime("%Y%m%d_%H%M%S"),
}
)
async def get_saved_conversation_filenames(params: FunctionCallParams):
# Construct the full pattern including the BASE_FILENAME
full_pattern = f"{BASE_FILENAME}*.json"
# Use glob to find all matching files
matching_files = glob.glob(full_pattern)
logger.debug(f"matching files: {matching_files}")
await params.result_callback({"filenames": matching_files})
async def save_conversation(params: FunctionCallParams):
timestamp = datetime.now().strftime("%Y-%m-%d_%H:%M:%S")
filename = f"{BASE_FILENAME}{timestamp}.json"
logger.debug(
f"writing conversation to {filename}\n{json.dumps(params.context.get_messages(), indent=4)}"
)
try:
with open(filename, "w") as file:
messages = params.context.get_messages()
# remove the last message, which is the instruction we just gave to save the conversation
messages.pop()
json.dump(messages, file, indent=2)
await params.result_callback({"success": True})
except Exception as e:
await params.result_callback({"success": False, "error": str(e)})
async def load_conversation(params: FunctionCallParams):
global tts
filename = params.arguments["filename"]
logger.debug(f"loading conversation from {filename}")
try:
with open(filename, "r") as file:
params.context.set_messages(json.load(file))
logger.debug(
f"loaded conversation from {filename}\n{json.dumps(params.context.get_messages(), indent=4)}"
)
await params.llm.queue_frame(TTSSpeakFrame("Ok, I've loaded that conversation."))
except Exception as e:
await params.result_callback({"success": False, "error": str(e)})
system_instruction = "You are a helpful assistant in a voice conversation. Your responses will be spoken aloud, so avoid emojis, bullet points, or other formatting that can't be spoken. Respond to what the user said in a creative, helpful, and brief way."
weather_function = FunctionSchema(
name="get_current_weather",
description="Get the current weather",
properties={
"location": {
"type": "string",
"description": "The city and state, e.g. San Francisco, CA",
},
"format": {
"type": "string",
"enum": ["celsius", "fahrenheit"],
"description": "The temperature unit to use. Infer this from the users location.",
},
},
required=["location", "format"],
)
save_conversation_function = FunctionSchema(
name="save_conversation",
description="Save the current conversation. Use this function to persist the current conversation to external storage.",
properties={},
required=[],
)
get_filenames_function = FunctionSchema(
name="get_saved_conversation_filenames",
description="Get a list of saved conversation histories. Returns a list of filenames. Each filename includes a date and timestamp. Each file is conversation history that can be loaded into this session.",
properties={},
required=[],
)
load_conversation_function = FunctionSchema(
name="load_conversation",
description="Load a conversation history. Use this function to load a conversation history into the current session.",
properties={
"filename": {
"type": "string",
"description": "The filename of the conversation history to load.",
}
},
required=["filename"],
)
tools = ToolsSchema(
standard_tools=[
weather_function,
save_conversation_function,
get_filenames_function,
load_conversation_function,
]
)
# We use lambdas to defer transport parameter creation until the transport
# type is selected at runtime.
transport_params = {
"daily": lambda: DailyParams(
audio_in_enabled=True,
audio_out_enabled=True,
),
"twilio": lambda: FastAPIWebsocketParams(
audio_in_enabled=True,
audio_out_enabled=True,
),
"webrtc": lambda: TransportParams(
audio_in_enabled=True,
audio_out_enabled=True,
),
}
async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
logger.info(f"Starting bot")
stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))
tts = CartesiaTTSService(
api_key=os.getenv("CARTESIA_API_KEY"),
settings=CartesiaTTSService.Settings(
voice="71a7ad14-091c-4e8e-a314-022ece01c121", # British Reading Lady
),
)
llm = OpenAIResponsesHttpLLMService(
api_key=os.getenv("OPENAI_API_KEY"),
settings=OpenAIResponsesHttpLLMService.Settings(
system_instruction=system_instruction,
),
)
# you can either register a single function for all function calls, or specific functions
# llm.register_function(None, fetch_weather_from_api)
llm.register_function("get_current_weather", fetch_weather_from_api)
llm.register_function("save_conversation", save_conversation)
llm.register_function("get_saved_conversation_filenames", get_saved_conversation_filenames)
llm.register_function("load_conversation", load_conversation)
context = LLMContext(tools=tools)
user_aggregator, assistant_aggregator = LLMContextAggregatorPair(
context,
user_params=LLMUserAggregatorParams(vad_analyzer=SileroVADAnalyzer()),
)
pipeline = Pipeline(
[
transport.input(), # Transport user input
stt, # STT
user_aggregator,
llm, # LLM
tts,
transport.output(), # Transport bot output
assistant_aggregator,
]
)
task = PipelineTask(
pipeline,
params=PipelineParams(
enable_metrics=True,
enable_usage_metrics=True,
),
idle_timeout_secs=runner_args.pipeline_idle_timeout_secs,
)
@transport.event_handler("on_client_connected")
async def on_client_connected(transport, client):
logger.info(f"Client connected")
# Kick off the conversation.
await task.queue_frames([LLMRunFrame()])
@transport.event_handler("on_client_disconnected")
async def on_client_disconnected(transport, client):
logger.info(f"Client disconnected")
await task.cancel()
runner = PipelineRunner(handle_sigint=runner_args.handle_sigint)
await runner.run(task)
async def bot(runner_args: RunnerArguments):
"""Main bot entry point compatible with Pipecat Cloud."""
transport = await create_transport(runner_args, transport_params)
await run_bot(transport, runner_args)
if __name__ == "__main__":
from pipecat.runner.run import main
main()

129
examples/foundational/55zi-update-settings-openai-responses-http-llm.py Normal file
View File

@@ -0,0 +1,129 @@
#
# Copyright (c) 2024-2026, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import asyncio
import os
from dotenv import load_dotenv
from loguru import logger
from pipecat.audio.vad.silero import SileroVADAnalyzer
from pipecat.frames.frames import LLMRunFrame, LLMUpdateSettingsFrame
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineParams, PipelineTask
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.cartesia.tts import CartesiaTTSService
from pipecat.services.deepgram.stt import DeepgramSTTService
from pipecat.services.openai.responses.llm import OpenAIResponsesHttpLLMService
from pipecat.transports.base_transport import BaseTransport, TransportParams
from pipecat.transports.daily.transport import DailyParams
from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams
load_dotenv(override=True)
transport_params = {
"daily": lambda: DailyParams(
audio_in_enabled=True,
audio_out_enabled=True,
),
"twilio": lambda: FastAPIWebsocketParams(
audio_in_enabled=True,
audio_out_enabled=True,
),
"webrtc": lambda: TransportParams(
audio_in_enabled=True,
audio_out_enabled=True,
),
}
async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
logger.info(f"Starting bot")
stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))
tts = CartesiaTTSService(
api_key=os.getenv("CARTESIA_API_KEY"),
settings=CartesiaTTSService.Settings(
voice="71a7ad14-091c-4e8e-a314-022ece01c121", # British Reading Lady
),
)
llm = OpenAIResponsesHttpLLMService(
api_key=os.getenv("OPENAI_API_KEY"),
settings=OpenAIResponsesHttpLLMService.Settings(
system_instruction="You are a helpful assistant in a voice conversation. Your responses will be spoken aloud, so avoid emojis, bullet points, or other formatting that can't be spoken. Respond to what the user said in a creative, helpful, and brief way.",
),
)
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,
]
)
task = PipelineTask(
pipeline,
params=PipelineParams(
enable_metrics=True,
enable_usage_metrics=True,
),
idle_timeout_secs=runner_args.pipeline_idle_timeout_secs,
)
@transport.event_handler("on_client_connected")
async def on_client_connected(transport, client):
logger.info(f"Client connected")
context.add_message(
{"role": "developer", "content": "Please introduce yourself to the user."}
)
await task.queue_frames([LLMRunFrame()])
await asyncio.sleep(10)
logger.info("Updating OpenAI LLM settings: temperature=0.1")
await task.queue_frame(
LLMUpdateSettingsFrame(delta=OpenAIResponsesHttpLLMService.Settings(temperature=0.1))
)
@transport.event_handler("on_client_disconnected")
async def on_client_disconnected(transport, client):
logger.info(f"Client disconnected")
await task.cancel()
runner = PipelineRunner(handle_sigint=runner_args.handle_sigint)
await runner.run(task)
async def bot(runner_args: RunnerArguments):
"""Main bot entry point compatible with Pipecat Cloud."""
transport = await create_transport(runner_args, transport_params)
await run_bot(transport, runner_args)
if __name__ == "__main__":
from pipecat.runner.run import main
main()

5
scripts/evals/run-release-evals.py
View File

@@ -149,6 +149,7 @@ TESTS_07 = [
("07zk-interruptible-resembleai.py", EVAL_SIMPLE_MATH),
("07zl-interruptible-smallest.py", EVAL_SIMPLE_MATH),
("07-interruptible-openai-responses.py", EVAL_SIMPLE_MATH),
("07-interruptible-openai-responses-http.py", EVAL_SIMPLE_MATH),
# Needs a local XTTS docker instance running.
# ("07i-interruptible-xtts.py", EVAL_SIMPLE_MATH),
]
@@ -156,6 +157,7 @@ TESTS_07 = [
TESTS_12 = [
("12-describe-image-openai.py", EVAL_VISION_IMAGE(eval_speaks_first=True)),
("12-describe-image-openai-responses.py", EVAL_VISION_IMAGE(eval_speaks_first=True)),
("12-describe-image-openai-responses-http.py", EVAL_VISION_IMAGE(eval_speaks_first=True)),
("12a-describe-image-anthropic.py", EVAL_VISION_IMAGE(eval_speaks_first=True)),
("12b-describe-image-aws.py", EVAL_VISION_IMAGE(eval_speaks_first=True)),
("12c-describe-image-gemini-flash.py", EVAL_VISION_IMAGE(eval_speaks_first=True)),
@@ -170,6 +172,8 @@ TESTS_14 = [
("14-function-calling.py", EVAL_WEATHER_AND_RESTAURANT),
("14-function-calling-openai-responses.py", EVAL_WEATHER),
("14-function-calling-openai-responses.py", EVAL_WEATHER_AND_RESTAURANT),
("14-function-calling-openai-responses-http.py", EVAL_WEATHER),
("14-function-calling-openai-responses-http.py", EVAL_WEATHER_AND_RESTAURANT),
("14a-function-calling-anthropic.py", EVAL_WEATHER),
("14a-function-calling-anthropic.py", EVAL_WEATHER_AND_RESTAURANT),
("14b-function-calling-openai.py", EVAL_WEATHER),
@@ -199,6 +203,7 @@ TESTS_14 = [
("14d-function-calling-moondream-video.py", EVAL_VISION_CAMERA),
("14d-function-calling-openai-video.py", EVAL_VISION_CAMERA),
("14d-function-calling-openai-responses-video.py", EVAL_VISION_CAMERA),
("14d-function-calling-openai-responses-video-http.py", EVAL_VISION_CAMERA),
# Currently not working.
# ("14c-function-calling-together.py", EVAL_WEATHER),
# ("14l-function-calling-deepseek.py", EVAL_WEATHER),

25
src/pipecat/adapters/services/open_ai_responses_adapter.py
View File

@@ -85,16 +85,21 @@ class OpenAIResponsesLLMAdapter(BaseLLMAdapter[OpenAIResponsesLLMInvocationParam
# OpenAILLMService (system_instruction + empty messages) need the
# instructions converted to an initial developer message.
#
# NOTE: if/when we support `previous_response_id` and/or
# `conversation_id`, we'll need to revisit this logic, as it'll
# be legit to provide instructions without input items. Worth
# noting that OpenAI's docs suggest these parameters are primarily
# for development convenience rather than performance (the model
# still processes the full context), and come with the tradeoff
# of requiring OpenAI-side 30-day conversation storage, which may
# not be desirable for many users. But it could give folks an easy
# way to store/switch between conversations without needing to
# manage that storage themselves.
# NOTE: The service layer (OpenAIResponsesLLMService) internally
# manages `previous_response_id` for incremental context delivery
# over WebSocket. This runs post-adapter — the adapter always
# produces the full input list and the service determines what
# subset to send. This empty-input fallback is therefore only
# relevant for one-shot or initial calls.
#
# If we added support for user-provided explicit
# `previous_response_id` and/or `conversation_id` (overriding
# internal management), we'd need to revisit this logic, as it'd
# be legit to provide instructions without input items. Note that
# over HTTP, `previous_response_id` requires `store=True` (30-day
# OpenAI-side storage), which is why the HTTP variant doesn't use
# it. The WebSocket variant avoids this via a connection-local
# in-memory cache — see the class docstrings in llm.py.
if not input_items:
params["input"] = [{"role": "developer", "content": system_instruction}]
else:

950
src/pipecat/services/openai/responses/llm.py
View File

File diff suppressed because it is too large Load Diff

739
tests/test_openai_responses_websocket.py Normal file
View File

@@ -0,0 +1,739 @@
#
# Copyright (c) 2024-2026, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Tests for the WebSocket variant of OpenAIResponsesLLMService."""
import asyncio
import json
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.processors.frame_processor import FrameDirection
from pipecat.services.openai.responses.llm import OpenAIResponsesLLMService
def _make_service(**kwargs):
"""Create a service with the client mocked out."""
with patch.object(OpenAIResponsesLLMService, "_create_client"):
service = OpenAIResponsesLLMService(
api_key="test-key",
**kwargs,
)
service._client = AsyncMock()
return service
def _ws_events(*events):
"""Build a mock WebSocket that yields the given events from recv()."""
ws = AsyncMock()
# .recv() returns each event in order, then raises StopAsyncIteration
ws.recv = AsyncMock(side_effect=[json.dumps(e) for e in events])
ws.send = AsyncMock()
ws.close = AsyncMock()
ws.close_code = None
return ws
# ---------------------------------------------------------------------------
# Hash determinism
# ---------------------------------------------------------------------------
class TestHashInputItems:
def test_same_input_same_hash(self):
items = [{"role": "user", "content": "hello"}]
h1 = OpenAIResponsesLLMService._hash_input_items(items)
h2 = OpenAIResponsesLLMService._hash_input_items(items)
assert h1 == h2
def test_different_input_different_hash(self):
h1 = OpenAIResponsesLLMService._hash_input_items([{"role": "user", "content": "hello"}])
h2 = OpenAIResponsesLLMService._hash_input_items([{"role": "user", "content": "world"}])
assert h1 != h2
def test_order_independent_keys(self):
"""Keys within a dict should not affect hash (sort_keys=True)."""
h1 = OpenAIResponsesLLMService._hash_input_items([{"a": 1, "b": 2}])
h2 = OpenAIResponsesLLMService._hash_input_items([{"b": 2, "a": 1}])
assert h1 == h2
class TestStartsWithResponseOutput:
def test_text_message_matches_by_role(self):
response_output = [
{
"type": "message",
"role": "assistant",
"content": [{"type": "output_text", "text": "Hello!"}],
}
]
# Adapter produces a different format, but same role
items = [{"role": "assistant", "content": "Hello!"}, {"role": "user", "content": "hi"}]
assert OpenAIResponsesLLMService._starts_with_response_output(items, response_output)
def test_function_call_matches_by_call_id(self):
response_output = [
{
"type": "function_call",
"id": "fc_1",
"call_id": "call_1",
"name": "get_weather",
"arguments": '{"location": "SF"}',
}
]
# Adapter format (no "id" field)
items = [
{
"type": "function_call",
"call_id": "call_1",
"name": "get_weather",
"arguments": "{}",
},
{"type": "function_call_output", "call_id": "call_1", "output": "sunny"},
]
assert OpenAIResponsesLLMService._starts_with_response_output(items, response_output)
def test_mixed_output(self):
response_output = [
{
"type": "message",
"role": "assistant",
"content": [{"type": "output_text", "text": "Let me check."}],
},
{
"type": "function_call",
"id": "fc_1",
"call_id": "call_1",
"name": "get_weather",
"arguments": "{}",
},
]
items = [
{"role": "assistant", "content": "Let me check."},
{
"type": "function_call",
"call_id": "call_1",
"name": "get_weather",
"arguments": "{}",
},
{"type": "function_call_output", "call_id": "call_1", "output": "sunny"},
]
assert OpenAIResponsesLLMService._starts_with_response_output(items, response_output)
def test_role_mismatch(self):
response_output = [{"type": "message", "role": "assistant", "content": []}]
items = [{"role": "user", "content": "hi"}]
assert not OpenAIResponsesLLMService._starts_with_response_output(items, response_output)
def test_text_content_mismatch(self):
response_output = [
{
"type": "message",
"role": "assistant",
"content": [{"type": "output_text", "text": "Hello!"}],
}
]
items = [{"role": "assistant", "content": "Something completely different"}]
assert not OpenAIResponsesLLMService._starts_with_response_output(items, response_output)
def test_call_id_mismatch(self):
response_output = [{"type": "function_call", "call_id": "call_1", "name": "f"}]
items = [{"type": "function_call", "call_id": "call_999", "name": "f"}]
assert not OpenAIResponsesLLMService._starts_with_response_output(items, response_output)
def test_too_few_items(self):
response_output = [
{"type": "message", "role": "assistant", "content": []},
{"type": "function_call", "call_id": "call_1", "name": "f"},
]
items = [{"role": "assistant", "content": "hi"}]
assert not OpenAIResponsesLLMService._starts_with_response_output(items, response_output)
def test_empty_output_always_matches(self):
assert OpenAIResponsesLLMService._starts_with_response_output([], [])
assert OpenAIResponsesLLMService._starts_with_response_output([{"role": "user"}], [])
def test_unknown_output_type_rejects(self):
response_output = [{"type": "unknown_thing", "data": "something"}]
items = [{"role": "assistant", "content": "hi"}]
assert not OpenAIResponsesLLMService._starts_with_response_output(items, response_output)
# ---------------------------------------------------------------------------
# previous_response_id optimization
# ---------------------------------------------------------------------------
class TestPreviousResponseOptimization:
def test_no_previous_state_sends_full_input(self):
service = _make_service()
full_input = [{"role": "user", "content": "hi"}]
params = {"input": full_input, "model": "gpt-4.1"}
result = service._apply_previous_response_optimization(params, full_input)
assert result["input"] == full_input
assert "previous_response_id" not in result
def test_matching_prefix_sends_incremental(self):
service = _make_service()
# Simulate: sent [user_msg], got assistant reply "hello"
prev_input = [{"role": "user", "content": "hi"}]
prev_output = [
{
"type": "message",
"role": "assistant",
"content": [{"type": "output_text", "text": "hello"}],
}
]
service._store_previous_response_state("resp_123", prev_input, prev_output)
# Next call: adapter produces full context including assistant reply + new user msg
full_input = [
{"role": "user", "content": "hi"},
{"role": "assistant", "content": "hello"},
{"role": "user", "content": "how are you?"},
]
params = {"input": list(full_input), "model": "gpt-4.1"}
result = service._apply_previous_response_optimization(params, full_input)
assert result["previous_response_id"] == "resp_123"
# Only the new user message should be sent
assert result["input"] == [{"role": "user", "content": "how are you?"}]
def test_mismatched_prefix_sends_full(self):
service = _make_service()
prev_input = [{"role": "user", "content": "hi"}]
service._store_previous_response_state("resp_123", prev_input, [])
# Different first message
full_input = [
{"role": "user", "content": "different"},
{"role": "assistant", "content": "hello"},
]
params = {"input": list(full_input), "model": "gpt-4.1"}
result = service._apply_previous_response_optimization(params, full_input)
assert "previous_response_id" not in result
assert result["input"] == full_input
def test_same_length_sends_full(self):
"""When new input is same length as previous, no optimization."""
service = _make_service()
prev_input = [{"role": "user", "content": "hi"}]
service._store_previous_response_state("resp_123", prev_input, [])
full_input = [{"role": "user", "content": "hi"}]
params = {"input": list(full_input), "model": "gpt-4.1"}
result = service._apply_previous_response_optimization(params, full_input)
assert "previous_response_id" not in result
def test_output_mismatch_sends_full_context(self):
"""When prefix matches but output doesn't, fall back to full context."""
service = _make_service()
prev_input = [{"role": "user", "content": "hi"}]
prev_output = [
{
"type": "message",
"role": "assistant",
"content": [{"type": "output_text", "text": "hello"}],
}
]
service._store_previous_response_state("resp_123", prev_input, prev_output)
# Aggregator stored the output differently (e.g. different role)
full_input = [
{"role": "user", "content": "hi"},
{"role": "developer", "content": "something unexpected"},
{"role": "user", "content": "how are you?"},
]
params = {"input": list(full_input), "model": "gpt-4.1"}
result = service._apply_previous_response_optimization(params, full_input)
assert "previous_response_id" not in result
assert result["input"] == full_input
def test_clear_state(self):
service = _make_service()
service._store_previous_response_state("resp_123", [{"role": "user", "content": "hi"}], [])
service._clear_previous_response_state()
assert service._previous_response_id is None
assert service._previous_input_hash is None
assert service._previous_input_length is None
# ---------------------------------------------------------------------------
# _receive_response_events — text streaming
# ---------------------------------------------------------------------------
class TestReceiveResponseEventsText:
@pytest.mark.asyncio
async def test_text_deltas_pushed(self):
service = _make_service()
service._push_llm_text = AsyncMock()
service.stop_ttfb_metrics = AsyncMock()
service.start_llm_usage_metrics = AsyncMock()
ws = _ws_events(
{"type": "response.output_text.delta", "delta": "Hello"},
{"type": "response.output_text.delta", "delta": " world"},
{
"type": "response.completed",
"response": {
"id": "resp_1",
"model": "gpt-4.1",
"usage": {
"input_tokens": 10,
"output_tokens": 5,
"total_tokens": 15,
"input_tokens_details": {"cached_tokens": 0},
"output_tokens_details": {"reasoning_tokens": 0},
},
},
},
)
service._websocket = ws
context = MagicMock(spec=LLMContext)
full_input = [{"role": "user", "content": "hi"}]
await service._receive_response_events(context, full_input)
assert service._push_llm_text.call_count == 2
service._push_llm_text.assert_any_await("Hello")
service._push_llm_text.assert_any_await(" world")
@pytest.mark.asyncio
async def test_response_completed_stores_state(self):
service = _make_service()
service._push_llm_text = AsyncMock()
service.stop_ttfb_metrics = AsyncMock()
service.start_llm_usage_metrics = AsyncMock()
ws = _ws_events(
{
"type": "response.completed",
"response": {
"id": "resp_42",
"model": "gpt-4.1",
"output": [
{
"type": "message",
"role": "assistant",
"content": [{"type": "output_text", "text": "Hello!"}],
}
],
"usage": {
"input_tokens": 10,
"output_tokens": 5,
"total_tokens": 15,
"input_tokens_details": {"cached_tokens": 2},
"output_tokens_details": {"reasoning_tokens": 1},
},
},
},
)
service._websocket = ws
context = MagicMock(spec=LLMContext)
full_input = [{"role": "user", "content": "hi"}]
await service._receive_response_events(context, full_input)
assert service._previous_response_id == "resp_42"
assert service._previous_input_length == 1
assert service._previous_input_hash is not None
assert len(service._previous_response_output) == 1
assert service.start_llm_usage_metrics.called
@pytest.mark.asyncio
async def test_token_usage_metrics(self):
service = _make_service()
service._push_llm_text = AsyncMock()
service.stop_ttfb_metrics = AsyncMock()
service.start_llm_usage_metrics = AsyncMock()
ws = _ws_events(
{
"type": "response.completed",
"response": {
"id": "resp_1",
"model": "gpt-4.1",
"usage": {
"input_tokens": 100,
"output_tokens": 50,
"total_tokens": 150,
"input_tokens_details": {"cached_tokens": 20},
"output_tokens_details": {"reasoning_tokens": 10},
},
},
},
)
service._websocket = ws
context = MagicMock(spec=LLMContext)
await service._receive_response_events(context, [])
tokens = service.start_llm_usage_metrics.call_args[0][0]
assert tokens.prompt_tokens == 100
assert tokens.completion_tokens == 50
assert tokens.total_tokens == 150
assert tokens.cache_read_input_tokens == 20
assert tokens.reasoning_tokens == 10
# ---------------------------------------------------------------------------
# _receive_response_events — function calls
# ---------------------------------------------------------------------------
class TestReceiveResponseEventsFunctionCalls:
@pytest.mark.asyncio
async def test_function_call_sequence(self):
service = _make_service()
service._push_llm_text = AsyncMock()
service.stop_ttfb_metrics = AsyncMock()
service.start_llm_usage_metrics = AsyncMock()
service.run_function_calls = AsyncMock()
ws = _ws_events(
{
"type": "response.output_item.added",
"item": {
"type": "function_call",
"id": "fc_1",
"name": "get_weather",
"call_id": "call_1",
},
},
{
"type": "response.function_call_arguments.delta",
"item_id": "fc_1",
"delta": '{"loc',
},
{
"type": "response.function_call_arguments.delta",
"item_id": "fc_1",
"delta": 'ation": "SF"}',
},
{
"type": "response.function_call_arguments.done",
"item_id": "fc_1",
"arguments": '{"location": "SF"}',
},
{
"type": "response.output_item.done",
"item": {
"type": "function_call",
"id": "fc_1",
"name": "get_weather",
"call_id": "call_1",
"arguments": '{"location": "SF"}',
},
},
{
"type": "response.completed",
"response": {"id": "resp_1", "model": "gpt-4.1", "usage": None},
},
)
service._websocket = ws
context = MagicMock(spec=LLMContext)
await service._receive_response_events(context, [])
service.run_function_calls.assert_called_once()
fc_list = service.run_function_calls.call_args[0][0]
assert len(fc_list) == 1
assert fc_list[0].function_name == "get_weather"
assert fc_list[0].tool_call_id == "call_1"
assert fc_list[0].arguments == {"location": "SF"}
# ---------------------------------------------------------------------------
# _receive_response_events — errors
# ---------------------------------------------------------------------------
class TestReceiveResponseEventsErrors:
@pytest.mark.asyncio
async def test_response_failed_pushes_error(self):
service = _make_service()
service.stop_ttfb_metrics = AsyncMock()
service.start_llm_usage_metrics = AsyncMock()
service.push_error = AsyncMock()
ws = _ws_events(
{
"type": "response.failed",
"response": {
"id": "resp_1",
"status_details": {
"error": {"message": "Content filter triggered"},
},
},
},
)
service._websocket = ws
context = MagicMock(spec=LLMContext)
await service._receive_response_events(context, [])
service.push_error.assert_called_once()
assert "Content filter triggered" in service.push_error.call_args.kwargs["error_msg"]
@pytest.mark.asyncio
async def test_response_incomplete_pushes_error(self):
service = _make_service()
service.stop_ttfb_metrics = AsyncMock()
service.start_llm_usage_metrics = AsyncMock()
service.push_error = AsyncMock()
ws = _ws_events(
{
"type": "response.incomplete",
"response": {"id": "resp_1", "status_details": None},
},
)
service._websocket = ws
context = MagicMock(spec=LLMContext)
await service._receive_response_events(context, [])
service.push_error.assert_called_once()
@pytest.mark.asyncio
async def test_previous_response_not_found_raises(self):
from pipecat.services.openai.responses.llm import _PreviousResponseNotFoundError
service = _make_service()
service.stop_ttfb_metrics = AsyncMock()
ws = _ws_events(
{
"type": "error",
"error": {
"code": "previous_response_not_found",
"message": "Previous response with id 'resp_abc' not found.",
},
},
)
service._websocket = ws
context = MagicMock(spec=LLMContext)
with pytest.raises(_PreviousResponseNotFoundError):
await service._receive_response_events(context, [])
@pytest.mark.asyncio
async def test_connection_limit_reached_raises(self):
from pipecat.services.openai.responses.llm import _ConnectionLimitReachedError
service = _make_service()
service.stop_ttfb_metrics = AsyncMock()
ws = _ws_events(
{
"type": "error",
"error": {
"code": "websocket_connection_limit_reached",
"message": "Connection limit reached.",
},
},
)
service._websocket = ws
context = MagicMock(spec=LLMContext)
with pytest.raises(_ConnectionLimitReachedError):
await service._receive_response_events(context, [])
@pytest.mark.asyncio
async def test_generic_error_pushes_error(self):
service = _make_service()
service.stop_ttfb_metrics = AsyncMock()
service.start_llm_usage_metrics = AsyncMock()
service.push_error = AsyncMock()
ws = _ws_events(
{
"type": "error",
"error": {
"code": "server_error",
"message": "Internal server error",
},
},
)
service._websocket = ws
context = MagicMock(spec=LLMContext)
await service._receive_response_events(context, [])
service.push_error.assert_called_once()
assert "Internal server error" in service.push_error.call_args.kwargs["error_msg"]
class TestDrainCancelledResponse:
@pytest.mark.asyncio
async def test_drain_discards_events_until_terminal(self):
"""Draining should discard events until a terminal event arrives."""
service = _make_service()
service._needs_drain = True
ws = _ws_events(
{"type": "response.output_text.delta", "delta": "stale"},
{"type": "response.output_text.delta", "delta": "also stale"},
{"type": "response.completed", "response": {"id": "resp_old"}},
)
service._websocket = ws
await service._drain_cancelled_response()
assert not service._needs_drain
@pytest.mark.asyncio
async def test_drain_handles_pending_cancel(self):
"""If cancelled before response.created, drain should send cancel
once it sees the response.created, then continue draining."""
service = _make_service()
service._needs_drain = True
service._cancel_pending_response = True
mock_ws = AsyncMock()
mock_ws.recv = AsyncMock(
side_effect=[
json.dumps({"type": "response.created", "response": {"id": "resp_late"}}),
json.dumps({"type": "response.output_text.delta", "delta": "stale"}),
json.dumps({"type": "response.failed", "response": {"id": "resp_late"}}),
]
)
mock_ws.send = AsyncMock()
service._websocket = mock_ws
await service._drain_cancelled_response()
assert not service._needs_drain
assert not service._cancel_pending_response
# Should have sent response.cancel
cancel_calls = [
call for call in mock_ws.send.call_args_list if "response.cancel" in call.args[0]
]
assert len(cancel_calls) == 1
@pytest.mark.asyncio
async def test_drain_timeout_triggers_reconnect(self):
"""If draining takes too long, should fall back to reconnecting."""
service = _make_service()
service._needs_drain = True
service.stop_all_metrics = AsyncMock()
service.push_error = AsyncMock()
mock_ws = AsyncMock()
# recv() never returns a terminal event — times out
mock_ws.recv = AsyncMock(side_effect=asyncio.TimeoutError)
mock_ws.close = AsyncMock()
service._websocket = mock_ws
with patch(
"pipecat.services.openai.responses.llm.websocket_connect",
new_callable=AsyncMock,
return_value=AsyncMock(),
):
await service._drain_cancelled_response()
assert not service._needs_drain
# Should have reconnected (old ws closed)
mock_ws.close.assert_called_once()
# ---------------------------------------------------------------------------
# Connection lifecycle
# ---------------------------------------------------------------------------
class TestConnectionLifecycle:
@pytest.mark.asyncio
async def test_disconnect_clears_previous_response_state(self):
service = _make_service()
service._store_previous_response_state("resp_1", [{"role": "user", "content": "hi"}], [])
service.stop_all_metrics = AsyncMock()
await service._disconnect()
assert service._previous_response_id is None
assert service._previous_input_hash is None
assert service._previous_input_length is None
@pytest.mark.asyncio
async def test_reconnect_clears_state_and_reconnects(self):
service = _make_service()
service._store_previous_response_state("resp_1", [{"role": "user", "content": "hi"}], [])
service.stop_all_metrics = AsyncMock()
service.push_error = AsyncMock()
# Mock connect to set a websocket
mock_ws = AsyncMock()
mock_ws.close = AsyncMock()
service._websocket = mock_ws
with patch(
"pipecat.services.openai.responses.llm.websocket_connect",
new_callable=AsyncMock,
return_value=AsyncMock(),
):
await service._reconnect()
assert service._previous_response_id is None
mock_ws.close.assert_called_once()
@pytest.mark.asyncio
async def test_cancellation_preserves_connection_and_sets_drain(self):
"""When process_frame is cancelled (e.g. interruption), the WebSocket
connection should be preserved and _needs_drain set."""
service = _make_service()
service.stop_processing_metrics = AsyncMock()
service.push_frame = AsyncMock()
mock_ws = AsyncMock()
mock_ws.recv = AsyncMock(side_effect=asyncio.CancelledError)
mock_ws.send = AsyncMock()
service._websocket = mock_ws
context = MagicMock(spec=LLMContext)
context.tools = None
context.tool_choice = None
context.messages = [{"role": "user", "content": "hi"}]
from pipecat.frames.frames import LLMContextFrame
with pytest.raises(asyncio.CancelledError):
await service.process_frame(LLMContextFrame(context=context), FrameDirection.DOWNSTREAM)
# Connection should be preserved, not closed
assert service._websocket is mock_ws
# Should be flagged for draining before next inference
assert service._needs_drain
@pytest.mark.asyncio
async def test_ensure_connected_raises_on_failure(self):
from pipecat.services.openai.responses.llm import _RetryableError
service = _make_service()
service._websocket = None
service.push_error = AsyncMock()
# Mock connect to fail
with patch(
"pipecat.services.openai.responses.llm.websocket_connect",
new_callable=AsyncMock,
side_effect=Exception("Connection refused"),
):
with pytest.raises(_RetryableError):
await service._ensure_connected()

170
tests/test_run_inference.py
View File

@@ -20,7 +20,10 @@ from pipecat.services.anthropic.llm import AnthropicLLMService
from pipecat.services.aws.llm import AWSBedrockLLMService
from pipecat.services.google.llm import GoogleLLMService
from pipecat.services.openai.llm import OpenAILLMService
from pipecat.services.openai.responses.llm import OpenAIResponsesLLMService
from pipecat.services.openai.responses.llm import (
OpenAIResponsesHttpLLMService,
OpenAIResponsesLLMService,
)
@pytest.mark.asyncio
@@ -945,3 +948,168 @@ async def test_openai_responses_run_inference_system_instruction_param_with_empt
{"role": "developer", "content": "Summarize the conversation"}
]
assert "instructions" not in call_kwargs
# --- OpenAI Responses HTTP API tests ---
# These mirror the WebSocket variant tests above, verifying that the HTTP
# variant's run_inference (inherited from the shared base class) works
# identically.
@pytest.mark.asyncio
async def test_openai_responses_http_run_inference_with_llm_context():
"""Test run_inference with LLMContext returns expected response (HTTP variant)."""
with patch.object(OpenAIResponsesHttpLLMService, "_create_client"):
service = OpenAIResponsesHttpLLMService(
settings=OpenAIResponsesHttpLLMService.Settings(
model="gpt-4.1",
system_instruction="You are a helpful assistant",
temperature=0.7,
max_completion_tokens=100,
),
)
service._client = AsyncMock()
context = LLMContext(
messages=[
{"role": "user", "content": "Hello, world!"},
]
)
mock_response = MagicMock()
mock_response.output_text = "Hello! How can I help you today?"
service._client.responses.create = AsyncMock(return_value=mock_response)
result = await service.run_inference(context)
assert result == "Hello! How can I help you today?"
call_kwargs = service._client.responses.create.call_args.kwargs
assert call_kwargs["model"] == "gpt-4.1"
assert call_kwargs["stream"] is False
assert call_kwargs["store"] is False
assert call_kwargs["input"] == [{"role": "user", "content": "Hello, world!"}]
assert call_kwargs["instructions"] == "You are a helpful assistant"
assert call_kwargs["temperature"] == 0.7
assert call_kwargs["max_output_tokens"] == 100
@pytest.mark.asyncio
async def test_openai_responses_http_run_inference_client_exception():
"""Test that exceptions from the client are propagated (HTTP variant)."""
with patch.object(OpenAIResponsesHttpLLMService, "_create_client"):
service = OpenAIResponsesHttpLLMService()
service._client = AsyncMock()
context = LLMContext(messages=[{"role": "user", "content": "Hello"}])
service._client.responses.create = AsyncMock(side_effect=Exception("API Error"))
with pytest.raises(Exception, match="API Error"):
await service.run_inference(context)
@pytest.mark.asyncio
async def test_openai_responses_http_run_inference_system_instruction_overrides():
"""Test that system_instruction parameter overrides the settings instruction (HTTP variant)."""
with patch.object(OpenAIResponsesHttpLLMService, "_create_client"):
service = OpenAIResponsesHttpLLMService(
settings=OpenAIResponsesHttpLLMService.Settings(
model="gpt-4.1",
system_instruction="Original instruction",
),
)
service._client = AsyncMock()
context = LLMContext(
messages=[{"role": "user", "content": "Hello"}],
)
mock_response = MagicMock()
mock_response.output_text = "Response"
service._client.responses.create = AsyncMock(return_value=mock_response)
result = await service.run_inference(context, system_instruction="New system instruction")
assert result == "Response"
call_kwargs = service._client.responses.create.call_args.kwargs
assert call_kwargs["instructions"] == "New system instruction"
assert call_kwargs["input"] == [{"role": "user", "content": "Hello"}]
@pytest.mark.asyncio
async def test_openai_responses_http_run_inference_empty_context_with_instruction():
"""Test that system_instruction becomes a developer message when context is empty (HTTP)."""
with patch.object(OpenAIResponsesHttpLLMService, "_create_client"):
service = OpenAIResponsesHttpLLMService(
settings=OpenAIResponsesHttpLLMService.Settings(
model="gpt-4.1",
system_instruction="You are helpful",
),
)
service._client = AsyncMock()
context = LLMContext(messages=[])
mock_response = MagicMock()
mock_response.output_text = "Response"
service._client.responses.create = AsyncMock(return_value=mock_response)
result = await service.run_inference(context)
assert result == "Response"
call_kwargs = service._client.responses.create.call_args.kwargs
assert call_kwargs["input"] == [{"role": "developer", "content": "You are helpful"}]
assert "instructions" not in call_kwargs
@pytest.mark.asyncio
async def test_openai_responses_http_run_inference_max_tokens_override():
"""Test that max_tokens parameter overrides max_output_tokens (HTTP variant)."""
with patch.object(OpenAIResponsesHttpLLMService, "_create_client"):
service = OpenAIResponsesHttpLLMService(
settings=OpenAIResponsesHttpLLMService.Settings(
model="gpt-4.1",
max_completion_tokens=500,
),
)
service._client = AsyncMock()
context = LLMContext(
messages=[{"role": "user", "content": "Summarize this"}],
)
mock_response = MagicMock()
mock_response.output_text = "Summary"
service._client.responses.create = AsyncMock(return_value=mock_response)
result = await service.run_inference(context, max_tokens=200)
assert result == "Summary"
call_kwargs = service._client.responses.create.call_args.kwargs
assert call_kwargs["max_output_tokens"] == 200
@pytest.mark.asyncio
async def test_openai_responses_http_run_inference_system_instruction_param_with_empty_context():
"""Test system_instruction param becomes developer message for empty context (HTTP)."""
with patch.object(OpenAIResponsesHttpLLMService, "_create_client"):
service = OpenAIResponsesHttpLLMService(
settings=OpenAIResponsesHttpLLMService.Settings(model="gpt-4.1"),
)
service._client = AsyncMock()
context = LLMContext(messages=[])
mock_response = MagicMock()
mock_response.output_text = "Response"
service._client.responses.create = AsyncMock(return_value=mock_response)
result = await service.run_inference(
context, system_instruction="Summarize the conversation"
)
assert result == "Response"
call_kwargs = service._client.responses.create.call_args.kwargs
assert call_kwargs["input"] == [
{"role": "developer", "content": "Summarize the conversation"}
]
assert "instructions" not in call_kwargs