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

Add tracking to OpenAI Realtime

Browse Source
This commit is contained in:
Mark Backman
2025-05-29 22:35:06 -04:00
parent ec39e794d3
commit dd1f7d0875
3 changed files with 314 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

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

@@ -258,7 +258,7 @@ def add_llm_span_attributes(
span.set_attribute(key, value)
def add_gemini_live_span_attributes(
def add_multimodal_llm_span_attributes(
span: "Span",
service_name: str,
model: str,
@@ -361,3 +361,102 @@ def add_gemini_live_span_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)

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

@@ -24,8 +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_multimodal_llm_span_attributes,
add_openai_realtime_span_attributes,
add_stt_span_attributes,
add_tts_span_attributes,
)
@@ -480,7 +481,7 @@ def traced_llm(func: Optional[Callable] = None, *, name: Optional[str] = None) -
return decorator
def traced_gemini_live(operation: str) -> Callable:
def traced_multimodal_llm(operation: str) -> Callable:
"""Traces Gemini Live service methods with operation-specific attributes.
This decorator automatically captures relevant information based on the operation type:
@@ -626,32 +627,6 @@ def traced_gemini_live(operation: str) -> Callable:
f"Error extracting context system instructions: {e}"
)
elif operation == "input_transcription" and args:
# Extract input transcription
evt = args[0] if args else None
if (
evt
and hasattr(evt, "serverContent")
and evt.serverContent.inputTranscription
):
text = evt.serverContent.inputTranscription.text
if text:
operation_attrs["transcript"] = text
operation_attrs["is_input"] = True
elif operation == "output_transcription" and args:
# Extract output transcription
evt = args[0] if args else None
if (
evt
and hasattr(evt, "serverContent")
and evt.serverContent.outputTranscription
):
text = evt.serverContent.outputTranscription.text
if text:
operation_attrs["transcript"] = text
operation_attrs["is_input"] = False
elif operation == "tool_call" and args:
# Extract tool call information
evt = args[0] if args else None
@@ -738,7 +713,7 @@ def traced_gemini_live(operation: str) -> Callable:
operation_attrs["tokens.total"] = usage.totalTokenCount or 0
# Add all attributes to the span
add_gemini_live_span_attributes(
add_multimodal_llm_span_attributes(
span=current_span,
service_name=service_class_name,
model=model_name,
@@ -785,3 +760,199 @@ def traced_gemini_live(operation: str) -> Callable:
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:
- setup: Session configuration and tools
- transcription_completed: User transcription
- response_create: Context and input messages
- response_done: Usage metadata and output
- function_call: Function call information
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 for output field
assistant_transcript = ""
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 + " "
if assistant_transcript.strip():
operation_attrs["output"] = assistant_transcript.strip()
# 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_ms = getattr(getattr(self, "_metrics", None), "ttfb_ms", None)
if ttfb_ms is not None:
current_span.set_attribute("metrics.ttfb_ms", ttfb_ms)
# 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