from __future__ import annotations import base64 import io import json import os import re import time from collections import defaultdict, deque from pathlib import Path from threading import Lock from collections.abc import Iterator from typing import Any import gradio as gr from openai import OpenAI from PIL import Image, ImageDraw, UnidentifiedImageError MODEL = "qwen3.8-max" BASE_URL = "https://dashscope-intl.aliyuncs.com/api/v2/apps/protocols/compatible-mode/v1" MAX_IMAGES = 3 MAX_IMAGE_BYTES = 15 * 1024 * 1024 RATE_LIMIT_REQUESTS = 5 RATE_LIMIT_WINDOW_SECONDS = 5 * 60 QWEN_LOGO_DATA_URL = "data:image/svg+xml;base64," + base64.b64encode( (Path(__file__).parent / "assets" / "qwen-logo.svg").read_bytes() ).decode("ascii") EXAMPLES_DIR = Path(__file__).parent / "examples" VISUAL_EXAMPLES = [ { "title": "Count crowded objects", "image": str(EXAMPLES_DIR / "crowded_shapes.png"), "prompt": "Count each kind of shape. Explain how you avoided double-counting, then return a compact JSON summary.", }, { "title": "Audit a receipt", "image": str(EXAMPLES_DIR / "mini_receipt.png"), "prompt": "Read this receipt, independently recompute the subtotal and total, and flag any inconsistency.", }, { "title": "Solve a visual pattern", "image": str(EXAMPLES_DIR / "logic_grid.png"), "prompt": "Solve the visual logic puzzle. State the rule you infer and identify the missing tile.", }, ] BOX_EXAMPLES = [ { "title": "Blue circles, not squares", "image": str(EXAMPLES_DIR / "crowded_shapes.png"), "prompt": "Find every other blue circle. Ignore blue squares and shapes of other colors.", "boxes": [ {"box": [761, 81, 849, 169], "kind": "positive"}, {"box": [205, 96, 281, 172], "kind": "negative"}, ], }, { "title": "Matching mint bottles", "image": str(EXAMPLES_DIR / "box_product_shelf.png"), "prompt": "Find every other teal MINT bottle. Ignore the purple MINT bottles and all cans.", "boxes": [ {"box": [80, 118, 142, 292], "kind": "positive"}, {"box": [780, 118, 842, 292], "kind": "negative"}, ], }, { "title": "Standard green cars", "image": str(EXAMPLES_DIR / "box_parking_lot.png"), "prompt": "Find the other standard-size green cars. Do not include the oversized green vehicle.", "boxes": [ {"box": [88, 112, 172, 262], "kind": "positive"}, {"box": [793, 340, 905, 518], "kind": "negative"}, ], }, ] OBJECT_EXAMPLES = [ { "title": "Detect every circle", "image": str(EXAMPLES_DIR / "crowded_shapes.png"), "prompt": "Detect every circle. Label each one by its color, such as red circle or blue circle.", }, { "title": "Detect all vehicles", "image": str(EXAMPLES_DIR / "box_parking_lot.png"), "prompt": "Detect every vehicle. Distinguish standard cars from the oversized vehicle.", }, { "title": "Inventory the shelf", "image": str(EXAMPLES_DIR / "box_product_shelf.png"), "prompt": "Detect every product and label its color and type, for example teal bottle or red can.", }, ] BOX_CANVAS_TEMPLATE = """ ${value && value.image ? `
Image for drawing bounding boxes
` : `
Upload an image or choose an example, then drag directly over the image to draw a box.
`} """ BOX_CANVAS_CSS = """ .box-canvas-stage { position: relative; width: 100%; overflow: hidden; border: 1px solid var(--border-color-primary); border-radius: var(--radius-lg); background: var(--block-background-fill); user-select: none; } .box-canvas-stage img { display: block; width: 100%; height: auto; pointer-events: none; } .box-canvas-stage svg, .box-canvas-stage canvas { position: absolute; inset: 0; width: 100%; height: 100%; } .box-canvas-stage svg { z-index: 1; pointer-events: none; } .box-canvas-stage canvas { z-index: 2; cursor: crosshair; touch-action: none; } .box-canvas-empty { min-height: 320px; display: grid; place-items: center; padding: 2rem; border: 2px dashed var(--border-color-primary); border-radius: var(--radius-lg); color: var(--body-text-color-subdued); text-align: center; } """ BOX_CANVAS_JS = """ document.__qwenBoxCanvasProps = props; document.__qwenBoxCanvasTrigger = trigger; if (!document.__qwenBoxCanvasReady) { document.__qwenBoxCanvasReady = true; let activeCanvas = null; let start = null; let current = null; const isBoxCanvas = (target) => target instanceof HTMLCanvasElement && target.matches("#box-drawing-canvas canvas"); const canvasDetails = (canvas) => { const width = Number(canvas.dataset.width); const height = Number(canvas.dataset.height); const context = canvas.getContext("2d"); if (canvas.width !== width) canvas.width = width; if (canvas.height !== height) canvas.height = height; return {width, height, context}; }; const pointFromEvent = (canvas, event) => { const {width, height} = canvasDetails(canvas); const rect = canvas.getBoundingClientRect(); return { x: Math.max(0, Math.min(width, Math.round((event.clientX - rect.left) * width / rect.width))), y: Math.max(0, Math.min(height, Math.round((event.clientY - rect.top) * height / rect.height))), }; }; const redrawPreview = () => { if (!activeCanvas) return; const {width, height, context} = canvasDetails(activeCanvas); context.clearRect(0, 0, width, height); if (!start || !current) return; const selected = document.querySelector("#box-kind input:checked"); const positive = (selected?.value || "Positive").toLowerCase() === "positive"; const color = positive ? "#16a34a" : "#dc2626"; const lineWidth = Math.max(3, Math.round(Math.min(width, height) / 180)); context.save(); context.strokeStyle = color; context.lineWidth = lineWidth; context.setLineDash([lineWidth * 2, lineWidth * 1.5]); context.strokeRect(start.x, start.y, current.x - start.x, current.y - start.y); context.restore(); }; document.addEventListener("pointerdown", (event) => { if (!isBoxCanvas(event.target)) return; event.preventDefault(); activeCanvas = event.target; activeCanvas.setPointerCapture(event.pointerId); start = pointFromEvent(activeCanvas, event); current = start; redrawPreview(); }); document.addEventListener("pointermove", (event) => { if (!activeCanvas || !start) return; current = pointFromEvent(activeCanvas, event); redrawPreview(); }); document.addEventListener("pointerup", (event) => { if (!activeCanvas || !start) return; current = pointFromEvent(activeCanvas, event); const x1 = Math.min(start.x, current.x); const y1 = Math.min(start.y, current.y); const x2 = Math.max(start.x, current.x); const y2 = Math.max(start.y, current.y); if (x2 - x1 >= 4 && y2 - y1 >= 4) { const selected = document.querySelector("#box-kind input:checked"); const kind = (selected?.value || "Positive").toLowerCase(); const boxes = JSON.parse(decodeURIComponent(activeCanvas.dataset.boxes || "%5B%5D")); boxes.push({box: [x1, y1, x2, y2], kind}); document.__qwenBoxCanvasProps.value = { ...document.__qwenBoxCanvasProps.value, boxes, }; document.__qwenBoxCanvasTrigger("input"); } const {width, height, context} = canvasDetails(activeCanvas); context.clearRect(0, 0, width, height); activeCanvas = null; start = null; current = null; }); document.addEventListener("pointercancel", () => { if (activeCanvas) { const {width, height, context} = canvasDetails(activeCanvas); context.clearRect(0, 0, width, height); } activeCanvas = null; start = null; current = null; }); } """ _requests_by_session: dict[str, deque[float]] = defaultdict(deque) _rate_limit_lock = Lock() def _client() -> OpenAI: api_key = os.getenv("DASHSCOPE_API_KEY") if not api_key: raise gr.Error( "This Space is waiting for its DASHSCOPE_API_KEY secret. " "The owner needs to finish setup before the demo can answer." ) return OpenAI(api_key=api_key, base_url=BASE_URL, timeout=180.0, max_retries=2) def _stream_terminal_problem(event: Any) -> tuple[str, str] | None: event_type = getattr(event, "type", "") response = getattr(event, "response", None) if event_type == "response.incomplete": details = getattr(response, "incomplete_details", None) reason = str(getattr(details, "reason", None) or "unknown_reason") return reason, f"Qwen returned an incomplete response ({reason})." if event_type == "response.failed": error = getattr(response, "error", None) or getattr(event, "error", None) return "response_failed", f"Qwen reported a failed response: {error or 'unknown error'}" return None def _detection_stream( input_messages: list[dict[str, Any]], enable_thinking: bool, max_output_tokens: int, ) -> Any: return _client().responses.create( model=MODEL, input=input_messages, max_output_tokens=int(max_output_tokens), stream=True, extra_body={"enable_thinking": bool(enable_thinking)}, ) def _check_rate_limit(session_hash: str | None) -> None: session = session_hash or "anonymous" now = time.monotonic() cutoff = now - RATE_LIMIT_WINDOW_SECONDS with _rate_limit_lock: recent = _requests_by_session[session] while recent and recent[0] < cutoff: recent.popleft() if len(recent) >= RATE_LIMIT_REQUESTS: retry_after = max(1, int(recent[0] + RATE_LIMIT_WINDOW_SECONDS - now)) raise gr.Error( f"You've reached the demo limit of {RATE_LIMIT_REQUESTS} requests " f"per 5 minutes. Please try again in about {retry_after} seconds." ) recent.append(now) def _image_data_url(file_path: str) -> str: path = Path(file_path) if not path.is_file(): raise gr.Error(f"The uploaded image {path.name!r} is no longer available.") if path.stat().st_size > MAX_IMAGE_BYTES: raise gr.Error(f"{path.name!r} is larger than the 15 MB per-image limit.") try: with Image.open(path) as image: image.verify() image_format = (image.format or "").lower() except (UnidentifiedImageError, OSError) as error: raise gr.Error(f"{path.name!r} is not a readable image.") from error mime_by_format = { "avif": "image/avif", "gif": "image/gif", "jpeg": "image/jpeg", "jpg": "image/jpeg", "png": "image/png", "webp": "image/webp", } mime_type = mime_by_format.get(image_format) if not mime_type: raise gr.Error( f"{path.name!r} uses an unsupported image format. " "Please upload PNG, JPEG, WEBP, GIF, or AVIF." ) encoded = base64.b64encode(path.read_bytes()).decode("ascii") return f"data:{mime_type};base64,{encoded}" def _pil_data_url(image: Image.Image) -> str: buffer = io.BytesIO() image.convert("RGB").save(buffer, format="JPEG", quality=95) encoded = base64.b64encode(buffer.getvalue()).decode("ascii") return f"data:image/jpeg;base64,{encoded}" def build_user_content(text: str, files: list[str] | None) -> list[dict[str, str]]: clean_text = text.strip() image_paths = files or [] if not clean_text and not image_paths: raise gr.Error("Write a message or attach an image first.") if len(image_paths) > MAX_IMAGES: raise gr.Error(f"Please attach at most {MAX_IMAGES} images per message.") content: list[dict[str, str]] = [] if clean_text: content.append({"type": "input_text", "text": clean_text}) elif image_paths: content.append({"type": "input_text", "text": "Describe and analyze the image(s)."}) content.extend( {"type": "input_image", "image_url": _image_data_url(path)} for path in image_paths ) return content def parse_response(response: Any) -> tuple[str, str]: reasoning_parts: list[str] = [] answer_parts: list[str] = [] for item in getattr(response, "output", []): if getattr(item, "type", None) == "reasoning": reasoning_parts.extend( summary.text for summary in getattr(item, "summary", []) if getattr(summary, "text", None) ) elif getattr(item, "type", None) == "message": answer_parts.extend( part.text for part in getattr(item, "content", []) if getattr(part, "text", None) ) answer = "\n\n".join(answer_parts).strip() if not answer: answer = str(getattr(response, "output_text", "")).strip() if not answer: answer = "The model returned no text response." return "\n\n".join(reasoning_parts).strip(), answer def _display_user_message(text: str, files: list[str] | None) -> str: image_count = len(files or []) attachment_note = "" if image_count: noun = "image" if image_count == 1 else "images" attachment_note = f"\n\n📎 *Attached {image_count} {noun}*" return (text.strip() or "Analyze the attached image(s).") + attachment_note def _streamed_assistant_message(reasoning: str, answer: str) -> str: parts: list[str] = [] if reasoning: parts.append(f"{reasoning}") parts.append(answer or "▌") return "\n\n".join(parts) def chat( message: dict[str, Any] | None, history: list[dict[str, Any]] | None, api_history: list[dict[str, Any]] | None, enable_thinking: bool, max_output_tokens: int, request: gr.Request, ) -> Iterator[ tuple[dict[str, Any], list[dict[str, Any]], list[dict[str, Any]]] ]: message = message or {} text = str(message.get("text") or "") files = [str(path) for path in (message.get("files") or [])] user_content = build_user_content(text, files) _check_rate_limit(getattr(request, "session_hash", None)) visible_history = list(history or []) conversation = list(api_history or []) visible_history.append({"role": "user", "content": _display_user_message(text, files)}) conversation.append({"role": "user", "content": user_content}) cleared_message = {"text": "", "files": []} # Show the submitted message immediately, before the upstream request connects. yield cleared_message, visible_history, conversation visible_history.append( {"role": "assistant", "content": _streamed_assistant_message("", "")} ) yield cleared_message, visible_history, conversation reasoning_parts: list[str] = [] answer_parts: list[str] = [] terminal_problem: tuple[str, str] | None = None try: stream = _client().responses.create( model=MODEL, input=conversation, max_output_tokens=int(max_output_tokens), stream=True, extra_body={"enable_thinking": bool(enable_thinking)}, ) for event in stream: event_type = getattr(event, "type", "") problem = _stream_terminal_problem(event) if problem: terminal_problem = problem continue if event_type == "response.reasoning_summary_text.delta": reasoning_parts.append(str(getattr(event, "delta", ""))) elif event_type == "response.output_text.delta": answer_parts.append(str(getattr(event, "delta", ""))) else: continue reasoning = "".join(reasoning_parts) answer = "".join(answer_parts) visible_history[-1] = { "role": "assistant", "content": _streamed_assistant_message(reasoning, answer), } yield cleared_message, visible_history, conversation except Exception as error: partial = "".join(answer_parts) visible_history[-1] = { "role": "assistant", "content": ( (partial + "\n\n" if partial else "") + "⚠️ Qwen could not finish this request. The demo may be busy, " + "the token pool may be exhausted, or the upstream API may be unavailable.\n\n" + f"`{error}`" ), } yield cleared_message, visible_history, conversation return if terminal_problem: reason, problem = terminal_problem reasoning = "".join(reasoning_parts).strip() partial = "".join(answer_parts).strip() guidance = ( " Increase **Maximum output tokens** or shorten the request." if reason == "max_output_tokens" else "" ) visible_history[-1] = { "role": "assistant", "content": _streamed_assistant_message( reasoning, (partial + "\n\n" if partial else "") + f"⚠️ {problem}{guidance}", ), } yield cleared_message, visible_history, conversation return reasoning = "".join(reasoning_parts).strip() answer = "".join(answer_parts).strip() or "The model returned no text response." visible_history[-1] = { "role": "assistant", "content": _streamed_assistant_message(reasoning, answer), } conversation.append({"role": "assistant", "content": answer}) yield cleared_message, visible_history, conversation def clear_chat() -> tuple[list[Any], list[Any], dict[str, Any]]: return [], [], {"text": "", "files": []} def load_visual_example(event: gr.SelectData) -> dict[str, Any]: example = VISUAL_EXAMPLES[int(event.index)] return {"text": example["prompt"], "files": [example["image"]]} def load_object_example( event: gr.SelectData, ) -> tuple[str, str, bool, None, str, str]: example = OBJECT_EXAMPLES[int(event.index)] return example["image"], example["prompt"], False, None, "", "" def clear_object_outputs(_: str | None) -> tuple[None, str, str]: return None, "", "" def _draw_prompt_boxes( original_path: str, boxes: list[dict[str, Any]] | None, pending_point: list[int] | None = None, ) -> Image.Image: with Image.open(original_path) as source: image = source.convert("RGB") draw = ImageDraw.Draw(image) line_width = max(3, round(min(image.size) / 180)) for index, annotation in enumerate(boxes or [], start=1): coordinates = tuple(int(value) for value in annotation["box"]) positive = annotation["kind"] == "positive" color = "#16a34a" if positive else "#dc2626" prefix = "+" if positive else "−" draw.rectangle(coordinates, outline=color, width=line_width) x1, y1, _, _ = coordinates label = f" {prefix}{index} " text_box = draw.textbbox((x1, y1), label) draw.rectangle(text_box, fill=color) draw.text((x1, y1), label, fill="white") if pending_point: x, y = pending_point radius = max(5, line_width * 2) draw.ellipse((x - radius, y - radius, x + radius, y + radius), fill="#f59e0b") return image def _box_canvas_value( original_path: str | None, boxes: list[dict[str, Any]] | None = None, ) -> dict[str, Any]: if not original_path: return {"image": None, "width": 0, "height": 0, "boxes": []} with Image.open(original_path) as source: width, height = source.size preview = source.convert("RGB") preview.thumbnail((1400, 1400), Image.Resampling.LANCZOS) return { "image": _pil_data_url(preview), "width": width, "height": height, "boxes": list(boxes or []), } def load_box_image( image_path: str | None, ) -> tuple[str | None, list[Any], dict[str, Any], str, str, None, str, str]: if not image_path: return ( None, [], _box_canvas_value(None), "Upload an image to begin.", "", None, "", "", ) return ( image_path, [], _box_canvas_value(image_path), "Choose Positive or Negative, then drag directly over an object to draw a box.", "", None, "", "", ) def load_box_example( event: gr.SelectData, ) -> tuple[ str, list[dict[str, Any]], str, dict[str, Any], str, str, None, str, str, ]: example = BOX_EXAMPLES[int(event.index)] original_path = example["image"] annotations = [ {"box": list(annotation["box"]), "kind": annotation["kind"]} for annotation in example["boxes"] ] positive_count = sum(box["kind"] == "positive" for box in annotations) negative_count = len(annotations) - positive_count status = ( f"Loaded **{example['title']}** with {positive_count} positive and " f"{negative_count} negative example box(es). Edit the prompt or run it as-is." ) return ( original_path, annotations, original_path, _box_canvas_value(original_path, annotations), status, example["prompt"], None, "", "", ) def sync_drawn_boxes( canvas_value: dict[str, Any] | None, original_path: str | None, ) -> tuple[list[dict[str, Any]], str]: if not original_path: raise gr.Error("Upload an image before drawing boxes.") with Image.open(original_path) as image: width, height = image.size annotations: list[dict[str, Any]] = [] for candidate in (canvas_value or {}).get("boxes", []): coordinates = candidate.get("box") kind = str(candidate.get("kind", "")).lower() if not isinstance(coordinates, list) or len(coordinates) != 4: continue if kind not in {"positive", "negative"}: continue x1, y1, x2, y2 = (int(round(float(value))) for value in coordinates) x1, x2 = sorted((max(0, min(width, x1)), max(0, min(width, x2)))) y1, y2 = sorted((max(0, min(height, y1)), max(0, min(height, y2)))) if x2 - x1 >= 4 and y2 - y1 >= 4: annotations.append({"box": [x1, y1, x2, y2], "kind": kind}) positive_count = sum(box["kind"] == "positive" for box in annotations) negative_count = len(annotations) - positive_count return annotations, ( f"{positive_count} positive and {negative_count} negative example(s) marked." ) def undo_box( original_path: str | None, boxes: list[dict[str, Any]] | None, ) -> tuple[dict[str, Any], list[dict[str, Any]], str]: annotations = list(boxes or []) if annotations: annotations.pop() return ( _box_canvas_value(original_path, annotations), annotations, f"{len(annotations)} example box(es) remaining.", ) def clear_boxes( original_path: str | None, ) -> tuple[dict[str, Any], list[Any], str]: return _box_canvas_value(original_path), [], "All example boxes cleared." def _normalized_boxes(original_path: str, boxes: list[dict[str, Any]]) -> str: with Image.open(original_path) as image: width, height = image.size descriptions = [] for annotation in boxes: x1, y1, x2, y2 = annotation["box"] normalized = [ round(1000 * x1 / width), round(1000 * y1 / height), round(1000 * x2 / width), round(1000 * y2 / height), ] descriptions.append(f"{annotation['kind']}: {normalized}") return "; ".join(descriptions) def _parse_detection_boxes( answer: str, width: int, height: int, preserve_labels: bool = False, ) -> list[tuple[tuple[int, int, int, int], str]]: fenced = re.search(r"```(?:json)?\s*([\s\S]*?)```", answer, flags=re.IGNORECASE) candidate = fenced.group(1) if fenced else answer object_match = re.search(r"\{[\s\S]*\}", candidate) list_match = re.search(r"\[[\s\S]*\]", candidate) payload = json.loads((object_match or list_match).group(0)) objects = payload.get("objects", []) if isinstance(payload, dict) else payload detections: list[tuple[tuple[int, int, int, int], str]] = [] for item in objects: coordinates = item.get("box_2d") or item.get("bbox") or item.get("box") if not isinstance(coordinates, list) or len(coordinates) != 4: continue x1, y1, x2, y2 = (float(value) for value in coordinates) x1, x2 = sorted((max(0, min(1000, x1)), max(0, min(1000, x2)))) y1, y2 = sorted((max(0, min(1000, y1)), max(0, min(1000, y2)))) pixel_box = ( round(x1 * width / 1000), round(y1 * height / 1000), round(x2 * width / 1000), round(y2 * height / 1000), ) label = ( str(item.get("label") or "Object")[:80] if preserve_labels else "Qwen match" ) detections.append((pixel_box, label)) return detections def detect_similar( original_path: str | None, boxes: list[dict[str, Any]] | None, target_prompt: str, enable_thinking: bool, max_output_tokens: int, request: gr.Request, ) -> Iterator[tuple[str, str, Any]]: if not original_path: raise gr.Error("Upload an image first.") annotations = list(boxes or []) if not any(box["kind"] == "positive" for box in annotations): raise gr.Error("Draw at least one green Positive box first.") _check_rate_limit(getattr(request, "session_hash", None)) annotated_image = _draw_prompt_boxes(original_path, annotations) requested_target = target_prompt.strip() or ( "Find every other unboxed object that visually matches the positive examples." ) prompt = f""" The image contains visual box prompts drawn by the user. Green boxes marked with + are positive examples of the object to find. Red boxes marked with − are negative examples that must be ignored. The user's instruction is: {requested_target} Find every other unboxed object in this same image that matches the positive examples while respecting the negative examples. Return only valid JSON in this exact shape: {{"objects": [{{"label": "match", "box_2d": [x_min, y_min, x_max, y_max]}}]}} Use XYXY coordinates normalized to integers from 0 to 1000. Do not return the already marked prompt boxes. If there are no additional matches, return {{"objects": []}}. For reference, the prompt boxes in normalized XYXY coordinates are: {_normalized_boxes(original_path, annotations)} """.strip() input_messages = [ { "role": "user", "content": [ {"type": "input_text", "text": prompt}, {"type": "input_image", "image_url": _pil_data_url(annotated_image)}, ], } ] reasoning_parts: list[str] = [] answer_parts: list[str] = [] terminal_problem: tuple[str, str] | None = None retried_without_thinking = False waiting_reasoning = "▌" if enable_thinking else "_Thinking mode is disabled._" yield waiting_reasoning, "**Connecting to Qwen…**\n\n```json\n▌\n```", None try: stream = _detection_stream(input_messages, enable_thinking, max_output_tokens) for event in stream: event_type = getattr(event, "type", "") problem = _stream_terminal_problem(event) if problem: terminal_problem = problem continue if event_type == "response.reasoning_summary_text.delta": reasoning_parts.append(str(getattr(event, "delta", ""))) elif event_type == "response.output_text.delta": answer_parts.append(str(getattr(event, "delta", ""))) else: continue reasoning = "".join(reasoning_parts) or waiting_reasoning answer = "".join(answer_parts) answer_markdown = ( "**Streaming model response…**\n\n" f"```json\n{answer or '▌'}\n```" ) yield reasoning, answer_markdown, None except Exception as error: reasoning = "".join(reasoning_parts) or waiting_reasoning partial_answer = "".join(answer_parts) error_markdown = ( "**Qwen could not finish the box-prompt request.**\n\n" + (f"```json\n{partial_answer}\n```\n\n" if partial_answer else "") + f"`{error}`" ) yield reasoning, error_markdown, None return should_retry = enable_thinking and ( (terminal_problem and terminal_problem[0] == "max_output_tokens") or not "".join(answer_parts).strip() ) if should_retry: retried_without_thinking = True terminal_problem = None answer_parts.clear() reasoning = "".join(reasoning_parts).strip() or waiting_reasoning yield ( reasoning, "**Reasoning reached the output limit. Retrying the final JSON without thinking…**\n\n" "```json\n▌\n```", None, ) try: retry_stream = _detection_stream(input_messages, False, max_output_tokens) for event in retry_stream: event_type = getattr(event, "type", "") problem = _stream_terminal_problem(event) if problem: terminal_problem = problem continue if event_type != "response.output_text.delta": continue answer_parts.append(str(getattr(event, "delta", ""))) answer = "".join(answer_parts) yield ( reasoning, "**Recovering final JSON…**\n\n" f"```json\n{answer or '▌'}\n```", None, ) except Exception as error: terminal_problem = ("retry_failed", f"The automatic JSON retry failed: {error}") if terminal_problem or not "".join(answer_parts).strip(): problem = terminal_problem[1] if terminal_problem else "Qwen returned no final JSON." partial_answer = "".join(answer_parts).strip() details = ( f"**⚠️ {problem}**\n\n" + (f"```json\n{partial_answer}\n```\n\n" if partial_answer else "") + "No detection result was fabricated. Try a larger token budget or turn off Thinking mode." ) yield "".join(reasoning_parts).strip() or waiting_reasoning, details, None return reasoning = "".join(reasoning_parts).strip() answer = "".join(answer_parts).strip() with Image.open(original_path) as image: width, height = image.size prompt_annotations = [ (tuple(box["box"]), "Positive prompt" if box["kind"] == "positive" else "Negative prompt") for box in annotations ] try: detections = _parse_detection_boxes(answer, width, height) note = f"Rendered {len(detections)} additional match(es)." except (AttributeError, TypeError, ValueError, json.JSONDecodeError): detections = [] note = "Qwen's response could not be parsed into boxes; the raw response is shown below." recovery_note = ( "\n\n_Automatic recovery: the reasoning pass exhausted its token budget, so the final JSON was retried without thinking._" if retried_without_thinking else "" ) details = f"**{note}**{recovery_note}\n\n### Model response\n\n```json\n{answer}\n```" reasoning_display = reasoning or "_The model returned no reasoning summary._" yield reasoning_display, details, (original_path, prompt_annotations + detections) def detect_objects( image_path: str | None, target_prompt: str, detect_everything: bool, enable_thinking: bool, max_output_tokens: int, request: gr.Request, ) -> Iterator[tuple[str, str, Any]]: if not image_path: raise gr.Error("Upload an image or choose an example first.") instruction = target_prompt.strip() if detect_everything: instruction = ( "Detect and label every distinct visible object. Use concise, specific labels " "and include separate boxes for separate object instances." ) elif not instruction: raise gr.Error("Describe the objects to detect, or enable Detect everything.") _check_rate_limit(getattr(request, "session_hash", None)) prompt = f""" Perform zero-shot object detection on the supplied image. The user's detection instruction is: {instruction} Return only valid JSON in this exact shape: {{"objects": [{{"label": "specific object label", "box_2d": [x_min, y_min, x_max, y_max]}}]}} Use tight XYXY bounding boxes with coordinates normalized to integers from 0 to 1000. Return one entry per visible object instance that satisfies the instruction. Use concise labels that distinguish requested categories or attributes. Do not invent objects. If there are no matches, return {{"objects": []}}. Return at most 100 objects. """.strip() input_messages = [ { "role": "user", "content": [ {"type": "input_text", "text": prompt}, {"type": "input_image", "image_url": _image_data_url(image_path)}, ], } ] reasoning_parts: list[str] = [] answer_parts: list[str] = [] terminal_problem: tuple[str, str] | None = None retried_without_thinking = False waiting_reasoning = "▌" if enable_thinking else "_Thinking mode is disabled._" yield waiting_reasoning, "**Connecting to Qwen…**\n\n```json\n▌\n```", None try: stream = _detection_stream(input_messages, enable_thinking, max_output_tokens) for event in stream: event_type = getattr(event, "type", "") problem = _stream_terminal_problem(event) if problem: terminal_problem = problem continue if event_type == "response.reasoning_summary_text.delta": reasoning_parts.append(str(getattr(event, "delta", ""))) elif event_type == "response.output_text.delta": answer_parts.append(str(getattr(event, "delta", ""))) else: continue reasoning = "".join(reasoning_parts) or waiting_reasoning answer = "".join(answer_parts) yield ( reasoning, "**Streaming model response…**\n\n" f"```json\n{answer or '▌'}\n```", None, ) except Exception as error: reasoning = "".join(reasoning_parts) or waiting_reasoning partial_answer = "".join(answer_parts) details = ( "**Qwen could not finish object detection.**\n\n" + (f"```json\n{partial_answer}\n```\n\n" if partial_answer else "") + f"`{error}`" ) yield reasoning, details, None return should_retry = enable_thinking and ( (terminal_problem and terminal_problem[0] == "max_output_tokens") or not "".join(answer_parts).strip() ) if should_retry: retried_without_thinking = True terminal_problem = None answer_parts.clear() reasoning = "".join(reasoning_parts).strip() or waiting_reasoning yield ( reasoning, "**Reasoning reached the output limit. Retrying the final JSON without thinking…**\n\n" "```json\n▌\n```", None, ) try: retry_stream = _detection_stream(input_messages, False, max_output_tokens) for event in retry_stream: event_type = getattr(event, "type", "") problem = _stream_terminal_problem(event) if problem: terminal_problem = problem continue if event_type != "response.output_text.delta": continue answer_parts.append(str(getattr(event, "delta", ""))) answer = "".join(answer_parts) yield ( reasoning, "**Recovering final JSON…**\n\n" f"```json\n{answer or '▌'}\n```", None, ) except Exception as error: terminal_problem = ("retry_failed", f"The automatic JSON retry failed: {error}") if terminal_problem or not "".join(answer_parts).strip(): problem = terminal_problem[1] if terminal_problem else "Qwen returned no final JSON." partial_answer = "".join(answer_parts).strip() details = ( f"**⚠️ {problem}**\n\n" + (f"```json\n{partial_answer}\n```\n\n" if partial_answer else "") + "No detection result was fabricated. Try a larger token budget or turn off Thinking mode." ) yield "".join(reasoning_parts).strip() or waiting_reasoning, details, None return reasoning = "".join(reasoning_parts).strip() answer = "".join(answer_parts).strip() with Image.open(image_path) as image: width, height = image.size try: detections = _parse_detection_boxes( answer, width, height, preserve_labels=True )[:100] note = f"Rendered {len(detections)} detected object(s)." except (AttributeError, TypeError, ValueError, json.JSONDecodeError): detections = [] note = "Qwen's response could not be parsed into boxes; the raw response is shown below." recovery_note = ( "\n\n_Automatic recovery: the reasoning pass exhausted its token budget, so the final JSON was retried without thinking._" if retried_without_thinking else "" ) details = f"**{note}**{recovery_note}\n\n### Model response\n\n```json\n{answer}\n```" reasoning_display = reasoning or "_The model returned no reasoning summary._" yield reasoning_display, details, (image_path, detections) CSS = """ html, body { height: auto !important; min-height: 100% !important; overflow-y: auto !important; } body > gradio-app, gradio-app { display: block !important; height: auto !important; min-height: 100vh !important; overflow: visible !important; } .gradio-container { height: auto !important; min-height: 100vh !important; overflow: visible !important; width: min(1080px, calc(100% - 32px)) !important; max-width: 1080px !important; margin-left: auto !important; margin-right: auto !important; padding-bottom: 2rem !important; } #hero { text-align: center; margin: 0 auto 0.75rem; } #hero .hero-lockup { display: flex; align-items: center; justify-content: center; gap: 0.85rem; } #hero img { width: clamp(54px, 8vw, 78px); height: auto; flex: 0 0 auto; } #hero h1 { font-size: clamp(2rem, 5vw, 3.8rem); margin: 0; line-height: 1.05; } #hero p { color: var(--body-text-color-subdued); font-size: 1.05rem; } #notice { border-left: 4px solid var(--color-accent); padding-left: 1rem; } #chat-shell { max-width: 840px; margin: 0 auto; width: 100%; } #box-shell { max-width: 960px; margin: 0 auto; width: 100%; } #object-shell { max-width: 960px; margin: 0 auto; width: 100%; } #examples-gallery { margin-top: 0.35rem; } #box-examples-gallery { margin: 0.35rem 0 1rem; } #example-prompts { color: var(--body-text-color-subdued); font-size: 0.92rem; } @media (max-width: 560px) { .gradio-container { width: calc(100% - 16px) !important; } #hero .hero-lockup { gap: 0.4rem; } #hero img { width: 48px; } } """ with gr.Blocks(title="Free Qwen 3.8 Max") as demo: api_history = gr.State([]) box_original = gr.State(None) box_annotations = gr.State([]) gr.HTML( f"""
Qwen logo

Free Qwen 3.8 Max

100 million tokens, shared with the community. Ask anything or attach up to three images to test Qwen's visual understanding.

""", elem_id="hero", ) gr.Markdown( """ This unofficial demo has a limited shared token pool. Please keep requests purposeful so more people get a turn. Uploaded images are sent to the DashScope API for inference. """, elem_id="notice", ) with gr.Tabs(): with gr.Tab("Chat + vision"): with gr.Column(elem_id="chat-shell"): chatbot = gr.Chatbot( label="Qwen 3.8 Max", height=560, placeholder="Ask a question, test a hard visual puzzle, or attach a screenshot…", buttons=["copy", "copy_all"], reasoning_tags=[("", "")], ) message = gr.MultimodalTextbox( placeholder="Message Qwen 3.8 Max…", file_types=["image"], file_count="multiple", submit_btn="Send", stop_btn=False, autofocus=True, max_plain_text_length=8_000, ) example_gallery = gr.Gallery( value=[(example["image"], example["title"]) for example in VISUAL_EXAMPLES], label="Try a visual challenge — click a sample to load it", columns=3, rows=1, height=255, object_fit="contain", allow_preview=False, buttons=[], elem_id="examples-gallery", ) gr.Markdown( """ **Sample prompts:** Count every shape without double-counting · Recompute the receipt total and flag errors · Infer the visual rule and solve the missing tile """, elem_id="example-prompts", ) with gr.Accordion("Generation settings", open=False): with gr.Row(): thinking = gr.Checkbox( value=True, label="Thinking mode", info="Shows Qwen's reasoning summary before the final answer.", ) max_tokens = gr.Slider( minimum=256, maximum=8_192, value=8_192, step=256, label="Maximum output tokens", ) with gr.Row(): clear = gr.Button("Clear conversation", size="sm") gr.Button( "Qwen announcement ↗", link="https://qwen.ai/blog?id=qwen3.8", link_target="_blank", size="sm", ) with gr.Tab("Box prompting"): with gr.Column(elem_id="box-shell"): gr.Markdown( """ ## Find objects by example Upload an image, choose **Positive** or **Negative**, then **click and drag directly over the image** to draw each box. Qwen will find other objects that look like the positive examples while avoiding the negatives. """ ) box_examples_gallery = gr.Gallery( value=[ (example["image"], example["title"]) for example in BOX_EXAMPLES ], label="Box-prompt examples — click one to load its image, boxes, and prompt", columns=3, rows=1, height=245, object_fit="contain", allow_preview=False, buttons=[], elem_id="box-examples-gallery", ) with gr.Row(): with gr.Column(scale=3): box_image = gr.Image( type="filepath", label="Upload or paste an image", sources=["upload", "clipboard"], height=180, interactive=True, ) box_canvas = gr.HTML( value=_box_canvas_value(None), html_template=BOX_CANVAS_TEMPLATE, css_template=BOX_CANVAS_CSS, js_on_load=BOX_CANVAS_JS, elem_id="box-drawing-canvas", ) box_status = gr.Markdown("Upload an image to begin.") with gr.Column(scale=1, min_width=220): box_target_prompt = gr.Textbox( label="What should Qwen find?", placeholder="For example: Find the other blue circles, but ignore blue squares.", lines=4, ) box_kind = gr.Radio( ["Positive", "Negative"], value="Positive", label="Next box", info="Green = find more like this. Red = ignore this kind.", elem_id="box-kind", interactive=True, ) undo = gr.Button("Undo last box") reset_boxes = gr.Button("Clear boxes") run_box_prompt = gr.Button("Find similar objects", variant="primary") with gr.Accordion("Reasoning stream", open=True): box_reasoning = gr.Markdown( value="Run a box prompt to see Qwen's reasoning summary stream here.", elem_id="box-reasoning-stream", ) box_details = gr.Markdown(elem_id="box-model-response") box_result = gr.AnnotatedImage( label="Qwen detections", height=520, color_map={ "Positive prompt": "#16a34a", "Negative prompt": "#dc2626", "Qwen match": "#2563eb", }, ) with gr.Tab("Object detection"): with gr.Column(elem_id="object-shell"): gr.Markdown( """ ## Detect objects from a prompt Upload an image and describe what Qwen should locate. Unlike box prompting, this is **zero-shot**: no example boxes are required. """ ) object_examples_gallery = gr.Gallery( value=[ (example["image"], example["title"]) for example in OBJECT_EXAMPLES ], label="Object-detection examples — click one to load it", columns=3, rows=1, height=245, object_fit="contain", allow_preview=False, buttons=[], ) with gr.Row(): object_image = gr.Image( type="filepath", label="Image to detect objects in", sources=["upload", "clipboard"], height=430, interactive=True, scale=3, ) with gr.Column(scale=2, min_width=260): object_prompt = gr.Textbox( label="What should Qwen detect?", placeholder="For example: Detect every person wearing a helmet.", lines=5, ) object_detect_all = gr.Checkbox( value=False, label="Detect everything", info="Ignores the custom prompt and inventories all visible objects.", ) object_run = gr.Button("Detect objects", variant="primary") with gr.Accordion("Reasoning stream", open=True): object_reasoning = gr.Markdown( value="Run object detection to see Qwen's reasoning summary stream here.", elem_id="object-reasoning-stream", ) object_details = gr.Markdown(elem_id="object-model-response") object_result = gr.AnnotatedImage( label="Detected objects", height=560, ) message.submit( chat, inputs=[message, chatbot, api_history, thinking, max_tokens], outputs=[message, chatbot, api_history], concurrency_limit=1, concurrency_id="qwen-api", stream_every=0.08, api_name=False, ) clear.click( clear_chat, outputs=[chatbot, api_history, message], queue=False, api_name=False, ) example_gallery.select( load_visual_example, outputs=message, queue=False, api_name=False, ) box_image.upload( load_box_image, inputs=box_image, outputs=[ box_original, box_annotations, box_canvas, box_status, box_target_prompt, box_result, box_reasoning, box_details, ], queue=False, api_name=False, ) box_examples_gallery.select( load_box_example, outputs=[ box_original, box_annotations, box_image, box_canvas, box_status, box_target_prompt, box_result, box_reasoning, box_details, ], queue=False, api_name=False, ) box_canvas.input( sync_drawn_boxes, inputs=[box_canvas, box_original], outputs=[box_annotations, box_status], queue=False, api_name=False, ) undo.click( undo_box, inputs=[box_original, box_annotations], outputs=[box_canvas, box_annotations, box_status], queue=False, api_name=False, ) reset_boxes.click( clear_boxes, inputs=box_original, outputs=[box_canvas, box_annotations, box_status], queue=False, api_name=False, ) run_box_prompt.click( detect_similar, inputs=[box_original, box_annotations, box_target_prompt, thinking, max_tokens], outputs=[box_reasoning, box_details, box_result], concurrency_limit=1, concurrency_id="qwen-api", stream_every=0.08, scroll_to_output=False, show_progress="hidden", api_name=False, ) object_image.upload( clear_object_outputs, inputs=object_image, outputs=[object_result, object_reasoning, object_details], queue=False, api_name=False, ) object_examples_gallery.select( load_object_example, outputs=[ object_image, object_prompt, object_detect_all, object_result, object_reasoning, object_details, ], queue=False, api_name=False, ) object_run.click( detect_objects, inputs=[ object_image, object_prompt, object_detect_all, thinking, max_tokens, ], outputs=[object_reasoning, object_details, object_result], concurrency_limit=1, concurrency_id="qwen-api", stream_every=0.08, scroll_to_output=False, show_progress="hidden", api_name=False, ) demo.queue(default_concurrency_limit=1, max_size=20) if __name__ == "__main__": demo.launch( css=CSS, max_file_size=MAX_IMAGE_BYTES, state_session_capacity=100, )