Addressing the comments left in the PR review.

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()
+            await self._stream.cancel()
-            self.chunk_stream = None
+            self._stream = None
 
         await super()._start_interruption()