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

Merge pull request #4074 from pipecat-ai/pk/openai-responses-llm-service

Browse Source 
feat: add OpenAI Responses API LLM service
This commit is contained in:
kompfner
2026-03-19 15:44:26 -04:00
committed by GitHub
parent f6e7599e49 dafbb2eb66
commit 488dc1d07e
18 changed files with 2202 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -0,0 +1 @@
- Added `OpenAIResponsesLLMService`, a new LLM service that uses the OpenAI Responses API. Supports streaming text, function calling, usage metrics, and out-of-band inference. Works with the universal `LLMContext` and `LLMContextAggregatorPair`. See `examples/foundational/07-interruptible-openai-responses.py` and `14-function-calling-openai-responses.py`.

125
examples/foundational/07-interruptible-openai-responses.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 OpenAIResponsesLLMService
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 = OpenAIResponsesLLMService(
api_key=os.getenv("OPENAI_API_KEY"),
settings=OpenAIResponsesLLMService.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.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 OpenAIResponsesLLMService
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 = OpenAIResponsesLLMService(
api_key=os.getenv("OPENAI_API_KEY"),
settings=OpenAIResponsesLLMService.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.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 OpenAIResponsesLLMService
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 = OpenAIResponsesLLMService(
api_key=os.getenv("OPENAI_API_KEY"),
settings=OpenAIResponsesLLMService.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()

195
examples/foundational/14d-function-calling-openai-responses-video.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 OpenAIResponsesLLMService
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 = OpenAIResponsesLLMService(
api_key=os.getenv("OPENAI_API_KEY"),
settings=OpenAIResponsesLLMService.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": "user",
"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.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 OpenAIResponsesLLMService
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 = OpenAIResponsesLLMService(
api_key=os.getenv("OPENAI_API_KEY"),
settings=OpenAIResponsesLLMService.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()

2
examples/foundational/20a-persistent-context-openai.py
View File

@@ -116,7 +116,7 @@ weather_function = FunctionSchema(
save_conversation_function = FunctionSchema(
name="save_conversation",
description="Save the current conversatione. Use this function to persist the current conversation to external storage.",
description="Save the current conversation. Use this function to persist the current conversation to external storage.",
properties={},
required=[],
)

2
examples/foundational/20b-persistent-context-openai-realtime-beta.py
View File

@@ -119,7 +119,7 @@ tools = [
{
"type": "function",
"name": "save_conversation",
"description": "Save the current conversatione. Use this function to persist the current conversation to external storage.",
"description": "Save the current conversation. Use this function to persist the current conversation to external storage.",
"parameters": {
"type": "object",
"properties": {},

2
examples/foundational/20b-persistent-context-openai-realtime.py
View File

@@ -125,7 +125,7 @@ tools = ToolsSchema(
),
FunctionSchema(
name="save_conversation",
description="Save the current conversatione. Use this function to persist the current conversation to external storage.",
description="Save the current conversation. Use this function to persist the current conversation to external storage.",
properties={},
required=[],
),

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

@@ -0,0 +1,127 @@
#
# 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 OpenAIResponsesLLMService
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 = OpenAIResponsesLLMService(
api_key=os.getenv("OPENAI_API_KEY"),
settings=OpenAIResponsesLLMService.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": "user", "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=OpenAIResponsesLLMService.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

@@ -147,12 +147,14 @@ TESTS_07 = [
("07zi-interruptible-piper.py", EVAL_SIMPLE_MATH),
("07zj-interruptible-kokoro.py", EVAL_SIMPLE_MATH),
("07zk-interruptible-resembleai.py", EVAL_SIMPLE_MATH),
("07-interruptible-openai-responses.py", EVAL_SIMPLE_MATH),
# Needs a local XTTS docker instance running.
# ("07i-interruptible-xtts.py", EVAL_SIMPLE_MATH),
]
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)),
("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)),
@@ -184,12 +186,15 @@ TESTS_14 = [
("14v-function-calling-openai.py", EVAL_WEATHER),
("14w-function-calling-mistral.py", EVAL_WEATHER),
("14x-function-calling-openpipe.py", EVAL_WEATHER),
("14-function-calling-openai-responses.py", EVAL_WEATHER),
("14-function-calling-openai-responses.py", EVAL_WEATHER_AND_RESTAURANT),
# Video
("14d-function-calling-anthropic-video.py", EVAL_VISION_CAMERA),
("14d-function-calling-aws-video.py", EVAL_VISION_CAMERA),
("14d-function-calling-gemini-flash-video.py", EVAL_VISION_CAMERA),
("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),
# Currently not working.
# ("14c-function-calling-together.py", EVAL_WEATHER),
# ("14l-function-calling-deepseek.py", EVAL_WEATHER),

254
src/pipecat/adapters/services/open_ai_responses_adapter.py Normal file
View File

@@ -0,0 +1,254 @@
#
# Copyright (c) 2024-2026, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
"""OpenAI Responses API adapter for Pipecat."""
import copy
from typing import Any, Dict, List, Optional, TypedDict
from loguru import logger
from openai._types import NotGiven as OpenAINotGiven
from openai.types.responses import FunctionToolParam, ResponseInputItemParam
from pipecat.adapters.base_llm_adapter import BaseLLMAdapter
from pipecat.adapters.schemas.tools_schema import ToolsSchema
from pipecat.processors.aggregators.llm_context import (
LLMContext,
LLMContextMessage,
LLMSpecificMessage,
NotGiven,
)
class OpenAIResponsesLLMInvocationParams(TypedDict, total=False):
"""Context-based parameters for invoking OpenAI Responses API."""
input: List[ResponseInputItemParam]
tools: List[FunctionToolParam] | OpenAINotGiven
instructions: str
class OpenAIResponsesLLMAdapter(BaseLLMAdapter[OpenAIResponsesLLMInvocationParams]):
"""OpenAI Responses API adapter for Pipecat.
Handles:
- Converting LLMContext messages to Responses API input items
- Converting Pipecat's standardized tools schema to Responses API function tool format
- Extracting and sanitizing messages from the LLM context for logging
"""
def __init__(self):
"""Initialize the adapter."""
super().__init__()
self._warned_system_instruction = False
@property
def id_for_llm_specific_messages(self) -> str:
"""Get the identifier used in LLMSpecificMessage instances."""
return "openai_responses"
def get_llm_invocation_params(
self,
context: LLMContext,
*,
system_instruction: Optional[str] = None,
) -> OpenAIResponsesLLMInvocationParams:
"""Get Responses API invocation parameters from a universal LLM context.
Args:
context: The LLM context containing messages, tools, etc.
system_instruction: Optional system instruction from service settings.
Returns:
Dictionary of parameters for the Responses API.
"""
messages = self.get_messages(context)
input_items = self._convert_messages_to_input(messages)
params: OpenAIResponsesLLMInvocationParams = {
"input": input_items,
"tools": self.from_standard_tools(context.tools),
}
if system_instruction:
# Compatibility: The Responses API requires at least one input
# message when instructions are provided. Contexts that worked with
# 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.
if not input_items:
params["input"] = [{"role": "developer", "content": system_instruction}]
else:
params["instructions"] = system_instruction
return params
def to_provider_tools_format(self, tools_schema: ToolsSchema) -> List[FunctionToolParam]:
"""Convert function schemas to Responses API function tool format.
Args:
tools_schema: The Pipecat tools schema to convert.
Returns:
List of Responses API function tool definitions.
"""
functions_schema = tools_schema.standard_tools
result = []
for func in functions_schema:
d = func.to_default_dict()
tool: FunctionToolParam = {
"type": "function",
"name": d["name"],
"parameters": d.get("parameters", {}),
"strict": d.get("strict", None),
}
if "description" in d:
tool["description"] = d["description"]
result.append(tool)
return result
def get_messages_for_logging(self, context: LLMContext) -> List[Dict[str, Any]]:
"""Get messages from context in a format ready for logging.
Removes or truncates sensitive data like image content for safe logging.
Args:
context: The LLM context containing messages.
Returns:
List of messages in a format ready for logging.
"""
msgs = []
for message in self.get_messages(context):
msg = copy.deepcopy(message)
if "content" in msg:
if isinstance(msg["content"], list):
for item in msg["content"]:
if item.get("type") == "image_url":
if item["image_url"]["url"].startswith("data:image/"):
item["image_url"]["url"] = "data:image/..."
if item.get("type") == "input_audio":
item["input_audio"]["data"] = "..."
msgs.append(msg)
return msgs
def _convert_messages_to_input(
self, messages: List[LLMContextMessage]
) -> List[ResponseInputItemParam]:
"""Convert LLMContext messages to Responses API input items.
Args:
messages: Messages from the LLMContext.
Returns:
List of Responses API input items.
"""
result: List[ResponseInputItemParam] = []
is_first = True
for message in messages:
if isinstance(message, LLMSpecificMessage):
result.append(message.message)
is_first = False
continue
role = message.get("role")
if role == "system":
if is_first and not self._warned_system_instruction:
logger.warning(
"System messages in LLMContext are converted to 'developer' role for the "
"Responses API. Consider using settings.system_instruction instead, which "
"maps to the 'instructions' parameter."
)
self._warned_system_instruction = True
content = message.get("content", "")
if isinstance(content, list):
content = self._convert_multimodal_content(content)
result.append({"role": "developer", "content": content})
elif role == "user":
content = message.get("content", "")
if isinstance(content, list):
content = self._convert_multimodal_content(content)
result.append({"role": "user", "content": content})
elif role == "assistant":
tool_calls = message.get("tool_calls")
if tool_calls:
for tc in tool_calls:
func = tc.get("function", {})
result.append(
{
"type": "function_call",
"call_id": tc.get("id", ""),
"name": func.get("name", ""),
"arguments": func.get("arguments", ""),
}
)
else:
content = message.get("content", "")
if isinstance(content, list):
content = self._convert_multimodal_content(content)
result.append({"role": "assistant", "content": content})
elif role == "tool":
content = message.get("content", "")
if not isinstance(content, str):
content = str(content)
result.append(
{
"type": "function_call_output",
"call_id": message.get("tool_call_id", ""),
"output": content,
}
)
is_first = False
return result
def _convert_multimodal_content(self, content: list) -> list:
"""Convert multimodal content parts to Responses API format.
Args:
content: List of content parts from the LLMContext message.
Returns:
List of content parts in Responses API format.
"""
result = []
for part in content:
part_type = part.get("type")
if part_type == "text":
result.append({"type": "input_text", "text": part.get("text", "")})
elif part_type == "image_url":
image_url_obj = part.get("image_url", {})
result.append(
{
"type": "input_image",
"image_url": image_url_obj.get("url", ""),
"detail": image_url_obj.get("detail", "auto"),
}
)
else:
# Pass through other types as-is. Note: "input_audio" is not
# yet supported by the Responses API (coming soon per OpenAI
# docs) but the LLMContext format already matches the expected
# shape, so it should work once support is enabled.
result.append(part)
return result

1
src/pipecat/services/openai/__init__.py
View File

@@ -11,6 +11,7 @@ from pipecat.services import DeprecatedModuleProxy
from .image import *
from .llm import *
from .realtime import *
from .responses.llm import *
from .stt import *
from .tts import *

5
src/pipecat/services/openai/responses/__init__.py Normal file
View File

@@ -0,0 +1,5 @@
#
# Copyright (c) 2024-2026, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#

400
src/pipecat/services/openai/responses/llm.py Normal file
View File

@@ -0,0 +1,400 @@
#
# Copyright (c) 2024-2026, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
"""OpenAI Responses API LLM service implementation."""
import json
from contextlib import asynccontextmanager
from dataclasses import dataclass, field
from typing import Any, Dict, List, Mapping, Optional
import httpx
from loguru import logger
from openai import NOT_GIVEN, AsyncOpenAI, AsyncStream, DefaultAsyncHttpxClient
from openai.types.responses import (
ResponseCompletedEvent,
ResponseFunctionCallArgumentsDeltaEvent,
ResponseFunctionCallArgumentsDoneEvent,
ResponseFunctionToolCall,
ResponseOutputItemAddedEvent,
ResponseOutputItemDoneEvent,
ResponseStreamEvent,
ResponseTextDeltaEvent,
)
from pipecat.adapters.services.open_ai_responses_adapter import (
OpenAIResponsesLLMAdapter,
OpenAIResponsesLLMInvocationParams,
)
from pipecat.frames.frames import (
Frame,
LLMContextFrame,
LLMFullResponseEndFrame,
LLMFullResponseStartFrame,
)
from pipecat.metrics.metrics import LLMTokenUsage
from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.processors.frame_processor import FrameDirection
from pipecat.services.llm_service import FunctionCallFromLLM, LLMService
from pipecat.services.settings import NOT_GIVEN as _NOT_GIVEN
from pipecat.services.settings import LLMSettings, _NotGiven
from pipecat.utils.tracing.service_decorators import traced_llm
@dataclass
class OpenAIResponsesLLMSettings(LLMSettings):
"""Settings for OpenAIResponsesLLMService.
Parameters:
max_completion_tokens: Maximum completion tokens to generate.
"""
max_completion_tokens: int | _NotGiven = field(default_factory=lambda: _NOT_GIVEN)
class OpenAIResponsesLLMService(LLMService):
"""OpenAI Responses API LLM service.
This service works with the universal LLMContext and LLMContextAggregatorPair.
Example::
llm = OpenAIResponsesLLMService(
api_key=os.getenv("OPENAI_API_KEY"),
settings=OpenAIResponsesLLMService.Settings(
model="gpt-4.1",
system_instruction="You are a helpful assistant.",
),
)
"""
Settings = OpenAIResponsesLLMSettings
_settings: Settings
adapter_class = OpenAIResponsesLLMAdapter
def __init__(
self,
*,
api_key=None,
base_url=None,
organization=None,
project=None,
default_headers: Optional[Mapping[str, str]] = None,
service_tier: Optional[str] = None,
settings: Optional[Settings] = None,
**kwargs,
):
"""Initialize the OpenAI Responses API LLM service.
Args:
api_key: OpenAI API key. If None, uses environment variable.
base_url: Custom base URL for OpenAI API. If None, uses default.
organization: OpenAI organization ID.
project: OpenAI project ID.
default_headers: Additional HTTP headers to include in requests.
service_tier: Service tier to use (e.g., "auto", "flex", "priority").
settings: Runtime-updatable settings.
**kwargs: Additional arguments passed to the parent LLMService.
"""
default_settings = self.Settings(
model="gpt-4.1",
system_instruction=None,
frequency_penalty=None,
presence_penalty=None,
seed=None,
temperature=NOT_GIVEN,
top_p=NOT_GIVEN,
top_k=None,
max_tokens=None,
max_completion_tokens=NOT_GIVEN,
filter_incomplete_user_turns=False,
user_turn_completion_config=None,
extra={},
)
if settings is not None:
default_settings.apply_update(settings)
super().__init__(
settings=default_settings,
**kwargs,
)
self._service_tier = service_tier
self._client = self._create_client(
api_key=api_key,
base_url=base_url,
organization=organization,
project=project,
default_headers=default_headers,
)
if self._settings.system_instruction:
logger.debug(f"{self}: Using system instruction: {self._settings.system_instruction}")
def _create_client(
self,
api_key=None,
base_url=None,
organization=None,
project=None,
default_headers=None,
) -> AsyncOpenAI:
"""Create an AsyncOpenAI client instance.
Args:
api_key: OpenAI API key.
base_url: Custom base URL for the API.
organization: OpenAI organization ID.
project: OpenAI project ID.
default_headers: Additional HTTP headers.
Returns:
Configured AsyncOpenAI client instance.
"""
return AsyncOpenAI(
api_key=api_key,
base_url=base_url,
organization=organization,
project=project,
http_client=DefaultAsyncHttpxClient(
limits=httpx.Limits(
max_keepalive_connections=100, max_connections=1000, keepalive_expiry=None
)
),
default_headers=default_headers,
)
def can_generate_metrics(self) -> bool:
"""Check if this service can generate processing metrics."""
return True
async def process_frame(self, frame: Frame, direction: FrameDirection):
"""Process frames for LLM completion requests.
Args:
frame: The frame to process.
direction: The direction of frame processing.
"""
await super().process_frame(frame, direction)
context = None
if isinstance(frame, LLMContextFrame):
context = frame.context
else:
await self.push_frame(frame, direction)
if context:
try:
await self.push_frame(LLMFullResponseStartFrame())
await self.start_processing_metrics()
await self._process_context(context)
except httpx.TimeoutException as e:
await self._call_event_handler("on_completion_timeout")
await self.push_error(error_msg="LLM completion timeout", exception=e)
except Exception as e:
await self.push_error(error_msg=f"Error during completion: {e}", exception=e)
finally:
await self.stop_processing_metrics()
await self.push_frame(LLMFullResponseEndFrame())
@traced_llm
async def _process_context(self, context: LLMContext):
adapter: OpenAIResponsesLLMAdapter = self.get_llm_adapter()
logger.debug(
f"{self}: Generating response from universal context "
f"{adapter.get_messages_for_logging(context)}"
)
invocation_params = adapter.get_llm_invocation_params(
context, system_instruction=self._settings.system_instruction
)
params = self._build_response_params(invocation_params)
await self.start_ttfb_metrics()
stream: AsyncStream[ResponseStreamEvent] = await self._client.responses.create(**params)
# Track function calls across stream events
function_calls: Dict[str, Dict[str, str]] = {} # item_id -> {name, call_id, arguments}
current_arguments: Dict[str, str] = {} # item_id -> accumulated arguments
# Ensure stream and its async iterator are closed on cancellation/exception
# to prevent socket leaks and uvloop crashes. Closing the iterator first
# cascades cleanup through nested async generators (httpx/httpcore internals),
# preventing uvloop's broken asyncgen finalizer from firing on Python 3.12+
# (MagicStack/uvloop#699).
@asynccontextmanager
async def _closing(stream):
chunk_iter = stream.__aiter__()
try:
yield chunk_iter
finally:
# Close the iterator first to cascade cleanup through
# nested async generators (httpx/httpcore internals).
if hasattr(chunk_iter, "aclose"):
await chunk_iter.aclose()
# Then close the stream to release HTTP resources.
if hasattr(stream, "close"):
await stream.close()
elif hasattr(stream, "aclose"):
await stream.aclose()
async with _closing(stream) as event_iter:
async for event in event_iter:
if isinstance(event, ResponseTextDeltaEvent):
await self.stop_ttfb_metrics()
await self._push_llm_text(event.delta)
elif isinstance(event, ResponseOutputItemAddedEvent):
await self.stop_ttfb_metrics()
item = event.item
if isinstance(item, ResponseFunctionToolCall):
item_id = item.id or ""
function_calls[item_id] = {
"name": item.name,
"call_id": item.call_id,
"arguments": "",
}
current_arguments[item_id] = ""
elif isinstance(event, ResponseFunctionCallArgumentsDeltaEvent):
item_id = event.item_id
if item_id in current_arguments:
current_arguments[item_id] += event.delta
elif isinstance(event, ResponseFunctionCallArgumentsDoneEvent):
item_id = event.item_id
if item_id in function_calls:
function_calls[item_id]["arguments"] = event.arguments
elif isinstance(event, ResponseOutputItemDoneEvent):
item = event.item
if isinstance(item, ResponseFunctionToolCall):
item_id = item.id or ""
if item_id in function_calls:
function_calls[item_id]["name"] = item.name
function_calls[item_id]["call_id"] = item.call_id
function_calls[item_id]["arguments"] = item.arguments
elif isinstance(event, ResponseCompletedEvent):
response = event.response
if response.usage:
tokens = LLMTokenUsage(
prompt_tokens=response.usage.input_tokens,
completion_tokens=response.usage.output_tokens,
total_tokens=response.usage.total_tokens,
cache_read_input_tokens=response.usage.input_tokens_details.cached_tokens,
reasoning_tokens=response.usage.output_tokens_details.reasoning_tokens,
)
await self.start_llm_usage_metrics(tokens)
# This field is used by @traced_llm for more detailed
# model name in tracing spans
self._full_model_name = response.model
# Process any function calls
if function_calls:
fc_list: List[FunctionCallFromLLM] = []
for item_id, fc in function_calls.items():
try:
arguments = json.loads(fc["arguments"]) if fc["arguments"] else {}
except json.JSONDecodeError:
logger.warning(
f"{self}: Failed to parse function call arguments: {fc['arguments']}"
)
arguments = {}
fc_list.append(
FunctionCallFromLLM(
context=context,
tool_call_id=fc["call_id"],
function_name=fc["name"],
arguments=arguments,
)
)
await self.run_function_calls(fc_list)
def _build_response_params(self, invocation_params: OpenAIResponsesLLMInvocationParams) -> dict:
"""Build parameters for the responses.create() call.
Args:
invocation_params: Parameters derived from the LLM context.
Returns:
Dictionary of parameters for the Responses API call.
"""
params: Dict[str, Any] = {
"model": self._settings.model,
"stream": True,
"store": False,
"input": invocation_params["input"],
}
# instructions (set by the adapter when input is non-empty)
if "instructions" in invocation_params:
params["instructions"] = invocation_params["instructions"]
# Optional parameters - only include if given
if isinstance(self._settings.temperature, (int, float)):
params["temperature"] = self._settings.temperature
if isinstance(self._settings.top_p, (int, float)):
params["top_p"] = self._settings.top_p
if isinstance(self._settings.max_completion_tokens, int):
params["max_output_tokens"] = self._settings.max_completion_tokens
if self._service_tier is not None:
params["service_tier"] = self._service_tier
# Tools
tools = invocation_params.get("tools")
if tools is not None and not isinstance(tools, type(NOT_GIVEN)):
params["tools"] = tools
# Extra settings
params.update(self._settings.extra)
return params
async def run_inference(
self,
context: LLMContext,
max_tokens: Optional[int] = None,
system_instruction: Optional[str] = None,
) -> Optional[str]:
"""Run a one-shot, out-of-band inference with the given LLM context.
Args:
context: The LLM context containing conversation history.
max_tokens: Optional maximum number of tokens to generate.
system_instruction: Optional system instruction for this inference.
Returns:
The LLM's response as a string, or None if no response is generated.
"""
adapter: OpenAIResponsesLLMAdapter = self.get_llm_adapter()
effective_instruction = system_instruction or self._settings.system_instruction
invocation_params = adapter.get_llm_invocation_params(
context, system_instruction=effective_instruction
)
params = self._build_response_params(invocation_params)
# Override for non-streaming
params["stream"] = False
if max_tokens is not None:
params["max_output_tokens"] = max_tokens
response = await self._client.responses.create(**params)
return response.output_text
__all__ = ["OpenAIResponsesLLMService", "OpenAIResponsesLLMSettings"]

6
src/pipecat/utils/tracing/service_decorators.py
View File

@@ -51,8 +51,10 @@ def _get_model_name(service) -> str:
check all the places we used to store it.
"""
return (
getattr(getattr(service, "_settings", None), "model", None)
or getattr(service, "_full_model_name", None)
# Some services store an API-response-provided detailed "full" name,
# which is distinct from the user-provided model name
getattr(service, "_full_model_name", None)
or getattr(getattr(service, "_settings", None), "model", None)
or getattr(service, "model_name", None)
or getattr(service, "_model_name", None)
or "unknown"

349
tests/test_openai_responses_adapter.py Normal file
View File

@@ -0,0 +1,349 @@
#
# Copyright (c) 2024-2026, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Unit tests for the OpenAI Responses API adapter.
Tests the conversion from LLMContext messages to Responses API input items, including:
1. Simple user/assistant text messages pass through (with correct role)
2. System role converted to developer role
3. First-message system role triggers a warning
4. Assistant messages with tool_calls produce function_call input items
5. Tool messages produce function_call_output input items
6. Mixed conversations with text + function calls convert correctly
7. Multimodal content conversion (text -> input_text, image_url -> input_image)
8. Tools schema flattening (nested function dict -> flat format)
9. Empty messages list
10. LLMSpecificMessage with llm="openai_responses" passes through
"""
import unittest
from unittest.mock import patch
from pipecat.adapters.schemas.function_schema import FunctionSchema
from pipecat.adapters.schemas.tools_schema import ToolsSchema
from pipecat.adapters.services.open_ai_responses_adapter import OpenAIResponsesLLMAdapter
from pipecat.processors.aggregators.llm_context import LLMContext, LLMStandardMessage
class TestOpenAIResponsesAdapter(unittest.TestCase):
def setUp(self):
self.adapter = OpenAIResponsesLLMAdapter()
def test_simple_user_assistant_messages(self):
"""Simple user/assistant text messages are converted correctly."""
messages: list[LLMStandardMessage] = [
{"role": "user", "content": "Hello"},
{"role": "assistant", "content": "Hi there!"},
]
context = LLMContext(messages=messages)
params = self.adapter.get_llm_invocation_params(context)
self.assertEqual(len(params["input"]), 2)
self.assertEqual(params["input"][0], {"role": "user", "content": "Hello"})
self.assertEqual(params["input"][1], {"role": "assistant", "content": "Hi there!"})
def test_system_role_converted_to_developer(self):
"""System role messages are converted to developer role."""
messages: list[LLMStandardMessage] = [
{"role": "system", "content": "You are helpful."},
{"role": "user", "content": "Hello"},
]
context = LLMContext(messages=messages)
params = self.adapter.get_llm_invocation_params(context)
self.assertEqual(params["input"][0]["role"], "developer")
self.assertEqual(params["input"][0]["content"], "You are helpful.")
def test_first_system_message_triggers_warning(self):
"""First system message triggers a warning about using system_instruction."""
# Use a fresh adapter so the warning hasn't been emitted yet
adapter = OpenAIResponsesLLMAdapter()
messages: list[LLMStandardMessage] = [
{"role": "system", "content": "You are helpful."},
{"role": "user", "content": "Hello"},
]
context = LLMContext(messages=messages)
with patch("pipecat.adapters.services.open_ai_responses_adapter.logger") as mock_logger:
adapter.get_llm_invocation_params(context)
mock_logger.warning.assert_called_once()
warning_msg = mock_logger.warning.call_args[0][0]
self.assertIn("system_instruction", warning_msg)
def test_non_initial_system_message_no_warning(self):
"""Non-initial system messages are converted without a warning."""
messages: list[LLMStandardMessage] = [
{"role": "user", "content": "Hello"},
{"role": "system", "content": "New instruction"},
]
context = LLMContext(messages=messages)
adapter = OpenAIResponsesLLMAdapter()
with patch("pipecat.adapters.services.open_ai_responses_adapter.logger") as mock_logger:
params = adapter.get_llm_invocation_params(context)
mock_logger.warning.assert_not_called()
self.assertEqual(params["input"][1]["role"], "developer")
self.assertEqual(params["input"][1]["content"], "New instruction")
def test_first_system_message_warning_fires_only_once(self):
"""The first-system-message warning fires only once per adapter instance."""
messages: list[LLMStandardMessage] = [
{"role": "system", "content": "You are helpful."},
{"role": "user", "content": "Hello"},
]
context = LLMContext(messages=messages)
adapter = OpenAIResponsesLLMAdapter()
with patch("pipecat.adapters.services.open_ai_responses_adapter.logger") as mock_logger:
adapter.get_llm_invocation_params(context)
adapter.get_llm_invocation_params(context)
# Warning should have been emitted exactly once, not twice
mock_logger.warning.assert_called_once()
def test_assistant_tool_calls_to_function_call(self):
"""Assistant messages with tool_calls produce function_call input items."""
messages = [
{
"role": "assistant",
"tool_calls": [
{
"id": "call_123",
"function": {
"name": "get_weather",
"arguments": '{"location": "SF"}',
},
"type": "function",
}
],
}
]
context = LLMContext(messages=messages)
params = self.adapter.get_llm_invocation_params(context)
self.assertEqual(len(params["input"]), 1)
fc = params["input"][0]
self.assertEqual(fc["type"], "function_call")
self.assertEqual(fc["call_id"], "call_123")
self.assertEqual(fc["name"], "get_weather")
self.assertEqual(fc["arguments"], '{"location": "SF"}')
def test_multiple_tool_calls(self):
"""Multiple tool calls in one assistant message produce multiple function_call items."""
messages = [
{
"role": "assistant",
"tool_calls": [
{
"id": "call_1",
"function": {"name": "get_weather", "arguments": '{"location": "SF"}'},
"type": "function",
},
{
"id": "call_2",
"function": {"name": "get_restaurant", "arguments": '{"location": "SF"}'},
"type": "function",
},
],
}
]
context = LLMContext(messages=messages)
params = self.adapter.get_llm_invocation_params(context)
self.assertEqual(len(params["input"]), 2)
self.assertEqual(params["input"][0]["name"], "get_weather")
self.assertEqual(params["input"][1]["name"], "get_restaurant")
def test_tool_message_to_function_call_output(self):
"""Tool role messages produce function_call_output input items."""
messages = [
{
"role": "tool",
"content": '{"temperature": "72"}',
"tool_call_id": "call_123",
}
]
context = LLMContext(messages=messages)
params = self.adapter.get_llm_invocation_params(context)
self.assertEqual(len(params["input"]), 1)
fco = params["input"][0]
self.assertEqual(fco["type"], "function_call_output")
self.assertEqual(fco["call_id"], "call_123")
self.assertEqual(fco["output"], '{"temperature": "72"}')
def test_mixed_conversation(self):
"""Mixed conversation with text + function calls converts correctly."""
messages = [
{"role": "user", "content": "What's the weather in SF?"},
{
"role": "assistant",
"tool_calls": [
{
"id": "call_abc",
"function": {"name": "get_weather", "arguments": '{"location": "SF"}'},
"type": "function",
}
],
},
{
"role": "tool",
"content": '{"temp": "72"}',
"tool_call_id": "call_abc",
},
{"role": "assistant", "content": "It's 72 degrees in SF."},
]
context = LLMContext(messages=messages)
params = self.adapter.get_llm_invocation_params(context)
self.assertEqual(len(params["input"]), 4)
self.assertEqual(params["input"][0]["role"], "user")
self.assertEqual(params["input"][1]["type"], "function_call")
self.assertEqual(params["input"][2]["type"], "function_call_output")
self.assertEqual(params["input"][3]["role"], "assistant")
def test_multimodal_text_conversion(self):
"""Multimodal text content parts are converted to input_text."""
messages = [
{
"role": "user",
"content": [
{"type": "text", "text": "What's in this image?"},
],
}
]
context = LLMContext(messages=messages)
params = self.adapter.get_llm_invocation_params(context)
content = params["input"][0]["content"]
self.assertEqual(len(content), 1)
self.assertEqual(content[0]["type"], "input_text")
self.assertEqual(content[0]["text"], "What's in this image?")
def test_multimodal_image_conversion(self):
"""Multimodal image_url content parts are converted to input_image."""
messages = [
{
"role": "user",
"content": [
{"type": "text", "text": "Describe this:"},
{
"type": "image_url",
"image_url": {"url": "data:image/jpeg;base64,abc123"},
},
],
}
]
context = LLMContext(messages=messages)
params = self.adapter.get_llm_invocation_params(context)
content = params["input"][0]["content"]
self.assertEqual(len(content), 2)
self.assertEqual(content[0]["type"], "input_text")
self.assertEqual(content[1]["type"], "input_image")
self.assertEqual(content[1]["image_url"], "data:image/jpeg;base64,abc123")
self.assertEqual(content[1]["detail"], "auto")
def test_multimodal_image_with_detail(self):
"""Image content parts preserve the detail setting when provided."""
messages = [
{
"role": "user",
"content": [
{
"type": "image_url",
"image_url": {"url": "https://example.com/img.png", "detail": "high"},
},
],
}
]
context = LLMContext(messages=messages)
params = self.adapter.get_llm_invocation_params(context)
content = params["input"][0]["content"]
self.assertEqual(content[0]["detail"], "high")
def test_tools_schema_flattening(self):
"""Tools schema with nested function dict is flattened to Responses API format."""
weather_fn = FunctionSchema(
name="get_weather",
description="Get the current weather",
properties={
"location": {"type": "string", "description": "The city"},
},
required=["location"],
)
tools = ToolsSchema(standard_tools=[weather_fn])
context = LLMContext(tools=tools)
params = self.adapter.get_llm_invocation_params(context)
tool_list = params["tools"]
self.assertEqual(len(tool_list), 1)
tool = tool_list[0]
self.assertEqual(tool["type"], "function")
self.assertEqual(tool["name"], "get_weather")
self.assertEqual(tool["description"], "Get the current weather")
self.assertIn("properties", tool["parameters"])
def test_empty_messages(self):
"""Empty messages list produces empty input list."""
context = LLMContext(messages=[])
params = self.adapter.get_llm_invocation_params(context)
self.assertEqual(params["input"], [])
def test_llm_specific_message_passthrough(self):
"""LLMSpecificMessage with llm='openai_responses' passes through."""
specific_msg = self.adapter.create_llm_specific_message(
{"type": "function_call", "call_id": "x", "name": "foo", "arguments": "{}"}
)
messages = [
{"role": "user", "content": "Hello"},
specific_msg,
]
context = LLMContext(messages=messages)
params = self.adapter.get_llm_invocation_params(context)
self.assertEqual(len(params["input"]), 2)
self.assertEqual(params["input"][0]["role"], "user")
self.assertEqual(params["input"][1]["type"], "function_call")
def test_id_for_llm_specific_messages(self):
"""Adapter identifier is 'openai_responses'."""
self.assertEqual(self.adapter.id_for_llm_specific_messages, "openai_responses")
def test_system_instruction_with_messages_sets_instructions(self):
"""When system_instruction is provided and input is non-empty, sets instructions."""
messages: list[LLMStandardMessage] = [
{"role": "user", "content": "Hello"},
]
context = LLMContext(messages=messages)
params = self.adapter.get_llm_invocation_params(context, system_instruction="Be helpful.")
self.assertEqual(params["instructions"], "Be helpful.")
self.assertEqual(len(params["input"]), 1)
self.assertEqual(params["input"][0]["role"], "user")
def test_system_instruction_with_empty_input_becomes_developer_message(self):
"""When system_instruction is provided but input is empty, it becomes a developer message."""
context = LLMContext(messages=[])
params = self.adapter.get_llm_invocation_params(context, system_instruction="Be helpful.")
self.assertNotIn("instructions", params)
self.assertEqual(len(params["input"]), 1)
self.assertEqual(params["input"][0]["role"], "developer")
self.assertEqual(params["input"][0]["content"], "Be helpful.")
def test_no_system_instruction_omits_instructions(self):
"""When no system_instruction is provided, instructions key is absent."""
context = LLMContext(messages=[{"role": "user", "content": "Hi"}])
params = self.adapter.get_llm_invocation_params(context)
self.assertNotIn("instructions", params)
if __name__ == "__main__":
unittest.main()

170
tests/test_run_inference.py
View File

@@ -20,6 +20,7 @@ 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
@pytest.mark.asyncio
@@ -765,3 +766,172 @@ async def test_aws_bedrock_run_inference_system_instruction_none_unchanged():
assert result == "Response"
call_kwargs = mock_client.converse.call_args.kwargs
assert call_kwargs["system"] == [{"text": "Original system"}]
# --- OpenAI Responses API tests ---
@pytest.mark.asyncio
async def test_openai_responses_run_inference_with_llm_context():
"""Test run_inference with LLMContext returns expected response."""
with patch.object(OpenAIResponsesLLMService, "_create_client"):
service = OpenAIResponsesLLMService(
settings=OpenAIResponsesLLMService.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_run_inference_client_exception():
"""Test that exceptions from the client are propagated."""
with patch.object(OpenAIResponsesLLMService, "_create_client"):
service = OpenAIResponsesLLMService()
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_run_inference_system_instruction_overrides():
"""Test that system_instruction parameter overrides the settings instruction."""
with patch.object(OpenAIResponsesLLMService, "_create_client"):
service = OpenAIResponsesLLMService(
settings=OpenAIResponsesLLMService.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_run_inference_empty_context_with_instruction():
"""Test that system_instruction becomes a developer message when context is empty."""
with patch.object(OpenAIResponsesLLMService, "_create_client"):
service = OpenAIResponsesLLMService(
settings=OpenAIResponsesLLMService.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
# With empty context, instruction should become a developer message
assert call_kwargs["input"] == [{"role": "developer", "content": "You are helpful"}]
assert "instructions" not in call_kwargs
@pytest.mark.asyncio
async def test_openai_responses_run_inference_max_tokens_override():
"""Test that max_tokens parameter overrides max_output_tokens."""
with patch.object(OpenAIResponsesLLMService, "_create_client"):
service = OpenAIResponsesLLMService(
settings=OpenAIResponsesLLMService.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_run_inference_system_instruction_param_with_empty_context():
"""Test that system_instruction param becomes a developer message when context is empty.
The Responses API rejects requests with instructions but no input items.
When run_inference is called with an explicit system_instruction and an
empty context, the instruction must become a developer message — not be
sent as the instructions parameter.
"""
with patch.object(OpenAIResponsesLLMService, "_create_client"):
service = OpenAIResponsesLLMService(
settings=OpenAIResponsesLLMService.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