Spaces:
Sleeping
Sleeping
File size: 3,969 Bytes
0fd441a 962ef72 0fd441a 653f79c 0fd441a 962ef72 0fd441a 962ef72 0fd441a 962ef72 0fd441a 962ef72 0fd441a |
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 |
from __future__ import annotations
from typing import Optional #Iterable, Literal
#import os
#import time
import traceback
#from huggingface_hub import InferenceClient, login, logout as hf_logout
from llm.llm_login import login_huggingface, is_loggedin_huggingface #, is_login_huggingface
import dotenv
#dotenv.load_dotenv(".env")
from utils.logger import get_logger
## Get logger instance
logger = get_logger(__name__)
class OpenAIChatClient:
"""
Provider‐agnostic OpenAI-based LLM client interface.
Compatible with `huggingface_hub.InferenceClient` setup and chat calls.
- base_url="https://router.huggingface.co/v1",
- api_key=os.environ["HF_TOKEN"],
"""
def __init__(self,
model_id: Optional[str] = None,
hf_provider: Optional[str] = None,
base_url: Optional[str] = "https://router.huggingface.co/v1", #None,
api_token: Optional[str] = None,
temperature: Optional[float] = 0.2,
top_p: Optional[float] = 0.2,
) -> None:
try:
openai_api_key_env = dotenv.get_key(".env", "OPENAI_API_KEY") or dotenv.get_key(".env", "GEMINI_API_KEY")
self.model_id = f"{model_id}:{hf_provider}" if hf_provider is not None else model_id ##concatenate so HF can pipe to Hf provider
self.hf_provider = hf_provider
self.base_url = base_url #"https://router.huggingface.co/v1" #%22" #HF API proxy
self.token = api_token if api_token else openai_api_key_env ##None ##debug
#self.token = openai_api_key_env if openai_api_key_env else api_token #dotenv.get_key(".env", "OPENAI_API_KEY")
#self.token = token or os.getenv("HF_TOKEN") or os.getenv("HUGGINGFACEHUB_API_TOKEN") ## not preferred
#self.fake_token = api_token or "a1b2c3" #or os.getenv("HF_TOKEN") or os.getenv("HUGGINGFACEHUB_API_TOKEN")
self.openai_api_key = self.token #self.fake_token
self.temperature = temperature
self.top_p = top_p
self.islogged_in = is_loggedin_huggingface()
##SMY: log in now handled at higher entry level.
#login_huggingface(self.token) if not is_loggedin_huggingface() else logger.log(level=20, msg=f"You are logged in to HF Hub already") ## attempt login if not already logged in. NB: HF CLI login prompt would not display in Process Worker.
logger.log(level=2, msg="initialised OpenAIChatClient:", extra={"base_url": self.base_url, "openai_api_key": self.openai_api_key})
except Exception as exc:
#logger.error(f"OpenAI client_init_failed", extra={"error": str(exc)}")
tb = traceback.format_exc()
logger.exception(f'✗ OpenAI client_init_failed", extra={"error": str(exc)}\n{tb}', exc_info=True)
raise RuntimeError(f"✗ Failed to initialise OpenAI client: {exc}\n{tb}")
#login_huggingface(self.token) if not is_loggedin_huggingface() else logger.log(level=20, msg=f"logged in to HF Hub already") ## attempt login if not already logged in. NB: HF CLI login prompt would not display in Process Worker.
####IN PROGRESS
#
"""
## HuggingFace API-proxy Inference Provider - https://huggingface.co/docs/inference-providers/index?python-clients=openai
## https://huggingface.co/openai/gpt-oss-20b?inference_api=true&inference_provider=fireworks-ai&language=python&client=openai
import os
from openai import OpenAI
client = OpenAI(
base_url="https://router.huggingface.co/v1",
api_key=os.environ["HF_TOKEN"],
)
stream = client.chat.completions.create(
model="openai/gpt-oss-20b:fireworks-ai",
messages=[
{
"role": "user",
"content": "What is the capital of France?"
}
],
stream=True,
)
for chunk in stream:
print(chunk.choices[0].delta.content, end="")
"""
|