Merge pull request #1264 from pipecat-ai/aleix/add-log-observers

add initial log observers

CHANGELOG.md

@@ -9,6 +9,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- Added new log observers `LLMLogObserver` and `TranscriptionLogObserver` that
+  can be useful for debugging your pipelines.
+
 - Added `room_url` property to `DailyTransport`.
 
 - Added `addons` argument to `DeepgramSTTService`.
@@ -45,6 +48,10 @@ stt = DeepgramSTTService(..., live_options=LiveOptions(model="nova-2-general"))
 - Fixed an issue where `EndTaskFrame` was not triggering
   `on_client_disconnected` or closing the WebSocket in FastAPI.
 
+- Fixed an issue in `DeepgramSTTService` where the `sample_rate` passed to the
+  `LiveOptions` was not being used, causing the service to use the default
+  sample rate of pipeline.
+
 - Fixed a context aggregator issue that would not append the LLM text response
   to the context if a function call happened in the same LLM turn.
 
@@ -63,10 +70,6 @@ stt = DeepgramSTTService(..., live_options=LiveOptions(model="nova-2-general"))
 - Fixed a `STTMuteFilter` issue that would not mute user audio frames causing
   transcriptions to be generated by the STT service.
 
-- Fixes an issue in `DeepgramSTTService` where `sample_rate` passed to the
-  `LiveOptions` was not being used, causing the service to use the default
-  sample rate of pipeline.
-
 ### Other
 
 - Added Gemini support to `examples/phone-chatbot`.

examples/foundational/30-observer.py

@@ -18,12 +18,10 @@ from pipecat.frames.frames import (
     BotStartedSpeakingFrame,
     BotStoppedSpeakingFrame,
     Frame,
-    LLMFullResponseEndFrame,
-    LLMFullResponseStartFrame,
-    LLMTextFrame,
     StartInterruptionFrame,
 )
 from pipecat.observers.base_observer import BaseObserver
+from pipecat.observers.loggers.llm_log_observer import LLMLogObserver
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
 from pipecat.pipeline.task import PipelineParams, PipelineTask
@@ -73,38 +71,6 @@ class DebugObserver(BaseObserver):
             logger.info(f"🤖 BOT STOP SPEAKING: {src} {arrow} {dst} at {time_sec:.2f}s")
 
 
-class LLMLogObserver(BaseObserver):
-    """Observer to log LLM activity to the console.
-
-    Logs all frame instances of:
-    - LLMFullResponseStartFrame (only from LLM service)
-    - LLMTextFrame
-    - LLMFullResponseEndFrame (only from LLM service)
-
-    This allows you to track when the LLM starts responding, what it generates, and when it finishes.
-    Log format: [LLM EVENT]: [details] at [timestamp]s
-    """
-
-    async def on_push_frame(
-        self,
-        src: FrameProcessor,
-        dst: FrameProcessor,
-        frame: Frame,
-        direction: FrameDirection,
-        timestamp: int,
-    ):
-        time_sec = timestamp / 1_000_000_000
-
-        # Only log start/end frames from OpenAILLMService
-        if isinstance(frame, (LLMFullResponseStartFrame, LLMFullResponseEndFrame)):
-            if isinstance(src, OpenAILLMService):
-                event = "START" if isinstance(frame, LLMFullResponseStartFrame) else "END"
-                logger.info(f"🧠 LLM {event} RESPONSE at {time_sec:.2f}s")
-        # Log all LLMTextFrames
-        elif isinstance(frame, LLMTextFrame):
-            logger.info(f"🧠 LLM GENERATING: {frame.text!r} at {time_sec:.2f}s")
-
-
 async def main():
     async with aiohttp.ClientSession() as session:
         (room_url, token) = await configure(session)

src/pipecat/observers/loggers/llm_log_observer.py

@@ -0,0 +1,85 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+from loguru import logger
+
+from pipecat.frames.frames import (
+    Frame,
+    FunctionCallInProgressFrame,
+    FunctionCallResultFrame,
+    LLMFullResponseEndFrame,
+    LLMFullResponseStartFrame,
+    LLMMessagesFrame,
+    LLMTextFrame,
+)
+from pipecat.observers.base_observer import BaseObserver
+from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContextFrame
+from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
+from pipecat.services.ai_services import LLMService
+
+
+class LLMLogObserver(BaseObserver):
+    """Observer to log LLM activity to the console.
+
+    Logs all frame instances (only from/to LLM service) of:
+
+    - LLMFullResponseStartFrame
+    - LLMFullResponseEndFrame
+    - LLMTextFrame
+    - FunctionCallInProgressFrame
+    - LLMMessagesFrame
+    - OpenAILLMContextFrame
+
+    This allows you to track when the LLM starts responding, what it generates,
+    and when it finishes.
+
+    """
+
+    async def on_push_frame(
+        self,
+        src: FrameProcessor,
+        dst: FrameProcessor,
+        frame: Frame,
+        direction: FrameDirection,
+        timestamp: int,
+    ):
+        if not isinstance(src, LLMService) and not isinstance(dst, LLMService):
+            return
+
+        time_sec = timestamp / 1_000_000_000
+
+        arrow = "→"
+
+        # Log LLM start/end frames (output)
+        if isinstance(frame, (LLMFullResponseStartFrame, LLMFullResponseEndFrame)):
+            event = "START" if isinstance(frame, LLMFullResponseStartFrame) else "END"
+            logger.debug(f"🧠 {src} {arrow} LLM {event} RESPONSE at {time_sec:.2f}s")
+        # Log all LLMTextFrames (output)
+        elif isinstance(frame, LLMTextFrame):
+            logger.debug(f"🧠 {src} {arrow} LLM GENERATING: {frame.text!r} at {time_sec:.2f}s")
+        # Log function calling (output)
+        elif (
+            isinstance(frame, FunctionCallInProgressFrame)
+            and direction != FrameDirection.DOWNSTREAM
+        ):
+            logger.debug(
+                f"🧠 {src} {arrow} LLM FUNCTION CALL ({frame.tool_call_id}): {frame.function_name!r}({frame.arguments}) at {time_sec:.2f}s"
+            )
+        # Log LLMMessagesFrame (input)
+        elif isinstance(frame, LLMMessagesFrame):
+            logger.debug(
+                f"🧠 {arrow} {dst} LLM MESSAGES FRAME: {frame.messages} at {time_sec:.2f}s"
+            )
+        # Log OpenAILLMContextFrame (input)
+        elif isinstance(frame, OpenAILLMContextFrame):
+            logger.debug(
+                f"🧠 {arrow} {dst} LLM CONTEXT FRAME: {frame.context.messages} at {time_sec:.2f}s"
+            )
+        # Log function call result (input)
+        elif isinstance(frame, FunctionCallResultFrame):
+            logger.debug(
+                f"🧠 {arrow} {src} LLM FUNCTION CALL RESULT ({frame.tool_call_id}): {frame.result} at {time_sec:.2f}s"
+            )

src/pipecat/observers/loggers/transcription_log_observer.py

@@ -0,0 +1,54 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+from loguru import logger
+
+from pipecat.frames.frames import (
+    Frame,
+    InterimTranscriptionFrame,
+    TranscriptionFrame,
+)
+from pipecat.observers.base_observer import BaseObserver
+from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
+from pipecat.services.ai_services import STTService
+
+
+class TranscriptionLogObserver(BaseObserver):
+    """Observer to log transcription activity to the console.
+
+    Logs all frame instances (only from STT service) of:
+
+    - TranscriptionFrame
+    - InterimTranscriptionFrame
+
+    This allows you to track when the LLM starts responding, what it generates,
+    and when it finishes.
+
+    """
+
+    async def on_push_frame(
+        self,
+        src: FrameProcessor,
+        dst: FrameProcessor,
+        frame: Frame,
+        direction: FrameDirection,
+        timestamp: int,
+    ):
+        if not isinstance(src, STTService):
+            return
+
+        time_sec = timestamp / 1_000_000_000
+
+        arrow = "→"
+
+        if isinstance(frame, TranscriptionFrame):
+            logger.debug(
+                f"💬 {src} {arrow} TRANSCRIPTION: {frame.text!r} from {frame.user_id!r} at {time_sec:.2f}s"
+            )
+        elif isinstance(frame, InterimTranscriptionFrame):
+            logger.debug(
+                f"💬 {src} {arrow} INTERIM TRANSCRIPTION: {frame.text!r} from {frame.user_id!r} at {time_sec:.2f}s"
+            )