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

refactor: use direct attribute access for typed stream events

Browse Source 
Replace getattr() calls with direct attribute access and isinstance()
checks on the strongly-typed OpenAI SDK event models.
This commit is contained in:
Paul Kompfner
2026-03-19 14:19:10 -04:00
parent a3431d3b01
commit 0533ea7b7f
1 changed files with 17 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files

36
src/pipecat/services/openai/responses/llm.py
View File

@@ -18,6 +18,7 @@ from openai.types.responses import (
ResponseCompletedEvent, ResponseCompletedEvent,
ResponseFunctionCallArgumentsDeltaEvent, ResponseFunctionCallArgumentsDeltaEvent,
ResponseFunctionCallArgumentsDoneEvent, ResponseFunctionCallArgumentsDoneEvent,
ResponseFunctionToolCall,
ResponseOutputItemAddedEvent, ResponseOutputItemAddedEvent,
ResponseOutputItemDoneEvent, ResponseOutputItemDoneEvent,
ResponseStreamEvent, ResponseStreamEvent,
@@ -251,11 +252,11 @@ class OpenAIResponsesLLMService(LLMService):
elif isinstance(event, ResponseOutputItemAddedEvent): elif isinstance(event, ResponseOutputItemAddedEvent):
await self.stop_ttfb_metrics() await self.stop_ttfb_metrics()
item = event.item item = event.item
if getattr(item, "type", None) == "function_call": if isinstance(item, ResponseFunctionToolCall):
item_id = getattr(item, "id", "") or "" item_id = item.id or ""
function_calls[item_id] = { function_calls[item_id] = {
"name": getattr(item, "name", ""), "name": item.name,
"call_id": getattr(item, "call_id", ""), "call_id": item.call_id,
"arguments": "", "arguments": "",
} }
current_arguments[item_id] = "" current_arguments[item_id] = ""
@@ -272,29 +273,26 @@ class OpenAIResponsesLLMService(LLMService):
elif isinstance(event, ResponseOutputItemDoneEvent): elif isinstance(event, ResponseOutputItemDoneEvent):
item = event.item item = event.item
if getattr(item, "type", None) == "function_call": if isinstance(item, ResponseFunctionToolCall):
item_id = getattr(item, "id", "") or "" item_id = item.id or ""
if item_id in function_calls: if item_id in function_calls:
function_calls[item_id]["name"] = getattr(item, "name", "") function_calls[item_id]["name"] = item.name
function_calls[item_id]["call_id"] = getattr(item, "call_id", "") function_calls[item_id]["call_id"] = item.call_id
function_calls[item_id]["arguments"] = getattr(item, "arguments", "") function_calls[item_id]["arguments"] = item.arguments
elif isinstance(event, ResponseCompletedEvent): elif isinstance(event, ResponseCompletedEvent):
response = event.response response = event.response
usage = getattr(response, "usage", None) if response.usage:
if usage:
tokens = LLMTokenUsage( tokens = LLMTokenUsage(
prompt_tokens=getattr(usage, "input_tokens", 0), prompt_tokens=response.usage.input_tokens,
completion_tokens=getattr(usage, "output_tokens", 0), completion_tokens=response.usage.output_tokens,
total_tokens=getattr(usage, "total_tokens", 0), total_tokens=response.usage.total_tokens,
) )
await self.start_llm_usage_metrics(tokens) await self.start_llm_usage_metrics(tokens)
model = getattr(response, "model", None) # This field is used by @traced_llm for more detailed
if model: # model name in tracing spans
# This field is used by @traced_llm for more detailed self._full_model_name = response.model
# model name in tracing spans
self._full_model_name = model
# Process any function calls # Process any function calls
if function_calls: if function_calls: