add cloud structured inference wrapper

infer_cloud.py

@@ -0,0 +1,110 @@
+import argparse
+import json
+import os
+import time
+
+import requests
+from huggingface_hub import InferenceClient, get_token
+
+from infer_local import build_instruction_prompt, build_structured_result
+
+
+def normalize_hf_response(response):
+    if isinstance(response, str):
+        return response
+
+    generated_text = getattr(response, "generated_text", None)
+    if generated_text is not None:
+        return generated_text
+
+    if isinstance(response, list) and response:
+        first = response[0]
+        if isinstance(first, dict):
+            return str(first.get("generated_text", ""))
+        return str(first)
+
+    if isinstance(response, dict):
+        return str(response.get("generated_text", response.get("text", "")))
+
+    return str(response)
+
+
+def call_direct_inference_api(repo_id, token, prompt_text, generation_kwargs):
+    headers = {}
+    if token:
+        headers["Authorization"] = f"Bearer {token}"
+
+    payload = {
+        "inputs": prompt_text,
+        "parameters": generation_kwargs,
+        "options": {"wait_for_model": True},
+    }
+    response = requests.post(
+        f"https://api-inference.huggingface.co/models/{repo_id}",
+        headers=headers,
+        json=payload,
+        timeout=120,
+    )
+    try:
+        body = response.json()
+    except ValueError:
+        body = response.text
+
+    if response.status_code >= 400:
+        raise RuntimeError(f"Hugging Face API error {response.status_code}: {body}")
+    if isinstance(body, dict) and body.get("error"):
+        raise RuntimeError(f"Hugging Face API error: {body['error']}")
+    return body
+
+
+def main():
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--repo-id", type=str, required=True)
+    parser.add_argument("--prompt", type=str, required=True)
+    parser.add_argument("--token", type=str, default=os.getenv("HF_TOKEN"))
+    parser.add_argument("--max-new-tokens", type=int, default=320)
+    parser.add_argument("--temperature", type=float, default=0.25)
+    parser.add_argument("--top-p", type=float, default=0.9)
+    parser.add_argument("--do-sample", action="store_true")
+    args = parser.parse_args()
+
+    token = args.token or get_token()
+    client = InferenceClient(model=args.repo_id, token=token)
+    prompt_text = build_instruction_prompt(args.prompt)
+
+    generation_kwargs = {
+        "max_new_tokens": args.max_new_tokens,
+        "return_full_text": False,
+    }
+    if args.do_sample:
+        generation_kwargs["temperature"] = args.temperature
+        generation_kwargs["top_p"] = args.top_p
+    else:
+        generation_kwargs["temperature"] = 0.01
+
+    start_time = time.perf_counter()
+    try:
+        response = client.text_generation(prompt_text, **generation_kwargs)
+    except TypeError:
+        generation_kwargs.pop("return_full_text", None)
+        response = client.text_generation(prompt_text, **generation_kwargs)
+    except Exception:
+        response = call_direct_inference_api(args.repo_id, token, prompt_text, generation_kwargs)
+    latency_ms = int((time.perf_counter() - start_time) * 1000)
+
+    generated_text = normalize_hf_response(response).strip()
+    if generated_text.startswith(prompt_text):
+        generated_text = generated_text[len(prompt_text) :].strip()
+    generated_text = generated_text.replace("<|im_end|>", "").strip()
+
+    result = build_structured_result(
+        args.prompt,
+        generated_text,
+        latency_ms,
+        default_confidence=0.0,
+    )
+    print(json.dumps(result, indent=2, ensure_ascii=False))
+
+
+if __name__ == "__main__":
+    main()