#
# Copyright (c) 2024-2026, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
"""LLM logging observer for Pipecat."""
from loguru import logger
from pipecat.frames.frames import (
FunctionCallInProgressFrame,
FunctionCallResultFrame,
LLMContextFrame,
LLMFullResponseEndFrame,
LLMFullResponseStartFrame,
LLMTextFrame,
)
from pipecat.observers.base_observer import BaseObserver, FramePushed
from pipecat.processors.frame_processor import FrameDirection
from pipecat.services.llm_service import LLMService
[docs]
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
This allows you to track when the LLM starts responding, what it generates,
and when it finishes.
"""
[docs]
async def on_push_frame(self, data: FramePushed):
"""Handle frame push events and log LLM-related activities.
Args:
data: The frame push event data containing source, destination,
frame, direction, and timestamp information.
"""
src = data.source
dst = data.destination
frame = data.frame
direction = data.direction
timestamp = data.timestamp
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 LLMContextFrame (input)
elif isinstance(frame, LLMContextFrame):
messages = frame.context.get_messages()
logger.debug(f"🧠 {arrow} {dst} LLM CONTEXT FRAME: {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"
)