gpt4free/g4f/api/__init__.py

265 lines
9.9 KiB
Python
Raw Normal View History

from __future__ import annotations
import logging
2023-10-12 04:35:11 +03:00
import json
2023-11-02 04:27:35 +03:00
import uvicorn
2024-04-28 12:02:44 +03:00
import secrets
from fastapi import FastAPI, Response, Request
2024-02-23 04:35:13 +03:00
from fastapi.responses import StreamingResponse, RedirectResponse, HTMLResponse, JSONResponse
from fastapi.exceptions import RequestValidationError
2024-04-28 12:02:44 +03:00
from fastapi.security import APIKeyHeader
from starlette.exceptions import HTTPException
from starlette.status import HTTP_422_UNPROCESSABLE_ENTITY, HTTP_401_UNAUTHORIZED, HTTP_403_FORBIDDEN
from fastapi.encoders import jsonable_encoder
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
from typing import Union, Optional, Iterator
2023-11-05 00:16:09 +03:00
import g4f
2024-02-23 04:35:13 +03:00
import g4f.debug
from g4f.client import Client, ChatCompletion, ChatCompletionChunk, ImagesResponse
2024-02-23 04:35:13 +03:00
from g4f.typing import Messages
2024-04-29 21:21:47 +03:00
from g4f.cookies import read_cookie_files
2024-02-23 04:35:13 +03:00
def create_app(g4f_api_key: str = None):
app = FastAPI()
# Add CORS middleware
app.add_middleware(
CORSMiddleware,
allow_origin_regex=".*",
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
api = Api(app, g4f_api_key=g4f_api_key)
api.register_routes()
2024-04-28 12:02:44 +03:00
api.register_authorization()
api.register_validation_exception_handler()
# Read cookie files if not ignored
2024-04-29 21:21:47 +03:00
if not AppConfig.ignore_cookie_files:
read_cookie_files()
return app
2024-04-29 17:56:56 +03:00
class ChatCompletionsConfig(BaseModel):
2024-02-23 04:35:13 +03:00
messages: Messages
model: str
provider: Optional[str] = None
2024-02-23 04:35:13 +03:00
stream: bool = False
temperature: Optional[float] = None
max_tokens: Optional[int] = None
stop: Union[list[str], str, None] = None
api_key: Optional[str] = None
web_search: Optional[bool] = None
proxy: Optional[str] = None
2023-11-02 04:27:35 +03:00
class ImageGenerationConfig(BaseModel):
prompt: str
model: Optional[str] = None
response_format: str = "url"
class AppConfig:
ignored_providers: Optional[list[str]] = None
2024-04-29 21:21:47 +03:00
g4f_api_key: Optional[str] = None
ignore_cookie_files: bool = False
2024-05-06 17:42:56 +03:00
defaults: dict = {}
2024-04-29 21:21:47 +03:00
@classmethod
2024-05-06 09:16:49 +03:00
def set_config(cls, **data):
for key, value in data.items():
setattr(cls, key, value)
2024-04-29 17:56:56 +03:00
list_ignored_providers: list[str] = None
def set_list_ignored_providers(ignored: list[str]):
global list_ignored_providers
list_ignored_providers = ignored
2023-11-05 00:16:09 +03:00
class Api:
def __init__(self, app: FastAPI, g4f_api_key=None) -> None:
self.app = app
self.client = Client()
self.g4f_api_key = g4f_api_key
2024-04-28 12:02:44 +03:00
self.get_g4f_api_key = APIKeyHeader(name="g4f-api-key")
def register_authorization(self):
@self.app.middleware("http")
async def authorization(request: Request, call_next):
if self.g4f_api_key and request.url.path in ["/v1/chat/completions", "/v1/completions", "/v1/images/generate"]:
2024-04-28 12:02:44 +03:00
try:
user_g4f_api_key = await self.get_g4f_api_key(request)
except HTTPException as e:
if e.status_code == 403:
return JSONResponse(
status_code=HTTP_401_UNAUTHORIZED,
content=jsonable_encoder({"detail": "G4F API key required"}),
)
if not secrets.compare_digest(self.g4f_api_key, user_g4f_api_key):
2024-04-28 12:02:44 +03:00
return JSONResponse(
2024-04-29 17:56:56 +03:00
status_code=HTTP_403_FORBIDDEN,
content=jsonable_encoder({"detail": "Invalid G4F API key"}),
)
response = await call_next(request)
return response
def register_validation_exception_handler(self):
@self.app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
details = exc.errors()
modified_details = []
for error in details:
modified_details.append({
"loc": error["loc"],
"message": error["msg"],
"type": error["type"],
})
return JSONResponse(
status_code=HTTP_422_UNPROCESSABLE_ENTITY,
content=jsonable_encoder({"detail": modified_details}),
)
2023-11-05 00:16:09 +03:00
def register_routes(self):
@self.app.get("/")
2023-11-05 00:16:09 +03:00
async def read_root():
2024-02-23 04:35:13 +03:00
return RedirectResponse("/v1", 302)
2023-11-05 00:16:09 +03:00
@self.app.get("/v1")
2023-11-05 00:16:09 +03:00
async def read_root_v1():
2024-02-23 04:35:13 +03:00
return HTMLResponse('g4f API: Go to '
'<a href="/v1/chat/completions">chat/completions</a>, '
'<a href="/v1/models">models</a>, or '
'<a href="/v1/images/generate">images/generate</a>.')
2023-11-05 00:16:09 +03:00
@self.app.get("/v1/models")
2023-11-05 00:16:09 +03:00
async def models():
model_list = dict(
(model, g4f.models.ModelUtils.convert[model])
2024-02-23 04:35:13 +03:00
for model in g4f.Model.__all__()
)
2024-02-23 04:35:13 +03:00
model_list = [{
'id': model_id,
2023-11-05 00:16:09 +03:00
'object': 'model',
'created': 0,
2024-02-23 04:35:13 +03:00
'owned_by': model.base_provider
} for model_id, model in model_list.items()]
return JSONResponse(model_list)
2023-11-05 00:16:09 +03:00
@self.app.get("/v1/models/{model_name}")
2023-11-05 00:16:09 +03:00
async def model_info(model_name: str):
try:
model_info = g4f.models.ModelUtils.convert[model_name]
2024-02-23 04:35:13 +03:00
return JSONResponse({
2023-11-05 00:16:09 +03:00
'id': model_name,
'object': 'model',
'created': 0,
'owned_by': model_info.base_provider
2024-02-23 04:35:13 +03:00
})
2023-11-05 00:16:09 +03:00
except:
2024-02-23 04:35:13 +03:00
return JSONResponse({"error": "The model does not exist."})
2023-11-05 00:16:09 +03:00
@self.app.post("/v1/chat/completions")
async def chat_completions(config: ChatCompletionsConfig, request: Request = None, provider: str = None):
2023-11-05 00:16:09 +03:00
try:
2024-02-23 04:35:13 +03:00
config.provider = provider if config.provider is None else config.provider
if config.api_key is None and request is not None:
2024-02-23 04:35:13 +03:00
auth_header = request.headers.get("Authorization")
if auth_header is not None:
auth_header = auth_header.split(None, 1)[-1]
if auth_header and auth_header != "Bearer":
config.api_key = auth_header
# Create the completion response
response = self.client.chat.completions.create(
**{
**AppConfig.defaults,
**config.dict(exclude_none=True),
},
ignored=AppConfig.ignored_providers
2023-12-23 22:50:56 +03:00
)
# Check if the response is synchronous or asynchronous
if isinstance(response, ChatCompletion):
# Synchronous response
return JSONResponse(response.to_json())
if not config.stream:
# If the response is an iterator but not streaming, collect the result
response_list = list(response) if isinstance(response, Iterator) else [response]
return JSONResponse(response_list[0].to_json())
# Streaming response
async def streaming():
try:
async for chunk in response:
yield f"data: {json.dumps(chunk.to_json())}\n\n"
except GeneratorExit:
pass
except Exception as e:
logging.exception(e)
yield f'data: {format_exception(e, config)}\n\n'
yield "data: [DONE]\n\n"
return StreamingResponse(streaming(), media_type="text/event-stream")
except Exception as e:
logging.exception(e)
2024-02-23 04:35:13 +03:00
return Response(content=format_exception(e, config), status_code=500, media_type="application/json")
@self.app.post("/v1/images/generate")
async def generate_image(config: ImageGenerationConfig):
try:
response: ImagesResponse = await self.client.images.async_generate(
prompt=config.prompt,
model=config.model,
response_format=config.response_format
)
# Convert Image objects to dictionaries
response_data = [image.to_dict() for image in response.data]
return JSONResponse({"data": response_data})
except Exception as e:
logging.exception(e)
return Response(content=format_exception(e, config), status_code=500, media_type="application/json")
2023-11-02 04:27:35 +03:00
@self.app.post("/v1/completions")
async def completions():
return Response(content=json.dumps({'info': 'Not working yet.'}, indent=4), media_type="application/json")
def format_exception(e: Exception, config: Union[ChatCompletionsConfig, ImageGenerationConfig]) -> str:
2024-02-23 04:35:13 +03:00
last_provider = g4f.get_last_provider(True)
return json.dumps({
"error": {"message": f"{e.__class__.__name__}: {e}"},
"model": last_provider.get("model") if last_provider else getattr(config, 'model', None),
"provider": last_provider.get("name") if last_provider else getattr(config, 'provider', None)
})
def run_api(
host: str = '0.0.0.0',
port: int = 1337,
bind: str = None,
debug: bool = False,
workers: int = None,
use_colors: bool = None,
g4f_api_key: str = None
) -> None:
print(f'Starting server... [g4f v-{g4f.version.utils.current_version}]' + (" (debug)" if debug else ""))
if use_colors is None:
use_colors = debug
if bind is not None:
host, port = bind.split(":")
if debug:
g4f.debug.logging = True
2024-04-29 17:56:56 +03:00
uvicorn.run(
"g4f.api:create_app",
host=host,
port=int(port),
workers=workers,
use_colors=use_colors,
factory=True,
2024-04-29 17:56:56 +03:00
reload=debug
)