scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-07 02:59:05 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 103

test: handle internal server errors

Browse source
This commit is contained in:
Krrish Dholakia 2025-08-23 11:00:18 -07:00
parent eb7e5e3a71
commit cbb161f10b
1 changed files with 114 additions and 110 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

224
tests/llm_responses_api_testing/base_responses_api.py
View file

@ -101,11 +101,11 @@ def validate_responses_api_response(response, final_chunk: bool = False):
return True # Return True if validation passes
class BaseResponsesAPITest(ABC):
"""
Abstract base test class that enforces a common test across all test classes.
"""
@abstractmethod
def get_base_completion_call_args(self) -> dict:
"""Must return the base completion call args"""
@ -115,32 +115,32 @@ class BaseResponsesAPITest(ABC):
"""Must return the base completion reasoning call args"""
return None
@pytest.mark.parametrize("sync_mode", [True, False])
@pytest.mark.asyncio
async def test_basic_openai_responses_api(self, sync_mode):
litellm._turn_on_debug()
litellm.set_verbose = True
base_completion_call_args = self.get_base_completion_call_args()
try:
try:
if sync_mode:
response = litellm.responses(
input="Basic ping", max_output_tokens=20,
**base_completion_call_args
input="Basic ping",
max_output_tokens=20,
**base_completion_call_args,
)
else:
response = await litellm.aresponses(
input="Basic ping", max_output_tokens=20,
**base_completion_call_args
input="Basic ping",
max_output_tokens=20,
**base_completion_call_args,
)
except litellm.InternalServerError:
except litellm.InternalServerError:
pytest.skip("Skipping test due to litellm.InternalServerError")
print("litellm response=", json.dumps(response, indent=4, default=str))
# Use the helper function to validate the response
validate_responses_api_response(response, final_chunk=True)
@pytest.mark.parametrize("sync_mode", [True, False])
@pytest.mark.asyncio
@pytest.mark.flaky(retries=3, delay=2)
@ -151,9 +151,7 @@ class BaseResponsesAPITest(ABC):
response_completed_event = None
if sync_mode:
response = litellm.responses(
input="Basic ping",
stream=True,
**base_completion_call_args
input="Basic ping", stream=True, **base_completion_call_args
)
for event in response:
print("litellm response=", json.dumps(event, indent=4, default=str))
@ -163,9 +161,7 @@ class BaseResponsesAPITest(ABC):
response_completed_event = event
else:
response = await litellm.aresponses(
input="Basic ping",
stream=True,
**base_completion_call_args
input="Basic ping", stream=True, **base_completion_call_args
)
async for event in response:
print("litellm response=", json.dumps(event, indent=4, default=str))
@ -188,15 +184,29 @@ class BaseResponsesAPITest(ABC):
assert response_completed_event.response.usage is not None
# basic test assert the usage seems reasonable
print("response_completed_event.response.usage=", response_completed_event.response.usage)
assert response_completed_event.response.usage.input_tokens > 0 and response_completed_event.response.usage.input_tokens < 100
assert response_completed_event.response.usage.output_tokens > 0 and response_completed_event.response.usage.output_tokens < 2000
assert response_completed_event.response.usage.total_tokens > 0 and response_completed_event.response.usage.total_tokens < 2000
print(
"response_completed_event.response.usage=",
response_completed_event.response.usage,
)
assert (
response_completed_event.response.usage.input_tokens > 0
and response_completed_event.response.usage.input_tokens < 100
)
assert (
response_completed_event.response.usage.output_tokens > 0
and response_completed_event.response.usage.output_tokens < 2000
)
assert (
response_completed_event.response.usage.total_tokens > 0
and response_completed_event.response.usage.total_tokens < 2000
)
# total tokens should be the sum of input and output tokens
assert response_completed_event.response.usage.total_tokens == response_completed_event.response.usage.input_tokens + response_completed_event.response.usage.output_tokens
assert (
response_completed_event.response.usage.total_tokens
== response_completed_event.response.usage.input_tokens
+ response_completed_event.response.usage.output_tokens
)
@pytest.mark.parametrize("sync_mode", [False, True])
@pytest.mark.asyncio
@ -206,48 +216,44 @@ class BaseResponsesAPITest(ABC):
base_completion_call_args = self.get_base_completion_call_args()
if sync_mode:
response = litellm.responses(
input="Basic ping", max_output_tokens=20,
**base_completion_call_args
input="Basic ping", max_output_tokens=20, **base_completion_call_args
)
# delete the response
if isinstance(response, ResponsesAPIResponse):
litellm.delete_responses(
response_id=response.id,
**base_completion_call_args
response_id=response.id, **base_completion_call_args
)
else:
raise ValueError("response is not a ResponsesAPIResponse")
else:
response = await litellm.aresponses(
input="Basic ping", max_output_tokens=20,
**base_completion_call_args
input="Basic ping", max_output_tokens=20, **base_completion_call_args
)
# async delete the response
if isinstance(response, ResponsesAPIResponse):
await litellm.adelete_responses(
response_id=response.id,
**base_completion_call_args
response_id=response.id, **base_completion_call_args
)
else:
raise ValueError("response is not a ResponsesAPIResponse")
@pytest.mark.parametrize("sync_mode", [True, False])
@pytest.mark.flaky(retries=3, delay=2)
@pytest.mark.asyncio
async def test_basic_openai_responses_streaming_delete_endpoint(self, sync_mode):
#litellm._turn_on_debug()
#litellm.set_verbose = True
# litellm._turn_on_debug()
# litellm.set_verbose = True
base_completion_call_args = self.get_base_completion_call_args()
response_id = None
if sync_mode:
response_id = None
response = litellm.responses(
input="Basic ping", max_output_tokens=20,
input="Basic ping",
max_output_tokens=20,
stream=True,
**base_completion_call_args
**base_completion_call_args,
)
for event in response:
print("litellm response=", json.dumps(event, indent=4, default=str))
@ -260,14 +266,14 @@ class BaseResponsesAPITest(ABC):
# delete the response
assert response_id is not None
litellm.delete_responses(
response_id=response_id,
**base_completion_call_args
response_id=response_id, **base_completion_call_args
)
else:
response = await litellm.aresponses(
input="Basic ping", max_output_tokens=20,
input="Basic ping",
max_output_tokens=20,
stream=True,
**base_completion_call_args
**base_completion_call_args,
)
async for event in response:
print("litellm response=", json.dumps(event, indent=4, default=str))
@ -280,8 +286,7 @@ class BaseResponsesAPITest(ABC):
# delete the response
assert response_id is not None
await litellm.adelete_responses(
response_id=response_id,
**base_completion_call_args
response_id=response_id, **base_completion_call_args
)
@pytest.mark.parametrize("sync_mode", [False, True])
@ -293,15 +298,13 @@ class BaseResponsesAPITest(ABC):
base_completion_call_args = self.get_base_completion_call_args()
if sync_mode:
response = litellm.responses(
input="Basic ping", max_output_tokens=20,
**base_completion_call_args
input="Basic ping", max_output_tokens=20, **base_completion_call_args
)
# get the response
if isinstance(response, ResponsesAPIResponse):
result = litellm.get_responses(
response_id=response.id,
**base_completion_call_args
response_id=response.id, **base_completion_call_args
)
assert result is not None
assert result.id == response.id
@ -310,14 +313,12 @@ class BaseResponsesAPITest(ABC):
raise ValueError("response is not a ResponsesAPIResponse")
else:
response = await litellm.aresponses(
input="Basic ping", max_output_tokens=20,
**base_completion_call_args
input="Basic ping", max_output_tokens=20, **base_completion_call_args
)
# async get the response
if isinstance(response, ResponsesAPIResponse):
result = await litellm.aget_responses(
response_id=response.id,
**base_completion_call_args
response_id=response.id, **base_completion_call_args
)
assert result is not None
assert result.id == response.id
@ -351,58 +352,60 @@ class BaseResponsesAPITest(ABC):
json.dumps(list_items_response, indent=4, default=str),
)
@pytest.mark.asyncio
async def test_multiturn_responses_api(self):
litellm._turn_on_debug()
litellm.set_verbose = True
base_completion_call_args = self.get_base_completion_call_args()
response_1 = await litellm.aresponses(
input="Basic ping", max_output_tokens=20, **base_completion_call_args
)
try:
base_completion_call_args = self.get_base_completion_call_args()
response_1 = await litellm.aresponses(
input="Basic ping", max_output_tokens=20, **base_completion_call_args
)
# follow up with a second request
response_1_id = response_1.id
response_2 = await litellm.aresponses(
input="Basic ping",
max_output_tokens=20,
previous_response_id=response_1_id,
**base_completion_call_args
)
# follow up with a second request
response_1_id = response_1.id
response_2 = await litellm.aresponses(
input="Basic ping",
max_output_tokens=20,
previous_response_id=response_1_id,
**base_completion_call_args,
)
# assert the response is not None
assert response_1 is not None
assert response_2 is not None
except litellm.InternalServerError:
pytest.skip("Skipping test due to litellm.InternalServerError")
# assert the response is not None
assert response_1 is not None
assert response_2 is not None
@pytest.mark.asyncio
async def test_responses_api_with_tool_calls(self):
"""Test that calls the Responses API with tool calls including function call and output"""
litellm._turn_on_debug()
litellm.set_verbose = True
base_completion_call_args = self.get_base_completion_call_args()
# Define the input with message, function call, and function call output
input_data: ResponseInputParam = [
{
"type": "message",
"role": "user",
"content": "How is the weather in São Paulo today ?"
"content": "How is the weather in São Paulo today ?",
},
{
"type": "function_call",
"arguments": "{\"location\": \"São Paulo, Brazil\"}",
"arguments": '{"location": "São Paulo, Brazil"}',
"call_id": "fc_1fe70e2a-a596-45ef-b72c-9b8567c460e5",
"name": "get_weather",
"id": "fc_1fe70e2a-a596-45ef-b72c-9b8567c460e5",
"status": "completed"
"status": "completed",
},
{
"type": "function_call_output",
"call_id": "fc_1fe70e2a-a596-45ef-b72c-9b8567c460e5",
"output": "Rainy"
}
"output": "Rainy",
},
]
# Define the tools
tools = [
{
@ -414,71 +417,67 @@ class BaseResponsesAPITest(ABC):
"properties": {
"location": {
"type": "string",
"description": "City and country e.g. Bogotá, Colombia"
"description": "City and country e.g. Bogotá, Colombia",
}
},
"required": ["location"],
"additionalProperties": False
}
"additionalProperties": False,
},
}
]
try:
# Make the responses API call
response = await litellm.aresponses(
input=input_data,
store=False,
tools=tools,
**base_completion_call_args
input=input_data, store=False, tools=tools, **base_completion_call_args
)
except litellm.InternalServerError:
pytest.skip("Skipping test due to litellm.InternalServerError")
print("litellm response=", json.dumps(response, indent=4, default=str))
# Validate the response structure
validate_responses_api_response(response, final_chunk=True)
# Additional assertions specific to tool calls
assert response is not None
assert "output" in response
assert len(response["output"]) > 0
@pytest.mark.asyncio
async def test_responses_api_multi_turn_with_reasoning_and_structured_output(self):
"""
Test multi-turn conversation with reasoning, structured output, and tool calls.
This test validates:
- First call: Model uses reasoning to process a question and makes a tool call
- Tool call handling: Function call output is properly processed
- Tool call handling: Function call output is properly processed
- Second call: Model produces structured output incorporating tool results
- Structured output: Response conforms to defined Pydantic model schema
"""
from pydantic import BaseModel
litellm._turn_on_debug()
litellm.set_verbose = True
base_completion_call_args = self.get_base_completion_reasoning_call_args()
if base_completion_call_args is None:
pytest.skip("Skipping test due to no base completion reasoning call args")
# Define tools for the conversation
tools = [{"type": "function", "name": "get_today"}]
# Define structured output schema
class Output(BaseModel):
today: str
number_of_r: str
# Initial conversation input
input_messages = [
{
"role": "user",
"role": "user",
"content": "How many r in strrawberrry? While you're thinking, you should call tool get_today. Then you output the today and number of r",
}
]
# First call - should trigger reasoning and tool call
response = await litellm.aresponses(
@ -486,49 +485,54 @@ class BaseResponsesAPITest(ABC):
tools=tools,
reasoning={"effort": "low", "summary": "detailed"},
text_format=Output,
**base_completion_call_args
**base_completion_call_args,
)
print("First call output:")
print(json.dumps(response.output, indent=4, default=str))
# Validate first response structure
validate_responses_api_response(response, final_chunk=True)
assert response.output is not None
assert len(response.output) > 0
# Extend input with first response output
input_messages.extend(response.output)
# Process any tool calls and add function outputs
function_outputs = []
for item in response.output:
if hasattr(item, 'type') and item.type in ["function_call", "custom_tool_call"]:
if hasattr(item, 'name') and item.name == "get_today":
function_outputs.append({
"type": "function_call_output",
"call_id": item.call_id,
"output": "2025-01-15"
})
if hasattr(item, "type") and item.type in [
"function_call",
"custom_tool_call",
]:
if hasattr(item, "name") and item.name == "get_today":
function_outputs.append(
{
"type": "function_call_output",
"call_id": item.call_id,
"output": "2025-01-15",
}
)
# Add function outputs to conversation
input_messages.extend(function_outputs)
print("Second call input:")
print(json.dumps(input_messages, indent=4, default=str))
# Second call - should produce structured output
final_response = await litellm.aresponses(
input=input_messages,
tools=tools,
reasoning={"effort": "low", "summary": "detailed"},
text_format=Output,
**base_completion_call_args
**base_completion_call_args,
)
print("Second call output:")
print(json.dumps(final_response.output, indent=4, default=str))
# Validate final response structure
validate_responses_api_response(final_response, final_chunk=True)
assert final_response.output is not None