fix(realtime): allow null transcripts in stream logging payloads (#29625)

Allow realtime event transcript fields to be nullable so GA conversation.item payloads with transcript=null don't fail logging normalization and suppress success callbacks.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
milan-berri 2026-06-04 00:59:18 +03:00 committed by GitHub
parent 97ba7e1a30
commit 5ee526d78e
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
2 changed files with 40 additions and 2 deletions

View File

@ -1892,7 +1892,7 @@ class OpenAIRealtimeStreamResponseOutputItemContent(TypedDict, total=False):
"""The ID of the previous conversation item for reference"""
text: str
"""The text content, used for 'input_text' / 'text' / 'output_text' content types"""
transcript: str
transcript: Optional[str]
"""The transcript content, used for 'input_audio' / 'audio' content types"""
type: Literal[
"input_audio",
@ -1998,7 +1998,7 @@ class OpenAIRealtimeResponseContentPart(TypedDict, total=False):
text: str
"""The text content, if type is 'text' or 'output_text'"""
transcript: str
transcript: Optional[str]
"""The transcript content, if type is 'audio' or 'output_audio'"""
type: Union[

View File

@ -12,6 +12,7 @@ from pydantic import BaseModel
import litellm
from litellm.cost_calculator import (
RealtimeAPITokenUsageProcessor,
completion_cost,
cost_per_token,
handle_realtime_stream_cost_calculation,
@ -385,6 +386,43 @@ def test_handle_realtime_stream_cost_calculation():
assert cost == 0.0 # No usage, no cost
def test_realtime_logging_object_allows_null_transcript_in_conversation_item_added():
results: OpenAIRealtimeStreamList = [
{
"type": "conversation.item.added",
"event_id": "event_added",
"item": {
"id": "item_123",
"type": "message",
"role": "assistant",
"status": "in_progress",
"content": [{"type": "audio", "transcript": None}],
},
},
{
"type": "response.done",
"event_id": "event_done",
"response": {
"id": "resp_123",
"object": "realtime.response",
"status": "completed",
"usage": {"input_tokens": 11, "output_tokens": 7, "total_tokens": 18},
},
},
]
usage = RealtimeAPITokenUsageProcessor.collect_and_combine_usage_from_realtime_stream_results(
results=results
)
logging_result = RealtimeAPITokenUsageProcessor.create_logging_realtime_object(
usage=usage,
results=results,
)
assert logging_result.usage.total_tokens == 18
assert logging_result.results[0]["item"]["content"][0]["transcript"] is None
def test_custom_pricing_with_router_model_id():
from litellm import Router