From b29ab8c608c2f852d5d10f25ff42f2e18e9a5ef2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Thu, 20 Feb 2025 23:15:09 -0800 Subject: [PATCH 1/2] observers: add LLMLogObserver and TranscriptionLogObserver --- CHANGELOG.md | 11 ++- src/pipecat/observers/loggers/__init__.py | 0 .../observers/loggers/llm_log_observer.py | 85 +++++++++++++++++++ .../loggers/transcription_log_observer.py | 54 ++++++++++++ 4 files changed, 146 insertions(+), 4 deletions(-) create mode 100644 src/pipecat/observers/loggers/__init__.py create mode 100644 src/pipecat/observers/loggers/llm_log_observer.py create mode 100644 src/pipecat/observers/loggers/transcription_log_observer.py diff --git a/CHANGELOG.md b/CHANGELOG.md index f2375c8e5..a71ae5f02 100644 --- a/CHANGELOG.md +++ b/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`. diff --git a/src/pipecat/observers/loggers/__init__.py b/src/pipecat/observers/loggers/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/src/pipecat/observers/loggers/llm_log_observer.py b/src/pipecat/observers/loggers/llm_log_observer.py new file mode 100644 index 000000000..907dce70b --- /dev/null +++ b/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" + ) diff --git a/src/pipecat/observers/loggers/transcription_log_observer.py b/src/pipecat/observers/loggers/transcription_log_observer.py new file mode 100644 index 000000000..630f7ab33 --- /dev/null +++ b/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" + ) From 65f548b2ece23442acbb50e8e98517dd5b94312e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Thu, 20 Feb 2025 23:15:30 -0800 Subject: [PATCH 2/2] examples(30-observer): update to use LLMLogObserver --- examples/foundational/30-observer.py | 36 +--------------------------- 1 file changed, 1 insertion(+), 35 deletions(-) diff --git a/examples/foundational/30-observer.py b/examples/foundational/30-observer.py index 4b672ba0e..0988be76d 100644 --- a/examples/foundational/30-observer.py +++ b/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)