Merge branch 'litellm_internal_staging' into pragnyan/fix-xiaomi-output-config

This commit is contained in:
Pragnyan Ramtha 2026-05-19 09:11:21 +05:30 • committed by GitHub
commit 414693488e
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
6 changed files with 388 additions and 16 deletions

View file

@ -2,7 +2,7 @@
Streaming iterator for transforming Responses API stream to Interactions API stream.
"""
from typing import Any, AsyncIterator, Dict, Iterator, Optional, cast
from typing import Any, AsyncIterator, Dict, Iterator, List, Optional, cast
from litellm.responses.streaming_iterator import (
BaseResponsesAPIStreamingIterator,
@ -15,6 +15,7 @@ from litellm.types.interactions import (
InteractionsAPIStreamingResponse,
)
from litellm.types.llms.openai import (
ContentPartAddedEvent,
OutputTextDeltaEvent,
ResponseCompletedEvent,
ResponseCreatedEvent,
@ -51,6 +52,7 @@ class LiteLLMResponsesInteractionsStreamingIterator:
self.collected_text = ""
self.sent_interaction_start = False
self.sent_content_start = False
self._pending_events: List[InteractionsAPIStreamingResponse] = []
def _transform_responses_chunk_to_interactions_chunk(
self,
@ -80,7 +82,49 @@ class LiteLLMResponsesInteractionsStreamingIterator:
)
self.collected_text += delta_text
# Send interaction.start if not sent
# Fallback: emit interaction.start, and queue content.start carrying this
# delta so the first token is preserved in the stream.
if not self.sent_interaction_start:
self.sent_interaction_start = True
self.sent_content_start = True
self._pending_events.append(
InteractionsAPIStreamingResponse(
event_type="content.start",
id=getattr(responses_chunk, "item_id", None),
object="content",
delta={"type": "text", "text": delta_text},
)
)
return InteractionsAPIStreamingResponse(
event_type="interaction.start",
id=getattr(responses_chunk, "item_id", None)
or f"interaction_{id(self)}",
object="interaction",
status="in_progress",
model=self.model,
)
# Fallback: emit content.start if ContentPartAddedEvent never arrived
if not self.sent_content_start:
self.sent_content_start = True
return InteractionsAPIStreamingResponse(
event_type="content.start",
id=getattr(responses_chunk, "item_id", None),
object="content",
delta={"type": "text", "text": delta_text},
)
# Normal path: emit content.delta with type field
return InteractionsAPIStreamingResponse(
event_type="content.delta",
id=getattr(responses_chunk, "item_id", None),
object="content",
delta={"type": "text", "text": delta_text},
)
# Handle ContentPartAddedEvent -> content.start (arrives before text deltas)
if isinstance(responses_chunk, ContentPartAddedEvent):
# Fallback: emit interaction.start if ResponseCreatedEvent never arrived
if not self.sent_interaction_start:
self.sent_interaction_start = True
return InteractionsAPIStreamingResponse(
@ -91,8 +135,6 @@ class LiteLLMResponsesInteractionsStreamingIterator:
status="in_progress",
model=self.model,
)
# Send content.start if not sent
if not self.sent_content_start:
self.sent_content_start = True
return InteractionsAPIStreamingResponse(
@ -101,14 +143,7 @@ class LiteLLMResponsesInteractionsStreamingIterator:
object="content",
delta={"type": "text", "text": ""},
)
# Send content.delta
return InteractionsAPIStreamingResponse(
event_type="content.delta",
id=getattr(responses_chunk, "item_id", None),
object="content",
delta={"text": delta_text},
)
return None
# Handle ResponseCreatedEvent or ResponseInProgressEvent -> interaction.start
if isinstance(responses_chunk, (ResponseCreatedEvent, ResponseInProgressEvent)):
@ -172,6 +207,10 @@ class LiteLLMResponsesInteractionsStreamingIterator:
delattr(self, "_pending_interaction_complete")
return pending
# Drain events queued from a prior chunk (e.g. content.start emitted alongside
# the interaction.start fallback for the first OutputTextDeltaEvent).
if self._pending_events:
return self._pending_events.pop(0)
# Use a loop instead of recursion to avoid stack overflow
sync_iterator = cast(
SyncResponsesAPIStreamingIterator, self.responses_stream_iterator
@ -237,6 +276,10 @@ class LiteLLMResponsesInteractionsStreamingIterator:
delattr(self, "_pending_interaction_complete")
return pending
# Drain events queued from a prior chunk (e.g. content.start emitted alongside
# the interaction.start fallback for the first OutputTextDeltaEvent).
if self._pending_events:
return self._pending_events.pop(0)
# Use a loop instead of recursion to avoid stack overflow
async_iterator = cast(
ResponsesAPIStreamingIterator, self.responses_stream_iterator

View file

@ -9,15 +9,24 @@ Covers credential leak prevention changes:
import os
import sys
from unittest.mock import patch
from unittest.mock import MagicMock, patch
import pytest
sys.path.insert(0, os.path.abspath("../../.."))
from litellm.interactions.litellm_responses_transformation.streaming_iterator import (
LiteLLMResponsesInteractionsStreamingIterator,
)
from litellm.llms.gemini.interactions.transformation import (
GoogleAIStudioInteractionsConfig,
)
from litellm.types.llms.openai import (
ContentPartAddedEvent,
OutputTextDeltaEvent,
ResponseCompletedEvent,
ResponseCreatedEvent,
)
from litellm.types.router import GenericLiteLLMParams
_PATCH_GET_API_KEY = "litellm.llms.gemini.common_utils.GeminiModelInfo.get_api_key"
@ -113,6 +122,186 @@ class TestGetCompleteUrl:
)
class TestStreamingIterator:
def _make_iterator(self) -> LiteLLMResponsesInteractionsStreamingIterator:
return LiteLLMResponsesInteractionsStreamingIterator(
model="gpt-5.4",
litellm_custom_stream_wrapper=MagicMock(),
request_input="hi",
optional_params={},
)
def _make_text_delta(
self, text: str, item_id: str = "item_1"
) -> OutputTextDeltaEvent:
event = MagicMock(spec=OutputTextDeltaEvent)
event.delta = text
event.item_id = item_id
return event
def _make_part_added(self, item_id: str = "item_1") -> ContentPartAddedEvent:
event = MagicMock(spec=ContentPartAddedEvent)
event.item_id = item_id
return event
def _make_response_created(self) -> ResponseCreatedEvent:
event = MagicMock(spec=ResponseCreatedEvent)
event.response = MagicMock(id="resp_123")
return event
def test_content_delta_includes_type_field(self):
"""content.delta events must carry delta.type='text' so the UI can display them."""
it = self._make_iterator()
it.sent_interaction_start = True
it.sent_content_start = True
chunk = it._transform_responses_chunk_to_interactions_chunk(
self._make_text_delta("Hello")
)
assert chunk is not None
assert chunk.event_type == "content.delta"
assert chunk.delta == {"type": "text", "text": "Hello"}
def test_response_part_added_emits_content_start(self):
"""ContentPartAddedEvent (arrives before text deltas) should emit content.start
so the first OutputTextDeltaEvent immediately emits content.delta without dropping text.
"""
it = self._make_iterator()
it.sent_interaction_start = True
chunk = it._transform_responses_chunk_to_interactions_chunk(
self._make_part_added()
)
assert chunk is not None
assert chunk.event_type == "content.start"
assert it.sent_content_start is True
def test_first_text_delta_not_dropped_when_part_added_seen(self):
"""After ContentPartAddedEvent, the first text delta must yield content.delta
(not content.start), preserving the token text."""
it = self._make_iterator()
it.sent_interaction_start = True
it._transform_responses_chunk_to_interactions_chunk(self._make_part_added())
chunk = it._transform_responses_chunk_to_interactions_chunk(
self._make_text_delta("Hello")
)
assert chunk is not None
assert chunk.event_type == "content.delta"
assert chunk.delta is not None
assert chunk.delta.get("text") == "Hello"
def test_part_added_emits_interaction_start_fallback_when_not_sent(self):
"""If ContentPartAddedEvent arrives before any ResponseCreatedEvent,
the iterator must emit interaction.start before content.start to honor
the documented event ordering contract."""
it = self._make_iterator()
chunk = it._transform_responses_chunk_to_interactions_chunk(
self._make_part_added(item_id="item_42")
)
assert chunk is not None
assert chunk.event_type == "interaction.start"
assert chunk.id == "item_42"
assert chunk.status == "in_progress"
assert chunk.model == "gpt-5.4"
assert it.sent_interaction_start is True
assert it.sent_content_start is False
def test_part_added_returns_none_when_already_started(self):
"""A second ContentPartAddedEvent (after content.start was already emitted)
should be a no-op so we don't re-emit content.start."""
it = self._make_iterator()
it.sent_interaction_start = True
it.sent_content_start = True
chunk = it._transform_responses_chunk_to_interactions_chunk(
self._make_part_added()
)
assert chunk is None
def test_part_added_without_item_id_falls_back_to_self_id(self):
"""When ContentPartAddedEvent has no item_id and we emit the interaction.start
fallback, the id must default to an interaction_<id(self)> string."""
it = self._make_iterator()
event = MagicMock(spec=ContentPartAddedEvent)
event.item_id = None
chunk = it._transform_responses_chunk_to_interactions_chunk(event)
assert chunk is not None
assert chunk.event_type == "interaction.start"
assert chunk.id == f"interaction_{id(it)}"
def test_first_text_delta_not_dropped_when_no_prior_start_events(self):
"""When OutputTextDeltaEvent arrives before any ResponseCreatedEvent or
ContentPartAddedEvent, the iterator must emit interaction.start *and*
immediately follow with a content.start that carries this delta's text,
so the first token is never silently dropped from the stream."""
events = [
self._make_text_delta("Hello"),
self._make_text_delta(" World"),
]
wrapper = MagicMock()
wrapper.__iter__ = lambda self: iter(events)
wrapper.__next__ = lambda self, _it=iter(events): next(_it)
it = LiteLLMResponsesInteractionsStreamingIterator(
model="gpt-5.4",
litellm_custom_stream_wrapper=wrapper,
request_input="hi",
optional_params={},
)
first = it._transform_responses_chunk_to_interactions_chunk(events[0])
assert first is not None
assert first.event_type == "interaction.start"
assert it.sent_interaction_start is True
assert it.sent_content_start is True
assert len(it._pending_events) == 1
pending = it._pending_events[0]
assert pending.event_type == "content.start"
assert pending.delta == {"type": "text", "text": "Hello"}
second = it._transform_responses_chunk_to_interactions_chunk(events[1])
assert second is not None
assert second.event_type == "content.delta"
assert second.delta == {"type": "text", "text": " World"}
class TestTransformRequest:
def test_stream_param_included_in_request_body(self, config):
"""When stream=True is in optional_params, the request body must include it
so the proxy forwards the SSE streaming flag to Google's backend."""
body = config.transform_request(
model="gemini-2.5-flash",
agent=None,
input="Hello",
optional_params={"stream": True},
litellm_params=GenericLiteLLMParams(api_key="test-key"),
headers={},
)
assert body.get("stream") is True
assert body.get("input") == "Hello"
def test_stream_false_not_included_when_absent(self, config):
body = config.transform_request(
model="gemini-2.5-flash",
agent=None,
input="Hello",
optional_params={},
litellm_params=GenericLiteLLMParams(api_key="test-key"),
headers={},
)
assert "stream" not in body
class TestInteractionOperationUrls:
"""Test that get/delete/cancel interaction URLs exclude API key."""

View file

@ -49,6 +49,7 @@ import { fetchAvailableModels, ModelGroup } from "../llm_calls/fetch_models";
import { makeOpenAIImageEditsRequest } from "../llm_calls/image_edits";
import { makeOpenAIImageGenerationRequest } from "../llm_calls/image_generation";
import { makeOpenAIResponsesRequest } from "../llm_calls/responses_api";
import { makeInteractionsRequest } from "../llm_calls/interactions_api";
import A2AMetrics from "./A2AMetrics";
import AdditionalModelSettings from "./AdditionalModelSettings";
import AudioRenderer from "./AudioRenderer";
@ -649,6 +650,7 @@ const ChatUI: React.FC<ChatUIProps> = ({
EndpointType.ANTHROPIC_MESSAGES,
EndpointType.EMBEDDINGS,
EndpointType.TRANSCRIPTION,
EndpointType.INTERACTIONS,
];
if (modelRequiredEndpoints.includes(endpointType as EndpointType) && !selectedModel) {
@ -914,6 +916,16 @@ const ChatUI: React.FC<ChatUIProps> = ({
customProxyBaseUrl || undefined,
);
}
} else if (endpointType === EndpointType.INTERACTIONS) {
await makeInteractionsRequest(
inputMessage,
(text, model) => updateTextUI("assistant", text, model),
selectedModel,
effectiveApiKey,
selectedTags,
signal,
customProxyBaseUrl || undefined,
);
}
}
@ -1241,10 +1253,11 @@ const ChatUI: React.FC<ChatUIProps> = ({
return true;
}
const optionEndpoint = getEndpointType(option.mode);
// Show chat models for responses/anthropic_messages endpoints as they are compatible
// Show chat models for responses/anthropic_messages/interactions endpoints as they are compatible
if (
endpointType === EndpointType.RESPONSES ||
endpointType === EndpointType.ANTHROPIC_MESSAGES
endpointType === EndpointType.ANTHROPIC_MESSAGES ||
endpointType === EndpointType.INTERACTIONS
) {
return optionEndpoint === endpointType || optionEndpoint === EndpointType.CHAT;
}
@ -2089,7 +2102,8 @@ const ChatUI: React.FC<ChatUIProps> = ({
endpointType === EndpointType.CHAT ||
endpointType === EndpointType.EMBEDDINGS ||
endpointType === EndpointType.RESPONSES ||
endpointType === EndpointType.ANTHROPIC_MESSAGES
endpointType === EndpointType.ANTHROPIC_MESSAGES ||
endpointType === EndpointType.INTERACTIONS
? "Type your message... (Shift+Enter for new line)"
: endpointType === EndpointType.A2A_AGENTS
? "Send a message to the A2A agent..."

View file

@ -45,4 +45,5 @@ export const ENDPOINT_OPTIONS = [
{ value: EndpointType.A2A_AGENTS, label: "/v1/a2a/message/send" },
{ value: EndpointType.MCP, label: "/mcp-rest/tools/call" },
{ value: EndpointType.REALTIME, label: "/v1/realtime" },
{ value: EndpointType.INTERACTIONS, label: "/v1beta/interactions" },
];

View file

@ -28,6 +28,7 @@ export enum EndpointType {
A2A_AGENTS = "a2a_agents",
MCP = "mcp",
REALTIME = "realtime",
INTERACTIONS = "interactions",
}
// Create a mapping between the model mode and the corresponding endpoint type

View file

@ -0,0 +1,124 @@
import NotificationManager from "@/components/molecules/notifications_manager";
import { getGlobalLitellmHeaderName, getProxyBaseUrl } from "@/components/networking";
export async function makeInteractionsRequest(
input: string,
updateUI: (text: string, model?: string) => void,
selectedModel: string,
accessToken: string,
tags?: string[],
signal?: AbortSignal,
customBaseUrl?: string,
previousInteractionId?: string,
): Promise<void> {
if (!accessToken) {
throw new Error("Virtual Key is required");
}
const isLocal = process.env.NODE_ENV === "development";
if (isLocal !== true) {
console.log = function () {};
}
const proxyBaseUrl = customBaseUrl || getProxyBaseUrl();
const normalizedBaseUrl = proxyBaseUrl.endsWith("/") ? proxyBaseUrl.slice(0, -1) : proxyBaseUrl;
const requestUrl = `${normalizedBaseUrl}/v1beta/interactions`;
const headers: Record<string, string> = {
"Content-Type": "application/json",
[getGlobalLitellmHeaderName()]: `Bearer ${accessToken}`,
};
if (tags && tags.length > 0) {
headers["x-litellm-tags"] = tags.join(",");
}
const body: Record<string, unknown> = {
model: selectedModel,
input,
stream: true,
};
if (previousInteractionId) {
body.previous_interaction_id = previousInteractionId;
}
try {
const response = await fetch(requestUrl, {
method: "POST",
headers,
body: JSON.stringify(body),
signal,
});
if (!response.ok) {
const errorText = await response.text();
throw new Error(errorText || `Request failed with status ${response.status}`);
}
if (!response.body) {
throw new Error("No response body received");
}
const reader = response.body.getReader();
const decoder = new TextDecoder();
let responseModel: string | undefined;
let buffer = "";
while (true) {
const { done, value } = await reader.read();
if (done) break;
buffer += decoder.decode(value, { stream: true });
// SSE lines are separated by double newlines; split on single newlines and
// look for "data: " prefixed lines.
const lines = buffer.split("\n");
// Keep the last (potentially incomplete) line in the buffer
buffer = lines.pop() ?? "";
for (const line of lines) {
const trimmed = line.trim();
if (!trimmed.startsWith("data:")) continue;
const jsonStr = trimmed.slice("data:".length).trim();
if (!jsonStr || jsonStr === "[DONE]") continue;
let event: Record<string, unknown>;
try {
event = JSON.parse(jsonStr);
} catch {
continue;
}
const eventType = event.event_type as string | undefined;
if (eventType === "interaction.start" || eventType === "interaction.complete") {
// Capture model from either the native Gemini shape (nested under
// `interaction`) or the bridge shape (top-level `model` field).
const interaction = event.interaction as Record<string, unknown> | undefined;
if (typeof interaction?.model === "string" && interaction.model) {
responseModel = interaction.model;
} else if (typeof event.model === "string" && event.model) {
responseModel = event.model;
}
} else if (eventType === "content.delta" || eventType === "content.start") {
const delta = event.delta as Record<string, unknown> | undefined;
// Accept both native Gemini format {"type":"text","text":"..."} and bridge
// format {"text":"..."} (no type discriminator)
if (typeof delta?.text === "string" && delta.text) {
updateUI(delta.text, responseModel ?? selectedModel);
}
}
// content.start, content.stop, interaction.status_update — no UI action needed
}
}
} catch (error: unknown) {
if (signal?.aborted) {
console.log("Interactions request was cancelled");
throw error;
}
NotificationManager.fromBackend(
`Error occurred while making Interactions API request. Error: ${error}`,
);
throw error;
}
}