mirror of
https://github.com/BerriAI/litellm.git
synced 2026-09-11 22:51:28 +00:00
Merge pull request #15432 from BerriAI/litellm_dev_10_10_2025_p2
Responses API - enable calling anthropic/gemini models in Responses API streaming in openai ruby sdk + DB - sanity check pending migrations before startup
This commit is contained in:
Krish Dholakia
GitHub
commit
a4f14adefe
10 changed files with 743 additions and 237 deletions
|
|
@ -131,7 +131,9 @@ class ProxyExtrasDBManager:
|
|||
)
|
||||
|
||||
@staticmethod
|
||||
def _resolve_all_migrations(migrations_dir: str, schema_path: str):
|
||||
def _resolve_all_migrations(
|
||||
migrations_dir: str, schema_path: str, mark_all_applied: bool = True
|
||||
):
|
||||
"""
|
||||
1. Compare the current database state to schema.prisma and generate a migration for the diff.
|
||||
2. Run prisma migrate deploy to apply any pending migrations.
|
||||
|
|
@ -210,6 +212,8 @@ class ProxyExtrasDBManager:
|
|||
logger.warning("Migration diff application timed out.")
|
||||
|
||||
# 3. Mark all migrations as applied
|
||||
if not mark_all_applied:
|
||||
return
|
||||
migration_names = ProxyExtrasDBManager._get_migration_names(migrations_dir)
|
||||
logger.info(f"Resolving {len(migration_names)} migrations")
|
||||
for migration_name in migration_names:
|
||||
|
|
@ -263,6 +267,13 @@ class ProxyExtrasDBManager:
|
|||
logger.info(f"prisma migrate deploy stdout: {result.stdout}")
|
||||
|
||||
logger.info("prisma migrate deploy completed")
|
||||
|
||||
# Run sanity check to ensure DB matches schema
|
||||
logger.info("Running post-migration sanity check...")
|
||||
ProxyExtrasDBManager._resolve_all_migrations(
|
||||
migrations_dir, schema_path, mark_all_applied=False
|
||||
)
|
||||
logger.info("✅ Post-migration sanity check completed")
|
||||
return True
|
||||
except subprocess.CalledProcessError as e:
|
||||
logger.info(f"prisma db error: {e.stderr}, e: {e.stdout}")
|
||||
|
|
|
|||
|
|
@ -57,22 +57,22 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge):
|
|||
) -> Tuple[Optional[Any], int]:
|
||||
"""
|
||||
Handle raw dict response items from Responses API (e.g., GPT-5 Codex format).
|
||||
|
||||
|
||||
Args:
|
||||
item: Raw dict response item with 'type' field
|
||||
index: Current choice index
|
||||
|
||||
|
||||
Returns:
|
||||
Tuple of (Choice object or None, updated index)
|
||||
"""
|
||||
from litellm.types.utils import Choices, Message
|
||||
|
||||
item_type = item.get("type")
|
||||
|
||||
|
||||
# Ignore reasoning items for now
|
||||
if item_type == "reasoning":
|
||||
return None, index
|
||||
|
||||
|
||||
# Handle message items with output_text content
|
||||
if item_type == "message":
|
||||
content_list = item.get("content", [])
|
||||
|
|
@ -83,13 +83,11 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge):
|
|||
response_text = content_item.get("text", "")
|
||||
msg = Message(
|
||||
role=item.get("role", "assistant"),
|
||||
content=response_text if response_text else ""
|
||||
)
|
||||
choice = Choices(
|
||||
message=msg, finish_reason="stop", index=index
|
||||
content=response_text if response_text else "",
|
||||
)
|
||||
choice = Choices(message=msg, finish_reason="stop", index=index)
|
||||
return choice, index + 1
|
||||
|
||||
|
||||
# Unknown or unsupported type
|
||||
return None, index
|
||||
|
||||
|
|
@ -294,8 +292,8 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge):
|
|||
|
||||
if isinstance(item, ResponseReasoningItem):
|
||||
|
||||
for content in item.summary:
|
||||
response_text = getattr(content, "text", "")
|
||||
for summary_item in item.summary:
|
||||
response_text = getattr(summary_item, "text", "")
|
||||
reasoning_content = response_text if response_text else ""
|
||||
|
||||
elif isinstance(item, ResponseOutputMessage):
|
||||
|
|
@ -340,7 +338,9 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge):
|
|||
index += 1
|
||||
elif isinstance(item, dict):
|
||||
# Handle raw dict responses (e.g., from GPT-5 Codex)
|
||||
choice, index = self._handle_raw_dict_response_item(item=item, index=index)
|
||||
choice, index = self._handle_raw_dict_response_item(
|
||||
item=item, index=index
|
||||
)
|
||||
if choice is not None:
|
||||
choices.append(choice)
|
||||
else:
|
||||
|
|
|
|||
|
|
@ -1,4 +1,4 @@
|
|||
model_list:
|
||||
- model_name: gpt-5-codex
|
||||
litellm_params:
|
||||
model: gpt-5-codex
|
||||
model: gpt-5-codex
|
||||
|
|
|
|||
|
|
@ -58,7 +58,7 @@ class LiteLLMCompletionTransformationHandler:
|
|||
responses_api_request=responses_api_request,
|
||||
**kwargs,
|
||||
)
|
||||
|
||||
|
||||
completion_args = {}
|
||||
completion_args.update(kwargs)
|
||||
completion_args.update(litellm_completion_request)
|
||||
|
|
@ -83,6 +83,7 @@ class LiteLLMCompletionTransformationHandler:
|
|||
|
||||
elif isinstance(litellm_completion_response, litellm.CustomStreamWrapper):
|
||||
return LiteLLMCompletionStreamingIterator(
|
||||
model=model,
|
||||
litellm_custom_stream_wrapper=litellm_completion_response,
|
||||
request_input=input,
|
||||
responses_api_request=responses_api_request,
|
||||
|
|
@ -106,7 +107,7 @@ class LiteLLMCompletionTransformationHandler:
|
|||
previous_response_id=previous_response_id,
|
||||
litellm_completion_request=litellm_completion_request,
|
||||
)
|
||||
|
||||
|
||||
acompletion_args = {}
|
||||
acompletion_args.update(kwargs)
|
||||
acompletion_args.update(litellm_completion_request)
|
||||
|
|
@ -130,9 +131,12 @@ class LiteLLMCompletionTransformationHandler:
|
|||
|
||||
elif isinstance(litellm_completion_response, litellm.CustomStreamWrapper):
|
||||
return LiteLLMCompletionStreamingIterator(
|
||||
model=litellm_completion_request.get("model") or "",
|
||||
litellm_custom_stream_wrapper=litellm_completion_response,
|
||||
request_input=request_input,
|
||||
responses_api_request=responses_api_request,
|
||||
custom_llm_provider=litellm_completion_request.get("custom_llm_provider"),
|
||||
custom_llm_provider=litellm_completion_request.get(
|
||||
"custom_llm_provider"
|
||||
),
|
||||
litellm_metadata=kwargs.get("litellm_metadata", {}),
|
||||
)
|
||||
|
|
|
|||
|
|
@ -1,4 +1,6 @@
|
|||
from typing import List, Optional, Union
|
||||
import time
|
||||
import uuid
|
||||
from typing import List, Optional, Union, cast
|
||||
|
||||
import litellm
|
||||
from litellm.main import stream_chunk_builder
|
||||
|
|
@ -8,11 +10,23 @@ from litellm.responses.litellm_completion_transformation.transformation import (
|
|||
from litellm.responses.streaming_iterator import ResponsesAPIStreamingIterator
|
||||
from litellm.responses.utils import ResponsesAPIRequestUtils
|
||||
from litellm.types.llms.openai import (
|
||||
PART_UNION_TYPES,
|
||||
BaseLiteLLMOpenAIResponseObject,
|
||||
ContentPartAddedEvent,
|
||||
ContentPartDoneEvent,
|
||||
ContentPartDonePartOutputText,
|
||||
ContentPartDonePartReasoningText,
|
||||
OutputItemAddedEvent,
|
||||
OutputItemDoneEvent,
|
||||
OutputTextDeltaEvent,
|
||||
OutputTextDoneEvent,
|
||||
ReasoningSummaryTextDeltaEvent,
|
||||
ResponseCompletedEvent,
|
||||
ResponseCreatedEvent,
|
||||
ResponseInProgressEvent,
|
||||
ResponseInputParam,
|
||||
ResponsesAPIOptionalRequestParams,
|
||||
ResponsesAPIResponse,
|
||||
ResponsesAPIStreamEvents,
|
||||
ResponsesAPIStreamingResponse,
|
||||
)
|
||||
|
|
@ -32,12 +46,14 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator):
|
|||
|
||||
def __init__(
|
||||
self,
|
||||
model: str,
|
||||
litellm_custom_stream_wrapper: litellm.CustomStreamWrapper,
|
||||
request_input: Union[str, ResponseInputParam],
|
||||
responses_api_request: ResponsesAPIOptionalRequestParams,
|
||||
custom_llm_provider: Optional[str] = None,
|
||||
litellm_metadata: Optional[dict] = None,
|
||||
):
|
||||
self.model: str = model
|
||||
self.litellm_custom_stream_wrapper: litellm.CustomStreamWrapper = (
|
||||
litellm_custom_stream_wrapper
|
||||
)
|
||||
|
|
@ -50,14 +66,276 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator):
|
|||
self.collected_chat_completion_chunks: List[ModelResponseStream] = []
|
||||
self.finished: bool = False
|
||||
self.litellm_logging_obj = litellm_custom_stream_wrapper.logging_obj
|
||||
self.sent_response_created_event: bool = False
|
||||
self.sent_response_in_progress_event: bool = False
|
||||
self.sent_output_item_added_event: bool = False
|
||||
self.sent_content_part_added_event: bool = False
|
||||
self.sent_output_text_done_event: bool = False
|
||||
self.sent_output_content_part_done_event: bool = False
|
||||
self.sent_output_item_done_event: bool = False
|
||||
self.litellm_model_response: Optional[
|
||||
Union[ModelResponse, TextCompletionResponse]
|
||||
] = None
|
||||
self.final_text: str = ""
|
||||
|
||||
def _default_response_created_event_data(self) -> dict:
|
||||
response_created_event_data = {
|
||||
"id": f"resp_{str(uuid.uuid4())}",
|
||||
"object": "response",
|
||||
"created_at": int(time.time()),
|
||||
"status": "in_progress",
|
||||
"error": None,
|
||||
"incomplete_details": None,
|
||||
"instructions": self.request_input,
|
||||
"max_output_tokens": None,
|
||||
"model": self.model,
|
||||
"output": [],
|
||||
"parallel_tool_calls": True,
|
||||
"previous_response_id": None,
|
||||
"reasoning": {"effort": None, "summary": None},
|
||||
"store": True,
|
||||
}
|
||||
if "temperature" in self.responses_api_request:
|
||||
response_created_event_data["temperature"] = self.responses_api_request[
|
||||
"temperature"
|
||||
]
|
||||
if "text" in self.responses_api_request:
|
||||
response_created_event_data["text"] = self.responses_api_request["text"]
|
||||
if "tool_choice" in self.responses_api_request:
|
||||
response_created_event_data["tool_choice"] = self.responses_api_request[
|
||||
"tool_choice"
|
||||
]
|
||||
else:
|
||||
response_created_event_data["tool_choice"] = "auto"
|
||||
if "tools" in self.responses_api_request:
|
||||
response_created_event_data["tools"] = self.responses_api_request["tools"]
|
||||
else:
|
||||
response_created_event_data["tools"] = []
|
||||
if "top_p" in self.responses_api_request:
|
||||
response_created_event_data["top_p"] = self.responses_api_request["top_p"]
|
||||
else:
|
||||
response_created_event_data["top_p"] = 1.0
|
||||
if "truncation" in self.responses_api_request:
|
||||
response_created_event_data["truncation"] = self.responses_api_request[
|
||||
"truncation"
|
||||
]
|
||||
if "usage" in self.responses_api_request:
|
||||
response_created_event_data["usage"] = self.responses_api_request["usage"]
|
||||
if "user" in self.responses_api_request:
|
||||
response_created_event_data["user"] = self.responses_api_request["user"]
|
||||
if "metadata" in self.responses_api_request:
|
||||
response_created_event_data["metadata"] = self.responses_api_request[
|
||||
"metadata"
|
||||
]
|
||||
return response_created_event_data
|
||||
|
||||
def create_response_created_event(self) -> ResponseCreatedEvent:
|
||||
"""
|
||||
data: {"type":"response.created","response":{"id":"resp_67c9fdcecf488190bdd9a0409de3a1ec07b8b0ad4e5eb654","object":"response","created_at":1741290958,"status":"in_progress","error":null,"incomplete_details":null,"instructions":"You are a helpful assistant.","max_output_tokens":null,"model":"gpt-4.1-2025-04-14","output":[],"parallel_tool_calls":true,"previous_response_id":null,"reasoning":{"effort":null,"summary":null},"store":true,"temperature":1.0,"text":{"format":{"type":"text"}},"tool_choice":"auto","tools":[],"top_p":1.0,"truncation":"disabled","usage":null,"user":null,"metadata":{}}}
|
||||
|
||||
"""
|
||||
response_created_event_data = self._default_response_created_event_data()
|
||||
return ResponseCreatedEvent(
|
||||
type=ResponsesAPIStreamEvents.RESPONSE_CREATED,
|
||||
response=ResponsesAPIResponse(**response_created_event_data),
|
||||
)
|
||||
|
||||
def create_response_in_progress_event(self) -> ResponseInProgressEvent:
|
||||
response_in_progress_event_data = self._default_response_created_event_data()
|
||||
response_in_progress_event_data["status"] = "in_progress"
|
||||
return ResponseInProgressEvent(
|
||||
type=ResponsesAPIStreamEvents.RESPONSE_IN_PROGRESS,
|
||||
response=ResponsesAPIResponse(**response_in_progress_event_data),
|
||||
)
|
||||
|
||||
def create_output_item_added_event(self) -> OutputItemAddedEvent:
|
||||
return OutputItemAddedEvent(
|
||||
type=ResponsesAPIStreamEvents.OUTPUT_ITEM_ADDED,
|
||||
output_index=0,
|
||||
item=BaseLiteLLMOpenAIResponseObject(
|
||||
**{
|
||||
"id": f"msg_{str(uuid.uuid4())}",
|
||||
"type": "message",
|
||||
"status": "in_progress",
|
||||
"role": "assistant",
|
||||
"content": [],
|
||||
}
|
||||
),
|
||||
)
|
||||
|
||||
def create_content_part_added_event(self) -> ContentPartAddedEvent:
|
||||
return ContentPartAddedEvent(
|
||||
type=ResponsesAPIStreamEvents.CONTENT_PART_ADDED,
|
||||
item_id=f"msg_{str(uuid.uuid4())}",
|
||||
output_index=0,
|
||||
content_index=0,
|
||||
part=BaseLiteLLMOpenAIResponseObject(
|
||||
**{"type": "output_text", "text": "", "annotations": []}
|
||||
),
|
||||
)
|
||||
|
||||
def create_litellm_model_response(
|
||||
self,
|
||||
) -> Optional[ModelResponse]:
|
||||
return cast(
|
||||
Optional[ModelResponse],
|
||||
stream_chunk_builder(
|
||||
chunks=self.collected_chat_completion_chunks,
|
||||
logging_obj=self.litellm_logging_obj,
|
||||
),
|
||||
)
|
||||
|
||||
def create_output_text_done_event(
|
||||
self, litellm_complete_object: ModelResponse
|
||||
) -> OutputTextDoneEvent:
|
||||
return OutputTextDoneEvent(
|
||||
type=ResponsesAPIStreamEvents.OUTPUT_TEXT_DONE,
|
||||
item_id=f"msg_{str(uuid.uuid4())}",
|
||||
output_index=0,
|
||||
content_index=0,
|
||||
text=getattr(litellm_complete_object.choices[0].message, "content", "") # type: ignore
|
||||
or "",
|
||||
)
|
||||
|
||||
def create_output_content_part_done_event(
|
||||
self, litellm_complete_object: ModelResponse
|
||||
) -> ContentPartDoneEvent:
|
||||
|
||||
text = getattr(litellm_complete_object.choices[0].message, "content", "") or "" # type: ignore
|
||||
reasoning_content = getattr(litellm_complete_object.choices[0].message, "reasoning_content", "") or "" # type: ignore
|
||||
|
||||
part: Optional[PART_UNION_TYPES] = None
|
||||
if reasoning_content:
|
||||
part = ContentPartDonePartReasoningText(
|
||||
type="reasoning_text",
|
||||
reasoning=reasoning_content,
|
||||
)
|
||||
|
||||
else:
|
||||
part = ContentPartDonePartOutputText(
|
||||
type="output_text",
|
||||
text=text,
|
||||
annotations=[],
|
||||
logprobs=None,
|
||||
)
|
||||
|
||||
return ContentPartDoneEvent(
|
||||
type=ResponsesAPIStreamEvents.CONTENT_PART_DONE,
|
||||
item_id=f"msg_{str(uuid.uuid4())}",
|
||||
output_index=0,
|
||||
content_index=0,
|
||||
part=part,
|
||||
)
|
||||
|
||||
def create_output_item_done_event(
|
||||
self, litellm_complete_object: ModelResponse
|
||||
) -> OutputItemDoneEvent:
|
||||
text = self.litellm_model_response.choices[0].message.content or "" # type: ignore
|
||||
return OutputItemDoneEvent(
|
||||
type=ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE,
|
||||
output_index=0,
|
||||
sequence_number=1,
|
||||
item=BaseLiteLLMOpenAIResponseObject(
|
||||
**{
|
||||
"id": f"msg_{str(uuid.uuid4())}",
|
||||
"status": "completed",
|
||||
"type": "message",
|
||||
"role": "assistant",
|
||||
"content": [
|
||||
{
|
||||
"type": "output_text",
|
||||
"text": text,
|
||||
"annotations": [],
|
||||
}
|
||||
],
|
||||
}
|
||||
),
|
||||
)
|
||||
|
||||
def return_default_done_events(
|
||||
self, litellm_complete_object: ModelResponse
|
||||
) -> Optional[BaseLiteLLMOpenAIResponseObject]:
|
||||
if self.sent_output_text_done_event is False:
|
||||
self.sent_output_text_done_event = True
|
||||
return self.create_output_text_done_event(litellm_complete_object)
|
||||
if self.sent_output_content_part_done_event is False:
|
||||
self.sent_output_content_part_done_event = True
|
||||
return self.create_output_content_part_done_event(litellm_complete_object)
|
||||
if self.sent_output_item_done_event is False:
|
||||
self.sent_output_item_done_event = True
|
||||
return self.create_output_item_done_event(litellm_complete_object)
|
||||
return None
|
||||
|
||||
def return_default_initial_events(
|
||||
self,
|
||||
) -> Optional[BaseLiteLLMOpenAIResponseObject]:
|
||||
if self.sent_response_created_event is False:
|
||||
self.sent_response_created_event = True
|
||||
return self.create_response_created_event()
|
||||
elif self.sent_response_in_progress_event is False:
|
||||
self.sent_response_in_progress_event = True
|
||||
return self.create_response_in_progress_event()
|
||||
elif self.sent_output_item_added_event is False:
|
||||
self.sent_output_item_added_event = True
|
||||
return self.create_output_item_added_event()
|
||||
elif self.sent_content_part_added_event is False:
|
||||
self.sent_content_part_added_event = True
|
||||
return self.create_content_part_added_event()
|
||||
return None
|
||||
|
||||
def is_stream_finished(self) -> bool:
|
||||
if (
|
||||
self.sent_output_text_done_event is True
|
||||
and self.sent_output_content_part_done_event is True
|
||||
and self.sent_output_item_done_event is True
|
||||
):
|
||||
return True
|
||||
return False
|
||||
|
||||
def common_done_event_logic(
|
||||
self, sync_mode: bool = True
|
||||
) -> BaseLiteLLMOpenAIResponseObject:
|
||||
if not self.litellm_model_response or isinstance(
|
||||
self.litellm_model_response, TextCompletionResponse
|
||||
):
|
||||
self.litellm_model_response = self.create_litellm_model_response()
|
||||
if self.litellm_model_response:
|
||||
done_event = self.return_default_done_events(self.litellm_model_response)
|
||||
if done_event:
|
||||
return done_event
|
||||
else:
|
||||
if sync_mode:
|
||||
raise StopIteration
|
||||
else:
|
||||
raise StopAsyncIteration
|
||||
|
||||
self.finished = self.is_stream_finished()
|
||||
response_completed_event = self._emit_response_completed_event(
|
||||
self.litellm_model_response
|
||||
)
|
||||
if response_completed_event:
|
||||
return response_completed_event
|
||||
else:
|
||||
if sync_mode:
|
||||
raise StopIteration
|
||||
else:
|
||||
raise StopAsyncIteration
|
||||
|
||||
async def __anext__(
|
||||
self,
|
||||
) -> Union[ResponsesAPIStreamingResponse, ResponseCompletedEvent]:
|
||||
) -> Union[
|
||||
ResponsesAPIStreamingResponse,
|
||||
ResponseCompletedEvent,
|
||||
BaseLiteLLMOpenAIResponseObject,
|
||||
]:
|
||||
try:
|
||||
while True:
|
||||
if self.finished is True:
|
||||
raise StopAsyncIteration
|
||||
|
||||
result = self.return_default_initial_events()
|
||||
if result:
|
||||
return result
|
||||
# Get the next chunk from the stream
|
||||
try:
|
||||
chunk = await self.litellm_custom_stream_wrapper.__anext__()
|
||||
|
|
@ -70,12 +348,7 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator):
|
|||
if response_api_chunk:
|
||||
return response_api_chunk
|
||||
except StopAsyncIteration:
|
||||
self.finished = True
|
||||
response_completed_event = self._emit_response_completed_event()
|
||||
if response_completed_event:
|
||||
return response_completed_event
|
||||
else:
|
||||
raise StopAsyncIteration
|
||||
return self.common_done_event_logic(sync_mode=False)
|
||||
|
||||
except Exception as e:
|
||||
# Handle HTTP errors
|
||||
|
|
@ -87,12 +360,20 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator):
|
|||
|
||||
def __next__(
|
||||
self,
|
||||
) -> Union[ResponsesAPIStreamingResponse, ResponseCompletedEvent]:
|
||||
) -> Union[
|
||||
ResponsesAPIStreamingResponse,
|
||||
ResponseCompletedEvent,
|
||||
BaseLiteLLMOpenAIResponseObject,
|
||||
]:
|
||||
try:
|
||||
while True:
|
||||
if self.finished is True:
|
||||
raise StopIteration
|
||||
# Get the next chunk from the stream
|
||||
|
||||
result = self.return_default_initial_events()
|
||||
if result:
|
||||
return result
|
||||
try:
|
||||
chunk = self.litellm_custom_stream_wrapper.__next__()
|
||||
self.collected_chat_completion_chunks.append(chunk)
|
||||
|
|
@ -104,13 +385,7 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator):
|
|||
if response_api_chunk:
|
||||
return response_api_chunk
|
||||
except StopIteration:
|
||||
self.finished = True
|
||||
response_completed_event = self._emit_response_completed_event()
|
||||
if response_completed_event:
|
||||
return response_completed_event
|
||||
else:
|
||||
raise StopIteration
|
||||
|
||||
return self.common_done_event_logic(sync_mode=True)
|
||||
except Exception as e:
|
||||
# Handle HTTP errors
|
||||
self.finished = True
|
||||
|
|
@ -165,26 +440,33 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator):
|
|||
chat_completion_delta: ChatCompletionDelta = choice.delta
|
||||
return chat_completion_delta.content or ""
|
||||
|
||||
def _emit_response_completed_event(self) -> Optional[ResponseCompletedEvent]:
|
||||
litellm_model_response: Optional[
|
||||
Union[ModelResponse, TextCompletionResponse]
|
||||
] = stream_chunk_builder(chunks=self.collected_chat_completion_chunks, logging_obj=self.litellm_logging_obj)
|
||||
if litellm_model_response and isinstance(litellm_model_response, ModelResponse):
|
||||
def _emit_response_completed_event(
|
||||
self, litellm_model_response: ModelResponse
|
||||
) -> Optional[ResponseCompletedEvent]:
|
||||
|
||||
if litellm_model_response:
|
||||
# Add cost to usage object if include_cost_in_streaming_usage is True
|
||||
if litellm.include_cost_in_streaming_usage and self.litellm_logging_obj is not None:
|
||||
if (
|
||||
litellm.include_cost_in_streaming_usage
|
||||
and self.litellm_logging_obj is not None
|
||||
):
|
||||
usage = getattr(litellm_model_response, "usage", None)
|
||||
if usage is not None:
|
||||
setattr(
|
||||
usage, "cost", self.litellm_logging_obj._response_cost_calculator(result=litellm_model_response)
|
||||
usage,
|
||||
"cost",
|
||||
self.litellm_logging_obj._response_cost_calculator(
|
||||
result=litellm_model_response
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
# Transform the response
|
||||
responses_api_response = LiteLLMCompletionResponsesConfig.transform_chat_completion_response_to_responses_api_response(
|
||||
request_input=self.request_input,
|
||||
chat_completion_response=litellm_model_response,
|
||||
responses_api_request=self.responses_api_request,
|
||||
)
|
||||
|
||||
|
||||
# Encode the response ID to match non-streaming behavior
|
||||
encoded_response = ResponsesAPIRequestUtils._update_responses_api_response_id_with_model_id(
|
||||
responses_api_response=responses_api_response,
|
||||
|
|
|
|||
|
|
@ -1,19 +1,10 @@
|
|||
from litellm._uuid import uuid
|
||||
from typing import (
|
||||
TYPE_CHECKING,
|
||||
Any,
|
||||
Dict,
|
||||
List,
|
||||
Optional,
|
||||
Union,
|
||||
cast,
|
||||
)
|
||||
from typing import TYPE_CHECKING, Any, Dict, List, Optional, Union, cast
|
||||
|
||||
from litellm._logging import verbose_logger
|
||||
from litellm.responses.streaming_iterator import (
|
||||
BaseResponsesAPIStreamingIterator,
|
||||
)
|
||||
from litellm._uuid import uuid
|
||||
from litellm.responses.streaming_iterator import BaseResponsesAPIStreamingIterator
|
||||
from litellm.types.llms.openai import (
|
||||
BaseLiteLLMOpenAIResponseObject,
|
||||
MCPCallArgumentsDeltaEvent,
|
||||
MCPCallArgumentsDoneEvent,
|
||||
MCPCallCompletedEvent,
|
||||
|
|
@ -38,22 +29,24 @@ async def create_mcp_list_tools_events(
|
|||
mcp_tools_with_litellm_proxy: List[ToolParam],
|
||||
user_api_key_auth: Any,
|
||||
base_item_id: str,
|
||||
pre_processed_mcp_tools: List[Any]
|
||||
pre_processed_mcp_tools: List[Any],
|
||||
) -> List[ResponsesAPIStreamingResponse]:
|
||||
"""Create MCP discovery events using pre-processed tools from the parent"""
|
||||
|
||||
|
||||
events: List[ResponsesAPIStreamingResponse] = []
|
||||
|
||||
|
||||
try:
|
||||
# Extract MCP server names
|
||||
mcp_servers = []
|
||||
for tool in mcp_tools_with_litellm_proxy:
|
||||
if isinstance(tool, dict) and "server_url" in tool:
|
||||
server_url = tool.get("server_url")
|
||||
if isinstance(server_url, str) and server_url.startswith("litellm_proxy/mcp/"):
|
||||
if isinstance(server_url, str) and server_url.startswith(
|
||||
"litellm_proxy/mcp/"
|
||||
):
|
||||
server_name = server_url.split("/")[-1]
|
||||
mcp_servers.append(server_name)
|
||||
|
||||
|
||||
# Emit list tools in progress event
|
||||
in_progress_event = MCPListToolsInProgressEvent(
|
||||
type=ResponsesAPIStreamEvents.MCP_LIST_TOOLS_IN_PROGRESS,
|
||||
|
|
@ -62,21 +55,21 @@ async def create_mcp_list_tools_events(
|
|||
item_id=base_item_id,
|
||||
)
|
||||
events.append(in_progress_event)
|
||||
|
||||
|
||||
# Use the pre-processed MCP tools that were already fetched, filtered, and deduplicated by the parent
|
||||
filtered_mcp_tools = pre_processed_mcp_tools
|
||||
|
||||
|
||||
# Convert tools to dict format for the event
|
||||
mcp_tools_dict = []
|
||||
for tool in filtered_mcp_tools:
|
||||
if hasattr(tool, 'model_dump') and callable(getattr(tool, 'model_dump')):
|
||||
if hasattr(tool, "model_dump") and callable(getattr(tool, "model_dump")):
|
||||
# Type cast to help mypy understand this is safe after hasattr check
|
||||
mcp_tools_dict.append(cast(Any, tool).model_dump())
|
||||
elif hasattr(tool, '__dict__'):
|
||||
elif hasattr(tool, "__dict__"):
|
||||
mcp_tools_dict.append(tool.__dict__)
|
||||
else:
|
||||
mcp_tools_dict.append({"name": getattr(tool, 'name', str(tool))})
|
||||
|
||||
mcp_tools_dict.append({"name": getattr(tool, "name", str(tool))})
|
||||
|
||||
# Emit list tools completed event
|
||||
completed_event = MCPListToolsCompletedEvent(
|
||||
type=ResponsesAPIStreamEvents.MCP_LIST_TOOLS_COMPLETED,
|
||||
|
|
@ -85,7 +78,7 @@ async def create_mcp_list_tools_events(
|
|||
item_id=base_item_id,
|
||||
)
|
||||
events.append(completed_event)
|
||||
|
||||
|
||||
# Add output_item.done event with the actual tools list (matching OpenAI format)
|
||||
from litellm.types.llms.openai import OutputItemDoneEvent
|
||||
|
||||
|
|
@ -95,45 +88,50 @@ async def create_mcp_list_tools_events(
|
|||
first_tool = mcp_tools_with_litellm_proxy[0]
|
||||
if isinstance(first_tool, dict):
|
||||
server_label_value = first_tool.get("server_label", "")
|
||||
server_label = str(server_label_value) if server_label_value is not None else ""
|
||||
|
||||
server_label = (
|
||||
str(server_label_value) if server_label_value is not None else ""
|
||||
)
|
||||
|
||||
# Format tools for OpenAI output_item.done format
|
||||
formatted_tools = []
|
||||
for tool in filtered_mcp_tools:
|
||||
tool_dict = {
|
||||
"name": getattr(tool, 'name', 'unknown'),
|
||||
"description": getattr(tool, 'description', ''),
|
||||
"name": getattr(tool, "name", "unknown"),
|
||||
"description": getattr(tool, "description", ""),
|
||||
"annotations": {"read_only": False},
|
||||
}
|
||||
|
||||
|
||||
# Add input_schema if available
|
||||
if hasattr(tool, 'inputSchema'):
|
||||
tool_dict["input_schema"] = getattr(tool, 'inputSchema')
|
||||
elif hasattr(tool, 'input_schema'):
|
||||
tool_dict["input_schema"] = getattr(tool, 'input_schema')
|
||||
|
||||
if hasattr(tool, "inputSchema"):
|
||||
tool_dict["input_schema"] = getattr(tool, "inputSchema")
|
||||
elif hasattr(tool, "input_schema"):
|
||||
tool_dict["input_schema"] = getattr(tool, "input_schema")
|
||||
|
||||
formatted_tools.append(tool_dict)
|
||||
|
||||
|
||||
# Create the output_item.done event with MCP tools list
|
||||
output_item_done_event = OutputItemDoneEvent(
|
||||
type=ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE,
|
||||
output_index=0,
|
||||
item={
|
||||
"id": base_item_id,
|
||||
"type": "mcp_list_tools",
|
||||
"server_label": server_label,
|
||||
"tools": formatted_tools
|
||||
}
|
||||
item=BaseLiteLLMOpenAIResponseObject(
|
||||
**{
|
||||
"id": base_item_id,
|
||||
"type": "mcp_list_tools",
|
||||
"server_label": server_label,
|
||||
"tools": formatted_tools,
|
||||
}
|
||||
),
|
||||
)
|
||||
events.append(output_item_done_event)
|
||||
|
||||
|
||||
verbose_logger.debug(f"Created {len(events)} MCP discovery events")
|
||||
|
||||
|
||||
except Exception as e:
|
||||
verbose_logger.error(f"Error creating MCP list tools events: {e}")
|
||||
import traceback
|
||||
|
||||
traceback.print_exc()
|
||||
|
||||
|
||||
# Emit failed event on error
|
||||
failed_event = MCPListToolsFailedEvent(
|
||||
type=ResponsesAPIStreamEvents.MCP_LIST_TOOLS_FAILED,
|
||||
|
|
@ -142,37 +140,39 @@ async def create_mcp_list_tools_events(
|
|||
item_id=base_item_id,
|
||||
)
|
||||
events.append(failed_event)
|
||||
|
||||
|
||||
# Still emit output_item.done event even on failure (with empty tools list)
|
||||
from litellm.types.llms.openai import OutputItemDoneEvent
|
||||
|
||||
|
||||
output_item_done_event = OutputItemDoneEvent(
|
||||
type=ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE,
|
||||
output_index=0,
|
||||
item={
|
||||
"id": base_item_id,
|
||||
"type": "mcp_list_tools",
|
||||
"server_label": "",
|
||||
"tools": []
|
||||
}
|
||||
item=BaseLiteLLMOpenAIResponseObject(
|
||||
**{
|
||||
"id": base_item_id,
|
||||
"type": "mcp_list_tools",
|
||||
"server_label": "",
|
||||
"tools": [],
|
||||
}
|
||||
),
|
||||
)
|
||||
events.append(output_item_done_event)
|
||||
|
||||
|
||||
return events
|
||||
|
||||
|
||||
def create_mcp_call_events(
|
||||
tool_name: str,
|
||||
tool_call_id: str,
|
||||
tool_name: str,
|
||||
tool_call_id: str,
|
||||
arguments: str,
|
||||
result: Optional[str] = None,
|
||||
base_item_id: Optional[str] = None,
|
||||
sequence_start: int = 1
|
||||
sequence_start: int = 1,
|
||||
) -> List[ResponsesAPIStreamingResponse]:
|
||||
"""Create MCP call events following OpenAI's specification"""
|
||||
events: List[ResponsesAPIStreamingResponse] = []
|
||||
item_id = base_item_id or f"mcp_{uuid.uuid4().hex[:8]}"
|
||||
|
||||
|
||||
# MCP call in progress event
|
||||
in_progress_event = MCPCallInProgressEvent(
|
||||
type=ResponsesAPIStreamEvents.MCP_CALL_IN_PROGRESS,
|
||||
|
|
@ -181,7 +181,7 @@ def create_mcp_call_events(
|
|||
item_id=item_id,
|
||||
)
|
||||
events.append(in_progress_event)
|
||||
|
||||
|
||||
# MCP call arguments delta event (streaming the arguments)
|
||||
arguments_delta_event = MCPCallArgumentsDeltaEvent(
|
||||
type=ResponsesAPIStreamEvents.MCP_CALL_ARGUMENTS_DELTA,
|
||||
|
|
@ -191,7 +191,7 @@ def create_mcp_call_events(
|
|||
sequence_number=sequence_start + 1,
|
||||
)
|
||||
events.append(arguments_delta_event)
|
||||
|
||||
|
||||
# MCP call arguments done event
|
||||
arguments_done_event = MCPCallArgumentsDoneEvent(
|
||||
type=ResponsesAPIStreamEvents.MCP_CALL_ARGUMENTS_DONE,
|
||||
|
|
@ -201,7 +201,7 @@ def create_mcp_call_events(
|
|||
sequence_number=sequence_start + 2,
|
||||
)
|
||||
events.append(arguments_done_event)
|
||||
|
||||
|
||||
# MCP call completed event (or failed if result indicates failure)
|
||||
if result is not None:
|
||||
completed_event = MCPCallCompletedEvent(
|
||||
|
|
@ -211,23 +211,25 @@ def create_mcp_call_events(
|
|||
output_index=0,
|
||||
)
|
||||
events.append(completed_event)
|
||||
|
||||
|
||||
# Add output_item.done event with the tool call result
|
||||
from litellm.types.llms.openai import OutputItemDoneEvent
|
||||
|
||||
|
||||
output_item_done_event = OutputItemDoneEvent(
|
||||
type=ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE,
|
||||
output_index=0,
|
||||
item={
|
||||
"id": item_id,
|
||||
"type": "mcp_call",
|
||||
"approval_request_id": f"mcpr_{uuid.uuid4().hex[:8]}",
|
||||
"arguments": arguments,
|
||||
"error": None,
|
||||
"name": tool_name,
|
||||
"output": result,
|
||||
"server_label": "litellm"
|
||||
},
|
||||
item=BaseLiteLLMOpenAIResponseObject(
|
||||
**{
|
||||
"id": item_id,
|
||||
"type": "mcp_call",
|
||||
"approval_request_id": f"mcpr_{uuid.uuid4().hex[:8]}",
|
||||
"arguments": arguments,
|
||||
"error": None,
|
||||
"name": tool_name,
|
||||
"output": result,
|
||||
"server_label": "litellm",
|
||||
}
|
||||
),
|
||||
)
|
||||
events.append(output_item_done_event)
|
||||
else:
|
||||
|
|
@ -238,7 +240,7 @@ def create_mcp_call_events(
|
|||
output_index=0,
|
||||
)
|
||||
events.append(failed_event)
|
||||
|
||||
|
||||
return events
|
||||
|
||||
|
||||
|
|
@ -250,51 +252,60 @@ class MCPEnhancedStreamingIterator(BaseResponsesAPIStreamingIterator):
|
|||
3. Handles tool execution and follow-up calls for auto-execute tools
|
||||
4. Emits tool execution events in the stream
|
||||
"""
|
||||
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
base_iterator: Any, # Can be None - will be created internally
|
||||
mcp_events: List[ResponsesAPIStreamingResponse],
|
||||
mcp_tools_with_litellm_proxy: Optional[List[Any]] = None,
|
||||
user_api_key_auth: Any = None,
|
||||
original_request_params: Optional[Dict[str, Any]] = None
|
||||
original_request_params: Optional[Dict[str, Any]] = None,
|
||||
):
|
||||
# MCP setup
|
||||
self.mcp_tools_with_litellm_proxy = mcp_tools_with_litellm_proxy or []
|
||||
self.user_api_key_auth = user_api_key_auth
|
||||
self.original_request_params = original_request_params or {}
|
||||
self.should_auto_execute = self._should_auto_execute_tools()
|
||||
|
||||
|
||||
# Streaming state management
|
||||
self.phase = "mcp_discovery" # mcp_discovery -> initial_response -> tool_execution -> follow_up_response -> finished
|
||||
self.finished = False
|
||||
|
||||
|
||||
# Event queues and generation flags
|
||||
self.mcp_discovery_events: List[ResponsesAPIStreamingResponse] = mcp_events # Pre-generated MCP discovery events
|
||||
self.mcp_discovery_events: List[ResponsesAPIStreamingResponse] = (
|
||||
mcp_events # Pre-generated MCP discovery events
|
||||
)
|
||||
self.tool_execution_events: List[ResponsesAPIStreamingResponse] = []
|
||||
self.mcp_discovery_generated = True # Events are already generated
|
||||
self.mcp_events = mcp_events # Store the initial MCP events for backward compatibility
|
||||
|
||||
self.mcp_events = (
|
||||
mcp_events # Store the initial MCP events for backward compatibility
|
||||
)
|
||||
|
||||
# Iterator references
|
||||
self.base_iterator: Optional[Union[Any, ResponsesAPIResponse]] = base_iterator # Will be created when needed
|
||||
self.base_iterator: Optional[Union[Any, ResponsesAPIResponse]] = (
|
||||
base_iterator # Will be created when needed
|
||||
)
|
||||
self.follow_up_iterator: Optional[Any] = None
|
||||
|
||||
|
||||
# Response collection for tool execution
|
||||
self.collected_response: Optional[ResponsesAPIResponse] = None
|
||||
|
||||
|
||||
# Set up model metadata (will be updated when we get the real iterator)
|
||||
self.model = self.original_request_params.get('model', 'unknown')
|
||||
self.model = self.original_request_params.get("model", "unknown")
|
||||
self.litellm_metadata = {}
|
||||
self.custom_llm_provider = self.original_request_params.get('custom_llm_provider', None)
|
||||
|
||||
self.custom_llm_provider = self.original_request_params.get(
|
||||
"custom_llm_provider", None
|
||||
)
|
||||
|
||||
# Mark as async iterator
|
||||
self.is_async = True
|
||||
|
||||
|
||||
def _should_auto_execute_tools(self) -> bool:
|
||||
"""Check if tools should be auto-executed"""
|
||||
from litellm.responses.mcp.litellm_proxy_mcp_handler import (
|
||||
LiteLLM_Proxy_MCP_Handler,
|
||||
)
|
||||
|
||||
return LiteLLM_Proxy_MCP_Handler._should_auto_execute_tools(
|
||||
self.mcp_tools_with_litellm_proxy
|
||||
)
|
||||
|
|
@ -306,45 +317,49 @@ class MCPEnhancedStreamingIterator(BaseResponsesAPIStreamingIterator):
|
|||
"""
|
||||
Phase-based streaming:
|
||||
1. mcp_discovery - Emit MCP discovery events
|
||||
2. initial_response - Stream the first LLM response
|
||||
2. initial_response - Stream the first LLM response
|
||||
3. tool_execution - Emit tool execution events
|
||||
4. follow_up_response - Stream the follow-up response
|
||||
5. finished - End iteration
|
||||
"""
|
||||
|
||||
|
||||
# Phase 1: MCP Discovery Events
|
||||
if self.phase == "mcp_discovery":
|
||||
# Generate MCP discovery events if not already done
|
||||
# MCP discovery events are already generated and available
|
||||
|
||||
|
||||
# Emit MCP discovery events
|
||||
if self.mcp_discovery_events:
|
||||
return self.mcp_discovery_events.pop(0)
|
||||
|
||||
|
||||
# All MCP discovery events emitted, move to next phase
|
||||
verbose_logger.debug("MCP discovery phase complete, transitioning to initial_response")
|
||||
verbose_logger.debug(
|
||||
"MCP discovery phase complete, transitioning to initial_response"
|
||||
)
|
||||
self.phase = "initial_response"
|
||||
await self._create_initial_response_iterator()
|
||||
# Fall through to process the initial response immediately
|
||||
|
||||
|
||||
# Phase 2: Initial Response Stream
|
||||
if self.phase == "initial_response":
|
||||
if self.base_iterator:
|
||||
# Check if base_iterator is actually iterable
|
||||
if hasattr(self.base_iterator, '__anext__'):
|
||||
if hasattr(self.base_iterator, "__anext__"):
|
||||
try:
|
||||
chunk = await cast(Any, self.base_iterator).__anext__() # type: ignore[attr-defined]
|
||||
|
||||
|
||||
# If auto-execution is enabled, check for completed responses
|
||||
if self.should_auto_execute and self._is_response_completed(chunk):
|
||||
if self.should_auto_execute and self._is_response_completed(
|
||||
chunk
|
||||
):
|
||||
# Collect the response for tool execution
|
||||
response_obj = getattr(chunk, 'response', None)
|
||||
response_obj = getattr(chunk, "response", None)
|
||||
if isinstance(response_obj, ResponsesAPIResponse):
|
||||
self.collected_response = response_obj
|
||||
# Move to tool execution phase after emitting this chunk
|
||||
self.phase = "tool_execution"
|
||||
await self._generate_tool_execution_events()
|
||||
|
||||
|
||||
return chunk
|
||||
except StopAsyncIteration:
|
||||
# Initial response ended, move to next phase
|
||||
|
|
@ -357,24 +372,26 @@ class MCPEnhancedStreamingIterator(BaseResponsesAPIStreamingIterator):
|
|||
else:
|
||||
# base_iterator is not async iterable (likely a ResponsesAPIResponse)
|
||||
# Collect it for tool execution if needed
|
||||
if self.should_auto_execute and isinstance(self.base_iterator, ResponsesAPIResponse):
|
||||
if self.should_auto_execute and isinstance(
|
||||
self.base_iterator, ResponsesAPIResponse
|
||||
):
|
||||
self.collected_response = self.base_iterator
|
||||
self.phase = "tool_execution"
|
||||
await self._generate_tool_execution_events()
|
||||
else:
|
||||
self.phase = "finished"
|
||||
raise StopAsyncIteration
|
||||
|
||||
|
||||
# Phase 3: Tool Execution Events
|
||||
if self.phase == "tool_execution":
|
||||
# Emit any queued tool execution events
|
||||
if self.tool_execution_events:
|
||||
return self.tool_execution_events.pop(0)
|
||||
|
||||
|
||||
# Move to follow-up response phase
|
||||
self.phase = "follow_up_response"
|
||||
await self._create_follow_up_iterator()
|
||||
|
||||
|
||||
# Phase 4: Follow-up Response Stream
|
||||
if self.phase == "follow_up_response":
|
||||
if self.follow_up_iterator:
|
||||
|
|
@ -386,20 +403,22 @@ class MCPEnhancedStreamingIterator(BaseResponsesAPIStreamingIterator):
|
|||
else:
|
||||
self.phase = "finished"
|
||||
raise StopAsyncIteration
|
||||
|
||||
|
||||
# Phase 5: Finished
|
||||
if self.phase == "finished":
|
||||
raise StopAsyncIteration
|
||||
|
||||
|
||||
# Should not reach here
|
||||
raise StopAsyncIteration
|
||||
|
||||
|
||||
def _is_response_completed(self, chunk: ResponsesAPIStreamingResponse) -> bool:
|
||||
"""Check if this chunk indicates the response is completed"""
|
||||
from litellm.types.llms.openai import ResponsesAPIStreamEvents
|
||||
return getattr(chunk, 'type', None) == ResponsesAPIStreamEvents.RESPONSE_COMPLETED
|
||||
|
||||
|
||||
|
||||
return (
|
||||
getattr(chunk, "type", None) == ResponsesAPIStreamEvents.RESPONSE_COMPLETED
|
||||
)
|
||||
|
||||
async def _create_initial_response_iterator(self) -> None:
|
||||
"""Create the initial response iterator by making the first LLM call"""
|
||||
try:
|
||||
|
|
@ -408,38 +427,45 @@ class MCPEnhancedStreamingIterator(BaseResponsesAPIStreamingIterator):
|
|||
|
||||
# Make the initial response API call - but avoid the MCP wrapper
|
||||
params = self.original_request_params.copy()
|
||||
params['stream'] = True # Ensure streaming
|
||||
|
||||
params["stream"] = True # Ensure streaming
|
||||
|
||||
# Use the pre-fetched all_tools from original_request_params (no re-processing needed)
|
||||
params_for_llm = {}
|
||||
for key, value in params.items():
|
||||
params_for_llm[key] = value # Copy all params as-is since tools are already processed
|
||||
|
||||
tools_count = len(params_for_llm.get('tools', []))
|
||||
params_for_llm[key] = (
|
||||
value # Copy all params as-is since tools are already processed
|
||||
)
|
||||
|
||||
tools_count = len(params_for_llm.get("tools", []))
|
||||
verbose_logger.debug(f"Making LLM call with {tools_count} tools")
|
||||
response = await aresponses(**params_for_llm)
|
||||
|
||||
|
||||
# Set the base iterator
|
||||
if hasattr(response, '__aiter__') or hasattr(response, '__iter__'):
|
||||
if hasattr(response, "__aiter__") or hasattr(response, "__iter__"):
|
||||
self.base_iterator = response
|
||||
# Copy metadata from the real iterator
|
||||
self.model = getattr(response, 'model', self.model)
|
||||
self.litellm_metadata = getattr(response, 'litellm_metadata', {})
|
||||
self.custom_llm_provider = getattr(response, 'custom_llm_provider', self.custom_llm_provider)
|
||||
verbose_logger.debug(f"Created base iterator: {type(self.base_iterator)}")
|
||||
self.model = getattr(response, "model", self.model)
|
||||
self.litellm_metadata = getattr(response, "litellm_metadata", {})
|
||||
self.custom_llm_provider = getattr(
|
||||
response, "custom_llm_provider", self.custom_llm_provider
|
||||
)
|
||||
verbose_logger.debug(
|
||||
f"Created base iterator: {type(self.base_iterator)}"
|
||||
)
|
||||
else:
|
||||
# Non-streaming response - this shouldn't happen but handle it
|
||||
verbose_logger.warning(f"Got non-streaming response: {type(response)}")
|
||||
self.base_iterator = None
|
||||
self.phase = "finished"
|
||||
|
||||
|
||||
except Exception as e:
|
||||
verbose_logger.error(f"Error creating initial response iterator: {e}")
|
||||
import traceback
|
||||
|
||||
traceback.print_exc()
|
||||
self.base_iterator = None
|
||||
self.phase = "finished"
|
||||
|
||||
|
||||
async def _generate_tool_execution_events(self) -> None:
|
||||
"""Generate tool execution events and execute tools"""
|
||||
if not self.collected_response:
|
||||
|
|
@ -447,7 +473,7 @@ class MCPEnhancedStreamingIterator(BaseResponsesAPIStreamingIterator):
|
|||
from litellm.responses.mcp.litellm_proxy_mcp_handler import (
|
||||
LiteLLM_Proxy_MCP_Handler,
|
||||
)
|
||||
|
||||
|
||||
try:
|
||||
# Extract tool calls from the response
|
||||
if self.collected_response is not None:
|
||||
|
|
@ -456,9 +482,11 @@ class MCPEnhancedStreamingIterator(BaseResponsesAPIStreamingIterator):
|
|||
tool_calls = []
|
||||
if not tool_calls:
|
||||
return
|
||||
|
||||
|
||||
for tool_call in tool_calls:
|
||||
tool_name, tool_arguments, tool_call_id = LiteLLM_Proxy_MCP_Handler._extract_tool_call_details(tool_call)
|
||||
tool_name, tool_arguments, tool_call_id = (
|
||||
LiteLLM_Proxy_MCP_Handler._extract_tool_call_details(tool_call)
|
||||
)
|
||||
if tool_name and tool_call_id:
|
||||
# Create MCP call events for this tool execution
|
||||
call_events = create_mcp_call_events(
|
||||
|
|
@ -467,34 +495,35 @@ class MCPEnhancedStreamingIterator(BaseResponsesAPIStreamingIterator):
|
|||
arguments=tool_arguments or "{}", # JSON string with arguments
|
||||
result=None, # Will be set after execution
|
||||
base_item_id=f"mcp_{uuid.uuid4().hex[:8]}",
|
||||
sequence_start=len(self.tool_execution_events) + 1
|
||||
sequence_start=len(self.tool_execution_events) + 1,
|
||||
)
|
||||
# Add the in_progress and arguments events (not the completed event yet)
|
||||
self.tool_execution_events.extend(call_events[:-1])
|
||||
|
||||
|
||||
# Execute the tools
|
||||
tool_results = await LiteLLM_Proxy_MCP_Handler._execute_tool_calls(
|
||||
tool_calls=tool_calls,
|
||||
user_api_key_auth=self.user_api_key_auth
|
||||
tool_calls=tool_calls, user_api_key_auth=self.user_api_key_auth
|
||||
)
|
||||
|
||||
|
||||
# Create completion events and output_item.done events for tool execution
|
||||
for tool_result in tool_results:
|
||||
tool_call_id = tool_result.get("tool_call_id", "unknown")
|
||||
result_text = tool_result.get("result", "")
|
||||
|
||||
|
||||
# Find matching tool name and arguments
|
||||
tool_name = "unknown"
|
||||
tool_arguments = "{}"
|
||||
for tool_call in tool_calls:
|
||||
name, args, call_id = LiteLLM_Proxy_MCP_Handler._extract_tool_call_details(tool_call)
|
||||
name, args, call_id = (
|
||||
LiteLLM_Proxy_MCP_Handler._extract_tool_call_details(tool_call)
|
||||
)
|
||||
if call_id == tool_call_id:
|
||||
tool_name = name or "unknown"
|
||||
tool_arguments = args or "{}"
|
||||
break
|
||||
|
||||
|
||||
item_id = f"mcp_{uuid.uuid4().hex[:8]}"
|
||||
|
||||
|
||||
# Create the completion event
|
||||
completed_event = MCPCallCompletedEvent(
|
||||
type=ResponsesAPIStreamEvents.MCP_CALL_COMPLETED,
|
||||
|
|
@ -503,79 +532,84 @@ class MCPEnhancedStreamingIterator(BaseResponsesAPIStreamingIterator):
|
|||
output_index=0,
|
||||
)
|
||||
self.tool_execution_events.append(completed_event)
|
||||
|
||||
|
||||
# Create output_item.done event with the tool call result
|
||||
from litellm.types.llms.openai import OutputItemDoneEvent
|
||||
|
||||
|
||||
output_item_done_event = OutputItemDoneEvent(
|
||||
type=ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE,
|
||||
output_index=0,
|
||||
item={
|
||||
"id": item_id,
|
||||
"type": "mcp_call",
|
||||
"approval_request_id": f"mcpr_{uuid.uuid4().hex[:8]}",
|
||||
"arguments": tool_arguments,
|
||||
"error": None,
|
||||
"name": tool_name,
|
||||
"output": result_text,
|
||||
"server_label": "litellm" # or extract from tool config
|
||||
},
|
||||
item=BaseLiteLLMOpenAIResponseObject(
|
||||
**{
|
||||
"id": item_id,
|
||||
"type": "mcp_call",
|
||||
"approval_request_id": f"mcpr_{uuid.uuid4().hex[:8]}",
|
||||
"arguments": tool_arguments,
|
||||
"error": None,
|
||||
"name": tool_name,
|
||||
"output": result_text,
|
||||
"server_label": "litellm", # or extract from tool config
|
||||
}
|
||||
),
|
||||
)
|
||||
self.tool_execution_events.append(output_item_done_event)
|
||||
|
||||
|
||||
# Store tool results for follow-up call
|
||||
self.tool_results = tool_results
|
||||
|
||||
|
||||
except Exception as e:
|
||||
verbose_logger.error(f"Error in tool execution: {e}")
|
||||
import traceback
|
||||
|
||||
traceback.print_exc()
|
||||
self.tool_results = []
|
||||
|
||||
|
||||
async def _create_follow_up_iterator(self) -> None:
|
||||
"""Create the follow-up response iterator with tool results"""
|
||||
if not self.collected_response or not hasattr(self, 'tool_results'):
|
||||
if not self.collected_response or not hasattr(self, "tool_results"):
|
||||
return
|
||||
|
||||
|
||||
from litellm.responses.main import aresponses
|
||||
from litellm.responses.mcp.litellm_proxy_mcp_handler import (
|
||||
LiteLLM_Proxy_MCP_Handler,
|
||||
)
|
||||
|
||||
|
||||
try:
|
||||
# Create follow-up input
|
||||
if self.collected_response is not None:
|
||||
follow_up_input = LiteLLM_Proxy_MCP_Handler._create_follow_up_input(
|
||||
response=self.collected_response, # type: ignore[arg-type]
|
||||
tool_results=self.tool_results,
|
||||
original_input=self.original_request_params.get('input')
|
||||
original_input=self.original_request_params.get("input"),
|
||||
)
|
||||
|
||||
|
||||
# Make follow-up call with streaming
|
||||
follow_up_params = self.original_request_params.copy()
|
||||
follow_up_params.update({
|
||||
'input': follow_up_input,
|
||||
'previous_response_id': self.collected_response.id, # type: ignore[attr-defined]
|
||||
'stream': True
|
||||
})
|
||||
follow_up_params.update(
|
||||
{
|
||||
"input": follow_up_input,
|
||||
"previous_response_id": self.collected_response.id, # type: ignore[attr-defined]
|
||||
"stream": True,
|
||||
}
|
||||
)
|
||||
else:
|
||||
return
|
||||
# Remove tool_choice to avoid forcing more tool calls
|
||||
follow_up_params.pop('tool_choice', None)
|
||||
|
||||
follow_up_params.pop("tool_choice", None)
|
||||
|
||||
follow_up_response = await aresponses(**follow_up_params)
|
||||
|
||||
|
||||
# Set up the follow-up iterator
|
||||
if hasattr(follow_up_response, '__aiter__'):
|
||||
if hasattr(follow_up_response, "__aiter__"):
|
||||
self.follow_up_iterator = follow_up_response
|
||||
|
||||
|
||||
except Exception as e:
|
||||
verbose_logger.error(f"Error creating follow-up iterator: {e}")
|
||||
import traceback
|
||||
|
||||
traceback.print_exc()
|
||||
self.follow_up_iterator = None
|
||||
|
||||
|
||||
def __iter__(self):
|
||||
return self
|
||||
|
||||
|
|
@ -583,11 +617,11 @@ class MCPEnhancedStreamingIterator(BaseResponsesAPIStreamingIterator):
|
|||
# First, emit any queued MCP events
|
||||
if self.mcp_events: # type: ignore[attr-defined]
|
||||
return self.mcp_events.pop(0) # type: ignore[attr-defined]
|
||||
|
||||
|
||||
# Then delegate to the base iterator
|
||||
if not self.is_async:
|
||||
try:
|
||||
if self.base_iterator and hasattr(self.base_iterator, '__next__'):
|
||||
if self.base_iterator and hasattr(self.base_iterator, "__next__"):
|
||||
return next(cast(Any, self.base_iterator)) # type: ignore[arg-type]
|
||||
else:
|
||||
raise StopIteration
|
||||
|
|
|
|||
|
|
@ -93,24 +93,35 @@ class BaseResponsesAPIStreamingIterator:
|
|||
# Store the completed response
|
||||
if (
|
||||
openai_responses_api_chunk
|
||||
and openai_responses_api_chunk.type
|
||||
and getattr(openai_responses_api_chunk, "type", None)
|
||||
== ResponsesAPIStreamEvents.RESPONSE_COMPLETED
|
||||
):
|
||||
self.completed_response = openai_responses_api_chunk
|
||||
# Add cost to usage object if include_cost_in_streaming_usage is True
|
||||
if litellm.include_cost_in_streaming_usage and self.logging_obj is not None:
|
||||
response_obj: Optional[ResponsesAPIResponse] = getattr(openai_responses_api_chunk, "response", None)
|
||||
if (
|
||||
litellm.include_cost_in_streaming_usage
|
||||
and self.logging_obj is not None
|
||||
):
|
||||
response_obj: Optional[ResponsesAPIResponse] = getattr(
|
||||
openai_responses_api_chunk, "response", None
|
||||
)
|
||||
if response_obj:
|
||||
usage_obj: Optional[ResponseAPIUsage] = getattr(response_obj, "usage", None)
|
||||
usage_obj: Optional[ResponseAPIUsage] = getattr(
|
||||
response_obj, "usage", None
|
||||
)
|
||||
if usage_obj is not None:
|
||||
try:
|
||||
cost: Optional[float] = self.logging_obj._response_cost_calculator(result=response_obj)
|
||||
cost: Optional[float] = (
|
||||
self.logging_obj._response_cost_calculator(
|
||||
result=response_obj
|
||||
)
|
||||
)
|
||||
if cost is not None:
|
||||
setattr(usage_obj, "cost", cost)
|
||||
except Exception:
|
||||
# If cost calculation fails, continue without cost
|
||||
pass
|
||||
|
||||
|
||||
self._handle_logging_completed_response()
|
||||
|
||||
return openai_responses_api_chunk
|
||||
|
|
|
|||
|
|
@ -1,3 +1,4 @@
|
|||
import uuid
|
||||
from enum import Enum
|
||||
from os import PathLike
|
||||
from typing import IO, Any, Iterable, List, Literal, Mapping, Optional, Tuple, Union
|
||||
|
|
@ -44,7 +45,7 @@ from openai.types.responses.response import (
|
|||
# Handle OpenAI SDK version compatibility for Text type
|
||||
try:
|
||||
# fmt: off
|
||||
from openai.types.responses.response_create_params import ( # type: ignore[attr-defined]
|
||||
from openai.types.responses.response_create_params import (
|
||||
Text as ResponseText, # type: ignore[attr-defined]
|
||||
)
|
||||
|
||||
|
|
@ -992,7 +993,9 @@ class ResponsesAPIOptionalRequestParams(TypedDict, total=False):
|
|||
prompt_cache_key: Optional[str]
|
||||
stream_options: Optional[dict]
|
||||
top_logprobs: Optional[int]
|
||||
partial_images: Optional[int] # Number of partial images to generate (1-3) for streaming image generation
|
||||
partial_images: Optional[
|
||||
int
|
||||
] # Number of partial images to generate (1-3) for streaming image generation
|
||||
|
||||
|
||||
class ResponsesAPIRequestParams(ResponsesAPIOptionalRequestParams, total=False):
|
||||
|
|
@ -1056,7 +1059,9 @@ class ResponsesAPIResponse(BaseLiteLLMOpenAIResponseObject):
|
|||
parallel_tool_calls: Optional[bool] = None
|
||||
temperature: Optional[float] = None
|
||||
tool_choice: Optional[ToolChoice] = None
|
||||
tools: Optional[Union[List[Tool], List[ResponseFunctionToolCall], List[Dict[str, Any]]]] = None
|
||||
tools: Optional[
|
||||
Union[List[Tool], List[ResponseFunctionToolCall], List[Dict[str, Any]]]
|
||||
] = None
|
||||
top_p: Optional[float] = None
|
||||
max_output_tokens: Optional[int] = None
|
||||
previous_response_id: Optional[str] = None
|
||||
|
|
@ -1180,13 +1185,27 @@ class ReasoningSummaryTextDeltaEvent(BaseLiteLLMOpenAIResponseObject):
|
|||
class OutputItemAddedEvent(BaseLiteLLMOpenAIResponseObject):
|
||||
type: Literal[ResponsesAPIStreamEvents.OUTPUT_ITEM_ADDED]
|
||||
output_index: int
|
||||
item: Optional[dict]
|
||||
item: Optional[BaseLiteLLMOpenAIResponseObject]
|
||||
|
||||
|
||||
class OutputItemDoneEvent(BaseLiteLLMOpenAIResponseObject):
|
||||
type: Literal[ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE]
|
||||
output_index: int
|
||||
item: dict
|
||||
sequence_number: int = 1
|
||||
item: BaseLiteLLMOpenAIResponseObject
|
||||
|
||||
|
||||
class OpenAIChatCompletionLogprobsContentTopLogprobs(TypedDict, total=False):
|
||||
bytes: List
|
||||
logprob: Required[float]
|
||||
token: Required[str]
|
||||
|
||||
|
||||
class OpenAIChatCompletionLogprobsContent(TypedDict, total=False):
|
||||
bytes: List
|
||||
logprob: Required[float]
|
||||
token: Required[str]
|
||||
top_logprobs: List[OpenAIChatCompletionLogprobsContentTopLogprobs]
|
||||
|
||||
|
||||
class ContentPartAddedEvent(BaseLiteLLMOpenAIResponseObject):
|
||||
|
|
@ -1194,7 +1213,31 @@ class ContentPartAddedEvent(BaseLiteLLMOpenAIResponseObject):
|
|||
item_id: str
|
||||
output_index: int
|
||||
content_index: int
|
||||
part: dict
|
||||
part: BaseLiteLLMOpenAIResponseObject
|
||||
|
||||
|
||||
class ContentPartDonePartOutputText(BaseLiteLLMOpenAIResponseObject):
|
||||
type: Literal["output_text"]
|
||||
text: str
|
||||
annotations: List[BaseLiteLLMOpenAIResponseObject]
|
||||
logprobs: Optional[List[OpenAIChatCompletionLogprobsContent]]
|
||||
|
||||
|
||||
class ContentPartDonePartRefusal(BaseLiteLLMOpenAIResponseObject):
|
||||
type: Literal["refusal"]
|
||||
refusal: str
|
||||
|
||||
|
||||
class ContentPartDonePartReasoningText(BaseLiteLLMOpenAIResponseObject):
|
||||
type: Literal["reasoning_text"]
|
||||
reasoning: str
|
||||
|
||||
|
||||
PART_UNION_TYPES = Union[
|
||||
ContentPartDonePartOutputText,
|
||||
ContentPartDonePartRefusal,
|
||||
ContentPartDonePartReasoningText,
|
||||
]
|
||||
|
||||
|
||||
class ContentPartDoneEvent(BaseLiteLLMOpenAIResponseObject):
|
||||
|
|
@ -1202,7 +1245,7 @@ class ContentPartDoneEvent(BaseLiteLLMOpenAIResponseObject):
|
|||
item_id: str
|
||||
output_index: int
|
||||
content_index: int
|
||||
part: dict
|
||||
part: PART_UNION_TYPES
|
||||
|
||||
|
||||
class OutputTextDeltaEvent(BaseLiteLLMOpenAIResponseObject):
|
||||
|
|
@ -1413,6 +1456,7 @@ ResponsesAPIStreamingResponse = Annotated[
|
|||
ImageGenerationPartialImageEvent,
|
||||
ErrorEvent,
|
||||
GenericEvent,
|
||||
BaseLiteLLMOpenAIResponseObject,
|
||||
],
|
||||
Discriminator("type"),
|
||||
]
|
||||
|
|
@ -1731,19 +1775,6 @@ class OpenAIModerationResponse(BaseLiteLLMOpenAIResponseObject):
|
|||
_hidden_params: dict = PrivateAttr(default_factory=dict)
|
||||
|
||||
|
||||
class OpenAIChatCompletionLogprobsContentTopLogprobs(TypedDict, total=False):
|
||||
bytes: List
|
||||
logprob: Required[float]
|
||||
token: Required[str]
|
||||
|
||||
|
||||
class OpenAIChatCompletionLogprobsContent(TypedDict, total=False):
|
||||
bytes: List
|
||||
logprob: Required[float]
|
||||
token: Required[str]
|
||||
top_logprobs: List[OpenAIChatCompletionLogprobsContentTopLogprobs]
|
||||
|
||||
|
||||
class OpenAIChatCompletionLogprobs(TypedDict, total=False):
|
||||
content: List[OpenAIChatCompletionLogprobsContent]
|
||||
refusal: List[OpenAIChatCompletionLogprobsContent]
|
||||
|
|
|
|||
|
|
@ -361,6 +361,7 @@ def test_process_anthropic_headers_with_no_matching_headers():
|
|||
def test_anthropic_tool_use(tool_type, tool_config, message_content):
|
||||
"""Test Anthropic tool use with computer use and web fetch tools."""
|
||||
from litellm import completion
|
||||
|
||||
litellm._turn_on_debug()
|
||||
|
||||
tools = [tool_config]
|
||||
|
|
@ -1518,3 +1519,126 @@ def test_anthropic_streaming():
|
|||
role_set_count += 1
|
||||
|
||||
assert role_set_count == 1
|
||||
|
||||
|
||||
def test_anthropic_via_responses_api():
|
||||
from litellm.types.llms.openai import ResponsesAPIStreamEvents
|
||||
|
||||
response = litellm.responses(
|
||||
model="anthropic/claude-sonnet-4-5",
|
||||
input="Who won the World Cup in 2022?",
|
||||
max_output_tokens=100,
|
||||
stream=True,
|
||||
)
|
||||
|
||||
assert response is not None
|
||||
|
||||
# Expected event sequence
|
||||
expected_events = [
|
||||
ResponsesAPIStreamEvents.RESPONSE_CREATED,
|
||||
ResponsesAPIStreamEvents.RESPONSE_IN_PROGRESS,
|
||||
ResponsesAPIStreamEvents.OUTPUT_ITEM_ADDED,
|
||||
ResponsesAPIStreamEvents.CONTENT_PART_ADDED,
|
||||
ResponsesAPIStreamEvents.OUTPUT_TEXT_DELTA, # Can occur multiple times
|
||||
ResponsesAPIStreamEvents.OUTPUT_TEXT_DONE,
|
||||
ResponsesAPIStreamEvents.CONTENT_PART_DONE,
|
||||
ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE,
|
||||
ResponsesAPIStreamEvents.RESPONSE_COMPLETED,
|
||||
]
|
||||
|
||||
events_seen = []
|
||||
text_delta_count = 0
|
||||
|
||||
for chunk in response:
|
||||
print(f"chunk: {chunk}")
|
||||
|
||||
# Each chunk should have a type attribute
|
||||
assert hasattr(chunk, "type"), f"Chunk missing 'type' attribute: {chunk}"
|
||||
|
||||
event_type = chunk.type
|
||||
|
||||
# Track events seen
|
||||
if event_type == ResponsesAPIStreamEvents.OUTPUT_TEXT_DELTA:
|
||||
text_delta_count += 1
|
||||
if ResponsesAPIStreamEvents.OUTPUT_TEXT_DELTA not in events_seen:
|
||||
events_seen.append(event_type)
|
||||
else:
|
||||
events_seen.append(event_type)
|
||||
|
||||
# Assert specific structures for each event type
|
||||
if event_type == ResponsesAPIStreamEvents.RESPONSE_CREATED:
|
||||
assert chunk.type == ResponsesAPIStreamEvents.RESPONSE_CREATED
|
||||
assert hasattr(chunk, "response")
|
||||
assert chunk.response.status == "in_progress"
|
||||
assert hasattr(chunk.response, "id")
|
||||
assert hasattr(chunk.response, "model")
|
||||
|
||||
elif event_type == ResponsesAPIStreamEvents.RESPONSE_IN_PROGRESS:
|
||||
assert chunk.type == ResponsesAPIStreamEvents.RESPONSE_IN_PROGRESS
|
||||
assert hasattr(chunk, "response")
|
||||
assert chunk.response.status == "in_progress"
|
||||
|
||||
elif event_type == ResponsesAPIStreamEvents.OUTPUT_ITEM_ADDED:
|
||||
assert chunk.type == ResponsesAPIStreamEvents.OUTPUT_ITEM_ADDED
|
||||
assert hasattr(chunk, "output_index")
|
||||
assert hasattr(chunk, "item")
|
||||
assert chunk.item.type == "message"
|
||||
assert chunk.item.role == "assistant"
|
||||
|
||||
elif event_type == ResponsesAPIStreamEvents.CONTENT_PART_ADDED:
|
||||
assert chunk.type == ResponsesAPIStreamEvents.CONTENT_PART_ADDED
|
||||
assert hasattr(chunk, "item_id")
|
||||
assert hasattr(chunk, "output_index")
|
||||
assert hasattr(chunk, "content_index")
|
||||
assert hasattr(chunk, "part")
|
||||
assert chunk.part.type == "output_text"
|
||||
|
||||
elif event_type == ResponsesAPIStreamEvents.OUTPUT_TEXT_DELTA:
|
||||
assert chunk.type == ResponsesAPIStreamEvents.OUTPUT_TEXT_DELTA
|
||||
assert hasattr(chunk, "item_id")
|
||||
assert hasattr(chunk, "output_index")
|
||||
assert hasattr(chunk, "content_index")
|
||||
assert hasattr(chunk, "delta")
|
||||
assert isinstance(chunk.delta, str)
|
||||
|
||||
elif event_type == ResponsesAPIStreamEvents.OUTPUT_TEXT_DONE:
|
||||
assert chunk.type == ResponsesAPIStreamEvents.OUTPUT_TEXT_DONE
|
||||
assert hasattr(chunk, "item_id")
|
||||
assert hasattr(chunk, "output_index")
|
||||
assert hasattr(chunk, "content_index")
|
||||
assert hasattr(chunk, "text")
|
||||
|
||||
elif event_type == ResponsesAPIStreamEvents.CONTENT_PART_DONE:
|
||||
assert chunk.type == ResponsesAPIStreamEvents.CONTENT_PART_DONE
|
||||
assert hasattr(chunk, "item_id")
|
||||
assert hasattr(chunk, "output_index")
|
||||
assert hasattr(chunk, "content_index")
|
||||
assert hasattr(chunk, "part")
|
||||
assert chunk.part.type == "output_text"
|
||||
|
||||
elif event_type == ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE:
|
||||
assert chunk.type == ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE
|
||||
assert hasattr(chunk, "output_index")
|
||||
assert hasattr(chunk, "item")
|
||||
assert chunk.item.status == "completed"
|
||||
|
||||
elif event_type == ResponsesAPIStreamEvents.RESPONSE_COMPLETED:
|
||||
assert chunk.type == ResponsesAPIStreamEvents.RESPONSE_COMPLETED
|
||||
assert hasattr(chunk, "response")
|
||||
assert chunk.response.status == "completed"
|
||||
assert hasattr(chunk.response, "usage")
|
||||
assert hasattr(chunk.response, "output")
|
||||
|
||||
# Assert we saw all expected events
|
||||
print(f"Events seen: {events_seen}")
|
||||
assert (
|
||||
events_seen == expected_events
|
||||
), f"Event sequence mismatch. Expected: {expected_events}, Got: {events_seen}"
|
||||
|
||||
# Assert we saw at least one text delta
|
||||
assert (
|
||||
text_delta_count > 0
|
||||
), f"Expected at least one response.output_text.delta event, got {text_delta_count}"
|
||||
|
||||
print(f"✓ All {len(events_seen)} events matched expected structure")
|
||||
print(f"✓ Received {text_delta_count} text delta chunks")
|
||||
|
|
|
|||
|
|
@ -4,14 +4,20 @@ Test reasoning content preservation in Responses API transformation
|
|||
|
||||
from unittest.mock import AsyncMock
|
||||
|
||||
from litellm.types.utils import ModelResponseStream, StreamingChoices, Delta
|
||||
from litellm.responses.litellm_completion_transformation.streaming_iterator import (
|
||||
LiteLLMCompletionStreamingIterator,
|
||||
)
|
||||
from litellm.responses.litellm_completion_transformation.transformation import (
|
||||
LiteLLMCompletionResponsesConfig,
|
||||
)
|
||||
from litellm.types.utils import ModelResponse, Choices, Message
|
||||
from litellm.types.utils import (
|
||||
Choices,
|
||||
Delta,
|
||||
Message,
|
||||
ModelResponse,
|
||||
ModelResponseStream,
|
||||
StreamingChoices,
|
||||
)
|
||||
|
||||
|
||||
class TestReasoningContentStreaming:
|
||||
|
|
@ -41,6 +47,7 @@ class TestReasoningContentStreaming:
|
|||
mock_stream = AsyncMock()
|
||||
|
||||
iterator = LiteLLMCompletionStreamingIterator(
|
||||
model="test-model",
|
||||
litellm_custom_stream_wrapper=mock_stream,
|
||||
request_input="Test input",
|
||||
responses_api_request={},
|
||||
|
|
@ -78,6 +85,7 @@ class TestReasoningContentStreaming:
|
|||
|
||||
mock_stream = AsyncMock()
|
||||
iterator = LiteLLMCompletionStreamingIterator(
|
||||
model="test-model",
|
||||
litellm_custom_stream_wrapper=mock_stream,
|
||||
request_input="Test input",
|
||||
responses_api_request={},
|
||||
|
|
@ -114,6 +122,7 @@ class TestReasoningContentStreaming:
|
|||
|
||||
mock_stream = AsyncMock()
|
||||
iterator = LiteLLMCompletionStreamingIterator(
|
||||
model="test-model",
|
||||
litellm_custom_stream_wrapper=mock_stream,
|
||||
request_input="Test input",
|
||||
responses_api_request={},
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue