Merge pull request #1280 from pipecat-ai/aleix/add-completion-timeout

services(llm): add on_completion_timeout event

CHANGELOG.md

@@ -9,6 +9,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- Added new `on_completion_timeout` event for LLM services (all OpenAI-based
+  services, Anthropic and Google). Note that this event will only get triggered
+  if LLM timeouts are setup and if the timeout was reached. It can be useful to
+  retrigger another completion and see if the timeout was just a blip.
+
 - Added new log observers `LLMLogObserver` and `TranscriptionLogObserver` that
   can be useful for debugging your pipelines.
 

src/pipecat/services/ai_services.py

@@ -142,6 +142,8 @@ class LLMService(AIService):
         self._callbacks = {}
         self._start_callbacks = {}
 
+        self._register_event_handler("on_completion_timeout")
+
     # TODO-CB: callback function type
     def register_function(self, function_name: Optional[str], callback, start_callback=None):
         # Registering a function with the function_name set to None will run that callback

src/pipecat/services/anthropic.py

@@ -4,15 +4,16 @@
 # SPDX-License-Identifier: BSD 2-Clause License
 #
 
+import asyncio
 import base64
 import copy
 import io
 import json
 import re
-from asyncio import CancelledError
 from dataclasses import dataclass
 from typing import Any, Dict, List, Optional, Union
 
+import httpx
 from loguru import logger
 from PIL import Image
 from pydantic import BaseModel, Field
@@ -251,12 +252,14 @@ class AnthropicLLMService(LLMService):
                     if total_input_tokens >= 1024:
                         context.turns_above_cache_threshold += 1
 
-        except CancelledError:
+        except asyncio.CancelledError:
             # If we're interrupted, we won't get a complete usage report. So set our flag to use the
             # token estimate. The reraise the exception so all the processors running in this task
             # also get cancelled.
             use_completion_tokens_estimate = True
             raise
+        except httpx.TimeoutException:
+            await self._call_event_handler("on_completion_timeout")
         except Exception as e:
             logger.exception(f"{self} exception: {e}")
         finally:

src/pipecat/services/google/google.py

@@ -11,6 +11,8 @@ import json
 import os
 import time
 
+from google.api_core.exceptions import DeadlineExceeded
+
 # Suppress gRPC fork warnings
 os.environ["GRPC_ENABLE_FORK_SUPPORT"] = "false"
 
@@ -1126,6 +1128,8 @@ class GoogleLLMService(LLMService):
                     else:
                         logger.exception(f"{self} error: {e}")
 
+        except DeadlineExceeded:
+            await self._call_event_handler("on_completion_timeout")
         except Exception as e:
             logger.exception(f"{self} exception: {e}")
         finally:

src/pipecat/services/openai.py

@@ -310,11 +310,15 @@ class BaseOpenAILLMService(LLMService):
             await self.push_frame(frame, direction)
 
         if context:
-            await self.push_frame(LLMFullResponseStartFrame())
+            try:
-            await self.start_processing_metrics()
+                await self.push_frame(LLMFullResponseStartFrame())
-            await self._process_context(context)
+                await self.start_processing_metrics()
-            await self.stop_processing_metrics()
+                await self._process_context(context)
-            await self.push_frame(LLMFullResponseEndFrame())
+            except httpx.TimeoutException:
+                await self._call_event_handler("on_completion_timeout")
+            finally:
+                await self.stop_processing_metrics()
+                await self.push_frame(LLMFullResponseEndFrame())
 
 
 @dataclass