你好，我是 Qwen VL

本地多模态助手，能看懂图片，也能陪你写代码、写文案

全程离线图片理解多轮上下文流式输出

内容由本地模型生成，请自行甄别 · Qwen2.5-VL 运行在你的电脑上

松手即可上传图片

""" # ---------- 后端 ---------- @app.route("/") def index(): return render_template_string(PAGE) @app.route("/manifest.webmanifest") def manifest(): """让手机可以「添加到主屏幕」，像 App 一样全屏使用。""" data = { "name": "Qwen VL · 本地多模态助手", "short_name": "Qwen VL", "description": "本地运行的 Qwen2.5-VL 多模态对话助手", "start_url": "/", "scope": "/", "display": "standalone", "orientation": "portrait", "background_color": "#FFFFFF", "theme_color": "#FFFFFF", "icons": [ {"src": "/icon/512", "sizes": "512x512", "type": "image/png", "purpose": "any"}, {"src": "/icon/180", "sizes": "180x180", "type": "image/png"}, ], } return Response( json.dumps(data, ensure_ascii=False), mimetype="application/manifest+json", headers={"Cache-Control": "public, max-age=86400"}, ) @app.route("/icon/") def icon(size): """返回预生成的应用图标（缺失时回退到内联 SVG）。""" path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "static", f"icon-{size}.png") if os.path.exists(path): with open(path, "rb") as f: return Response(f.read(), mimetype="image/png", headers={"Cache-Control": "public, max-age=604800"}) svg = ( "" ) return Response(svg, mimetype="image/svg+xml") def _model_context_length(model): """从 Ollama 读取模型的上下文窗口大小（带缓存）。""" if model in _CTX_CACHE: return _CTX_CACHE[model] ctx = 32768 try: r = requests.post(f"{OLLAMA_URL}/api/show", json={"model": model}, timeout=3) info = r.json().get("model_info", {}) or {} for k, v in info.items(): if k.endswith(".context_length") and isinstance(v, int) and v > 0: ctx = v break except Exception: # noqa: BLE001 pass _CTX_CACHE[model] = ctx return ctx @app.route("/api/status") def api_status(): """探测 Ollama 是否可用、模型是否已拉取，并返回上下文窗口信息。""" try: r = requests.get(f"{OLLAMA_URL}/api/tags", timeout=2.5) r.raise_for_status() names = [m.get("name", "") for m in r.json().get("models", [])] return jsonify({ "ok": True, "model": MODEL, "available": any(n == MODEL or n.split(":")[0] == MODEL for n in names), "models": names, "context_length": _model_context_length(MODEL), "num_ctx_default": DEFAULT_NUM_CTX, "system_chars": len(SYSTEM_PROMPT), }) except Exception as e: # noqa: BLE001 return jsonify({ "ok": False, "model": MODEL, "error": str(e), "context_length": 32768, "num_ctx_default": DEFAULT_NUM_CTX, "system_chars": len(SYSTEM_PROMPT), }) @app.route("/chat", methods=["POST"]) def chat(): data = request.get_json(force=True) user_text = (data.get("text") or "").strip() images = data.get("images") or [] history = data.get("history") or [] if not user_text and not images: return jsonify({"error": "请输入文字或附带图片"}), 400 # 上下文长度（前端滑杆传上来，做一次范围收敛） try: num_ctx = int(data.get("num_ctx") or DEFAULT_NUM_CTX) except (TypeError, ValueError): num_ctx = DEFAULT_NUM_CTX num_ctx = max(MIN_NUM_CTX, min(MAX_NUM_CTX, num_ctx)) # 构建 Ollama 消息（qwen2.5vl 图片放 messages 的 images 数组） content = user_text or ("请描述这张图片。" if images else "") msg = {"role": "user", "content": content} if images: b64s = [] for im in images: b = im.split(",", 1)[1] if "," in im else im b64s.append(b) # 体积兜底：过大的图既拖慢预填，也容易让 Ollama 直接报错 total_img = sum(len(b) for b in b64s) if total_img > 18 * 1024 * 1024: return jsonify({ "error": "图片总量过大（约 %.1f MB），请减少图片数量或降低分辨率" % (total_img / 1048576) }), 413 msg["images"] = b64s # 多轮上下文（仅文本，避免每轮重复上传图片） messages = [] for h in history[-12:]: if not isinstance(h, dict): continue role = h.get("role") text = (h.get("content") or "").strip() if role in ("user", "assistant") and text: messages.append({"role": role, "content": text}) while messages and messages[0]["role"] != "user": messages.pop(0) messages.insert(0, {"role": "system", "content": SYSTEM_PROMPT}) def generate(): usage = {} t_start = time.perf_counter() t_first = None try: resp = requests.post( f"{OLLAMA_URL}/api/chat", json={ "model": MODEL, "messages": messages, "stream": True, "options": {"num_ctx": num_ctx}, }, stream=True, timeout=600, ) resp.raise_for_status() for line in resp.iter_lines(decode_unicode=True): if not line: continue try: chunk = json.loads(line) except json.JSONDecodeError: continue if chunk.get("done"): eval_cnt = chunk.get("eval_count") or 0 eval_ns = chunk.get("eval_duration") or 0 prompt_ns = chunk.get("prompt_eval_duration") or 0 load_ns = chunk.get("load_duration") or 0 total_ns = chunk.get("total_duration") or 0 usage = { "prompt": chunk.get("prompt_eval_count") or 0, "eval": eval_cnt, # 以下均为纳秒 "eval_ns": eval_ns, "prompt_ns": prompt_ns, "load_ns": load_ns, "total_ns": total_ns, "num_ctx": num_ctx, # 生成速度 / 首字延迟（秒），后端先算好，前端直接用 "tps": round(eval_cnt / (eval_ns / 1e9), 1) if eval_ns and eval_cnt else 0, "ttft": round((prompt_ns + load_ns) / 1e9, 2) if prompt_ns else 0, # 墙钟兜底（Ollama 某些版本不返回 *_duration） "wall": round(time.perf_counter() - t_start, 2), "ttft_wall": round((t_first or time.perf_counter()) - t_start, 2), } delta = chunk.get("message", {}).get("content", "") if delta: if t_first is None: t_first = time.perf_counter() yield delta except requests.exceptions.RequestException as e: yield f"\n\n> **调用 Ollama 失败**：{e}\n>\n> 请确认 Ollama 已启动（`ollama serve`）。" finally: if usage: yield USAGE_MARK + json.dumps(usage) return Response( generate(), mimetype="text/plain; charset=utf-8", headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"}, )