from __future__ import annotations
import random
import json
import re
import requests
from requests.packages.urllib3.exceptions import InsecureRequestWarning
requests.packages.urllib3.disable_warnings(InsecureRequestWarning)
from ..typing import AsyncResult, Messages
from .base_provider import AsyncGeneratorProvider, ProviderModelMixin
from ..image import ImageResponse
from ..requests import StreamSession, raise_for_status
class Airforce(AsyncGeneratorProvider, ProviderModelMixin):
url = "https://llmplayground.net"
api_endpoint_completions = "https://api.airforce/chat/completions"
api_endpoint_imagine = "https://api.airforce/imagine2"
working = True
supports_system_message = True
supports_message_history = True
@classmethod
def fetch_completions_models(cls):
response = requests.get('https://api.airforce/models', verify=False)
response.raise_for_status()
data = response.json()
return [model['id'] for model in data['data']]
@classmethod
def fetch_imagine_models(cls):
response = requests.get('https://api.airforce/imagine/models', verify=False)
response.raise_for_status()
return response.json()
default_model = "gpt-4o-mini"
default_image_model = "flux"
additional_models_imagine = ["stable-diffusion-xl-base", "stable-diffusion-xl-lightning", "Flux-1.1-Pro"]
@classmethod
def get_models(cls):
if not cls.models:
cls.image_models = [*cls.fetch_imagine_models(), *cls.additional_models_imagine]
cls.models = [
*cls.fetch_completions_models(),
*cls.image_models
]
return cls.models
model_aliases = {
### completions ###
# openchat
"openchat-3.5": "openchat-3.5-0106",
# deepseek-ai
"deepseek-coder": "deepseek-coder-6.7b-instruct",
# NousResearch
"hermes-2-dpo": "Nous-Hermes-2-Mixtral-8x7B-DPO",
"hermes-2-pro": "hermes-2-pro-mistral-7b",
# teknium
"openhermes-2.5": "openhermes-2.5-mistral-7b",
# liquid
"lfm-40b": "lfm-40b-moe",
# DiscoResearch
"german-7b": "discolm-german-7b-v1",
# meta-llama
"llama-2-7b": "llama-2-7b-chat-int8",
"llama-2-7b": "llama-2-7b-chat-fp16",
"llama-3.1-70b": "llama-3.1-70b-chat",
"llama-3.1-8b": "llama-3.1-8b-chat",
"llama-3.1-70b": "llama-3.1-70b-turbo",
"llama-3.1-8b": "llama-3.1-8b-turbo",
# inferless
"neural-7b": "neural-chat-7b-v3-1",
# HuggingFaceH4
"zephyr-7b": "zephyr-7b-beta",
### imagine ###
"sdxl": "stable-diffusion-xl-base",
"sdxl": "stable-diffusion-xl-lightning",
"flux-pro": "Flux-1.1-Pro",
}
@classmethod
def create_async_generator(
cls,
model: str,
messages: Messages,
proxy: str = None,
seed: int = None,
size: str = "1:1", # "1:1", "16:9", "9:16", "21:9", "9:21", "1:2", "2:1"
stream: bool = False,
**kwargs
) -> AsyncResult:
model = cls.get_model(model)
if model in cls.image_models:
return cls._generate_image(model, messages, proxy, seed, size)
else:
return cls._generate_text(model, messages, proxy, stream, **kwargs)
@classmethod
async def _generate_image(
cls,
model: str,
messages: Messages,
proxy: str = None,
seed: int = None,
size: str = "1:1",
**kwargs
) -> AsyncResult:
headers = {
"accept": "*/*",
"accept-language": "en-US,en;q=0.9",
"cache-control": "no-cache",
"origin": "https://llmplayground.net",
"user-agent": "Mozilla/5.0"
}
if seed is None:
seed = random.randint(0, 100000)
prompt = messages[-1]['content']
async with StreamSession(headers=headers, proxy=proxy) as session:
params = {
"model": model,
"prompt": prompt,
"size": size,
"seed": seed
}
async with session.get(f"{cls.api_endpoint_imagine}", params=params) as response:
await raise_for_status(response)
content_type = response.headers.get('Content-Type', '').lower()
if 'application/json' in content_type:
raise RuntimeError(await response.json().get("error", {}).get("message"))
elif 'image' in content_type:
image_data = b""
async for chunk in response.iter_content():
if chunk:
image_data += chunk
image_url = f"{cls.api_endpoint_imagine}?model={model}&prompt={prompt}&size={size}&seed={seed}"
yield ImageResponse(images=image_url, alt=prompt)
@classmethod
async def _generate_text(
cls,
model: str,
messages: Messages,
proxy: str = None,
stream: bool = False,
max_tokens: int = 4096,
temperature: float = 1,
top_p: float = 1,
**kwargs
) -> AsyncResult:
headers = {
"accept": "*/*",
"accept-language": "en-US,en;q=0.9",
"authorization": "Bearer missing api key",
"content-type": "application/json",
"user-agent": "Mozilla/5.0"
}
async with StreamSession(headers=headers, proxy=proxy) as session:
data = {
"messages": messages,
"model": model,
"max_tokens": max_tokens,
"temperature": temperature,
"top_p": top_p,
"stream": stream
}
async with session.post(cls.api_endpoint_completions, json=data) as response:
await raise_for_status(response)
content_type = response.headers.get('Content-Type', '').lower()
if 'application/json' in content_type:
json_data = await response.json()
if json_data.get("model") == "error":
raise RuntimeError(json_data['choices'][0]['message'].get('content', ''))
if stream:
async for line in response.iter_lines():
if line:
line = line.decode('utf-8').strip()
if line.startswith("data: ") and line != "data: [DONE]":
json_data = json.loads(line[6:])
content = json_data['choices'][0]['delta'].get('content', '')
if content:
yield cls._filter_content(content)
else:
json_data = await response.json()
content = json_data['choices'][0]['message']['content']
yield cls._filter_content(content)
@classmethod
def _filter_content(cls, part_response: str) -> str:
part_response = re.sub(
r"One message exceeds the \d+chars per message limit\..+https:\/\/discord\.com\/invite\/\S+",
'',
part_response
)
part_response = re.sub(
r"Rate limit \(\d+\/minute\) exceeded\. Join our discord for more: .+https:\/\/discord\.com\/invite\/\S+",
'',
part_response
)
return part_response
