79 lines
3.0 KiB
Python
79 lines
3.0 KiB
Python
from collections.abc import Mapping
|
|
from typing import Any
|
|
from urllib.parse import urljoin
|
|
|
|
import httpx
|
|
|
|
|
|
class OpenAICompatibleGateway:
|
|
"""One adapter for OpenAI, lk666.ai and other OpenAI-compatible gateways."""
|
|
|
|
def __init__(self, values: Mapping[str, Any]) -> None:
|
|
self.values = values
|
|
|
|
def _url(self, path_key: str, fallback: str) -> str:
|
|
base_url = str(self.values.get("ai_base_url", "")).rstrip("/") + "/"
|
|
path = str(self.values.get(path_key, fallback)).lstrip("/")
|
|
return urljoin(base_url, path)
|
|
|
|
@property
|
|
def headers(self) -> dict[str, str]:
|
|
return {
|
|
"Authorization": f"Bearer {self.values.get('ai_api_key', '')}",
|
|
"Content-Type": "application/json",
|
|
}
|
|
|
|
async def list_models(self) -> list[str]:
|
|
async with httpx.AsyncClient(timeout=15) as client:
|
|
response = await client.get(self._url("ai_models_path", "/models"), headers=self.headers)
|
|
response.raise_for_status()
|
|
payload = response.json()
|
|
return [item["id"] for item in payload.get("data", []) if isinstance(item, dict) and item.get("id")]
|
|
|
|
async def chat(self, messages: list[dict[str, Any]], *, vision: bool = False) -> dict[str, Any]:
|
|
model_key = "vision_model" if vision else "orchestrator_model"
|
|
async with httpx.AsyncClient(timeout=120) as client:
|
|
response = await client.post(
|
|
self._url("ai_chat_path", "/chat/completions"),
|
|
headers=self.headers,
|
|
json={"model": self.values[model_key], "messages": messages},
|
|
)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
async def generate_image(self, prompt: str, **options: Any) -> dict[str, Any]:
|
|
payload = {"model": self.values["image_model"], "prompt": prompt, **options}
|
|
async with httpx.AsyncClient(timeout=180) as client:
|
|
response = await client.post(
|
|
self._url("ai_image_generation_path", "/images/generations"),
|
|
headers=self.headers,
|
|
json=payload,
|
|
)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
async def edit_image(
|
|
self,
|
|
prompt: str,
|
|
image: bytes,
|
|
*,
|
|
filename: str = "image.png",
|
|
mask: bytes | None = None,
|
|
**options: Any,
|
|
) -> dict[str, Any]:
|
|
files: dict[str, tuple[str, bytes, str]] = {
|
|
"image": (filename, image, "image/png"),
|
|
}
|
|
if mask is not None:
|
|
files["mask"] = ("mask.png", mask, "image/png")
|
|
data = {"model": self.values["image_model"], "prompt": prompt, **options}
|
|
headers = {"Authorization": self.headers["Authorization"]}
|
|
async with httpx.AsyncClient(timeout=180) as client:
|
|
response = await client.post(
|
|
self._url("ai_image_edit_path", "/images/edits"),
|
|
headers=headers,
|
|
data=data,
|
|
files=files,
|
|
)
|
|
response.raise_for_status()
|
|
return response.json()
|