Spaces:
Build error
Build error
File size: 1,884 Bytes
71b4454 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 | import os
import httpx
from typing import Dict, Any
DEFAULT_MODEL = os.getenv("DEFAULT_MODEL", "zai-glm-4.7")
CEREBRAS_API_KEY = os.getenv("CEREBRAS_API_KEY", "").strip()
CEREBRAS_URL = "https://api.cerebras.ai/v1/chat/completions"
class ModelGateway:
def __init__(self):
self.client = httpx.AsyncClient(
timeout=httpx.Timeout(180.0),
follow_redirects=True,
)
async def generate(
self,
model: str,
prompt: str,
system: str = "You are DOLOR3V Autonomous Builder.",
) -> Dict[str, Any]:
if not CEREBRAS_API_KEY:
raise RuntimeError("CEREBRAS_API_KEY is not configured")
payload = {
"model": model,
"messages": [
{
"role": "system",
"content": system,
},
{
"role": "user",
"content": prompt,
},
],
"temperature": 0.2,
"max_tokens": 4096,
}
response = await self.client.post(
CEREBRAS_URL,
headers={
"Authorization": f"Bearer {CEREBRAS_API_KEY}",
"Content-Type": "application/json",
},
json=payload,
)
response.raise_for_status()
data = response.json()
if "choices" not in data:
raise RuntimeError(f"Unexpected Cerebras response: {data}")
text = (
data["choices"][0]
.get("message", {})
.get("content", "")
.strip()
)
return {
"text": text,
"provider": "cerebras",
"model": model,
"raw": data,
}
async def close(self):
await self.client.aclose()
model_gateway = ModelGateway()
|