Spaces:
Sleeping
Sleeping
File size: 9,808 Bytes
2c4cd1d 469f08c 2c4cd1d a4acfba 2c4cd1d 469f08c 2c4cd1d 4c0f990 2c4cd1d 469f08c 4c0f990 469f08c 4c0f990 469f08c 2c4cd1d 4c0f990 2c4cd1d 4c0f990 2c4cd1d |
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 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 |
"""
GPTOSSWrapper - Simple integration wrapper for OpenAI or Hugging Face Inference API.
Usage:
from gptoss_wrapper import GPTOSSWrapper
w = GPTOSSWrapper(model="gpt-oss-120")
text = w.generate(prompt)
Behavior:
- Provider selection (priority):
1) If OPENAI_API_KEY is set -> use OpenAI Chat Completions (v1/chat/completions)
2) Else if HUGGINGFACE_API_TOKEN or HF_API_TOKEN is set -> use Hugging Face Inference API
3) Else -> generate() will raise a RuntimeError describing missing credentials.
Note for Spaces:
- Add the secret in your Space settings (Settings → Secrets & variables → Add secret):
- For OpenAI: key name = OPENAI_API_KEY, value = <your_openai_api_key>
- For Hugging Face: key name = HUGGINGFACE_API_TOKEN (or HF_API_TOKEN), value = <your_hf_token>
This file intentionally uses only the requests stdlib-friendly HTTP approach to avoid depending on extra SDKs.
"""
import os
import time
import requests
from typing import Optional
class GPTOSSWrapper:
"""
Lightweight wrapper that can call either OpenAI or Hugging Face inference endpoints.
Constructor:
GPTOSSWrapper(model="gpt-oss-120", provider="auto")
- model: model name to request (for OpenAI it must be an available model for your account;
for Hugging Face it should be a model id hosted on HF).
- provider: "auto" (default) | "openai" | "hf"
"""
def __init__(self, model: str = "gpt-oss-120", provider: str = "auto"):
# Allow overriding the model via env var MODEL_ID (useful in Spaces)
env_model = os.getenv("MODEL_ID")
if env_model:
self.model = env_model
else:
self.model = model
self.request_timeout = 30
self.openai_key = os.getenv("OPENAI_API_KEY")
# Accept multiple HF token environment variable names for compatibility:
# HUGGINGFACE_API_TOKEN, HF_API_TOKEN, or HF_TOKEN (used by some HF examples)
self.hf_token = (
os.getenv("HUGGINGFACE_API_TOKEN")
or os.getenv("HF_API_TOKEN")
or os.getenv("HF_TOKEN")
)
self.provider = provider.lower() if provider else "auto"
# If we have an HF token and the user didn't explicitly set a MODEL_ID,
# prefer the HF router and use a sensible default router model id.
if self.hf_token and not env_model and model == "gpt-oss-120":
# Default router model id; you can override via MODEL_ID env var in the Space
self.model = "openai/gpt-oss-120b:fireworks-ai"
if self.provider == "auto":
if self.openai_key:
self.provider = "openai"
elif self.hf_token:
self.provider = "hf"
else:
self.provider = "none"
def generate(self, prompt: str, max_tokens: int = 512, temperature: float = 0.2) -> str:
"""
Generate a textual response for the given prompt.
Returns:
A string with the generated text.
Raises:
RuntimeError if no credentials are found or the remote call fails.
"""
if self.provider == "openai":
return self._generate_openai(prompt, max_tokens=max_tokens, temperature=temperature)
elif self.provider == "hf":
return self._generate_hf(prompt, max_tokens=max_tokens, temperature=temperature)
else:
raise RuntimeError(
"No API key configured for GPT wrapper. Set OPENAI_API_KEY or HUGGINGFACE_API_TOKEN in the environment."
)
def _generate_openai(self, prompt: str, max_tokens: int, temperature: float) -> str:
if not self.openai_key:
raise RuntimeError("OPENAI_API_KEY not set in environment.")
url = "https://api.openai.com/v1/chat/completions"
headers = {
"Authorization": f"Bearer {self.openai_key}",
"Content-Type": "application/json",
}
# Build a simple chat conversation with a single system + user message
payload = {
"model": self.model,
"messages": [
{"role": "system", "content": "You are an expert inspection assistant for wind turbine blade images/videos."},
{"role": "user", "content": prompt},
],
"max_tokens": max_tokens,
"temperature": float(temperature),
"n": 1,
}
try:
r = requests.post(url, headers=headers, json=payload, timeout=self.request_timeout)
r.raise_for_status()
data = r.json()
# OpenAI API returns a list of choices
choices = data.get("choices", [])
if not choices:
raise RuntimeError(f"OpenAI returned empty choices: {data}")
# Extract the assistant message
msg = choices[0].get("message", {}).get("content")
if msg is None:
# Some deployments return text in 'text' or in other fields; fallback to stringifying response
return str(data)
return msg.strip()
except Exception as e:
# Surface a clear error for the calling code to handle (the app catches exceptions)
raise RuntimeError(f"OpenAI API call failed: {e}")
def _generate_hf(self, prompt: str, max_tokens: int, temperature: float) -> str:
if not self.hf_token:
raise RuntimeError("HUGGINGFACE_API_TOKEN (or HF_API_TOKEN / HF_TOKEN) not set in environment.")
# Prefer the HF router automatically when an HF token is present unless explicitly disabled.
use_router = False
# If HF token exists, default to using the router (unless HF_USE_ROUTER is set to a falsey value).
if self.hf_token:
hf_use_router_val = os.getenv("HF_USE_ROUTER", "").lower()
if hf_use_router_val in ("0", "false", "no"):
use_router = False
else:
use_router = True
# Explicit enable via HF_USE_ROUTER env var
if os.getenv("HF_USE_ROUTER", "").lower() in ("1", "true", "yes"):
use_router = True
# Also enable router if model id looks like an OpenAI-style id
if "openai/" in (self.model or "") or ":" in (self.model or ""):
use_router = True
try:
if use_router:
# Router (OpenAI-compatible) endpoint: accepts chat/completions style payloads
url = "https://router.huggingface.co/v1/chat/completions"
headers = {"Authorization": f"Bearer {self.hf_token}", "Content-Type": "application/json"}
payload = {
"model": self.model,
"messages": [
{"role": "system", "content": "You are an expert inspection assistant for wind turbine blade images/videos."},
{"role": "user", "content": prompt},
],
"max_tokens": max_tokens,
"temperature": float(temperature),
"n": 1,
}
r = requests.post(url, headers=headers, json=payload, timeout=self.request_timeout)
r.raise_for_status()
data = r.json()
# Try to extract OpenAI-style response
choices = data.get("choices", [])
if choices and isinstance(choices, list):
first = choices[0]
# OpenAI-compatible router usually returns message under 'message'
msg = first.get("message", {}).get("content") if isinstance(first, dict) else None
# Some router variants may return text under 'text' or 'content'
if not msg:
msg = first.get("text") or first.get("content")
if msg:
return msg.strip()
# Fallback stringify
return str(data)
else:
# Standard Hugging Face inference API
url = f"https://api-inference.huggingface.co/models/{self.model}"
headers = {"Authorization": f"Bearer {self.hf_token}"}
payload = {
"inputs": prompt,
"parameters": {
"max_new_tokens": max_tokens,
"temperature": float(temperature),
},
"options": {"wait_for_model": True},
}
r = requests.post(url, headers=headers, json=payload, timeout=self.request_timeout)
r.raise_for_status()
data = r.json()
# Hugging Face inference may return a list of generated outputs or a dict
if isinstance(data, list) and len(data) > 0 and isinstance(data[0], dict) and "generated_text" in data[0]:
return data[0]["generated_text"].strip()
elif isinstance(data, dict) and "generated_text" in data:
return data["generated_text"].strip()
elif isinstance(data, dict) and "error" in data:
raise RuntimeError(f"Hugging Face error: {data['error']}")
else:
# Some text-generation endpoints return a plain string or different struct; try to stringify
return str(data)
except Exception as e:
raise RuntimeError(f"Hugging Face API call failed: {e}")
# Backwards-compatible factory in case caller expects a function or attribute
def GPTOSSWrapperFactory(model: Optional[str] = None, provider: Optional[str] = None):
return GPTOSSWrapper(model=model or "gpt-oss-120", provider=provider or "auto") |