litellm/deprecated_litellm_server/main.py

# import os, traceback
# from fastapi import FastAPI, Request, HTTPException
# from fastapi.routing import APIRouter
# from fastapi.responses import StreamingResponse, FileResponse
# from fastapi.middleware.cors import CORSMiddleware
# import json, sys
# from typing import Optional
# sys.path.insert(
#     0, os.path.abspath("../")
# )  # Adds the parent directory to the system path - for litellm local dev
# import litellm

# try:
#     from litellm.deprecated_litellm_server.server_utils import set_callbacks, load_router_config, print_verbose
# except ImportError:
#     from litellm.deprecated_litellm_server.server_utils import set_callbacks, load_router_config, print_verbose
# import dotenv
# dotenv.load_dotenv() # load env variables

# app = FastAPI(docs_url="/", title="LiteLLM API")
# router = APIRouter()
# origins = ["*"]

# app.add_middleware(
#     CORSMiddleware,
#     allow_origins=origins,
#     allow_credentials=True,
#     allow_methods=["*"],
#     allow_headers=["*"],
# )
# #### GLOBAL VARIABLES ####
# llm_router: Optional[litellm.Router] = None
# llm_model_list: Optional[list] = None
# server_settings: Optional[dict] = None

# set_callbacks() # sets litellm callbacks for logging if they exist in the environment

# if "CONFIG_FILE_PATH" in os.environ:
#     llm_router, llm_model_list, server_settings = load_router_config(router=llm_router, config_file_path=os.getenv("CONFIG_FILE_PATH"))
# else:
#     llm_router, llm_model_list, server_settings = load_router_config(router=llm_router)
# #### API ENDPOINTS ####
# @router.get("/v1/models")
# @router.get("/models")  # if project requires model list
# def model_list():
#     all_models = litellm.utils.get_valid_models()
#     if llm_model_list:
#         all_models += llm_model_list
#     return dict(
#         data=[
#             {
#                 "id": model,
#                 "object": "model",
#                 "created": 1677610602,
#                 "owned_by": "openai",
#             }
#             for model in all_models
#         ],
#         object="list",
#     )
# # for streaming
# def data_generator(response):

#     for chunk in response:

#         yield f"data: {json.dumps(chunk)}\n\n"

# @router.post("/v1/completions")
# @router.post("/completions")
# async def completion(request: Request):
#     data = await request.json()
#     response = litellm.completion(
#         **data
#     )
#     if 'stream' in data and data['stream'] == True: # use generate_responses to stream responses
#             return StreamingResponse(data_generator(response), media_type='text/event-stream')
#     return response

# @router.post("/v1/embeddings")
# @router.post("/embeddings")
# async def embedding(request: Request):
#     try:
#         data = await request.json()
#         # default to always using the "ENV" variables, only if AUTH_STRATEGY==DYNAMIC then reads headers
#         if os.getenv("AUTH_STRATEGY", None) == "DYNAMIC" and "authorization" in request.headers: # if users pass LLM api keys as part of header
#             api_key = request.headers.get("authorization")
#             api_key = api_key.replace("Bearer", "").strip() # type: ignore
#             if len(api_key.strip()) > 0:
#                 api_key = api_key
#                 data["api_key"] = api_key
#         response = litellm.embedding(
#             **data
#         )
#         return response
#     except Exception as e:
#         error_traceback = traceback.format_exc()
#         error_msg = f"{str(e)}\n\n{error_traceback}"
#         return {"error": error_msg}

# @router.post("/v1/chat/completions")
# @router.post("/chat/completions")
# @router.post("/openai/deployments/{model:path}/chat/completions") # azure compatible endpoint
# async def chat_completion(request: Request, model: Optional[str] = None):
#     global llm_model_list, server_settings
#     try:
#         data = await request.json()
#         server_model = server_settings.get("completion_model", None) if server_settings else None
#         data["model"] = server_model or model or data["model"]
#         ## CHECK KEYS ##
#         # default to always using the "ENV" variables, only if AUTH_STRATEGY==DYNAMIC then reads headers
#         # env_validation = litellm.validate_environment(model=data["model"])
#         # if (env_validation['keys_in_environment'] is False or os.getenv("AUTH_STRATEGY", None) == "DYNAMIC") and ("authorization" in request.headers or "api-key" in request.headers): # if users pass LLM api keys as part of header
#         #     if "authorization" in request.headers:
#         #         api_key = request.headers.get("authorization")
#         #     elif "api-key" in request.headers:
#         #         api_key = request.headers.get("api-key")
#         #     print(f"api_key in headers: {api_key}")
#         #     if " " in api_key:
#         #         api_key = api_key.split(" ")[1]
#         #     print(f"api_key split: {api_key}")
#         #     if len(api_key) > 0:
#         #         api_key = api_key
#         #         data["api_key"] = api_key
#         #         print(f"api_key in data: {api_key}")
#         ## CHECK CONFIG ##
#         if llm_model_list and data["model"] in [m["model_name"] for m in llm_model_list]:
#             for m in llm_model_list:
#                 if data["model"] == m["model_name"]:
#                     for key, value in m["litellm_params"].items():
#                         data[key] = value
#                     break
#         response = litellm.completion(
#             **data
#         )
#         if 'stream' in data and data['stream'] == True: # use generate_responses to stream responses
#                 return StreamingResponse(data_generator(response), media_type='text/event-stream')
#         return response
#     except Exception as e:
#         error_traceback = traceback.format_exc()

#         error_msg = f"{str(e)}\n\n{error_traceback}"
#         # return {"error": error_msg}
#         raise HTTPException(status_code=500, detail=error_msg)

# @router.post("/router/completions")
# async def router_completion(request: Request):
#     global llm_router
#     try:
#         data = await request.json()
#         if "model_list" in data:
#             llm_router = litellm.Router(model_list=data.pop("model_list"))
#         if llm_router is None:
#             raise Exception("Save model list via config.yaml. Eg.: ` docker build -t myapp --build-arg CONFIG_FILE=myconfig.yaml .` or pass it in as model_list=[..] as part of the request body")

#         # openai.ChatCompletion.create replacement
#         response = await llm_router.acompletion(model="gpt-3.5-turbo",
#                         messages=[{"role": "user", "content": "Hey, how's it going?"}])

#         if 'stream' in data and data['stream'] == True: # use generate_responses to stream responses
#                 return StreamingResponse(data_generator(response), media_type='text/event-stream')
#         return response
#     except Exception as e:
#         error_traceback = traceback.format_exc()
#         error_msg = f"{str(e)}\n\n{error_traceback}"
#         return {"error": error_msg}

# @router.post("/router/embedding")
# async def router_embedding(request: Request):
#     global llm_router
#     try:
#         data = await request.json()
#         if "model_list" in data:
#             llm_router = litellm.Router(model_list=data.pop("model_list"))
#         if llm_router is None:
#             raise Exception("Save model list via config.yaml. Eg.: ` docker build -t myapp --build-arg CONFIG_FILE=myconfig.yaml .` or pass it in as model_list=[..] as part of the request body")

#         response = await llm_router.aembedding(model="gpt-3.5-turbo",  # type: ignore
#                         messages=[{"role": "user", "content": "Hey, how's it going?"}])

#         if 'stream' in data and data['stream'] == True: # use generate_responses to stream responses
#                 return StreamingResponse(data_generator(response), media_type='text/event-stream')
#         return response
#     except Exception as e:
#         error_traceback = traceback.format_exc()
#         error_msg = f"{str(e)}\n\n{error_traceback}"
#         return {"error": error_msg}

# @router.get("/")
# async def home(request: Request):
#     return "LiteLLM: RUNNING"


# app.include_router(router)