From ff0eb6d28612c665888565bd153bc957ce292e8a Mon Sep 17 00:00:00 2001 From: Luke Payyapilli Date: Wed, 28 Jan 2026 09:42:22 -0500 Subject: [PATCH] fix: emit ErrorFrame on LLM completion timeout --- changelog/3529.fixed.md | 1 + src/pipecat/services/openai/base_llm.py | 3 +- tests/test_openai_llm_timeout.py | 127 ++++++++++++++++++++++++ 3 files changed, 130 insertions(+), 1 deletion(-) create mode 100644 changelog/3529.fixed.md create mode 100644 tests/test_openai_llm_timeout.py diff --git a/changelog/3529.fixed.md b/changelog/3529.fixed.md new file mode 100644 index 000000000..91c9eeda0 --- /dev/null +++ b/changelog/3529.fixed.md @@ -0,0 +1 @@ +- Fixed OpenAI LLM services to emit `ErrorFrame` on completion timeout, enabling proper error handling and LLMSwitcher failover. diff --git a/src/pipecat/services/openai/base_llm.py b/src/pipecat/services/openai/base_llm.py index 419d5db1e..8e0a45f0d 100644 --- a/src/pipecat/services/openai/base_llm.py +++ b/src/pipecat/services/openai/base_llm.py @@ -492,8 +492,9 @@ class BaseOpenAILLMService(LLMService): await self.push_frame(LLMFullResponseStartFrame()) await self.start_processing_metrics() await self._process_context(context) - except httpx.TimeoutException: + except httpx.TimeoutException as e: await self._call_event_handler("on_completion_timeout") + await self.push_error(error_msg="LLM completion timeout", exception=e) finally: await self.stop_processing_metrics() await self.push_frame(LLMFullResponseEndFrame()) diff --git a/tests/test_openai_llm_timeout.py b/tests/test_openai_llm_timeout.py new file mode 100644 index 000000000..df9483289 --- /dev/null +++ b/tests/test_openai_llm_timeout.py @@ -0,0 +1,127 @@ +# +# Copyright (c) 2024-2026, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +"""Unit tests for OpenAI LLM timeout handling.""" + +from unittest.mock import AsyncMock, patch + +import httpx +import pytest + +from pipecat.frames.frames import ( + LLMContextFrame, + LLMFullResponseEndFrame, + LLMFullResponseStartFrame, +) +from pipecat.processors.aggregators.llm_context import LLMContext +from pipecat.processors.frame_processor import FrameDirection +from pipecat.services.openai.llm import OpenAILLMService + + +@pytest.mark.asyncio +async def test_openai_llm_emits_error_frame_on_timeout(): + """Test that OpenAI LLM service emits ErrorFrame when a timeout occurs. + + This enables LLMSwitcher to trigger failover to backup LLMs when the + primary LLM times out. + """ + with patch.object(OpenAILLMService, "create_client"): + service = OpenAILLMService(model="gpt-4") + service._client = AsyncMock() + + # Track pushed frames and errors + pushed_frames = [] + pushed_errors = [] + timeout_handler_called = False + + original_push_frame = service.push_frame + + async def mock_push_frame(frame, direction=FrameDirection.DOWNSTREAM): + pushed_frames.append(frame) + await original_push_frame(frame, direction) + + async def mock_push_error(error_msg, exception=None): + pushed_errors.append({"error_msg": error_msg, "exception": exception}) + + async def mock_timeout_handler(event_name): + nonlocal timeout_handler_called + if event_name == "on_completion_timeout": + timeout_handler_called = True + + service.push_frame = mock_push_frame + service.push_error = mock_push_error + service._call_event_handler = AsyncMock(side_effect=mock_timeout_handler) + + # Mock _process_context to raise TimeoutException + service._process_context = AsyncMock( + side_effect=httpx.TimeoutException("Connection timed out") + ) + + # Mock metrics methods + service.start_processing_metrics = AsyncMock() + service.stop_processing_metrics = AsyncMock() + service.start_ttfb_metrics = AsyncMock() + + # Create a context frame to process + context = LLMContext( + messages=[{"role": "user", "content": "Hello"}], + ) + frame = LLMContextFrame(context=context) + + # Process the frame + await service.process_frame(frame, FrameDirection.DOWNSTREAM) + + # Verify timeout handler was called + service._call_event_handler.assert_called_once_with("on_completion_timeout") + assert timeout_handler_called + + # Verify push_error was called with correct message + assert len(pushed_errors) == 1 + assert pushed_errors[0]["error_msg"] == "LLM completion timeout" + assert isinstance(pushed_errors[0]["exception"], httpx.TimeoutException) + + # Verify LLMFullResponseStartFrame and LLMFullResponseEndFrame were pushed + frame_types = [type(f) for f in pushed_frames] + assert LLMFullResponseStartFrame in frame_types + assert LLMFullResponseEndFrame in frame_types + + +@pytest.mark.asyncio +async def test_openai_llm_timeout_still_pushes_end_frame(): + """Test that LLMFullResponseEndFrame is pushed even when timeout occurs. + + The finally block should ensure proper cleanup regardless of timeout. + """ + with patch.object(OpenAILLMService, "create_client"): + service = OpenAILLMService(model="gpt-4") + service._client = AsyncMock() + + pushed_frames = [] + + async def mock_push_frame(frame, direction=FrameDirection.DOWNSTREAM): + pushed_frames.append(frame) + + service.push_frame = mock_push_frame + service.push_error = AsyncMock() + service._call_event_handler = AsyncMock() + service._process_context = AsyncMock(side_effect=httpx.TimeoutException("Timeout")) + service.start_processing_metrics = AsyncMock() + service.stop_processing_metrics = AsyncMock() + + context = LLMContext( + messages=[{"role": "user", "content": "Hello"}], + ) + frame = LLMContextFrame(context=context) + + await service.process_frame(frame, FrameDirection.DOWNSTREAM) + + # Verify both start and end frames are pushed + frame_types = [type(f) for f in pushed_frames] + assert LLMFullResponseStartFrame in frame_types + assert LLMFullResponseEndFrame in frame_types + + # Verify metrics were stopped + service.stop_processing_metrics.assert_called_once()