46 lines
1.5 KiB
Python
46 lines
1.5 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
import httpx
|
|
|
|
|
|
class OpenAICompatClient:
|
|
def __init__(self, timeout: float = 180.0) -> None:
|
|
self.timeout = timeout
|
|
|
|
async def chat_completion(
|
|
self,
|
|
*,
|
|
base_url: str,
|
|
api_key: str,
|
|
model: str,
|
|
system_prompt: str,
|
|
user_prompt: str,
|
|
temperature: float = 0.7,
|
|
) -> str:
|
|
url = base_url.rstrip("/") + "/chat/completions"
|
|
headers = {"Content-Type": "application/json"}
|
|
if api_key.strip():
|
|
headers["Authorization"] = f"Bearer {api_key.strip()}"
|
|
payload: dict[str, Any] = {
|
|
"model": model,
|
|
"temperature": temperature,
|
|
"messages": [
|
|
{"role": "system", "content": system_prompt},
|
|
{"role": "user", "content": user_prompt},
|
|
],
|
|
}
|
|
async with httpx.AsyncClient(timeout=self.timeout) as client:
|
|
response = await client.post(url, headers=headers, json=payload)
|
|
response.raise_for_status()
|
|
data = response.json()
|
|
choices = data.get("choices") or []
|
|
if not choices:
|
|
return ""
|
|
message = choices[0].get("message") or {}
|
|
content = message.get("content") or ""
|
|
if isinstance(content, list):
|
|
return "\n".join(str(item.get("text", "")) for item in content if isinstance(item, dict)).strip()
|
|
return str(content).strip()
|