from fastapi import FastAPI from pydantic import BaseModel from transformers import AutoProcessor, Qwen2VLForConditionalGeneration import torch, os, base64, io, logging, time from typing import Any, Dict, List, Tuple from PIL import Image MODEL_ID = "osunlp/UGround-V1-72B" CACHE_DIR = ( os.environ.get("HF_HUB_CACHE") or os.environ.get("HF_HOME") or "/data/huggingface" ) # PyTorch performance settings # 1) Ensure CUDA kernel cache directory is writable/persistent to avoid recompilation stalls KERNEL_CACHE_DIR = os.environ.get("PYTORCH_KERNEL_CACHE_PATH", "/tmp/torch_kernels") os.environ["PYTORCH_KERNEL_CACHE_PATH"] = KERNEL_CACHE_DIR try: os.makedirs(KERNEL_CACHE_DIR, exist_ok=True) except Exception: pass # 2) Enable TF32 for faster matmul on Ampere+ GPUs (minimal quality impact) try: torch.backends.cuda.matmul.allow_tf32 = True # type: ignore[attr-defined] torch.backends.cudnn.allow_tf32 = True # type: ignore[attr-defined] torch.set_float32_matmul_precision("high") # type: ignore[attr-defined] except Exception: pass processor = AutoProcessor.from_pretrained( MODEL_ID, trust_remote_code=True, cache_dir=CACHE_DIR, use_fast=False ) model = Qwen2VLForConditionalGeneration.from_pretrained( MODEL_ID, dtype=torch.bfloat16, device_map="auto", trust_remote_code=True, cache_dir=CACHE_DIR, ) model.eval() try: torch.set_grad_enabled(False) except Exception: pass app = FastAPI() # Configure basic logging for debugging logging.basicConfig( level=logging.DEBUG, format="%(asctime)s %(levelname)s %(name)s: %(message)s" ) logger = logging.getLogger(__name__) @app.get("/") async def root(): return {"status": "ok"} class ChatCompletionRequest(BaseModel): model: str messages: List[Dict[str, Any]] max_tokens: int = 256 MAX_IMAGE_WIDTH = 512 MAX_IMAGE_HEIGHT = 388 def _decode_base64_image(data_url: str) -> Image.Image: try: is_data_url = data_url.startswith("data:") if is_data_url: header, b64data = data_url.split(",", 1) logger.debug("Decoding image from data URL; header prefix=%r", header[:50]) else: b64data = data_url logger.debug("Decoding image from raw base64 string; length=%d", len(b64data)) img_bytes = base64.b64decode(b64data) img = Image.open(io.BytesIO(img_bytes)).convert("RGB") orig_w, orig_h = img.width, img.height # Downscale if larger than bounds, preserving aspect ratio if orig_w > MAX_IMAGE_WIDTH or orig_h > MAX_IMAGE_HEIGHT: target = (MAX_IMAGE_WIDTH, MAX_IMAGE_HEIGHT) img = img.copy() img.thumbnail(target, Image.LANCZOS) logger.debug( "Resized image from %sx%s to %sx%s (bounds %sx%s)", orig_w, orig_h, img.width, img.height, MAX_IMAGE_WIDTH, MAX_IMAGE_HEIGHT, ) try: logger.debug("Decoded image: size=%sx%s mode=%s", img.width, img.height, img.mode) except Exception: logger.debug("Decoded image but could not log image metadata") return img except Exception: logger.exception("Failed to decode base64 image") raise def _to_qwen_messages_and_images(messages: List[Dict[str, Any]]) -> Tuple[List[Dict[str, Any]], List[Any]]: qwen_msgs: List[Dict[str, Any]] = [] images: List[Any] = [] logger.debug("Begin parsing messages: count=%d", len(messages) if messages else 0) for idx, msg in enumerate(messages): role = msg.get("role", "user") content = msg.get("content") logger.debug("Processing message #%d role=%s content_type=%s", idx, role, type(content).__name__) q_content: List[Dict[str, Any]] = [] if isinstance(content, str): logger.debug("Message #%d text length=%d", idx, len(content)) q_content.append({"type": "text", "text": content}) elif isinstance(content, list): logger.debug("Message #%d has %d content parts", idx, len(content)) for pidx, part in enumerate(content): ptype = part.get("type") logger.debug("Part #%d type=%s", pidx, ptype) if ptype == "text": text_val = part.get("text") or part.get("content") or "" logger.debug("Part #%d text length=%d", pidx, len(text_val)) q_content.append({"type": "text", "text": text_val}) elif ptype in ("image", "image_url"): # OpenAI style: {type:"image_url", image_url:{url:"..."}} url = part.get("image") if url is None and isinstance(part.get("image_url"), dict): url = part["image_url"].get("url") if isinstance(url, str) and url.startswith("data:image"): logger.debug("Part #%d image provided as base64 data URL", pidx) img = _decode_base64_image(url) images.append(img) q_content.append({"type": "image", "image": img}) else: # URL or non-base64 string logger.debug("Part #%d image provided as URL or non-base64 string: %s", pidx, str(url)[:200]) images.append(url) q_content.append({"type": "image", "image": url}) else: # Unknown content; coerce to text logger.debug("Message #%d unknown content type; coercing to text", idx) q_content.append({"type": "text", "text": str(content)}) qwen_msgs.append({"role": role, "content": q_content}) logger.debug("Finished parsing messages: qwen_msgs=%d images=%d", len(qwen_msgs), len(images)) return qwen_msgs, images def _make_tiny_base64_png(size: Tuple[int, int] = (64, 48), color: Tuple[int, int, int] = (128, 128, 128)) -> str: buf = io.BytesIO() Image.new("RGB", size, color).save(buf, format="PNG") data = base64.b64encode(buf.getvalue()).decode("ascii") return f"data:image/png;base64,{data}" @app.on_event("startup") async def _startup_warmup(): if os.environ.get("DISABLE_WARMUP", "0") == "1": logger.info("Warmup disabled via DISABLE_WARMUP=1") return try: logger.info("Warmup start: compiling kernels (text + tiny image)") # Text-only warmup text_msgs: List[Dict[str, Any]] = [ {"role": "user", "content": "Hello"} ] qmsgs_t, _ = _to_qwen_messages_and_images(text_msgs) prompt_t = processor.apply_chat_template(qmsgs_t, tokenize=False, add_generation_prompt=True) inputs_t = processor(text=[prompt_t], images=None, padding=True, return_tensors="pt") inputs_t = inputs_t.to(model.device) _t0 = time.perf_counter() with torch.no_grad(): _ = model.generate(**inputs_t, max_new_tokens=int(os.environ.get("WARMUP_MAX_NEW_TOKENS", "4")), max_time=float(os.environ.get("WARMUP_MAX_TIME_SECONDS", "3"))) logger.info("Text warmup done in %.1f ms", (time.perf_counter() - _t0) * 1000.0) # Tiny image + text warmup tiny_url = _make_tiny_base64_png() viz_msgs: List[Dict[str, Any]] = [ {"role": "user", "content": [ {"type": "text", "text": "Describe the image"}, {"type": "image_url", "image_url": {"url": tiny_url}}, ]} ] qmsgs_v, images_v = _to_qwen_messages_and_images(viz_msgs) prompt_v = processor.apply_chat_template(qmsgs_v, tokenize=False, add_generation_prompt=True) inputs_v = processor(text=[prompt_v], images=images_v, padding=True, return_tensors="pt") inputs_v = inputs_v.to(model.device) _t1 = time.perf_counter() with torch.no_grad(): _ = model.generate(**inputs_v, max_new_tokens=int(os.environ.get("WARMUP_MAX_NEW_TOKENS", "4")), max_time=float(os.environ.get("WARMUP_MAX_TIME_SECONDS", "3"))) logger.info("Vision warmup done in %.1f ms", (time.perf_counter() - _t1) * 1000.0) logger.info("Warmup complete") except Exception: logger.exception("Warmup failed") @app.post("/v1/chat/completions") async def chat_completions(req: ChatCompletionRequest): logger.debug( "Request received: model=%s, max_tokens=%s, message_count=%d", req.model, req.max_tokens, len(req.messages) if req.messages is not None else 0, ) if req.messages: logger.debug("First message preview: %s", str(req.messages[0])[:300]) qwen_messages, image_inputs = _to_qwen_messages_and_images(req.messages) logger.debug( "Converted messages: qwen_count=%d, images_count=%d", len(qwen_messages), len(image_inputs) if image_inputs is not None else 0, ) if qwen_messages: logger.debug("First qwen message preview: %s", str(qwen_messages[0])[:300]) prompt_text = processor.apply_chat_template( qwen_messages, tokenize=False, add_generation_prompt=True ) logger.debug("Prompt length (chars)=%d; preview=%r", len(prompt_text), prompt_text[:200]) inputs = processor( text=[prompt_text], images=image_inputs if image_inputs else None, padding=True, return_tensors="pt", ) try: tensor_info_pre = { k: (tuple(v.shape), str(getattr(v, "dtype", ""))) for k, v in inputs.items() if hasattr(v, "shape") } logger.debug("Processor outputs (pre .to): %s", tensor_info_pre) except Exception: logger.debug("Could not summarize processor outputs before device move") inputs = inputs.to(model.device) try: tensor_info_post = { k: ( tuple(v.shape), str(getattr(v, "dtype", "")), str(getattr(v, "device", "")), ) for k, v in inputs.items() if torch.is_tensor(v) } logger.debug("Inputs moved to device=%s; tensor_info=%s", getattr(model, "device", ""), tensor_info_post) except Exception: logger.debug("Could not summarize inputs after device move") logger.debug("Starting generation: max_new_tokens=%d", req.max_tokens) _t0 = time.perf_counter() generated_ids = model.generate(**inputs, max_new_tokens=req.max_tokens) _elapsed_ms = (time.perf_counter() - _t0) * 1000.0 try: logger.debug( "Generation done in %.1f ms; generated_ids shape=%s dtype=%s device=%s", _elapsed_ms, tuple(generated_ids.shape) if hasattr(generated_ids, "shape") else "", str(getattr(generated_ids, "dtype", "")), str(getattr(generated_ids, "device", "")), ) except Exception: logger.debug("Could not summarize generated_ids") trimmed = [ out_ids[len(in_ids):] for in_ids, out_ids in zip(inputs.input_ids, generated_ids) ] try: lengths_in = [row.size(0) for row in inputs.input_ids] lengths_out = [row.size(0) for row in generated_ids] logger.debug("Token lengths: input=%s, output=%s", lengths_in, lengths_out) except Exception: logger.debug("Could not compute token length summaries") output_texts = processor.batch_decode( trimmed, skip_special_tokens=True, clean_up_tokenization_spaces=False ) text = output_texts[0] if output_texts else "" logger.debug( "Decoded %d sequences; first_text_len=%d", len(output_texts), len(text) if text else 0, ) if text: logger.debug("Output preview: %r", text[:500]) return { "id": "chatcmpl-uground72b", "object": "chat.completion", "choices": [{ "index": 0, "message": {"role": "assistant", "content": text}, "finish_reason": "stop" }] }