diff --git a/src/pipecat/services/openai/base_llm.py b/src/pipecat/services/openai/base_llm.py index da69f9a7c..8f040f8d7 100644 --- a/src/pipecat/services/openai/base_llm.py +++ b/src/pipecat/services/openai/base_llm.py @@ -41,7 +41,7 @@ from pipecat.processors.aggregators.openai_llm_context import ( ) from pipecat.processors.frame_processor import FrameDirection from pipecat.services.llm_service import FunctionCallFromLLM, LLMService -from pipecat.utils.cancellable_stream import CancellableStream +from pipecat.utils.asyncio.cancellable_stream import CancellableStream from pipecat.utils.tracing.service_decorators import traced_llm @@ -139,7 +139,7 @@ class BaseOpenAILLMService(LLMService): default_headers=default_headers, **kwargs, ) - self.chunk_stream: Optional[CancellableStream[AsyncStream[AsyncStream]]] = None + self._stream: Optional[CancellableStream[AsyncStream[AsyncStream]]] = None def create_client( self, @@ -331,7 +331,7 @@ class BaseOpenAILLMService(LLMService): await self.start_ttfb_metrics() # Generate chat completions using either OpenAILLMContext or universal LLMContext - self.chunk_stream = CancellableStream( + self._stream = CancellableStream( await ( self._stream_chat_completions_specific_context(context) if isinstance(context, OpenAILLMContext) @@ -339,7 +339,7 @@ class BaseOpenAILLMService(LLMService): ) ) - async for chunk in self.chunk_stream: + async for chunk in self._stream: if chunk.usage: tokens = LLMTokenUsage( prompt_tokens=chunk.usage.prompt_tokens, @@ -393,7 +393,7 @@ class BaseOpenAILLMService(LLMService): ): await self.push_frame(LLMTextFrame(chunk.choices[0].delta.audio["transcript"])) - self.chunk_stream = None + self._stream = None # if we got a function name and arguments, check to see if it's a function with # a registered handler. If so, run the registered callback, save the result to @@ -431,10 +431,10 @@ class BaseOpenAILLMService(LLMService): # 3. This allows the task to be cancelled cleanly afterwards # This approach ensures we don't get stuck in case there was a chunk processing loop # when cancellation is requested. - if self.chunk_stream: + if self._stream: logger.debug(f"{self}: Cancelling chunk stream due to interruption") - await self.chunk_stream.cancel() - self.chunk_stream = None + await self._stream.cancel() + self._stream = None await super()._start_interruption() diff --git a/src/pipecat/utils/cancellable_stream.py b/src/pipecat/utils/asyncio/cancellable_stream.py similarity index 100% rename from src/pipecat/utils/cancellable_stream.py rename to src/pipecat/utils/asyncio/cancellable_stream.py