wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fix Langfuse tracing for GoogleLLMService with universal LLMContext (#3025)

Browse Source 
* Fix Langfuse tracing for GoogleLLMService with universal LLMContext

- Fixed issue where input appeared as null in Langfuse dashboard for GoogleLLMService
- Added fallback to use adapter's get_messages_for_logging() for universal LLMContext
- Ensures proper message format conversion for Google/Gemini services
- Handles system message conversion to system_instruction format
- Also fixes serialization of empty message lists ([] now serializes correctly)

This fix ensures Langfuse tracing works correctly for Google services using
both OpenAILLMContext/GoogleLLMContext and the universal LLMContext.

* Add unit tests for Langfuse tracing with GoogleLLMService

- Test that tracing correctly captures messages with universal LLMContext
- Test that empty message lists are properly serialized
- Test that adapter's get_messages_for_logging is used instead of context method
- All tests verify that input is correctly added to Langfuse spans

* Fix test mocking to patch opentelemetry.trace.get_tracer correctly

The tests were failing in CI because they were trying to patch
'pipecat.utils.tracing.service_decorators.trace' which doesn't exist as
an attribute. The trace module is imported from opentelemetry, so we need
to patch 'opentelemetry.trace.get_tracer' instead.

* Skip tracing tests when opentelemetry is not installed

The tracing dependencies (opentelemetry) are optional in Pipecat and not
installed in the CI environment. Added a skipif marker to skip these tests
when opentelemetry is not available, preventing CI failures while still
allowing the tests to run when tracing dependencies are installed locally.

* Install tracing dependencies in GitHub Actions CI

Instead of skipping the tracing tests, install the 'tracing' extra
(opentelemetry) in the CI environment so the tests can run properly.
Removed the skipif condition from the tests since opentelemetry will
now be available in CI.

* Use the context type to determine which messages to use, fix tool_count and tools (#3032)

---------

Co-authored-by: Mark Backman <mark@daily.co>
This commit is contained in:
James Hush
2025-11-12 14:58:00 +01:00
committed by GitHub
parent eb36a1bc91
commit 2006a64def
2 changed files with 47 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
CHANGELOG.md
View File

@@ -5,10 +5,13 @@ All notable changes to **Pipecat** will be documented in this file.
The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
## Unreleased
## [Unreleased]
### Fixed
- Fixed an issue with OpenTelemetry where tracing wasn't correctly displaying
LLM completions and tools when using the universal `LLMContext`.
- Fixed issue where `DeepgramFluxSTTService` failed to connect if passing a
`keyterm` or `tag` containing a space.

70
src/pipecat/utils/tracing/service_decorators.py
View File

@@ -23,6 +23,8 @@ if TYPE_CHECKING:
from opentelemetry import context as context_api
from opentelemetry import trace
from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
from pipecat.utils.tracing.service_attributes import (
add_gemini_live_span_attributes,
add_llm_span_attributes,
@@ -382,43 +384,57 @@ def traced_llm(func: Optional[Callable] = None, *, name: Optional[str] = None) -
# Replace push_frame to capture output
self.push_frame = traced_push_frame
# Detect if we're using Google's service
is_google_service = "google" in service_class_name.lower()
# Try to get messages based on service type
# Get messages for logging
# For OpenAILLMContext: use context's own get_messages_for_logging() method
# For LLMContext: use adapter's get_messages_for_logging() which returns
# messages in provider's native format with sensitive data sanitized
messages = None
serialized_messages = None
# TODO: Revisit once we unify the messages across services
if is_google_service:
# Handle Google service specifically
if hasattr(context, "get_messages_for_logging"):
messages = context.get_messages_for_logging()
else:
# Handle other services like OpenAI
if hasattr(context, "get_messages"):
messages = context.get_messages()
elif hasattr(context, "messages"):
messages = context.messages
if isinstance(context, OpenAILLMContext):
# OpenAILLMContext and subclasses have their own method
messages = context.get_messages_for_logging()
elif isinstance(context, LLMContext):
# Universal LLMContext - use adapter for provider-native format
if hasattr(self, "get_llm_adapter"):
adapter = self.get_llm_adapter()
messages = adapter.get_messages_for_logging(context)
elif hasattr(context, "get_messages"):
# Fallback for unknown context types
messages = context.get_messages()
elif hasattr(context, "messages"):
messages = context.messages
# Serialize messages if available
if messages:
try:
serialized_messages = json.dumps(messages)
except Exception as e:
serialized_messages = f"Error serializing messages: {str(e)}"
serialized_messages = json.dumps(messages)
# Get tools, system message, etc. based on the service type
tools = getattr(context, "tools", None)
# Get tools
# For OpenAILLMContext: tools may need adapter conversion if set
# For LLMContext: use adapter's from_standard_tools() to convert ToolsSchema
tools = None
serialized_tools = None
tool_count = 0
if tools:
try:
serialized_tools = json.dumps(tools)
tool_count = len(tools) if isinstance(tools, list) else 1
except Exception as e:
serialized_tools = f"Error serializing tools: {str(e)}"
if isinstance(context, OpenAILLMContext):
# OpenAILLMContext: tools property handles adapter conversion internally
tools = context.tools
elif isinstance(context, LLMContext):
# Universal LLMContext - use adapter to convert ToolsSchema
if hasattr(self, "get_llm_adapter") and hasattr(context, "tools"):
adapter = self.get_llm_adapter()
tools = adapter.from_standard_tools(context.tools)
elif hasattr(context, "tools"):
# Fallback for unknown context types
tools = context.tools
# Serialize and count tools if available
# Check if tools is not None and not NOT_GIVEN (using attribute check as fallback)
if tools is not None and not (
hasattr(tools, "__name__") and tools.__name__ == "NOT_GIVEN"
):
serialized_tools = json.dumps(tools)
tool_count = len(tools) if isinstance(tools, list) else 1
# Handle system message for different services
system_message = None