From 45186cc4ce0024c414216d79ce8f5681f1a5f7b5 Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Wed, 18 Mar 2026 11:45:23 -0400 Subject: [PATCH 01/19] feat: add OpenAI Responses API LLM service MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add OpenAIResponsesLLMService using the Responses API, with a dedicated adapter that converts LLMContext messages to Responses API input items (system→developer, tool_calls→function_call, tool→function_call_output, multimodal content conversion, and tools schema flattening). - New adapter: open_ai_responses_adapter.py - New service: openai/responses/llm.py - Examples: 07-interruptible and 14-function-calling variants - 19 unit tests for adapter conversion logic - Eval entries for both examples --- .../07-interruptible-openai-responses.py | 125 ++++++ .../14-function-calling-openai-responses.py | 175 ++++++++ scripts/evals/run-release-evals.py | 3 + .../services/open_ai_responses_adapter.py | 240 +++++++++++ src/pipecat/services/openai/__init__.py | 1 + .../services/openai/responses/__init__.py | 5 + src/pipecat/services/openai/responses/llm.py | 393 ++++++++++++++++++ tests/test_openai_responses_adapter.py | 349 ++++++++++++++++ 8 files changed, 1291 insertions(+) create mode 100644 examples/foundational/07-interruptible-openai-responses.py create mode 100644 examples/foundational/14-function-calling-openai-responses.py create mode 100644 src/pipecat/adapters/services/open_ai_responses_adapter.py create mode 100644 src/pipecat/services/openai/responses/__init__.py create mode 100644 src/pipecat/services/openai/responses/llm.py create mode 100644 tests/test_openai_responses_adapter.py diff --git a/examples/foundational/07-interruptible-openai-responses.py b/examples/foundational/07-interruptible-openai-responses.py new file mode 100644 index 000000000..baae3754a --- /dev/null +++ b/examples/foundational/07-interruptible-openai-responses.py @@ -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() diff --git a/examples/foundational/14-function-calling-openai-responses.py b/examples/foundational/14-function-calling-openai-responses.py new file mode 100644 index 000000000..58cac774a --- /dev/null +++ b/examples/foundational/14-function-calling-openai-responses.py @@ -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() diff --git a/scripts/evals/run-release-evals.py b/scripts/evals/run-release-evals.py index 625f33564..9671703ba 100644 --- a/scripts/evals/run-release-evals.py +++ b/scripts/evals/run-release-evals.py @@ -147,6 +147,7 @@ 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), ] @@ -184,6 +185,8 @@ 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), diff --git a/src/pipecat/adapters/services/open_ai_responses_adapter.py b/src/pipecat/adapters/services/open_ai_responses_adapter.py new file mode 100644 index 000000000..f55ba6435 --- /dev/null +++ b/src/pipecat/adapters/services/open_ai_responses_adapter.py @@ -0,0 +1,240 @@ +# +# 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. + 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 unknown types as-is + result.append(part) + return result diff --git a/src/pipecat/services/openai/__init__.py b/src/pipecat/services/openai/__init__.py index e182264b1..3caa3c3cb 100644 --- a/src/pipecat/services/openai/__init__.py +++ b/src/pipecat/services/openai/__init__.py @@ -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 * diff --git a/src/pipecat/services/openai/responses/__init__.py b/src/pipecat/services/openai/responses/__init__.py new file mode 100644 index 000000000..c4d243b97 --- /dev/null +++ b/src/pipecat/services/openai/responses/__init__.py @@ -0,0 +1,5 @@ +# +# Copyright (c) 2024-2026, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# diff --git a/src/pipecat/services/openai/responses/llm.py b/src/pipecat/services/openai/responses/llm.py new file mode 100644 index 000000000..e72b8acdb --- /dev/null +++ b/src/pipecat/services/openai/responses/llm.py @@ -0,0 +1,393 @@ +# +# 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, + 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, + *, + model: Optional[str] = None, + api_key=None, + base_url=None, + organization=None, + project=None, + default_headers: Optional[Mapping[str, str]] = None, + settings: Optional[Settings] = None, + **kwargs, + ): + """Initialize the OpenAI Responses API LLM service. + + Args: + model: The OpenAI model name to use. Defaults to "gpt-4.1". + 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. + settings: Runtime-updatable settings. When provided alongside + other parameters, ``settings`` values take precedence. + **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 model is not None: + default_settings.model = model + + if settings is not None: + default_settings.apply_update(settings) + + super().__init__( + settings=default_settings, + **kwargs, + ) + + 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 = self.get_llm_adapter() + logger.debug( + f"{self}: Generating response from universal context " + f"{adapter.get_messages_for_logging(context)}" + ) + + invocation_params: OpenAIResponsesLLMInvocationParams = 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 + + @asynccontextmanager + async def _closing(stream): + chunk_iter = stream.__aiter__() + try: + yield chunk_iter + finally: + if hasattr(chunk_iter, "aclose"): + await chunk_iter.aclose() + 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 getattr(item, "type", None) == "function_call": + item_id = getattr(item, "id", "") or "" + function_calls[item_id] = { + "name": getattr(item, "name", ""), + "call_id": getattr(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 getattr(item, "type", None) == "function_call": + item_id = getattr(item, "id", "") or "" + if item_id in function_calls: + function_calls[item_id]["name"] = getattr(item, "name", "") + function_calls[item_id]["call_id"] = getattr(item, "call_id", "") + function_calls[item_id]["arguments"] = getattr(item, "arguments", "") + + elif isinstance(event, ResponseCompletedEvent): + response = event.response + usage = getattr(response, "usage", None) + if usage: + tokens = LLMTokenUsage( + prompt_tokens=getattr(usage, "input_tokens", 0), + completion_tokens=getattr(usage, "output_tokens", 0), + total_tokens=getattr(usage, "total_tokens", 0), + ) + await self.start_llm_usage_metrics(tokens) + + model = getattr(response, "model", None) + if model: + self._full_model_name = 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, + "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 + + # 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 = 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 + + # Override instructions if caller provided one explicitly + if system_instruction is not None: + params["instructions"] = system_instruction + + 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"] diff --git a/tests/test_openai_responses_adapter.py b/tests/test_openai_responses_adapter.py new file mode 100644 index 000000000..973c05c8c --- /dev/null +++ b/tests/test_openai_responses_adapter.py @@ -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() From eaccb964541d19496970557b732461a51667c706 Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Wed, 18 Mar 2026 11:46:49 -0400 Subject: [PATCH 02/19] docs: add changelog for OpenAI Responses API service --- changelog/4074.added.md | 1 + 1 file changed, 1 insertion(+) create mode 100644 changelog/4074.added.md diff --git a/changelog/4074.added.md b/changelog/4074.added.md new file mode 100644 index 000000000..c27a8e3cf --- /dev/null +++ b/changelog/4074.added.md @@ -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`. From a7167ad121f206ebb97f1ae120015d6254ab85b7 Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Wed, 18 Mar 2026 14:09:17 -0400 Subject: [PATCH 03/19] test: add run_inference tests for OpenAIResponsesLLMService Tests cover basic inference, client exception propagation, system_instruction override, and max_tokens override. --- src/pipecat/services/openai/responses/llm.py | 6 +- tests/test_run_inference.py | 148 +++++++++++++++++++ 2 files changed, 151 insertions(+), 3 deletions(-) diff --git a/src/pipecat/services/openai/responses/llm.py b/src/pipecat/services/openai/responses/llm.py index e72b8acdb..e0a3e4428 100644 --- a/src/pipecat/services/openai/responses/llm.py +++ b/src/pipecat/services/openai/responses/llm.py @@ -206,13 +206,13 @@ class OpenAIResponsesLLMService(LLMService): @traced_llm async def _process_context(self, context: LLMContext): - adapter = self.get_llm_adapter() + adapter: OpenAIResponsesLLMAdapter = self.get_llm_adapter() logger.debug( f"{self}: Generating response from universal context " f"{adapter.get_messages_for_logging(context)}" ) - invocation_params: OpenAIResponsesLLMInvocationParams = adapter.get_llm_invocation_params( + invocation_params = adapter.get_llm_invocation_params( context, system_instruction=self._settings.system_instruction ) @@ -367,7 +367,7 @@ class OpenAIResponsesLLMService(LLMService): Returns: The LLM's response as a string, or None if no response is generated. """ - adapter = self.get_llm_adapter() + 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 diff --git a/tests/test_run_inference.py b/tests/test_run_inference.py index 4f4021b5d..4b35aee2b 100644 --- a/tests/test_run_inference.py +++ b/tests/test_run_inference.py @@ -15,11 +15,13 @@ from pipecat.adapters.services.anthropic_adapter import AnthropicLLMInvocationPa from pipecat.adapters.services.bedrock_adapter import AWSBedrockLLMInvocationParams from pipecat.adapters.services.gemini_adapter import GeminiLLMInvocationParams from pipecat.adapters.services.open_ai_adapter import OpenAILLMInvocationParams +from pipecat.adapters.services.open_ai_responses_adapter import OpenAIResponsesLLMInvocationParams from pipecat.processors.aggregators.llm_context import LLMContext 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 +767,149 @@ 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() + + # Setup mocks + mock_context = MagicMock(spec=LLMContext) + mock_adapter = MagicMock() + test_input = [ + {"role": "developer", "content": "You are a helpful assistant"}, + {"role": "user", "content": "Hello, world!"}, + ] + mock_adapter.get_llm_invocation_params.return_value = OpenAIResponsesLLMInvocationParams( + input=test_input, + tools=OPENAI_NOT_GIVEN, + instructions="You are a helpful assistant", + ) + service.get_llm_adapter = MagicMock(return_value=mock_adapter) + + # Mock response + mock_response = MagicMock() + mock_response.output_text = "Hello! How can I help you today?" + service._client.responses.create = AsyncMock(return_value=mock_response) + + # Execute + result = await service.run_inference(mock_context) + + # Verify + assert result == "Hello! How can I help you today?" + service.get_llm_adapter.assert_called_once() + mock_adapter.get_llm_invocation_params.assert_called_once_with( + mock_context, system_instruction="You are a helpful assistant" + ) + service._client.responses.create.assert_called_once_with( + model="gpt-4.1", + stream=False, + input=test_input, + instructions="You are a helpful assistant", + temperature=0.7, + 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() + + mock_context = MagicMock(spec=LLMContext) + mock_adapter = MagicMock() + mock_adapter.get_llm_invocation_params.return_value = OpenAIResponsesLLMInvocationParams( + input=[], tools=OPENAI_NOT_GIVEN + ) + service.get_llm_adapter = MagicMock(return_value=mock_adapter) + service._client.responses.create = AsyncMock(side_effect=Exception("API Error")) + + with pytest.raises(Exception, match="API Error"): + await service.run_inference(mock_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() + + mock_context = MagicMock(spec=LLMContext) + mock_adapter = MagicMock() + test_input = [{"role": "user", "content": "Hello"}] + mock_adapter.get_llm_invocation_params.return_value = OpenAIResponsesLLMInvocationParams( + input=test_input, + tools=OPENAI_NOT_GIVEN, + instructions="New system instruction", + ) + service.get_llm_adapter = MagicMock(return_value=mock_adapter) + + mock_response = MagicMock() + mock_response.output_text = "Response" + service._client.responses.create = AsyncMock(return_value=mock_response) + + result = await service.run_inference( + mock_context, system_instruction="New system instruction" + ) + + assert result == "Response" + # The adapter should have been called with the override instruction + mock_adapter.get_llm_invocation_params.assert_called_once_with( + mock_context, system_instruction="New system instruction" + ) + # The final API call should have the override instruction + call_kwargs = service._client.responses.create.call_args.kwargs + assert call_kwargs["instructions"] == "New system instruction" + + +@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() + + mock_context = MagicMock(spec=LLMContext) + mock_adapter = MagicMock() + test_input = [{"role": "user", "content": "Summarize this"}] + mock_adapter.get_llm_invocation_params.return_value = OpenAIResponsesLLMInvocationParams( + input=test_input, tools=OPENAI_NOT_GIVEN + ) + service.get_llm_adapter = MagicMock(return_value=mock_adapter) + + mock_response = MagicMock() + mock_response.output_text = "Summary" + service._client.responses.create = AsyncMock(return_value=mock_response) + + result = await service.run_inference(mock_context, max_tokens=200) + + assert result == "Summary" + call_kwargs = service._client.responses.create.call_args.kwargs + # max_tokens override should take precedence + assert call_kwargs["max_output_tokens"] == 200 From c4f21ef76ba9d15b6fae045f6870cf1d789b6b2a Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Wed, 18 Mar 2026 14:17:21 -0400 Subject: [PATCH 04/19] test: add run_inference tests for OpenAIResponsesLLMService Uses real LLMContext and adapter (only HTTP client is mocked) to test basic inference, client exception propagation, system_instruction override, empty context fallback, and max_tokens override. --- tests/test_run_inference.py | 108 ++++++++++++++++-------------------- 1 file changed, 48 insertions(+), 60 deletions(-) diff --git a/tests/test_run_inference.py b/tests/test_run_inference.py index 4b35aee2b..e01d7b36d 100644 --- a/tests/test_run_inference.py +++ b/tests/test_run_inference.py @@ -15,7 +15,6 @@ from pipecat.adapters.services.anthropic_adapter import AnthropicLLMInvocationPa from pipecat.adapters.services.bedrock_adapter import AWSBedrockLLMInvocationParams from pipecat.adapters.services.gemini_adapter import GeminiLLMInvocationParams from pipecat.adapters.services.open_ai_adapter import OpenAILLMInvocationParams -from pipecat.adapters.services.open_ai_responses_adapter import OpenAIResponsesLLMInvocationParams from pipecat.processors.aggregators.llm_context import LLMContext from pipecat.services.anthropic.llm import AnthropicLLMService from pipecat.services.aws.llm import AWSBedrockLLMService @@ -786,42 +785,26 @@ async def test_openai_responses_run_inference_with_llm_context(): ) service._client = AsyncMock() - # Setup mocks - mock_context = MagicMock(spec=LLMContext) - mock_adapter = MagicMock() - test_input = [ - {"role": "developer", "content": "You are a helpful assistant"}, - {"role": "user", "content": "Hello, world!"}, - ] - mock_adapter.get_llm_invocation_params.return_value = OpenAIResponsesLLMInvocationParams( - input=test_input, - tools=OPENAI_NOT_GIVEN, - instructions="You are a helpful assistant", + context = LLMContext( + messages=[ + {"role": "user", "content": "Hello, world!"}, + ] ) - service.get_llm_adapter = MagicMock(return_value=mock_adapter) - # Mock response mock_response = MagicMock() mock_response.output_text = "Hello! How can I help you today?" service._client.responses.create = AsyncMock(return_value=mock_response) - # Execute - result = await service.run_inference(mock_context) + result = await service.run_inference(context) - # Verify assert result == "Hello! How can I help you today?" - service.get_llm_adapter.assert_called_once() - mock_adapter.get_llm_invocation_params.assert_called_once_with( - mock_context, system_instruction="You are a helpful assistant" - ) - service._client.responses.create.assert_called_once_with( - model="gpt-4.1", - stream=False, - input=test_input, - instructions="You are a helpful assistant", - temperature=0.7, - max_output_tokens=100, - ) + 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["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 @@ -831,16 +814,11 @@ async def test_openai_responses_run_inference_client_exception(): service = OpenAIResponsesLLMService() service._client = AsyncMock() - mock_context = MagicMock(spec=LLMContext) - mock_adapter = MagicMock() - mock_adapter.get_llm_invocation_params.return_value = OpenAIResponsesLLMInvocationParams( - input=[], tools=OPENAI_NOT_GIVEN - ) - service.get_llm_adapter = MagicMock(return_value=mock_adapter) + 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(mock_context) + await service.run_inference(context) @pytest.mark.asyncio @@ -855,32 +833,47 @@ async def test_openai_responses_run_inference_system_instruction_overrides(): ) service._client = AsyncMock() - mock_context = MagicMock(spec=LLMContext) - mock_adapter = MagicMock() - test_input = [{"role": "user", "content": "Hello"}] - mock_adapter.get_llm_invocation_params.return_value = OpenAIResponsesLLMInvocationParams( - input=test_input, - tools=OPENAI_NOT_GIVEN, - instructions="New system instruction", + context = LLMContext( + messages=[{"role": "user", "content": "Hello"}], ) - service.get_llm_adapter = MagicMock(return_value=mock_adapter) mock_response = MagicMock() mock_response.output_text = "Response" service._client.responses.create = AsyncMock(return_value=mock_response) - result = await service.run_inference( - mock_context, system_instruction="New system instruction" - ) + result = await service.run_inference(context, system_instruction="New system instruction") assert result == "Response" - # The adapter should have been called with the override instruction - mock_adapter.get_llm_invocation_params.assert_called_once_with( - mock_context, system_instruction="New system instruction" - ) - # The final API call should have the override instruction 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 @@ -895,21 +888,16 @@ async def test_openai_responses_run_inference_max_tokens_override(): ) service._client = AsyncMock() - mock_context = MagicMock(spec=LLMContext) - mock_adapter = MagicMock() - test_input = [{"role": "user", "content": "Summarize this"}] - mock_adapter.get_llm_invocation_params.return_value = OpenAIResponsesLLMInvocationParams( - input=test_input, tools=OPENAI_NOT_GIVEN + context = LLMContext( + messages=[{"role": "user", "content": "Summarize this"}], ) - service.get_llm_adapter = MagicMock(return_value=mock_adapter) mock_response = MagicMock() mock_response.output_text = "Summary" service._client.responses.create = AsyncMock(return_value=mock_response) - result = await service.run_inference(mock_context, max_tokens=200) + result = await service.run_inference(context, max_tokens=200) assert result == "Summary" call_kwargs = service._client.responses.create.call_args.kwargs - # max_tokens override should take precedence assert call_kwargs["max_output_tokens"] == 200 From 21b1812c71413cabdb7f2a836024d16e0e345d24 Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Wed, 18 Mar 2026 14:26:51 -0400 Subject: [PATCH 05/19] chore: add note about previous_response_id and empty input handling --- src/pipecat/adapters/services/open_ai_responses_adapter.py | 3 +++ 1 file changed, 3 insertions(+) diff --git a/src/pipecat/adapters/services/open_ai_responses_adapter.py b/src/pipecat/adapters/services/open_ai_responses_adapter.py index f55ba6435..01e6a7503 100644 --- a/src/pipecat/adapters/services/open_ai_responses_adapter.py +++ b/src/pipecat/adapters/services/open_ai_responses_adapter.py @@ -79,6 +79,9 @@ class OpenAIResponsesLLMAdapter(BaseLLMAdapter[OpenAIResponsesLLMInvocationParam # message when instructions are provided. Contexts that worked with # OpenAILLMService (system_instruction + empty messages) need the # instructions converted to an initial developer message. + # NOTE: once we support `previous_response_id`, we need to revisit + # this logic, as it'll be legit to provide instructions without input + # items if `previous_response_id` is provided. if not input_items: params["input"] = [{"role": "developer", "content": system_instruction}] else: From 951bb0c1a724712e03757f2315f76571bb8d378a Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Wed, 18 Mar 2026 14:47:12 -0400 Subject: [PATCH 06/19] feat: set store=False and add run_inference tests Set store=False in Responses API calls since we send full conversation history as input items and don't use previous_response_id. Add 5 run_inference tests for OpenAIResponsesLLMService using real LLMContext and adapter (only HTTP client mocked). --- .../adapters/services/open_ai_responses_adapter.py | 14 ++++++++++++-- src/pipecat/services/openai/responses/llm.py | 1 + tests/test_run_inference.py | 1 + 3 files changed, 14 insertions(+), 2 deletions(-) diff --git a/src/pipecat/adapters/services/open_ai_responses_adapter.py b/src/pipecat/adapters/services/open_ai_responses_adapter.py index 01e6a7503..4f88e028d 100644 --- a/src/pipecat/adapters/services/open_ai_responses_adapter.py +++ b/src/pipecat/adapters/services/open_ai_responses_adapter.py @@ -79,9 +79,19 @@ class OpenAIResponsesLLMAdapter(BaseLLMAdapter[OpenAIResponsesLLMInvocationParam # message when instructions are provided. Contexts that worked with # OpenAILLMService (system_instruction + empty messages) need the # instructions converted to an initial developer message. - # NOTE: once we support `previous_response_id`, we need to revisit + # + # NOTE: if/when we support `previous_response_id`, we'll need to revisit # this logic, as it'll be legit to provide instructions without input - # items if `previous_response_id` is provided. + # items if `previous_response_id` is provided. Though...OpenAI's docs + + # ChatGPT suggests that `previous_response_id` is primarily for + # development convenience, not performance (other than minor bandwidth + # savings from not transferring the full context), as the model still + # processes the full context from the previous response. The tradeoff + # of using `previous_response_id` is that it requires enabling OpenAI-side + # 30-day conversation storage (meaning we couldn't do `store=False` + # in the API call), which may not be desirable for all users. So, + # my guess is we won't need to support `previous_response_id` in the + # immediate future. if not input_items: params["input"] = [{"role": "developer", "content": system_instruction}] else: diff --git a/src/pipecat/services/openai/responses/llm.py b/src/pipecat/services/openai/responses/llm.py index e0a3e4428..fd1d711cc 100644 --- a/src/pipecat/services/openai/responses/llm.py +++ b/src/pipecat/services/openai/responses/llm.py @@ -324,6 +324,7 @@ class OpenAIResponsesLLMService(LLMService): params: Dict[str, Any] = { "model": self._settings.model, "stream": True, + "store": False, "input": invocation_params["input"], } diff --git a/tests/test_run_inference.py b/tests/test_run_inference.py index e01d7b36d..f67c725ae 100644 --- a/tests/test_run_inference.py +++ b/tests/test_run_inference.py @@ -801,6 +801,7 @@ async def test_openai_responses_run_inference_with_llm_context(): 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 From 0449df828cf6ae0959e36cf2e3374fe527fa81d0 Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Wed, 18 Mar 2026 15:06:54 -0400 Subject: [PATCH 07/19] chore: update previous_response_id comment --- .../services/open_ai_responses_adapter.py | 22 +++++++++---------- 1 file changed, 10 insertions(+), 12 deletions(-) diff --git a/src/pipecat/adapters/services/open_ai_responses_adapter.py b/src/pipecat/adapters/services/open_ai_responses_adapter.py index 4f88e028d..0d586fb12 100644 --- a/src/pipecat/adapters/services/open_ai_responses_adapter.py +++ b/src/pipecat/adapters/services/open_ai_responses_adapter.py @@ -80,18 +80,16 @@ 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`, we'll need to revisit - # this logic, as it'll be legit to provide instructions without input - # items if `previous_response_id` is provided. Though...OpenAI's docs + - # ChatGPT suggests that `previous_response_id` is primarily for - # development convenience, not performance (other than minor bandwidth - # savings from not transferring the full context), as the model still - # processes the full context from the previous response. The tradeoff - # of using `previous_response_id` is that it requires enabling OpenAI-side - # 30-day conversation storage (meaning we couldn't do `store=False` - # in the API call), which may not be desirable for all users. So, - # my guess is we won't need to support `previous_response_id` in the - # immediate future. + # 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: From 2001ab4577df414bf26128678b5980a8b0976e56 Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Wed, 18 Mar 2026 15:14:28 -0400 Subject: [PATCH 08/19] feat: add 20a persistent context example for OpenAI Responses --- ...20a-persistent-context-openai-responses.py | 249 ++++++++++++++++++ 1 file changed, 249 insertions(+) create mode 100644 examples/foundational/20a-persistent-context-openai-responses.py diff --git a/examples/foundational/20a-persistent-context-openai-responses.py b/examples/foundational/20a-persistent-context-openai-responses.py new file mode 100644 index 000000000..730e90197 --- /dev/null +++ b/examples/foundational/20a-persistent-context-openai-responses.py @@ -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 conversatione. 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() From 891966346ca88443f33b61921dd2a466d78f897b Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Wed, 18 Mar 2026 15:17:16 -0400 Subject: [PATCH 09/19] feat: add 55zi update-settings example for OpenAI Responses --- ...zi-update-settings-openai-responses-llm.py | 127 ++++++++++++++++++ 1 file changed, 127 insertions(+) create mode 100644 examples/foundational/55zi-update-settings-openai-responses-llm.py diff --git a/examples/foundational/55zi-update-settings-openai-responses-llm.py b/examples/foundational/55zi-update-settings-openai-responses-llm.py new file mode 100644 index 000000000..61bd8329e --- /dev/null +++ b/examples/foundational/55zi-update-settings-openai-responses-llm.py @@ -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() From 5de794e1da5389005d2d155c8311a1191b93479e Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Wed, 18 Mar 2026 15:29:04 -0400 Subject: [PATCH 10/19] feat: add service_tier support to OpenAIResponsesLLMService --- src/pipecat/services/openai/responses/llm.py | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/src/pipecat/services/openai/responses/llm.py b/src/pipecat/services/openai/responses/llm.py index fd1d711cc..f62c6e758 100644 --- a/src/pipecat/services/openai/responses/llm.py +++ b/src/pipecat/services/openai/responses/llm.py @@ -84,6 +84,7 @@ class OpenAIResponsesLLMService(LLMService): organization=None, project=None, default_headers: Optional[Mapping[str, str]] = None, + service_tier: Optional[str] = None, settings: Optional[Settings] = None, **kwargs, ): @@ -96,6 +97,7 @@ class OpenAIResponsesLLMService(LLMService): 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. When provided alongside other parameters, ``settings`` values take precedence. **kwargs: Additional arguments passed to the parent LLMService. @@ -127,6 +129,7 @@ class OpenAIResponsesLLMService(LLMService): **kwargs, ) + self._service_tier = service_tier self._client = self._create_client( api_key=api_key, base_url=base_url, @@ -342,6 +345,9 @@ class OpenAIResponsesLLMService(LLMService): 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)): From b1a8588209d407feebbffd1b7c40998bcfe8a33b Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Wed, 18 Mar 2026 15:39:06 -0400 Subject: [PATCH 11/19] feat: add 12- and 14d- image/video examples for OpenAI Responses --- .../12-describe-image-openai-responses.py | 139 +++++++++++++ ...function-calling-openai-responses-video.py | 195 ++++++++++++++++++ scripts/evals/run-release-evals.py | 2 + 3 files changed, 336 insertions(+) create mode 100644 examples/foundational/12-describe-image-openai-responses.py create mode 100644 examples/foundational/14d-function-calling-openai-responses-video.py diff --git a/examples/foundational/12-describe-image-openai-responses.py b/examples/foundational/12-describe-image-openai-responses.py new file mode 100644 index 000000000..a3c113cb2 --- /dev/null +++ b/examples/foundational/12-describe-image-openai-responses.py @@ -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() diff --git a/examples/foundational/14d-function-calling-openai-responses-video.py b/examples/foundational/14d-function-calling-openai-responses-video.py new file mode 100644 index 000000000..440c51cc1 --- /dev/null +++ b/examples/foundational/14d-function-calling-openai-responses-video.py @@ -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() diff --git a/scripts/evals/run-release-evals.py b/scripts/evals/run-release-evals.py index 9671703ba..19492ba62 100644 --- a/scripts/evals/run-release-evals.py +++ b/scripts/evals/run-release-evals.py @@ -154,6 +154,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)), ("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)), @@ -193,6 +194,7 @@ TESTS_14 = [ ("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), From 348df9d4ced25de0f6c36fb0fb38e1b31743d28b Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Thu, 19 Mar 2026 13:34:41 -0400 Subject: [PATCH 12/19] fix: remove redundant instructions override in run_inference The override would re-add `instructions` after the adapter had intentionally converted it to a developer message for empty contexts. Added a regression test. --- src/pipecat/services/openai/responses/llm.py | 4 --- tests/test_run_inference.py | 33 ++++++++++++++++++++ 2 files changed, 33 insertions(+), 4 deletions(-) diff --git a/src/pipecat/services/openai/responses/llm.py b/src/pipecat/services/openai/responses/llm.py index f62c6e758..4f0c81dc7 100644 --- a/src/pipecat/services/openai/responses/llm.py +++ b/src/pipecat/services/openai/responses/llm.py @@ -385,10 +385,6 @@ class OpenAIResponsesLLMService(LLMService): # Override for non-streaming params["stream"] = False - # Override instructions if caller provided one explicitly - if system_instruction is not None: - params["instructions"] = system_instruction - if max_tokens is not None: params["max_output_tokens"] = max_tokens diff --git a/tests/test_run_inference.py b/tests/test_run_inference.py index f67c725ae..cef13fb27 100644 --- a/tests/test_run_inference.py +++ b/tests/test_run_inference.py @@ -902,3 +902,36 @@ async def test_openai_responses_run_inference_max_tokens_override(): 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 From a3431d3b0176c4740dc5f27fe5f59977ae1d54bc Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Thu, 19 Mar 2026 13:55:26 -0400 Subject: [PATCH 13/19] fix: prefer _full_model_name over _settings.model in tracing The API-provided full model name is more specific than the user-provided model name (e.g. includes version/snapshot details). Reorder the lookup in _get_model_name and add a comment where the Responses service sets the field. --- src/pipecat/services/openai/responses/llm.py | 2 ++ src/pipecat/utils/tracing/service_decorators.py | 6 ++++-- 2 files changed, 6 insertions(+), 2 deletions(-) diff --git a/src/pipecat/services/openai/responses/llm.py b/src/pipecat/services/openai/responses/llm.py index 4f0c81dc7..8c7269323 100644 --- a/src/pipecat/services/openai/responses/llm.py +++ b/src/pipecat/services/openai/responses/llm.py @@ -292,6 +292,8 @@ class OpenAIResponsesLLMService(LLMService): model = getattr(response, "model", None) if model: + # This field is used by @traced_llm for more detailed + # model name in tracing spans self._full_model_name = model # Process any function calls diff --git a/src/pipecat/utils/tracing/service_decorators.py b/src/pipecat/utils/tracing/service_decorators.py index f4764d248..5292a5d3f 100644 --- a/src/pipecat/utils/tracing/service_decorators.py +++ b/src/pipecat/utils/tracing/service_decorators.py @@ -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" From 0533ea7b7fdd45a906cc05349109606b3314f236 Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Thu, 19 Mar 2026 14:19:10 -0400 Subject: [PATCH 14/19] refactor: use direct attribute access for typed stream events Replace getattr() calls with direct attribute access and isinstance() checks on the strongly-typed OpenAI SDK event models. --- src/pipecat/services/openai/responses/llm.py | 36 +++++++++----------- 1 file changed, 17 insertions(+), 19 deletions(-) diff --git a/src/pipecat/services/openai/responses/llm.py b/src/pipecat/services/openai/responses/llm.py index 8c7269323..a412d9627 100644 --- a/src/pipecat/services/openai/responses/llm.py +++ b/src/pipecat/services/openai/responses/llm.py @@ -18,6 +18,7 @@ from openai.types.responses import ( ResponseCompletedEvent, ResponseFunctionCallArgumentsDeltaEvent, ResponseFunctionCallArgumentsDoneEvent, + ResponseFunctionToolCall, ResponseOutputItemAddedEvent, ResponseOutputItemDoneEvent, ResponseStreamEvent, @@ -251,11 +252,11 @@ class OpenAIResponsesLLMService(LLMService): elif isinstance(event, ResponseOutputItemAddedEvent): await self.stop_ttfb_metrics() item = event.item - if getattr(item, "type", None) == "function_call": - item_id = getattr(item, "id", "") or "" + if isinstance(item, ResponseFunctionToolCall): + item_id = item.id or "" function_calls[item_id] = { - "name": getattr(item, "name", ""), - "call_id": getattr(item, "call_id", ""), + "name": item.name, + "call_id": item.call_id, "arguments": "", } current_arguments[item_id] = "" @@ -272,29 +273,26 @@ class OpenAIResponsesLLMService(LLMService): elif isinstance(event, ResponseOutputItemDoneEvent): item = event.item - if getattr(item, "type", None) == "function_call": - item_id = getattr(item, "id", "") or "" + if isinstance(item, ResponseFunctionToolCall): + item_id = item.id or "" if item_id in function_calls: - function_calls[item_id]["name"] = getattr(item, "name", "") - function_calls[item_id]["call_id"] = getattr(item, "call_id", "") - function_calls[item_id]["arguments"] = getattr(item, "arguments", "") + 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 - usage = getattr(response, "usage", None) - if usage: + if response.usage: tokens = LLMTokenUsage( - prompt_tokens=getattr(usage, "input_tokens", 0), - completion_tokens=getattr(usage, "output_tokens", 0), - total_tokens=getattr(usage, "total_tokens", 0), + prompt_tokens=response.usage.input_tokens, + completion_tokens=response.usage.output_tokens, + total_tokens=response.usage.total_tokens, ) await self.start_llm_usage_metrics(tokens) - model = getattr(response, "model", None) - if model: - # This field is used by @traced_llm for more detailed - # model name in tracing spans - self._full_model_name = model + # 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: From 4ec7be88507eaae006d41c283d80ce1f805de920 Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Thu, 19 Mar 2026 14:23:39 -0400 Subject: [PATCH 15/19] feat: include cached_tokens and reasoning_tokens in usage metrics --- src/pipecat/services/openai/responses/llm.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/src/pipecat/services/openai/responses/llm.py b/src/pipecat/services/openai/responses/llm.py index a412d9627..bcc6274b4 100644 --- a/src/pipecat/services/openai/responses/llm.py +++ b/src/pipecat/services/openai/responses/llm.py @@ -287,6 +287,8 @@ class OpenAIResponsesLLMService(LLMService): 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) From 05e344b9ec67fe94ab22f5fe8047bd9a41b40ed6 Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Thu, 19 Mar 2026 14:30:34 -0400 Subject: [PATCH 16/19] docs: port _closing comments from BaseOpenAILLMService --- src/pipecat/services/openai/responses/llm.py | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/src/pipecat/services/openai/responses/llm.py b/src/pipecat/services/openai/responses/llm.py index bcc6274b4..b12a33510 100644 --- a/src/pipecat/services/openai/responses/llm.py +++ b/src/pipecat/services/openai/responses/llm.py @@ -230,14 +230,22 @@ class OpenAIResponsesLLMService(LLMService): 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"): From 6424c36666bd7755c7fe5c26d342e5431bb5d3c4 Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Thu, 19 Mar 2026 14:37:46 -0400 Subject: [PATCH 17/19] refactor: remove model init param from OpenAIResponsesLLMService Model is only configurable via settings, matching the canonical API. --- src/pipecat/services/openai/responses/llm.py | 8 +------- 1 file changed, 1 insertion(+), 7 deletions(-) diff --git a/src/pipecat/services/openai/responses/llm.py b/src/pipecat/services/openai/responses/llm.py index b12a33510..e9e5d3a1f 100644 --- a/src/pipecat/services/openai/responses/llm.py +++ b/src/pipecat/services/openai/responses/llm.py @@ -79,7 +79,6 @@ class OpenAIResponsesLLMService(LLMService): def __init__( self, *, - model: Optional[str] = None, api_key=None, base_url=None, organization=None, @@ -92,15 +91,13 @@ class OpenAIResponsesLLMService(LLMService): """Initialize the OpenAI Responses API LLM service. Args: - model: The OpenAI model name to use. Defaults to "gpt-4.1". 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. When provided alongside - other parameters, ``settings`` values take precedence. + settings: Runtime-updatable settings. **kwargs: Additional arguments passed to the parent LLMService. """ default_settings = self.Settings( @@ -119,9 +116,6 @@ class OpenAIResponsesLLMService(LLMService): extra={}, ) - if model is not None: - default_settings.model = model - if settings is not None: default_settings.apply_update(settings) From ea1534f9f8e757ee0fa86c46aea79eafdcb86a95 Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Thu, 19 Mar 2026 15:36:23 -0400 Subject: [PATCH 18/19] docs: note input_audio coming soon, no conversion needed The LLMContext format already matches the expected Responses API shape for input_audio, so no adapter conversion will be needed once OpenAI enables support. --- src/pipecat/adapters/services/open_ai_responses_adapter.py | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/src/pipecat/adapters/services/open_ai_responses_adapter.py b/src/pipecat/adapters/services/open_ai_responses_adapter.py index 0d586fb12..70627fe5d 100644 --- a/src/pipecat/adapters/services/open_ai_responses_adapter.py +++ b/src/pipecat/adapters/services/open_ai_responses_adapter.py @@ -246,6 +246,9 @@ class OpenAIResponsesLLMAdapter(BaseLLMAdapter[OpenAIResponsesLLMInvocationParam } ) else: - # Pass through unknown types as-is + # 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 From dafbb2eb6628692b3cba8e7659c99abdf124e97d Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Thu, 19 Mar 2026 15:38:38 -0400 Subject: [PATCH 19/19] =?UTF-8?q?fix:=20typo=20"conversatione"=20=E2=86=92?= =?UTF-8?q?=20"conversation"=20in=2020-=20examples?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../foundational/20a-persistent-context-openai-responses.py | 2 +- examples/foundational/20a-persistent-context-openai.py | 2 +- .../foundational/20b-persistent-context-openai-realtime-beta.py | 2 +- examples/foundational/20b-persistent-context-openai-realtime.py | 2 +- 4 files changed, 4 insertions(+), 4 deletions(-) diff --git a/examples/foundational/20a-persistent-context-openai-responses.py b/examples/foundational/20a-persistent-context-openai-responses.py index 730e90197..5fd9c7657 100644 --- a/examples/foundational/20a-persistent-context-openai-responses.py +++ b/examples/foundational/20a-persistent-context-openai-responses.py @@ -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=[], ) diff --git a/examples/foundational/20a-persistent-context-openai.py b/examples/foundational/20a-persistent-context-openai.py index f6a4dc937..7f744fd46 100644 --- a/examples/foundational/20a-persistent-context-openai.py +++ b/examples/foundational/20a-persistent-context-openai.py @@ -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=[], ) diff --git a/examples/foundational/20b-persistent-context-openai-realtime-beta.py b/examples/foundational/20b-persistent-context-openai-realtime-beta.py index fa59e1674..4b05db618 100644 --- a/examples/foundational/20b-persistent-context-openai-realtime-beta.py +++ b/examples/foundational/20b-persistent-context-openai-realtime-beta.py @@ -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": {}, diff --git a/examples/foundational/20b-persistent-context-openai-realtime.py b/examples/foundational/20b-persistent-context-openai-realtime.py index de4215ab8..bceca410d 100644 --- a/examples/foundational/20b-persistent-context-openai-realtime.py +++ b/examples/foundational/20b-persistent-context-openai-realtime.py @@ -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=[], ),