SolusOps/Study-with-ChampAI

1from __future__ import annotations2import json3import base644 5try:6    from huggingface_hub import InferenceClient as _HFClient7    _HF_AVAILABLE = True8except ImportError:9    _HF_AVAILABLE = False10 11import urllib.request12 13HF_API_BASE = "https://api-inference.huggingface.co/models"14HF_ROUTER_BASE = "https://router.huggingface.co/hf-inference/models"15 16def _try_router_url(model_id: str, payload: dict, api_key: str, timeout: int = 60) -> str:17    """Try the new HF Router endpoint as fallback."""18    url = f"{HF_ROUTER_BASE}/{model_id}/v1/chat/completions"19    data = json.dumps(payload).encode("utf-8")20    headers = {"Content-Type": "application/json", "Authorization": f"Bearer {api_key}"}21    req = urllib.request.Request(url, data=data, headers=headers, method="POST")22    with urllib.request.urlopen(req, timeout=timeout) as resp:23        result = json.loads(resp.read().decode("utf-8"))24    return result["choices"][0]["message"]["content"]25 26def generate(model_id: str, prompt: str, system: str = "",27             api_key: str = "", max_tokens: int = 1024, temperature: float = 0.3) -> str:28    """Text generation — MiniCPM concept extraction and Tiny Aya translation."""29    if _HF_AVAILABLE and api_key:30        client = _HFClient(api_key=api_key)31        messages = []32        if system:33            messages.append({"role": "system", "content": system})34        messages.append({"role": "user", "content": prompt})35        response = client.chat_completion(36            messages=messages, model=model_id,37            max_tokens=max_tokens, temperature=temperature38        )39        return response.choices[0].message.content40 41    # Fallback: raw HTTP42    messages = []43    if system:44        messages.append({"role": "system", "content": system})45    messages.append({"role": "user", "content": prompt})46    payload = {"inputs": {"messages": messages},47               "parameters": {"max_new_tokens": max_tokens, "temperature": temperature}}48    data = json.dumps(payload).encode("utf-8")49    headers = {"Content-Type": "application/json"}50    if api_key:51        headers["Authorization"] = f"Bearer {api_key}"52    req = urllib.request.Request(f"{HF_API_BASE}/{model_id}", data=data, headers=headers, method="POST")53    with urllib.request.urlopen(req, timeout=60) as resp:54        result = json.loads(resp.read().decode("utf-8"))55    if isinstance(result, list): return result[0].get("generated_text", str(result[0]))56    if isinstance(result, dict): return result.get("generated_text", result.get("text", str(result)))57    return str(result)58 59def vision_generate(model_id: str, image_b64: str, prompt: str,60                    api_key: str = "", max_tokens: int = 1024) -> str:61    """Vision+language — MiniCPM-V for OCR and visual understanding."""62    if _HF_AVAILABLE and api_key:63        client = _HFClient(api_key=api_key)64        image_data_url = f"data:image/png;base64,{image_b64}"65        messages = [{"role": "user", "content": [66            {"type": "image_url", "image_url": {"url": image_data_url}},67            {"type": "text", "text": prompt},68        ]}]69        response = client.chat_completion(70            messages=messages, model=model_id, max_tokens=max_tokens71        )72        return response.choices[0].message.content73 74    # Fallback: raw HTTP75    url = f"{HF_API_BASE}/{model_id}"76    payload = {"inputs": {"image": image_b64, "question": prompt},77               "parameters": {"max_new_tokens": max_tokens}}78    data = json.dumps(payload).encode("utf-8")79    headers = {"Content-Type": "application/json"}80    if api_key:81        headers["Authorization"] = f"Bearer {api_key}"82    req = urllib.request.Request(url, data=data, headers=headers, method="POST")83    with urllib.request.urlopen(req, timeout=90) as resp:84        result = json.loads(resp.read().decode("utf-8"))85    if isinstance(result, list): return str(result[0])86    if isinstance(result, dict): return result.get("generated_text", result.get("text", str(result)))87    return str(result)88 89def transcribe(model_id: str, audio_bytes: bytes, api_key: str = "") -> str:90    """ASR — Whisper via HuggingFace for voice input."""91    if _HF_AVAILABLE and api_key:92        client = _HFClient(api_key=api_key)93        result = client.automatic_speech_recognition(audio=audio_bytes, model=model_id)94        return result.text if hasattr(result, "text") else str(result)95 96    # Fallback: raw HTTP97    url = f"{HF_API_BASE}/{model_id}"98    headers = {"Content-Type": "audio/wav"}99    if api_key:100        headers["Authorization"] = f"Bearer {api_key}"101    req = urllib.request.Request(url, data=audio_bytes, headers=headers, method="POST")102    with urllib.request.urlopen(req, timeout=60) as resp:103        result = json.loads(resp.read().decode("utf-8"))104    return result.get("text", "")105