""" LeyoAI Video Safety Assistant — 带 Supabase 鉴权 """ import gradio as gr import asyncio from fastapi import FastAPI, HTTPException from pydantic import BaseModel from typing import Optional import torch, os, sys import warnings warnings.filterwarnings("ignore") from leyoai_auth.auth import verify_token from leyoai_auth.quota import check_quota, get_user_plan, log_usage BASE_MODEL = "Qwen/Qwen2.5-1.5B-Instruct" LORA_MODEL = "FFZwai/qwen2.5-1.5b-video-safety" MODEL_READY = False try: from transformers import AutoModelForCausalLM, AutoTokenizer from peft import PeftModel print(f"📥 Loading base: {BASE_MODEL}") base_model = AutoModelForCausalLM.from_pretrained(BASE_MODEL, dtype=torch.float32) print(f"🔗 Loading LoRA: {LORA_MODEL}") try: model = PeftModel.from_pretrained(base_model, LORA_MODEL) model.eval() print("✅ LoRA loaded!") except Exception as e: print(f"⚠️ LoRA failed, using base: {e}") model = base_model model.eval() model = model.to("cpu") tokenizer = AutoTokenizer.from_pretrained(BASE_MODEL, trust_remote_code=True) if tokenizer.pad_token is None: tokenizer.pad_token = tokenizer.eos_token print("✅ Model ready!") MODEL_READY = True except Exception as e: print(f"⚠️ Model load failed: {e}") import traceback; traceback.print_exc() def _user_panel(token: str) -> str: if not token: return "🔓 **未登录**" payload = verify_token(token) if payload is None: return "❌ Token 无效" uid = payload.get("sub", "?") email = payload.get("email", "?") plan = get_user_plan(uid, token) if uid else "?" available, usage, limit = check_quota(uid, "video", token) pct = f"{usage}/{limit}" if limit > 0 else f"{usage}/∞" status = "✅ 可用" if available else "⚠️ 已用完" return f"🔐 `{email[:20]}…` | 📦 {plan} | 📊 {pct} | {status}" def answer(question, token): """带鉴权的回答函数 — token 从 Gradio token_box 直接传入""" if not token: return "❌ 请先登录：在上方输入您的 JWT Token" payload = verify_token(token) if payload is None: return "❌ Token 无效或已过期，请重新登录" uid = payload.get("sub", "") if not uid: return "❌ 无法识别用户" # 配额检查 available, usage, limit = check_quota(uid, "video", token) if not available: return f"❌ 本月配额已用完（{usage}/{limit}），请升级套餐或下月再试" if not MODEL_READY: return "⚠️ 模型未就绪，请稍后再试" if not question or not question.strip(): return "请输入问题" messages = [ {"role": "system", "content": ( "你是一个专业的视频内容安全分析助手。分析用户描述的视频场景，" "识别其中的安全风险、危险行为和隐患，给出简洁、专业的安全建议。" "用中文回答。" )}, {"role": "user", "content": question.strip()}, ] text = tokenizer.apply_chat_template(messages, tokenize=False, add_generation_prompt=True) inputs = tokenizer(text, return_tensors="pt", max_length=512, truncation=True) inputs = {k: v.to("cpu") for k, v in inputs.items()} with torch.no_grad(): outputs = model.generate( **inputs, max_new_tokens=250, do_sample=True, temperature=0.7, top_p=0.9, pad_token_id=tokenizer.pad_token_id, eos_token_id=tokenizer.eos_token_id, ) response = tokenizer.decode(outputs[0][inputs["input_ids"].shape[1]:], skip_special_tokens=True) # 记录用量 try: log_usage(user_id=uid, product="video", action="answer", tokens_used=0, token=token) except Exception: pass return response.strip() if response.strip() else "模型未返回有效回答。" examples = [ ["视频里电脑突然弹出\"系统中毒，立即联系客服\"的弹窗，这是诈骗吗？"], ["有人发来链接说点击领取红包，这个链接安全吗？"], ["油锅着火时，旁边有人用水去泼，正确吗？"], ["视频里看到有人被困在地铁屏蔽门和车厢之间，这时应该怎么做？"], ] def load_token(request: gr.Request): """从 URL query param 读取 token 并自动填充""" token = "" try: token = request.request.query_params.get("token", "") if request else "" except Exception: pass return token, _user_panel(token) def generate_response(message, api_key=""): """供 FastAPI 调用的核心推理函数（跳过用户鉴权，由 API Gateway 验证）""" import os INTERNAL_KEY = os.environ.get("LEYOAI_INTERNAL_KEY", "leyoai-internal-2026") if api_key != INTERNAL_KEY: raise ValueError("Invalid internal API key") msg = message.strip() if not msg: raise ValueError("Message is required") if not MODEL_READY: raise ValueError("Model not ready") messages = [ {"role": "system", "content": ( "你是一个专业的视频内容安全分析助手。分析用户描述的视频场景，" "识别其中的安全风险、危险行为和隐患，给出简洁、专业的安全建议。" "用中文回答。" )}, {"role": "user", "content": msg}, ] text = tokenizer.apply_chat_template(messages, tokenize=False, add_generation_prompt=True) inputs = tokenizer(text, return_tensors="pt", max_length=512, truncation=True) inputs = {k: v.to("cpu") for k, v in inputs.items()} with torch.no_grad(): outputs = model.generate( **inputs, max_new_tokens=250, do_sample=True, temperature=0.7, top_p=0.9, pad_token_id=tokenizer.pad_token_id, eos_token_id=tokenizer.eos_token_id, ) response = tokenizer.decode(outputs[0][inputs["input_ids"].shape[1]:], skip_special_tokens=True) return response.strip() if response.strip() else "模型未返回有效回答。" with gr.Blocks(title="LeyoAI Video Safety") as demo: gr.Markdown("# 🎬 LeyoAI Video Safety\n基于 Qwen2.5-1.5B + 视频安全 LoRA") gr.Markdown("> ⚠️ CPU 推理，首个回复约需 20-60 秒") token_box = gr.Textbox( label="🔑 JWT Token（从 leyoai.vercel.app/dashboard 获取）", placeholder="eyJhbGc...", type="password", lines=1, ) user_status = gr.Markdown("🔓 **未登录** — 请在上方输入 JWT Token") token_box.change(_user_panel, inputs=token_box, outputs=user_status) with gr.Row(): with gr.Column(scale=3): question = gr.Textbox( label="📹 描述视频场景或安全问题", placeholder="例：视频里有人接到电话说医保卡被冻结…", lines=4, ) with gr.Row(): submit_btn = gr.Button("🔍 分析安全风险", variant="primary") clear_btn = gr.Button("🗑️ 清除") with gr.Column(scale=2): gr.Markdown("**适用场景：** 🔥厨房安全 | 💰诈骗识别 | 🚇地铁安全 | 🏠火灾逃生") output = gr.Textbox(label="✅ 安全分析结果", lines=8) gr.Examples(examples=examples, inputs=question) submit_btn.click(answer, inputs=[question, token_box], outputs=output) question.submit(answer, inputs=[question, token_box], outputs=output) clear_btn.click(fn=lambda: ("", ""), inputs=[], outputs=[question, output]) demo.load(load_token, outputs=[token_box, user_status]) # ========== FastAPI 端点 ========== import os INTERNAL_API_KEY = os.environ.get("LEYOAI_INTERNAL_KEY", "leyoai-internal-2026") class ChatRequest(BaseModel): message: str system_prompt: Optional[str] = None temperature: Optional[float] = 0.7 max_tokens: Optional[int] = 512 api_key: str = "" class ChatResponse(BaseModel): response: str model: str = "video" status: str = "success" api_app = FastAPI(title="LeyoAI Video Assistant API") @api_app.post("/api/v1/chat", response_model=ChatResponse) async def api_chat(request: ChatRequest): try: if not request.api_key or request.api_key != INTERNAL_API_KEY: raise HTTPException(status_code=401, detail="Invalid internal API key") if not request.message or not request.message.strip(): raise HTTPException(status_code=400, detail="Message is required") # 获取该 Space 的 system_prompt sp = request.system_prompt if sp is None: # 从全局变量读取默认值 try: sp = SYSTEM_PROMPT except NameError: sp = "" loop = asyncio.get_event_loop() response = await loop.run_in_executor( None, generate_response, request.message, request.api_key ) return ChatResponse(response=response or "模型未返回有效回答。") except HTTPException: raise except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @api_app.get("/api/v1/health") async def health_check(): return {"status": "ok", "model": "video", "version": "2.0"} app = gr.mount_gradio_app(api_app, demo, path="/") if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=7860)