Spaces:

nonhuman
/

nnnn

Runtime error

App Files Files Community

nnnn / litellm /tests /test_exceptions.py

nonhuman

Upload 165 files

395201c 7 months ago

raw history blame contribute delete

No virus

11.6 kB

	from openai import AuthenticationError, BadRequestError, RateLimitError, OpenAIError
	import os
	import sys
	import traceback
	import subprocess

	sys.path.insert(
	0, os.path.abspath("../..")
	) # Adds the parent directory to the system path
	import litellm
	from litellm import (
	embedding,
	completion,
	# AuthenticationError,
	ContextWindowExceededError,
	# RateLimitError,
	# ServiceUnavailableError,
	# OpenAIError,
	)
	from concurrent.futures import ThreadPoolExecutor
	import pytest
	litellm.vertex_project = "pathrise-convert-1606954137718"
	litellm.vertex_location = "us-central1"

	# litellm.failure_callback = ["sentry"]
	#### What this tests ####
	# This tests exception mapping -> trigger an exception from an llm provider -> assert if output is of the expected type


	# 5 providers -> OpenAI, Azure, Anthropic, Cohere, Replicate

	# 3 main types of exceptions -> - Rate Limit Errors, Context Window Errors, Auth errors (incorrect/rotated key, etc.)

	# Approach: Run each model through the test -> assert if the correct error (always the same one) is triggered

	models = ["command-nightly"]

	# Test 1: Context Window Errors
	@pytest.mark.parametrize("model", models)
	def test_context_window(model):
	sample_text = "Say error 50 times" * 1000000
	messages = [{"content": sample_text, "role": "user"}]
	try:
	litellm.set_verbose = False
	response = completion(model=model, messages=messages)
	print(f"response: {response}")
	print("FAILED!")
	pytest.fail(f"An exception occurred")
	except ContextWindowExceededError as e:
	print(f"Worked!")
	except RateLimitError:
	print("RateLimited!")
	except Exception as e:
	print(f"{e}")
	pytest.fail(f"An error occcurred - {e}")

	@pytest.mark.parametrize("model", models)
	def test_context_window_with_fallbacks(model):
	ctx_window_fallback_dict = {"command-nightly": "claude-2", "gpt-3.5-turbo-instruct": "gpt-3.5-turbo-16k", "azure/chatgpt-v-2": "gpt-3.5-turbo-16k"}
	sample_text = "how does a court case get to the Supreme Court?" * 1000
	messages = [{"content": sample_text, "role": "user"}]

	completion(model=model, messages=messages, context_window_fallback_dict=ctx_window_fallback_dict)

	# for model in litellm.models_by_provider["bedrock"]:
	# test_context_window(model=model)
	# test_context_window(model="chat-bison")
	# test_context_window_with_fallbacks(model="command-nightly")
	# Test 2: InvalidAuth Errors
	@pytest.mark.parametrize("model", models)
	def invalid_auth(model): # set the model key to an invalid key, depending on the model
	messages = [{"content": "Hello, how are you?", "role": "user"}]
	temporary_key = None
	try:
	if model == "gpt-3.5-turbo" or model == "gpt-3.5-turbo-instruct":
	temporary_key = os.environ["OPENAI_API_KEY"]
	os.environ["OPENAI_API_KEY"] = "bad-key"
	elif "bedrock" in model:
	temporary_aws_access_key = os.environ["AWS_ACCESS_KEY_ID"]
	os.environ["AWS_ACCESS_KEY_ID"] = "bad-key"
	temporary_aws_region_name = os.environ["AWS_REGION_NAME"]
	os.environ["AWS_REGION_NAME"] = "bad-key"
	temporary_secret_key = os.environ["AWS_SECRET_ACCESS_KEY"]
	os.environ["AWS_SECRET_ACCESS_KEY"] = "bad-key"
	elif model == "azure/chatgpt-v-2":
	temporary_key = os.environ["AZURE_API_KEY"]
	os.environ["AZURE_API_KEY"] = "bad-key"
	elif model == "claude-instant-1":
	temporary_key = os.environ["ANTHROPIC_API_KEY"]
	os.environ["ANTHROPIC_API_KEY"] = "bad-key"
	elif model == "command-nightly":
	temporary_key = os.environ["COHERE_API_KEY"]
	os.environ["COHERE_API_KEY"] = "bad-key"
	elif "j2" in model:
	temporary_key = os.environ["AI21_API_KEY"]
	os.environ["AI21_API_KEY"] = "bad-key"
	elif "togethercomputer" in model:
	temporary_key = os.environ["TOGETHERAI_API_KEY"]
	os.environ["TOGETHERAI_API_KEY"] = "84060c79880fc49df126d3e87b53f8a463ff6e1c6d27fe64207cde25cdfcd1f24a"
	elif model in litellm.openrouter_models:
	temporary_key = os.environ["OPENROUTER_API_KEY"]
	os.environ["OPENROUTER_API_KEY"] = "bad-key"
	elif model in litellm.aleph_alpha_models:
	temporary_key = os.environ["ALEPH_ALPHA_API_KEY"]
	os.environ["ALEPH_ALPHA_API_KEY"] = "bad-key"
	elif model in litellm.nlp_cloud_models:
	temporary_key = os.environ["NLP_CLOUD_API_KEY"]
	os.environ["NLP_CLOUD_API_KEY"] = "bad-key"
	elif (
	model
	== "replicate/llama-2-70b-chat:2c1608e18606fad2812020dc541930f2d0495ce32eee50074220b87300bc16e1"
	):
	temporary_key = os.environ["REPLICATE_API_KEY"]
	os.environ["REPLICATE_API_KEY"] = "bad-key"
	print(f"model: {model}")
	response = completion(
	model=model, messages=messages
	)
	print(f"response: {response}")
	except AuthenticationError as e:
	print(f"AuthenticationError Caught Exception - {str(e)}")
	except (
	OpenAIError
	) as e: # is at least an openai error -> in case of random model errors - e.g. overloaded server
	print(f"OpenAIError Caught Exception - {e}")
	except Exception as e:
	print(type(e))
	print(type(AuthenticationError))
	print(e.__class__.__name__)
	print(f"Uncaught Exception - {e}")
	pytest.fail(f"Error occurred: {e}")
	if temporary_key != None: # reset the key
	if model == "gpt-3.5-turbo":
	os.environ["OPENAI_API_KEY"] = temporary_key
	elif model == "chatgpt-test":
	os.environ["AZURE_API_KEY"] = temporary_key
	azure = True
	elif model == "claude-instant-1":
	os.environ["ANTHROPIC_API_KEY"] = temporary_key
	elif model == "command-nightly":
	os.environ["COHERE_API_KEY"] = temporary_key
	elif (
	model
	== "replicate/llama-2-70b-chat:2c1608e18606fad2812020dc541930f2d0495ce32eee50074220b87300bc16e1"
	):
	os.environ["REPLICATE_API_KEY"] = temporary_key
	elif "j2" in model:
	os.environ["AI21_API_KEY"] = temporary_key
	elif ("togethercomputer" in model):
	os.environ["TOGETHERAI_API_KEY"] = temporary_key
	elif model in litellm.aleph_alpha_models:
	os.environ["ALEPH_ALPHA_API_KEY"] = temporary_key
	elif model in litellm.nlp_cloud_models:
	os.environ["NLP_CLOUD_API_KEY"] = temporary_key
	elif "bedrock" in model:
	os.environ["AWS_ACCESS_KEY_ID"] = temporary_aws_access_key
	os.environ["AWS_REGION_NAME"] = temporary_aws_region_name
	os.environ["AWS_SECRET_ACCESS_KEY"] = temporary_secret_key
	return

	# for model in litellm.models_by_provider["bedrock"]:
	# invalid_auth(model=model)
	# invalid_auth(model="command-nightly")

	# Test 3: Invalid Request Error
	@pytest.mark.parametrize("model", models)
	def test_invalid_request_error(model):
	messages = [{"content": "hey, how's it going?", "role": "user"}]

	with pytest.raises(BadRequestError):
	completion(model=model, messages=messages, max_tokens="hello world")



	def test_completion_azure_exception():
	try:
	import openai
	print("azure gpt-3.5 test\n\n")
	litellm.set_verbose=False
	## Test azure call
	old_azure_key = os.environ["AZURE_API_KEY"]
	os.environ["AZURE_API_KEY"] = "good morning"
	response = completion(
	model="azure/chatgpt-v-2",
	messages=[
	{
	"role": "user",
	"content": "hello"
	}
	],
	)
	print(f"response: {response}")
	print(response)
	except openai.AuthenticationError as e:
	os.environ["AZURE_API_KEY"] = old_azure_key
	print("good job got the correct error for azure when key not set")
	except Exception as e:
	pytest.fail(f"Error occurred: {e}")
	test_completion_azure_exception()

	async def asynctest_completion_azure_exception():
	try:
	import openai
	import litellm
	print("azure gpt-3.5 test\n\n")
	litellm.set_verbose=False
	## Test azure call
	old_azure_key = os.environ["AZURE_API_KEY"]
	os.environ["AZURE_API_KEY"] = "good morning"
	response = await litellm.acompletion(
	model="azure/chatgpt-v-2",
	messages=[
	{
	"role": "user",
	"content": "hello"
	}
	],
	)
	print(f"response: {response}")
	print(response)
	except openai.AuthenticationError as e:
	os.environ["AZURE_API_KEY"] = old_azure_key
	print("good job got the correct error for azure when key not set")
	print(e)
	except Exception as e:
	print("Got wrong exception")
	print("exception", e)
	pytest.fail(f"Error occurred: {e}")

	# import asyncio
	# asyncio.run(
	# asynctest_completion_azure_exception()
	# )


	def test_completion_openai_exception():
	# test if openai:gpt raises openai.AuthenticationError
	try:
	import openai
	print("openai gpt-3.5 test\n\n")
	litellm.set_verbose=False
	## Test azure call
	old_azure_key = os.environ["OPENAI_API_KEY"]
	os.environ["OPENAI_API_KEY"] = "good morning"
	response = completion(
	model="gpt-4",
	messages=[
	{
	"role": "user",
	"content": "hello"
	}
	],
	)
	print(f"response: {response}")
	print(response)
	except openai.AuthenticationError as e:
	os.environ["OPENAI_API_KEY"] = old_azure_key
	print("good job got the correct error for openai when key not set")
	except Exception as e:
	pytest.fail(f"Error occurred: {e}")
	# test_completion_openai_exception()




	# # test_invalid_request_error(model="command-nightly")
	# # Test 3: Rate Limit Errors
	# def test_model_call(model):
	# try:
	# sample_text = "how does a court case get to the Supreme Court?"
	# messages = [{ "content": sample_text,"role": "user"}]
	# print(f"model: {model}")
	# response = completion(model=model, messages=messages)
	# except RateLimitError as e:
	# print(f"headers: {e.response.headers}")
	# return True
	# # except OpenAIError: # is at least an openai error -> in case of random model errors - e.g. overloaded server
	# # return True
	# except Exception as e:
	# print(f"Uncaught Exception {model}: {type(e).__name__} - {e}")
	# traceback.print_exc()
	# pass
	# return False
	# # Repeat each model 500 times
	# # extended_models = [model for model in models for _ in range(250)]
	# extended_models = ["azure/chatgpt-v-2" for _ in range(250)]

	# def worker(model):
	# return test_model_call(model)

	# # Create a dictionary to store the results
	# counts = {True: 0, False: 0}

	# # Use Thread Pool Executor
	# with ThreadPoolExecutor(max_workers=500) as executor:
	# # Use map to start the operation in thread pool
	# results = executor.map(worker, extended_models)

	# # Iterate over results and count True/False
	# for result in results:
	# counts[result] += 1

	# accuracy_score = counts[True]/(counts[True] + counts[False])
	# print(f"accuracy_score: {accuracy_score}")