| import os |
| import time |
| import json |
| import random |
| import requests |
| import g4f |
| from fastapi import FastAPI,Response, status |
| from typing import Dict, NewType, Union, Optional, List, get_type_hints |
| from fastapi.middleware.cors import CORSMiddleware |
| from fastapi.responses import StreamingResponse,JSONResponse |
| from fastapi.openapi.utils import get_openapi |
| from starlette.exceptions import HTTPException as StarletteHTTPException |
| from pydantic import BaseModel |
|
|
|
|
|
|
| |
| app = FastAPI() |
| app.add_middleware( |
| CORSMiddleware, |
| allow_origins=["*"], |
| allow_methods=["*"], |
| allow_headers=["*"], |
| ) |
|
|
| class chat_completions_Item(BaseModel): |
| stream : bool = False |
| model : str = 'gpt-3.5-turbo' |
| messages : list = [{'role': 'user', 'content':"Say 'Hello World'."}] |
| provider: Union[str, None] = None |
| temperature : float = 0.8 |
| presence_penalty: float = 0 |
| frequency_penalty: float = 0 |
| top_p: float = 1 |
|
|
| class completions_Item(BaseModel): |
| stream : bool = False |
| model : str = 'gpt-3.5-turbo' |
| prompt : str = "Say 'Hello World'." |
| provider : Union[str, None] = None |
| temperature : float = 0.8 |
| presence_penalty: float = 0 |
| frequency_penalty: float = 0 |
| top_p: float = 1 |
|
|
| def auto_select(model:str='gpt-3.5-turbo',stream:bool=False): |
| r = requests.get('https://gpt.lemonsoftware.eu.org/v1/status') |
| data = r.json()['data'] |
| model_providers = set() |
| random.shuffle(data) |
| for provider_info in data: |
| for model_info in provider_info['model']: |
| if model in model_info: |
| model_providers.add(provider_info['provider']) |
| if model_info[model]['status'] == 'Active': |
| if stream == True and getattr(g4f.Provider,provider_info['provider']).supports_stream == False: |
| continue |
| return [getattr(g4f.Provider,provider_info['provider']),provider_info['provider']] |
| else: |
| continue |
| break |
| if not model_providers: |
| return None |
| active_providers = set() |
| for provider_info in data: |
| for model_info in provider_info['model']: |
| for model in model_info.values(): |
| if model['status'] == 'Active': |
| if stream == True and getattr(g4f.Provider,provider_info['provider']).supports_stream == False: |
| continue |
| active_providers.add(provider_info['provider']) |
| chooseable_providers = model_providers & active_providers |
| if not chooseable_providers: |
| return None |
| chooseable_provider = random.choice(list(chooseable_providers)) |
| return [getattr(g4f.Provider,chooseable_provider),chooseable_provider] |
|
|
| @app.post("/v1/chat/completions") |
| def chat_completions(item: chat_completions_Item,responses: Response): |
| stream = item.stream |
| model = item.model.lower() |
| messages = item.messages |
| provider_name = item.provider |
| temperature = item.temperature |
| presence_penalty = item.presence_penalty |
| frequency_penalty = item.frequency_penalty |
| top_p = item.top_p |
|
|
| if provider_name: |
| try: |
| response = g4f.ChatCompletion.create(model=model, provider=getattr(g4f.Provider,provider_name),stream=stream,messages=messages,temperature=temperature,presence_penalty=presence_penalty,frequency_penalty=frequency_penalty,top_p=top_p) |
| except: |
| return JSONResponse(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, content={"error": {"message": "There was an error","type": "invalid_request_error","param": None,"code": 500}}) |
| else: |
| provider = auto_select(model=model,stream=stream) |
| if provider == None: |
| return JSONResponse(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, content={"error": {"message": "The model is invalid or not working.","type": "invalid_request_error","param": None,"code": 500}}) |
| if stream and provider[0].supports_stream == False: |
| return JSONResponse(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, content={"error": {"message": "Stream is not supported.","type": "invalid_request_error","param": None,"code": 500}}) |
| provider_name = provider[1] |
| try: |
| response = g4f.ChatCompletion.create(model=model, provider=provider[0],stream=stream,messages=messages,temperature=temperature,presence_penalty=presence_penalty,frequency_penalty=frequency_penalty,top_p=top_p) |
| except: |
| provider = auto_select(model=model,stream=stream) |
| if provider == None: |
| return JSONResponse(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, content={"error": {"message": "The model is invalid or not working.","type": "invalid_request_error","param": None,"code": 500}}) |
| if stream and provider[0].supports_stream == False: |
| return JSONResponse(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, content={"error": {"message": "Stream is not supported.","type": "invalid_request_error","param": None,"code": 500}}) |
| provider_name = provider[1] |
| try: |
| response = g4f.ChatCompletion.create(model=model, provider=provider[0],stream=stream,messages=messages,temperature=temperature,presence_penalty=presence_penalty,frequency_penalty=frequency_penalty,top_p=top_p) |
| except: |
| return JSONResponse(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, content={"error": {"message": "There was an error.Please try again.","type": "server_error","param": None,"code": 500}}) |
| |
| |
| if not stream: |
| completion_timestamp = int(time.time()) |
| completion_id = ''.join(random.choices( |
| 'ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789', k=28)) |
| return { |
| 'id': 'chatcmpl-%s' % completion_id, |
| 'object': 'chat.completion', |
| 'created': completion_timestamp, |
| 'model': model, |
| 'provider':provider_name, |
| 'usage': { |
| 'prompt_tokens': len(messages), |
| 'completion_tokens': len(response), |
| 'total_tokens': len(messages)+len(response) |
| }, |
| 'choices': [{ |
| 'message': { |
| 'role': 'assistant', |
| 'content': response |
| }, |
| 'finish_reason': 'stop', |
| 'index': 0 |
| }] |
| } |
| |
| def stream(): |
| nonlocal response |
| for token in response: |
| completion_timestamp = int(time.time()) |
| completion_id = ''.join(random.choices( |
| 'ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789', k=28)) |
| completion_data = { |
| 'id': f'chatcmpl-{completion_id}', |
| 'object': 'chat.completion.chunk', |
| 'created': completion_timestamp, |
| 'model': model, |
| 'provider':provider_name, |
| 'choices': [ |
| { |
| 'delta': { |
| 'content': token |
| }, |
| 'index': 0, |
| 'finish_reason': None |
| } |
| ] |
| } |
| yield 'data: %s\n\n' % json.dumps(completion_data, separators=(',' ':')) |
| time.sleep(0.1) |
| return StreamingResponse(stream(), media_type='text/event-stream') |
|
|
| @app.post("/v1/completions") |
| def completions(item: completions_Item,responses: Response): |
| stream = item.stream |
| model = item.model.lower() |
| messages = [{'role': 'user', 'content':item.prompt}] |
| provider_name = item.provider |
| temperature = item.temperature |
| presence_penalty = item.presence_penalty |
| frequency_penalty = item.frequency_penalty |
| top_p = item.top_p |
| if provider_name: |
| try: |
| response = g4f.ChatCompletion.create(model=model, provider=getattr(g4f.Provider,provider_name),stream=stream,messages=messages,temperature=temperature,presence_penalty=presence_penalty,frequency_penalty=frequency_penalty,top_p=top_p) |
| except: |
| return JSONResponse(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, content={"error": {"message": "There was an error.","type": "invalid_request_error","param": None,"code": 500}}) |
|
|
| else: |
| provider = auto_select(model=model,stream=stream) |
| if provider == None: |
| return JSONResponse(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, content={"error": {"message": "The model is invalid or not working.","type": "invalid_request_error","param": None,"code": 500}}) |
| if stream and provider[0].supports_stream == False: |
| return JSONResponse(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, content={"error": {"message": "Stream is not supported.","type": "invalid_request_error","param": None,"code": 500}}) |
| provider_name = provider[1] |
| try: |
| response = g4f.ChatCompletion.create(model=model, provider=provider[0],stream=stream,messages=messages,temperature=temperature,presence_penalty=presence_penalty,frequency_penalty=frequency_penalty,top_p=top_p) |
| except: |
| provider = auto_select(model=model,stream=stream) |
| if provider == None: |
| return JSONResponse(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, content={"error": {"message": "The model is invalid or not working.","type": "invalid_request_error","param": None,"code": 500}}) |
| if stream and provider[0].supports_stream == False: |
| return JSONResponse(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, content={"error": {"message": "Stream is not supported.","type": "invalid_request_error","param": None,"code": 500}}) |
| provider_name = provider[1] |
| try: |
| response = g4f.ChatCompletion.create(model=model, provider=provider[0],stream=stream,messages=messages,temperature=temperature,presence_penalty=presence_penalty,frequency_penalty=frequency_penalty,top_p=top_p) |
| except: |
| return JSONResponse(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, content={"error": {"message": "There was an error.Please try again.","type": "server_error","param": None,"code": 500}}) |
| |
| if not stream: |
| completion_timestamp = int(time.time()) |
| completion_id = ''.join(random.choices( |
| 'ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789', k=28)) |
| return { |
| 'id': 'cmpl-%s' % completion_id, |
| 'object': 'text.completion', |
| 'created': completion_timestamp, |
| 'model': model, |
| 'provider':provider_name, |
| 'usage': { |
| 'prompt_tokens': len(messages), |
| 'completion_tokens': len(response), |
| 'total_tokens': len(messages)+len(response) |
| }, |
| 'choices': [{ |
| 'text': response, |
| 'finish_reason': 'length', |
| "logprobs": None, |
| 'index': 0 |
| }] |
| } |
| |
| def stream(): |
| nonlocal response |
| for token in response: |
| completion_timestamp = int(time.time()) |
| completion_id = ''.join(random.choices( |
| 'ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789', k=28)) |
| completion_data = { |
| 'id': f'cmpl-{completion_id}', |
| 'object': 'text.completion.chunk', |
| 'created': completion_timestamp, |
| 'model': model, |
| 'provider':provider_name, |
| 'choices': [ |
| { |
| 'delta': { |
| 'text': token |
| }, |
| 'index': 0, |
| 'finish_reason': None |
| } |
| ] |
| } |
| yield 'data: %s\n\n' % json.dumps(completion_data, separators=(',' ':')) |
| time.sleep(0.1) |
| return StreamingResponse(stream(), media_type='text/event-stream') |
|
|
| @app.get("/v1/dashboard/billing/subscription") |
| @app.get("/dashboard/billing/subscription") |
| async def billing_subscription(): |
| return { |
| "object": "billing_subscription", |
| "has_payment_method": True, |
| "canceled": False, |
| "canceled_at": None, |
| "delinquent": None, |
| "access_until": 2556028800, |
| "soft_limit": 6944500, |
| "hard_limit": 166666666, |
| "system_hard_limit": 166666666, |
| "soft_limit_usd": 416.67, |
| "hard_limit_usd": 9999.99996, |
| "system_hard_limit_usd": 9999.99996, |
| "plan": { |
| "title": "Pay-as-you-go", |
| "id": "payg" |
| }, |
| "primary": True, |
| "account_name": "OpenAI", |
| "po_number": None, |
| "billing_email": None, |
| "tax_ids": None, |
| "billing_address": { |
| "city": "New York", |
| "line1": "OpenAI", |
| "country": "US", |
| "postal_code": "NY10031" |
| }, |
| "business_address": None |
| } |
|
|
|
|
| @app.get("/v1/dashboard/billing/usage") |
| @app.get("/dashboard/billing/usage") |
| async def billing_usage(start_date:str='2023-01-01',end_date:str='2023-01-31'): |
| return { |
| "object": "list", |
| "daily_costs": [ |
| { |
| "timestamp": time.time(), |
| "line_items": [ |
| { |
| "name": "GPT-4", |
| "cost": 0.0 |
| }, |
| { |
| "name": "Chat models", |
| "cost": 1.01 |
| }, |
| { |
| "name": "InstructGPT", |
| "cost": 0.0 |
| }, |
| { |
| "name": "Fine-tuning models", |
| "cost": 0.0 |
| }, |
| { |
| "name": "Embedding models", |
| "cost": 0.0 |
| }, |
| { |
| "name": "Image models", |
| "cost": 16.0 |
| }, |
| { |
| "name": "Audio models", |
| "cost": 0.0 |
| } |
| ] |
| } |
| ], |
| "total_usage": 1.01 |
| } |
|
|
| @app.get("/v1/models") |
| def models(): |
| import g4f.models |
| model = {"data":[]} |
| for i in g4f.models.ModelUtils.convert: |
| model['data'].append({ |
| "id": i, |
| "object": "model", |
| "owned_by": g4f.models.ModelUtils.convert[i].base_provider, |
| "tokens": 99999, |
| "fallbacks": None, |
| "endpoints": [ |
| "/v1/chat/completions" |
| ], |
| "limits": None, |
| "permission": [] |
| }) |
| return model |
|
|
| @app.get("/v1/providers") |
| async def providers(): |
| files = os.listdir("g4f/Provider/Providers") |
| files = [f for f in files if os.path.isfile(os.path.join("g4f/Provider/Providers", f))] |
| files.sort(key=str.lower) |
| providers_data = {"data":[]} |
| for file in files: |
| if file.endswith(".py"): |
| name = file[:-3] |
| try: |
| p = getattr(g4f.Provider,name) |
| providers_data["data"].append({ |
| "provider": str(name), |
| "model": list(p.model), |
| "url": str(p.url), |
| "working": bool(p.working), |
| "supports_stream": bool(p.supports_stream) |
| }) |
| except: |
| pass |
| return providers_data |
|
|
| def custom_openapi(): |
| if app.openapi_schema: |
| return app.openapi_schema |
| openapi_schema = get_openapi( |
| title="GPT API", |
| version="1.0.0", |
| summary="GPT API", |
| description="[Try Online](https://chatgpt-next-web.lemonsoftware.eu.org/)", |
| routes=app.routes, |
| ) |
| openapi_schema["info"]["x-logo"] = { |
| "url": "https://gpt-status.lemonsoftware.eu.org/icon.svg" |
| } |
| app.openapi_schema = openapi_schema |
| return app.openapi_schema |
| app.openapi = custom_openapi |
|
|
| @app.exception_handler(StarletteHTTPException) |
|
|
| async def custom_http_exception_handler(request, exc): |
| return JSONResponse(status_code=status.HTTP_404_NOT_FOUND, content={"error": {"message": "Invalid URL","type": "invalid_request_error","param": None,"code": 404}}) |
|
|
| if __name__ == '__main__': |
| import uvicorn |
| uvicorn.run(app, host="0.0.0.0", port=7860) |
