Merge pull request #2115 from pipecat-ai/mb/docstring-cleanup

Docstring cleanup, fix missing examples imports

examples/daily-multi-translation/requirements.txt

@@ -2,4 +2,4 @@ aiofiles
 python-dotenv
 fastapi[all]
 uvicorn
-pipecat-ai[daily,deepgram,openai,silero,cartesia]
+pipecat-ai[daily,deepgram,openai,silero,cartesia,soundfile]

examples/open-telemetry/jaeger/requirements.txt

@@ -1,6 +1,6 @@
 fastapi
 uvicorn
 python-dotenv
-pipecat-ai[webrtc,silero,cartesia,deepgram,openai,tracing]
+pipecat-ai[daily,webrtc,silero,cartesia,deepgram,openai,tracing]
 pipecat-ai-small-webrtc-prebuilt
 opentelemetry-exporter-otlp-proto-grpc

examples/open-telemetry/langfuse/requirements.txt

@@ -1,6 +1,6 @@
 fastapi
 uvicorn
 python-dotenv
-pipecat-ai[webrtc,silero,cartesia,deepgram,openai,tracing]
+pipecat-ai[daily,webrtc,silero,cartesia,deepgram,openai,tracing]
 pipecat-ai-small-webrtc-prebuilt
 opentelemetry-exporter-otlp-proto-http

examples/phone-chatbot/daily-twilio-sip-dial-in/requirements.txt

@@ -1,4 +1,4 @@
-pipecat-ai[daily,elevenlabs,openai,silero]
+pipecat-ai[daily,cartesia,openai,silero]
 fastapi==0.115.6
 uvicorn
 python-dotenv

src/pipecat/observers/loggers/debug_log_observer.py

@@ -47,7 +47,7 @@ class DebugLogObserver(BaseObserver):
 
         Log specific frame types from any source/destination::
 
-            from pipecat.frames.frames import TranscriptionFrame, InterimTranscriptionFrame
+            from pipecat.frames.frames import LLMTextFrame, TranscriptionFrame
             observers=[
                 DebugLogObserver(frame_types=(LLMTextFrame,TranscriptionFrame,)),
             ]
@@ -55,7 +55,8 @@ class DebugLogObserver(BaseObserver):
         Log frames with specific source/destination filters::
 
             from pipecat.frames.frames import StartInterruptionFrame, UserStartedSpeakingFrame, LLMTextFrame
-            from pipecat.transports.base_output_transport import BaseOutputTransport
+            from pipecat.observers.loggers.debug_log_observer import DebugLogObserver, FrameEndpoint
+            from pipecat.transports.base_output import BaseOutputTransport
             from pipecat.services.stt_service import STTService
 
             observers=[

src/pipecat/processors/aggregators/gated.py

@@ -26,29 +26,6 @@ class GatedAggregator(FrameProcessor):
     until and not including the gate-closed frame. The aggregator maintains an
     internal gate state that controls whether frames are passed through immediately
     or accumulated for later release.
-
-    Doctest: FIXME to work with asyncio
-    >>> from pipecat.frames.frames import ImageRawFrame
-
-    >>> async def print_frames(aggregator, frame):
-    ...     async for frame in aggregator.process_frame(frame):
-    ...         if isinstance(frame, TextFrame):
-    ...             print(frame.text)
-    ...         else:
-    ...             print(frame.__class__.__name__)
-
-    >>> aggregator = GatedAggregator(
-    ...     gate_close_fn=lambda x: isinstance(x, LLMResponseStartFrame),
-    ...     gate_open_fn=lambda x: isinstance(x, ImageRawFrame),
-    ...     start_open=False)
-    >>> asyncio.run(print_frames(aggregator, TextFrame("Hello")))
-    >>> asyncio.run(print_frames(aggregator, TextFrame("Hello again.")))
-    >>> asyncio.run(print_frames(aggregator, ImageRawFrame(image=bytes([]), size=(0, 0))))
-    ImageRawFrame
-    Hello
-    Hello again.
-    >>> asyncio.run(print_frames(aggregator, TextFrame("Goodbye.")))
-    Goodbye.
     """
 
     def __init__(

src/pipecat/processors/aggregators/sentence.py

@@ -23,20 +23,10 @@ class SentenceAggregator(FrameProcessor):
     Useful for ensuring downstream processors receive coherent, complete sentences
     rather than fragmented text.
 
-    Frame input/output:
+    Frame input/output::
+
         TextFrame("Hello,") -> None
         TextFrame(" world.") -> TextFrame("Hello, world.")
-
-    Doctest: FIXME to work with asyncio
-    >>> import asyncio
-    >>> async def print_frames(aggregator, frame):
-    ...     async for frame in aggregator.process_frame(frame):
-    ...         print(frame.text)
-
-    >>> aggregator = SentenceAggregator()
-    >>> asyncio.run(print_frames(aggregator, TextFrame("Hello,")))
-    >>> asyncio.run(print_frames(aggregator, TextFrame(" world.")))
-    Hello, world.
     """
 
     def __init__(self):

src/pipecat/processors/aggregators/vision_image_frame.py

@@ -20,17 +20,6 @@ class VisionImageFrameAggregator(FrameProcessor):
     This aggregator waits for a consecutive TextFrame and an InputImageRawFrame.
     After the InputImageRawFrame arrives it will output a VisionImageRawFrame
     combining both the text and image data for multimodal processing.
-
-    >>> from pipecat.frames.frames import ImageFrame
-
-    >>> async def print_frames(aggregator, frame):
-    ...     async for frame in aggregator.process_frame(frame):
-    ...         print(frame)
-
-    >>> aggregator = VisionImageFrameAggregator()
-    >>> asyncio.run(print_frames(aggregator, TextFrame("What do you see?")))
-    >>> asyncio.run(print_frames(aggregator, ImageFrame(image=bytes([]), size=(0, 0))))
-    VisionImageFrame, text: What do you see?, image size: 0x0, buffer size: 0 B
     """
 
     def __init__(self):

src/pipecat/processors/text_transformer.py

@@ -18,14 +18,6 @@ class StatelessTextTransformer(FrameProcessor):
     This processor intercepts TextFrame objects and applies a user-provided
     transformation function to the text content. The function can be either
     synchronous or asynchronous (coroutine).
-
-    >>> async def print_frames(aggregator, frame):
-    ...     async for frame in aggregator.process_frame(frame):
-    ...         print(frame.text)
-
-    >>> aggregator = StatelessTextTransformer(lambda x: x.upper())
-    >>> asyncio.run(print_frames(aggregator, TextFrame("Hello")))
-    HELLO
     """
 
     def __init__(

src/pipecat/services/gemini_multimodal_live/file_api.py

@@ -4,6 +4,13 @@
 # SPDX-License-Identifier: BSD 2-Clause License
 #
 
+"""Gemini File API client for uploading and managing files.
+
+This module provides a client for Google's Gemini File API, enabling file
+uploads, metadata retrieval, listing, and deletion. Files uploaded through
+this API can be referenced in Gemini generative model calls.
+"""
+
 import mimetypes
 from typing import Any, Dict, Optional
 

src/pipecat/services/gemini_multimodal_live/gemini.py

@@ -72,7 +72,6 @@ from pipecat.utils.time import time_now_iso8601
 from pipecat.utils.tracing.service_decorators import traced_gemini_live, traced_stt
 
 from . import events
-
 from .file_api import GeminiFileAPI
 
 try:
@@ -237,7 +236,9 @@ class GeminiMultimodalLiveContext(OpenAILLMContext):
             parts.append({"type": "text", "text": text})
 
         # Add file reference part
-        parts.append({"type": "file_data", "file_data": {"mime_type": mime_type, "file_uri": file_uri}})
+        parts.append(
+            {"type": "file_data", "file_data": {"mime_type": mime_type, "file_uri": file_uri}}
+        )
 
         # Add to messages
         message = {"role": "user", "content": parts}
@@ -270,12 +271,14 @@ class GeminiMultimodalLiveContext(OpenAILLMContext):
                         parts.append({"text": part.get("text")})
                     elif part.get("type") == "file_data":
                         file_data = part.get("file_data", {})
-                        parts.append({
+                        parts.append(
-                            "fileData": {
+                            {
-                                "mimeType": file_data.get("mime_type"),
+                                "fileData": {
-                                "fileUri": file_data.get("file_uri")
+                                    "mimeType": file_data.get("mime_type"),
+                                    "fileUri": file_data.get("file_uri"),
+                                }
                             }
-                        })
+                        )
                     else:
                         logger.warning(f"Unsupported content type: {str(part)[:80]}")
             else:
@@ -379,7 +382,14 @@ class GeminiMultimodalModalities(Enum):
 
 
 class GeminiMediaResolution(str, Enum):
-    """Media resolution options for Gemini Multimodal Live."""
+    """Media resolution options for Gemini Multimodal Live.
+
+    Parameters:
+        UNSPECIFIED: Use default resolution setting.
+        LOW: Low resolution with 64 tokens.
+        MEDIUM: Medium resolution with 256 tokens.
+        HIGH: High resolution with zoomed reframing and 256 tokens.
+    """
 
     UNSPECIFIED = "MEDIA_RESOLUTION_UNSPECIFIED"  # Use default
     LOW = "MEDIA_RESOLUTION_LOW"  # 64 tokens
@@ -496,6 +506,7 @@ class GeminiMultimodalLiveLLMService(LLMService):
             params: Configuration parameters for the model. Defaults to InputParams().
             inference_on_context_initialization: Whether to generate a response when context
                 is first set. Defaults to True.
+            file_api_base_url: Base URL for the Gemini File API. Defaults to the official endpoint.
             **kwargs: Additional arguments passed to parent LLMService.
         """
         super().__init__(base_url=base_url, **kwargs)
@@ -757,6 +768,7 @@ class GeminiMultimodalLiveLLMService(LLMService):
         await self._ws_send(event.model_dump(exclude_none=True))
 
     async def _connect(self):
+        """Establish WebSocket connection to Gemini Live API."""
         if self._websocket:
             # Here we assume that if we have a websocket, we are connected. We
             # handle disconnections in the send/recv code paths.
@@ -861,6 +873,7 @@ class GeminiMultimodalLiveLLMService(LLMService):
             self._websocket = None
 
     async def _disconnect(self):
+        """Disconnect from Gemini Live API and clean up resources."""
         logger.info("Disconnecting from Gemini service")
         try:
             self._disconnecting = True
@@ -877,6 +890,7 @@ class GeminiMultimodalLiveLLMService(LLMService):
             logger.error(f"{self} error disconnecting: {e}")
 
     async def _ws_send(self, message):
+        """Send a message to the WebSocket connection."""
         # logger.debug(f"Sending message to websocket: {message}")
         try:
             if self._websocket:
@@ -897,6 +911,7 @@ class GeminiMultimodalLiveLLMService(LLMService):
     #
 
     async def _receive_task_handler(self):
+        """Handle incoming messages from the WebSocket connection."""
         async for message in WatchdogAsyncIterator(self._websocket, manager=self.task_manager):
             evt = events.parse_server_event(message)
             # logger.debug(f"Received event: {message[:500]}")
@@ -925,6 +940,7 @@ class GeminiMultimodalLiveLLMService(LLMService):
     #
 
     async def _send_user_audio(self, frame):
+        """Send user audio frame to Gemini Live API."""
         if self._audio_input_paused:
             return
         # Send all audio to Gemini
@@ -941,6 +957,7 @@ class GeminiMultimodalLiveLLMService(LLMService):
             self._user_audio_buffer = self._user_audio_buffer[-length:]
 
     async def _send_user_video(self, frame):
+        """Send user video frame to Gemini Live API."""
         if self._video_input_paused:
             return
 
@@ -954,6 +971,7 @@ class GeminiMultimodalLiveLLMService(LLMService):
         await self.send_client_event(evt)
 
     async def _create_initial_response(self):
+        """Create initial response based on context history."""
         if not self._api_session_ready:
             self._run_llm_when_api_session_ready = True
             return
@@ -979,6 +997,7 @@ class GeminiMultimodalLiveLLMService(LLMService):
             self._needs_turn_complete_message = True
 
     async def _create_single_response(self, messages_list):
+        """Create a single response from a list of messages."""
         # Refactor to combine this logic with same logic in GeminiMultimodalLiveContext
         messages = []
         for item in messages_list:
@@ -1000,12 +1019,14 @@ class GeminiMultimodalLiveLLMService(LLMService):
                         parts.append({"text": part.get("text")})
                     elif part.get("type") == "file_data":
                         file_data = part.get("file_data", {})
-                        parts.append({
+                        parts.append(
-                            "fileData": {
+                            {
-                                "mimeType": file_data.get("mime_type"),
+                                "fileData": {
-                                "fileUri": file_data.get("file_uri")
+                                    "mimeType": file_data.get("mime_type"),
+                                    "fileUri": file_data.get("file_uri"),
+                                }
                             }
-                        })
+                        )
                     else:
                         logger.warning(f"Unsupported content type: {str(part)[:80]}")
             else:
@@ -1029,6 +1050,7 @@ class GeminiMultimodalLiveLLMService(LLMService):
 
     @traced_gemini_live(operation="llm_tool_result")
     async def _tool_result(self, tool_result_message):
+        """Send tool result back to the API."""
         # For now we're shoving the name into the tool_call_id field, so this
         # will work until we revisit that.
         id = tool_result_message.get("tool_call_id")
@@ -1054,6 +1076,7 @@ class GeminiMultimodalLiveLLMService(LLMService):
 
     @traced_gemini_live(operation="llm_setup")
     async def _handle_evt_setup_complete(self, evt):
+        """Handle the setup complete event."""
         # If this is our first context frame, run the LLM
         self._api_session_ready = True
         # Now that we've configured the session, we can run the LLM if we need to.
@@ -1062,6 +1085,7 @@ class GeminiMultimodalLiveLLMService(LLMService):
             await self._create_initial_response()
 
     async def _handle_evt_model_turn(self, evt):
+        """Handle the model turn event."""
         part = evt.serverContent.modelTurn.parts[0]
         if not part:
             return
@@ -1103,6 +1127,7 @@ class GeminiMultimodalLiveLLMService(LLMService):
 
     @traced_gemini_live(operation="llm_tool_call")
     async def _handle_evt_tool_call(self, evt):
+        """Handle tool call events."""
         function_calls = evt.toolCall.functionCalls
         if not function_calls:
             return
@@ -1123,6 +1148,7 @@ class GeminiMultimodalLiveLLMService(LLMService):
 
     @traced_gemini_live(operation="llm_response")
     async def _handle_evt_turn_complete(self, evt):
+        """Handle the turn complete event."""
         self._bot_is_speaking = False
         text = self._bot_text_buffer
 
@@ -1206,6 +1232,7 @@ class GeminiMultimodalLiveLLMService(LLMService):
             )
 
     async def _handle_evt_output_transcription(self, evt):
+        """Handle the output transcription event."""
         if not evt.serverContent.outputTranscription:
             return
 
@@ -1224,6 +1251,7 @@ class GeminiMultimodalLiveLLMService(LLMService):
         await self.push_frame(TTSTextFrame(text=text))
 
     async def _handle_evt_usage_metadata(self, evt):
+        """Handle the usage metadata event."""
         if not evt.usageMetadata:
             return
 

src/pipecat/utils/utils.py

@@ -25,15 +25,6 @@ def obj_id() -> int:
 
     Returns:
         A unique integer identifier that increments globally across all objects.
-
-    Examples::
-
-        >>> obj_id()
-        0
-        >>> obj_id()
-        1
-        >>> obj_id()
-        2
     """
     with _ID_LOCK:
         return next(_ID)
@@ -47,16 +38,6 @@ def obj_count(obj) -> int:
 
     Returns:
         A unique integer count that increments per class type.
-
-    Examples::
-
-        >>> obj_count(object())
-        0
-        >>> obj_count(object())
-        1
-        >>> new_type = type('NewType', (object,), {})
-        >>> obj_count(new_type())
-        0
     """
     with _COUNTS_LOCK:
         return next(_COUNTS[obj.__class__.__name__])