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

Merge pull request #1927 from pipecat-ai/mb/gemini-tracing

Browse Source
This commit is contained in:
Mark Backman
2025-05-30 22:01:44 -04:00
committed by GitHub
parent 0fd38d8115 43719ec737
commit c647114bb9
5 changed files with 794 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
CHANGELOG.md
View File

@@ -9,6 +9,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
### Added
- Added OpenTelemetry tracing for `GeminiMultimodalLiveLLMService` and
`OpenAIRealtimeBetaLLMService`.
- Added `interruption_strategies` to `PipelineParams` using
`MinWordsInterruptionStrategy` to specify minimum words required to interrupt
the bot when it's speaking. Use
@@ -48,6 +51,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
`DailyTransport.stop_transcription()` to be able to start and stop Daily
transcription dynamically (maybe with different settings).
### Changed
- Updated OpenTelemetry tracing attribute `metrics.ttfb_ms` to `metrics.ttfb`.
The attribute reports TTFB in seconds.
### Deprecated
- `DailyTransport.send_dtmf()` is deprecated, push an `OutputDTMFFrame` or an

45
src/pipecat/services/gemini_multimodal_live/gemini.py
View File

@@ -60,6 +60,7 @@ from pipecat.services.openai.llm import (
from pipecat.transcriptions.language import Language
from pipecat.utils.string import match_endofsentence
from pipecat.utils.time import time_now_iso8601
from pipecat.utils.tracing.service_decorators import traced_gemini_live, traced_stt, traced_tts
from . import events
@@ -378,6 +379,7 @@ class GeminiMultimodalLiveLLMService(LLMService):
self._last_transcription_sent = ""
self._bot_audio_buffer = bytearray()
self._bot_text_buffer = ""
self._llm_output_buffer = ""
self._sample_rate = 24000
@@ -471,6 +473,7 @@ class GeminiMultimodalLiveLLMService(LLMService):
async def _handle_user_stopped_speaking(self, frame):
self._user_is_speaking = False
self._user_audio_buffer = bytearray()
await self.start_ttfb_metrics()
if self._needs_turn_complete_message:
self._needs_turn_complete_message = False
evt = events.ClientContentMessage.model_validate(
@@ -752,6 +755,8 @@ class GeminiMultimodalLiveLLMService(LLMService):
logger.debug(f"Creating initial response: {messages}")
await self.start_ttfb_metrics()
evt = events.ClientContentMessage.model_validate(
{
"clientContent": {
@@ -793,6 +798,8 @@ class GeminiMultimodalLiveLLMService(LLMService):
return
logger.debug(f"Creating response: {messages}")
await self.start_ttfb_metrics()
evt = events.ClientContentMessage.model_validate(
{
"clientContent": {
@@ -803,6 +810,7 @@ class GeminiMultimodalLiveLLMService(LLMService):
)
await self.send_client_event(evt)
@traced_gemini_live(operation="llm_tool_result")
async def _tool_result(self, tool_result_message):
# For now we're shoving the name into the tool_call_id field, so this
# will work until we revisit that.
@@ -827,6 +835,7 @@ class GeminiMultimodalLiveLLMService(LLMService):
await self._websocket.send(response_message)
# await self._websocket.send(json.dumps({"clientContent": {"turnComplete": True}}))
@traced_gemini_live(operation="llm_setup")
async def _handle_evt_setup_complete(self, evt):
# If this is our first context frame, run the LLM
self._api_session_ready = True
@@ -840,6 +849,8 @@ class GeminiMultimodalLiveLLMService(LLMService):
if not part:
return
await self.stop_ttfb_metrics()
# part.text is added when `modalities` is set to TEXT; otherwise, it's None
text = part.text
if text:
@@ -873,6 +884,7 @@ class GeminiMultimodalLiveLLMService(LLMService):
)
await self.push_frame(frame)
@traced_gemini_live(operation="llm_tool_call")
async def _handle_evt_tool_call(self, evt):
function_calls = evt.toolCall.functionCalls
if not function_calls:
@@ -887,12 +899,28 @@ class GeminiMultimodalLiveLLMService(LLMService):
arguments=call.args,
)
@traced_gemini_live(operation="llm_response")
async def _handle_evt_turn_complete(self, evt):
self._bot_is_speaking = False
text = self._bot_text_buffer
self._bot_text_buffer = ""
# Only push the TTSStoppedFrame the bot is outputting audio
# Determine output and modality for tracing
if text:
# TEXT modality
output_text = text
output_modality = "TEXT"
else:
# AUDIO modality
output_text = self._llm_output_buffer
output_modality = "AUDIO"
# Trace the complete LLM response (this will be handled by the decorator)
# The decorator will extract the output text and usage metadata from the event
self._bot_text_buffer = ""
self._llm_output_buffer = ""
# Only push the TTSStoppedFrame if the bot is outputting audio
# when text is found, modalities is set to TEXT and no audio
# is produced.
if not text:
@@ -900,6 +928,13 @@ class GeminiMultimodalLiveLLMService(LLMService):
await self.push_frame(LLMFullResponseEndFrame())
@traced_stt
async def _handle_user_transcription(
self, transcript: str, is_final: bool, language: Optional[Language] = None
):
"""Handle a transcription result with tracing."""
pass
async def _handle_evt_input_transcription(self, evt):
"""Handle the input transcription event.
@@ -935,6 +970,9 @@ class GeminiMultimodalLiveLLMService(LLMService):
# Send a TranscriptionFrame with the complete sentence
logger.debug(f"[Transcription:user] [{complete_sentence}]")
await self._handle_user_transcription(
complete_sentence, True, self._settings["language"]
)
await self.push_frame(
TranscriptionFrame(
text=complete_sentence,
@@ -957,6 +995,9 @@ class GeminiMultimodalLiveLLMService(LLMService):
if not text:
return
# Collect text for tracing
self._llm_output_buffer += text
await self.push_frame(LLMTextFrame(text=text))
await self.push_frame(TTSTextFrame(text=text))

14
src/pipecat/services/openai_realtime_beta/openai.py
View File

@@ -50,7 +50,9 @@ from pipecat.processors.aggregators.openai_llm_context import (
from pipecat.processors.frame_processor import FrameDirection
from pipecat.services.llm_service import LLMService
from pipecat.services.openai.llm import OpenAIContextAggregatorPair
from pipecat.transcriptions.language import Language
from pipecat.utils.time import time_now_iso8601
from pipecat.utils.tracing.service_decorators import traced_openai_realtime, traced_stt, traced_tts
from . import events
from .context import (
@@ -100,6 +102,7 @@ class OpenAIRealtimeBetaLLMService(LLMService):
self.api_key = api_key
self.base_url = full_url
self.set_model_name(model)
self._session_properties: events.SessionProperties = (
session_properties or events.SessionProperties()
@@ -402,6 +405,7 @@ class OpenAIRealtimeBetaLLMService(LLMService):
# errors are fatal, so exit the receive loop
return
@traced_openai_realtime(operation="llm_setup")
async def _handle_evt_session_created(self, evt):
# session.created is received right after connecting. Send a message
# to configure the session properties.
@@ -467,6 +471,13 @@ class OpenAIRealtimeBetaLLMService(LLMService):
InterimTranscriptionFrame(evt.delta, "", time_now_iso8601(), result=evt)
)
@traced_stt
async def _handle_user_transcription(
self, transcript: str, is_final: bool, language: Optional[Language] = None
):
"""Handle a transcription result with tracing."""
pass
async def handle_evt_input_audio_transcription_completed(self, evt):
await self._call_event_handler("on_conversation_item_updated", evt.item_id, None)
@@ -475,6 +486,7 @@ class OpenAIRealtimeBetaLLMService(LLMService):
# no way to get a language code?
TranscriptionFrame(evt.transcript, "", time_now_iso8601(), result=evt)
)
await self._handle_user_transcription(evt.transcript, True, Language.EN)
pair = self._user_and_response_message_tuple
if pair:
user, assistant = pair
@@ -493,6 +505,7 @@ class OpenAIRealtimeBetaLLMService(LLMService):
for future in futures:
future.set_result(evt.item)
@traced_openai_realtime(operation="llm_response")
async def _handle_evt_response_done(self, evt):
# todo: figure out whether there's anything we need to do for "cancelled" events
# usage metrics
@@ -609,6 +622,7 @@ class OpenAIRealtimeBetaLLMService(LLMService):
self._context.llm_needs_initial_messages = True
await self._connect()
@traced_openai_realtime(operation="llm_request")
async def _create_response(self):
if not self._api_session_ready:
self._run_llm_when_api_session_ready = True

206
src/pipecat/utils/tracing/service_attributes.py
View File

@@ -6,7 +6,7 @@
"""Functions for adding attributes to OpenTelemetry spans."""
from typing import TYPE_CHECKING, Any, Dict, Optional
from typing import TYPE_CHECKING, Any, Dict, List, Optional
# Import for type checking only
if TYPE_CHECKING:
@@ -256,3 +256,207 @@ def add_llm_span_attributes(
for key, value in kwargs.items():
if isinstance(value, (str, int, float, bool)):
span.set_attribute(key, value)
def add_gemini_live_span_attributes(
span: "Span",
service_name: str,
model: str,
operation_name: str,
voice_id: Optional[str] = None,
language: Optional[str] = None,
modalities: Optional[str] = None,
settings: Optional[Dict[str, Any]] = None,
tools: Optional[List[Dict]] = None,
tools_serialized: Optional[str] = None,
transcript: Optional[str] = None,
is_input: Optional[bool] = None,
text_output: Optional[str] = None,
audio_data_size: Optional[int] = None,
**kwargs,
) -> None:
"""Add Gemini Live specific attributes to a span.
Args:
span: The span to add attributes to
service_name: Name of the service
model: Model name/identifier
operation_name: Name of the operation (setup, model_turn, tool_call, etc.)
voice_id: Voice identifier used for output
language: Language code for the session
modalities: Supported modalities (e.g., "AUDIO", "TEXT")
settings: Service configuration settings
tools: Available tools/functions list
tools_serialized: JSON-serialized tools for detailed inspection
transcript: Transcription text
is_input: Whether transcript is input (True) or output (False)
text_output: Text output from model
audio_data_size: Size of audio data in bytes
**kwargs: Additional attributes to add
"""
# Add standard attributes
span.set_attribute("gen_ai.system", "gcp.gemini")
span.set_attribute("gen_ai.request.model", model)
span.set_attribute("gen_ai.operation.name", operation_name)
span.set_attribute("service.operation", operation_name)
# Add optional attributes
if voice_id:
span.set_attribute("voice_id", voice_id)
if language:
span.set_attribute("language", language)
if modalities:
span.set_attribute("modalities", modalities)
if transcript:
span.set_attribute("transcript", transcript)
if is_input is not None:
span.set_attribute("transcript.is_input", is_input)
if text_output:
span.set_attribute("text_output", text_output)
if audio_data_size is not None:
span.set_attribute("audio.data_size_bytes", audio_data_size)
if tools:
span.set_attribute("tools.count", len(tools))
span.set_attribute("tools.available", True)
# Add individual tool names for easier filtering
tool_names = []
for tool in tools:
if isinstance(tool, dict) and "name" in tool:
tool_names.append(tool["name"])
elif hasattr(tool, "name"):
tool_name = getattr(tool, "name", None)
if tool_name is not None:
tool_names.append(tool_name)
if tool_names:
span.set_attribute("tools.names", ",".join(tool_names))
if tools_serialized:
span.set_attribute("tools.definitions", tools_serialized)
# Add settings if provided
if settings:
for key, value in settings.items():
if isinstance(value, (str, int, float, bool)):
span.set_attribute(f"settings.{key}", value)
elif key == "vad" and value:
# Handle VAD settings specially
if hasattr(value, "disabled") and value.disabled is not None:
span.set_attribute("settings.vad.disabled", value.disabled)
if hasattr(value, "start_sensitivity") and value.start_sensitivity:
span.set_attribute(
"settings.vad.start_sensitivity", value.start_sensitivity.value
)
if hasattr(value, "end_sensitivity") and value.end_sensitivity:
span.set_attribute("settings.vad.end_sensitivity", value.end_sensitivity.value)
# Add any additional keyword arguments as attributes
for key, value in kwargs.items():
if isinstance(value, (str, int, float, bool)):
span.set_attribute(key, value)
def add_openai_realtime_span_attributes(
span: "Span",
service_name: str,
model: str,
operation_name: str,
session_properties: Optional[Dict[str, Any]] = None,
transcript: Optional[str] = None,
is_input: Optional[bool] = None,
context_messages: Optional[str] = None,
function_calls: Optional[List[Dict]] = None,
tools: Optional[List[Dict]] = None,
tools_serialized: Optional[str] = None,
audio_data_size: Optional[int] = None,
**kwargs,
) -> None:
"""Add OpenAI Realtime specific attributes to a span.
Args:
span: The span to add attributes to
service_name: Name of the service
model: Model name/identifier
operation_name: Name of the operation (setup, transcription, response, etc.)
session_properties: Session configuration properties
transcript: Transcription text
is_input: Whether transcript is input (True) or output (False)
context_messages: JSON-serialized context messages
function_calls: Function calls being made
tools: Available tools/functions list
tools_serialized: JSON-serialized tools for detailed inspection
audio_data_size: Size of audio data in bytes
**kwargs: Additional attributes to add
"""
# Add standard attributes
span.set_attribute("gen_ai.system", "openai")
span.set_attribute("gen_ai.request.model", model)
span.set_attribute("gen_ai.operation.name", operation_name)
span.set_attribute("service.operation", operation_name)
# Add optional attributes
if transcript:
span.set_attribute("transcript", transcript)
if is_input is not None:
span.set_attribute("transcript.is_input", is_input)
if context_messages:
span.set_attribute("input", context_messages)
if audio_data_size is not None:
span.set_attribute("audio.data_size_bytes", audio_data_size)
if tools:
span.set_attribute("tools.count", len(tools))
span.set_attribute("tools.available", True)
# Add individual tool names for easier filtering
tool_names = []
for tool in tools:
if isinstance(tool, dict) and "name" in tool:
tool_names.append(tool["name"])
elif hasattr(tool, "name"):
tool_names.append(tool.name)
elif isinstance(tool, dict) and "function" in tool and "name" in tool["function"]:
tool_names.append(tool["function"]["name"])
if tool_names:
span.set_attribute("tools.names", ",".join(tool_names))
if tools_serialized:
span.set_attribute("tools.definitions", tools_serialized)
if function_calls:
span.set_attribute("function_calls.count", len(function_calls))
if function_calls:
call = function_calls[0]
if hasattr(call, "name"):
span.set_attribute("function_calls.first_name", call.name)
elif isinstance(call, dict) and "name" in call:
span.set_attribute("function_calls.first_name", call["name"])
# Add session properties if provided
if session_properties:
for key, value in session_properties.items():
if isinstance(value, (str, int, float, bool)):
span.set_attribute(f"session.{key}", value)
elif key == "turn_detection" and value is not None:
if isinstance(value, bool):
span.set_attribute("session.turn_detection.enabled", value)
elif isinstance(value, dict):
span.set_attribute("session.turn_detection.enabled", True)
for td_key, td_value in value.items():
if isinstance(td_value, (str, int, float, bool)):
span.set_attribute(f"session.turn_detection.{td_key}", td_value)
# Add any additional keyword arguments as attributes
for key, value in kwargs.items():
if isinstance(value, (str, int, float, bool)):
span.set_attribute(key, value)

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

@@ -24,7 +24,9 @@ if TYPE_CHECKING:
from opentelemetry import trace
from pipecat.utils.tracing.service_attributes import (
add_gemini_live_span_attributes,
add_llm_span_attributes,
add_openai_realtime_span_attributes,
add_stt_span_attributes,
add_tts_span_attributes,
)
@@ -477,3 +479,525 @@ def traced_llm(func: Optional[Callable] = None, *, name: Optional[str] = None) -
if func is not None:
return decorator(func)
return decorator
def traced_gemini_live(operation: str) -> Callable:
"""Traces Gemini Live service methods with operation-specific attributes.
This decorator automatically captures relevant information based on the operation type:
- llm_setup: Configuration, tools definitions, and system instructions
- llm_tool_call: Function call information
- llm_tool_result: Function execution results
- llm_response: Complete LLM response with usage and output
Args:
operation: The operation name (matches the event type being handled)
Returns:
Wrapped method with Gemini Live specific tracing.
"""
if not is_tracing_available():
return _noop_decorator
def decorator(func):
@functools.wraps(func)
async def wrapper(self, *args, **kwargs):
try:
if not is_tracing_available():
return await func(self, *args, **kwargs)
service_class_name = self.__class__.__name__
span_name = f"{operation}"
# Get the parent context - turn context if available, otherwise service context
turn_context = get_current_turn_context()
parent_context = turn_context or _get_parent_service_context(self)
# Create a new span as child of the turn span or service span
tracer = trace.get_tracer("pipecat")
with tracer.start_as_current_span(
span_name, context=parent_context
) as current_span:
try:
# Base service attributes
model_name = getattr(
self, "model_name", getattr(self, "_model_name", "unknown")
)
voice_id = getattr(self, "_voice_id", None)
language_code = getattr(self, "_language_code", None)
settings = getattr(self, "_settings", {})
# Get modalities if available
modalities = None
if hasattr(self, "_settings") and "modalities" in self._settings:
modality_obj = self._settings["modalities"]
if hasattr(modality_obj, "value"):
modalities = modality_obj.value
else:
modalities = str(modality_obj)
# Operation-specific attribute collection
operation_attrs = {}
if operation == "llm_setup":
# Capture detailed tool information
tools = getattr(self, "_tools", None)
if tools:
# Handle different tool formats
tools_list = []
tools_serialized = None
try:
if hasattr(tools, "standard_tools"):
# ToolsSchema object
tools_list = tools.standard_tools
# Serialize the tools for detailed inspection
tools_serialized = json.dumps(
[
{
"name": tool.name
if hasattr(tool, "name")
else tool.get("name", "unknown"),
"description": tool.description
if hasattr(tool, "description")
else tool.get("description", ""),
"properties": tool.properties
if hasattr(tool, "properties")
else tool.get("properties", {}),
"required": tool.required
if hasattr(tool, "required")
else tool.get("required", []),
}
for tool in tools_list
]
)
elif isinstance(tools, list):
# List of tool dictionaries or objects
tools_list = tools
tools_serialized = json.dumps(
[
{
"name": tool.get("name", "unknown")
if isinstance(tool, dict)
else getattr(tool, "name", "unknown"),
"description": tool.get("description", "")
if isinstance(tool, dict)
else getattr(tool, "description", ""),
"properties": tool.get("properties", {})
if isinstance(tool, dict)
else getattr(tool, "properties", {}),
"required": tool.get("required", [])
if isinstance(tool, dict)
else getattr(tool, "required", []),
}
for tool in tools_list
]
)
if tools_list:
operation_attrs["tools"] = tools_list
operation_attrs["tools_serialized"] = tools_serialized
except Exception as e:
logging.warning(f"Error serializing tools for tracing: {e}")
# Fallback to basic tool count
if tools_list:
operation_attrs["tools"] = tools_list
# Capture system instruction information
system_instruction = getattr(self, "_system_instruction", None)
if system_instruction:
operation_attrs["system_instruction"] = system_instruction[
:500
] # Truncate if very long
# Capture context system instructions if available
if hasattr(self, "_context") and self._context:
try:
context_system = self._context.extract_system_instructions()
if context_system:
operation_attrs["context_system_instruction"] = (
context_system[:500]
) # Truncate if very long
except Exception as e:
logging.warning(
f"Error extracting context system instructions: {e}"
)
elif operation == "llm_tool_call" and args:
# Extract tool call information
evt = args[0] if args else None
if evt and hasattr(evt, "toolCall") and evt.toolCall.functionCalls:
function_calls = evt.toolCall.functionCalls
if function_calls:
# Add information about the first function call
call = function_calls[0]
operation_attrs["tool.function_name"] = call.name
operation_attrs["tool.call_id"] = call.id
operation_attrs["tool.calls_count"] = len(function_calls)
# Add all function names being called
all_function_names = [c.name for c in function_calls]
operation_attrs["tool.all_function_names"] = ",".join(
all_function_names
)
# Add arguments for the first call (truncated if too long)
try:
args_str = json.dumps(call.args) if call.args else "{}"
if len(args_str) > 1000:
args_str = args_str[:1000] + "..."
operation_attrs["tool.arguments"] = args_str
except Exception:
operation_attrs["tool.arguments"] = str(call.args)[:1000]
elif operation == "llm_tool_result" and args:
# Extract tool result information
tool_result_message = args[0] if args else None
if tool_result_message and isinstance(tool_result_message, dict):
# Extract the tool call information
tool_call_id = tool_result_message.get("tool_call_id")
tool_call_name = tool_result_message.get("tool_call_name")
result_content = tool_result_message.get("content")
if tool_call_id:
operation_attrs["tool.call_id"] = tool_call_id
if tool_call_name:
operation_attrs["tool.function_name"] = tool_call_name
# Parse and capture the result
if result_content:
try:
result = json.loads(result_content)
# Serialize the result, truncating if too long
result_str = json.dumps(result)
if len(result_str) > 2000: # Larger limit for results
result_str = result_str[:2000] + "..."
operation_attrs["tool.result"] = result_str
# Add result status/success indicator if present
if isinstance(result, dict):
if "error" in result:
operation_attrs["tool.result_status"] = "error"
elif "success" in result:
operation_attrs["tool.result_status"] = "success"
else:
operation_attrs["tool.result_status"] = "completed"
except json.JSONDecodeError as e:
operation_attrs["tool.result"] = (
f"Invalid JSON: {str(result_content)[:500]}"
)
operation_attrs["tool.result_status"] = "parse_error"
except Exception as e:
operation_attrs["tool.result"] = (
f"Error processing result: {str(e)}"
)
operation_attrs["tool.result_status"] = "processing_error"
elif operation == "llm_response" and args:
# Extract usage and response metadata from turn complete event
evt = args[0] if args else None
if evt and hasattr(evt, "usageMetadata") and evt.usageMetadata:
usage = evt.usageMetadata
# Token usage - basic attributes for span visibility
if hasattr(usage, "promptTokenCount"):
operation_attrs["tokens.prompt"] = usage.promptTokenCount or 0
if hasattr(usage, "responseTokenCount"):
operation_attrs["tokens.completion"] = (
usage.responseTokenCount or 0
)
if hasattr(usage, "totalTokenCount"):
operation_attrs["tokens.total"] = usage.totalTokenCount or 0
# Get output text and modality from service state
text = getattr(self, "_bot_text_buffer", "")
audio_text = getattr(self, "_llm_output_buffer", "")
if text:
# TEXT modality
operation_attrs["output"] = text
operation_attrs["output_modality"] = "TEXT"
elif audio_text:
# AUDIO modality
operation_attrs["output"] = audio_text
operation_attrs["output_modality"] = "AUDIO"
# Add turn completion status
if (
evt
and hasattr(evt, "serverContent")
and evt.serverContent.turnComplete
):
operation_attrs["turn_complete"] = True
# Add all attributes to the span
add_gemini_live_span_attributes(
span=current_span,
service_name=service_class_name,
model=model_name,
operation_name=operation,
voice_id=voice_id,
language=language_code,
modalities=modalities,
settings=settings,
**operation_attrs,
)
# For llm_response operation, also handle token usage metrics
if operation == "llm_response" and hasattr(self, "start_llm_usage_metrics"):
evt = args[0] if args else None
if evt and hasattr(evt, "usageMetadata") and evt.usageMetadata:
usage = evt.usageMetadata
# Create LLMTokenUsage object
from pipecat.metrics.metrics import LLMTokenUsage
tokens = LLMTokenUsage(
prompt_tokens=usage.promptTokenCount or 0,
completion_tokens=usage.responseTokenCount or 0,
total_tokens=usage.totalTokenCount or 0,
)
_add_token_usage_to_span(current_span, tokens)
# Capture TTFB metric if available
ttfb = getattr(getattr(self, "_metrics", None), "ttfb", None)
if ttfb is not None:
current_span.set_attribute("metrics.ttfb", ttfb)
# Run the original function
result = await func(self, *args, **kwargs)
return result
except Exception as e:
current_span.record_exception(e)
current_span.set_status(trace.Status(trace.StatusCode.ERROR, str(e)))
raise
except Exception as e:
logging.error(f"Error in Gemini Live tracing (continuing without tracing): {e}")
# If tracing fails, fall back to the original function
return await func(self, *args, **kwargs)
return wrapper
return decorator
def traced_openai_realtime(operation: str) -> Callable:
"""Traces OpenAI Realtime service methods with operation-specific attributes.
This decorator automatically captures relevant information based on the operation type:
- llm_setup: Session configuration and tools
- llm_request: Context and input messages
- llm_response: Usage metadata, output, and function calls
Args:
operation: The operation name (matches the event type being handled)
Returns:
Wrapped method with OpenAI Realtime specific tracing.
"""
if not is_tracing_available():
return _noop_decorator
def decorator(func):
@functools.wraps(func)
async def wrapper(self, *args, **kwargs):
try:
if not is_tracing_available():
return await func(self, *args, **kwargs)
service_class_name = self.__class__.__name__
span_name = f"{operation}"
# Get the parent context - turn context if available, otherwise service context
turn_context = get_current_turn_context()
parent_context = turn_context or _get_parent_service_context(self)
# Create a new span as child of the turn span or service span
tracer = trace.get_tracer("pipecat")
with tracer.start_as_current_span(
span_name, context=parent_context
) as current_span:
try:
# Base service attributes
model_name = getattr(
self, "model_name", getattr(self, "_model_name", "unknown")
)
# Operation-specific attribute collection
operation_attrs = {}
if operation == "llm_setup":
# Capture session properties and tools
session_properties = getattr(self, "_session_properties", None)
if session_properties:
try:
# Convert to dict for easier processing
if hasattr(session_properties, "model_dump"):
props_dict = session_properties.model_dump()
elif hasattr(session_properties, "__dict__"):
props_dict = session_properties.__dict__
else:
props_dict = {}
operation_attrs["session_properties"] = props_dict
# Extract tools if available
tools = props_dict.get("tools")
if tools:
operation_attrs["tools"] = tools
try:
operation_attrs["tools_serialized"] = json.dumps(tools)
except Exception as e:
logging.warning(f"Error serializing OpenAI tools: {e}")
# Extract instructions
instructions = props_dict.get("instructions")
if instructions:
operation_attrs["instructions"] = instructions[:500]
except Exception as e:
logging.warning(f"Error processing session properties: {e}")
# Also check context for tools
if hasattr(self, "_context") and self._context:
try:
context_tools = getattr(self._context, "tools", None)
if context_tools and not operation_attrs.get("tools"):
operation_attrs["tools"] = context_tools
operation_attrs["tools_serialized"] = json.dumps(
context_tools
)
except Exception as e:
logging.warning(f"Error extracting context tools: {e}")
elif operation == "llm_request":
# Capture context messages being sent
if hasattr(self, "_context") and self._context:
try:
messages = self._context.get_messages_for_logging()
if messages:
operation_attrs["context_messages"] = json.dumps(messages)
except Exception as e:
logging.warning(f"Error getting context messages: {e}")
elif operation == "llm_response" and args:
# Extract usage and response metadata
evt = args[0] if args else None
if evt and hasattr(evt, "response"):
response = evt.response
# Token usage - basic attributes for span visibility
if hasattr(response, "usage"):
usage = response.usage
if hasattr(usage, "input_tokens"):
operation_attrs["tokens.prompt"] = usage.input_tokens
if hasattr(usage, "output_tokens"):
operation_attrs["tokens.completion"] = usage.output_tokens
if hasattr(usage, "total_tokens"):
operation_attrs["tokens.total"] = usage.total_tokens
# Response status and metadata
if hasattr(response, "status"):
operation_attrs["response.status"] = response.status
if hasattr(response, "id"):
operation_attrs["response.id"] = response.id
# Output items and extract transcript for output field
if hasattr(response, "output") and response.output:
operation_attrs["response.output_items"] = len(response.output)
# Extract assistant transcript and function calls
assistant_transcript = ""
function_calls = []
for item in response.output:
if (
hasattr(item, "content")
and item.content
and hasattr(item, "role")
and item.role == "assistant"
):
for content in item.content:
if (
hasattr(content, "transcript")
and content.transcript
):
assistant_transcript += content.transcript + " "
elif hasattr(item, "type") and item.type == "function_call":
function_call_info = {
"name": getattr(item, "name", "unknown"),
"call_id": getattr(item, "call_id", "unknown"),
}
if hasattr(item, "arguments"):
args_str = item.arguments
if len(args_str) > 500:
args_str = args_str[:500] + "..."
function_call_info["arguments"] = args_str
function_calls.append(function_call_info)
if assistant_transcript.strip():
operation_attrs["output"] = assistant_transcript.strip()
if function_calls:
operation_attrs["function_calls"] = function_calls
operation_attrs["function_calls.count"] = len(
function_calls
)
all_names = [call["name"] for call in function_calls]
operation_attrs["function_calls.all_names"] = ",".join(
all_names
)
# Add all attributes to the span
add_openai_realtime_span_attributes(
span=current_span,
service_name=service_class_name,
model=model_name,
operation_name=operation,
**operation_attrs,
)
# For llm_response operation, also handle token usage metrics
if operation == "llm_response" and hasattr(self, "start_llm_usage_metrics"):
evt = args[0] if args else None
if evt and hasattr(evt, "response") and hasattr(evt.response, "usage"):
usage = evt.response.usage
# Create LLMTokenUsage object
from pipecat.metrics.metrics import LLMTokenUsage
tokens = LLMTokenUsage(
prompt_tokens=getattr(usage, "input_tokens", 0),
completion_tokens=getattr(usage, "output_tokens", 0),
total_tokens=getattr(usage, "total_tokens", 0),
)
_add_token_usage_to_span(current_span, tokens)
# Capture TTFB metric if available
ttfb = getattr(getattr(self, "_metrics", None), "ttfb", None)
if ttfb is not None:
current_span.set_attribute("metrics.ttfb", ttfb)
# Run the original function
result = await func(self, *args, **kwargs)
return result
except Exception as e:
current_span.record_exception(e)
current_span.set_status(trace.Status(trace.StatusCode.ERROR, str(e)))
raise
except Exception as e:
logging.error(f"Error in OpenAI Realtime tracing (continuing without tracing): {e}")
# If tracing fails, fall back to the original function
return await func(self, *args, **kwargs)
return wrapper
return decorator