import gradio as gr import requests import mimetypes import json, os import asyncio import aiohttp import subprocess # --- 1. 環境設定 --- # pip 升級 (通常 Space 啟動時跑一次即可) def upgrade_pip(): try: subprocess.check_call([os.sys.executable, "-m", "pip", "install", "--upgrade", "pip"]) print("pip 升級成功") except subprocess.CalledProcessError: print("pip 升級失敗") upgrade_pip() LLM_API = os.environ.get("LLM_API", "").strip() LLM_URL = os.environ.get("LLM_URL", "").strip() # 確保去除空格 USER_ID = "HuggingFace Space" # --- 2. 上傳檔案函式 (修正版) --- async def upload_file(LLM_URL, LLM_API, file_path, user_id): """ 將本地暫存檔案上傳到 LLM Server,取得 file_id """ if not os.path.exists(file_path): return {"error": f"File {file_path} not found"} mime_type, _ = mimetypes.guess_type(file_path) if mime_type is None: mime_type = 'application/octet-stream' filename = os.path.basename(file_path) print(f"正在上傳檔案: {filename} ({mime_type})") try: data = aiohttp.FormData() # 注意: 這裡必須再次 open file,aiohttp 會自動處理串流 data.add_field('file', open(file_path, 'rb'), filename=filename, content_type=mime_type) data.add_field('user', user_id) async with aiohttp.ClientSession() as session: async with session.post( f"{LLM_URL}/files/upload", headers={"Authorization": f"Bearer {LLM_API}"}, data=data ) as response: response_text = await response.text() print(f"上傳回應狀態: {response.status}") if response.status != 200 and response.status != 201: print(f"上傳失敗回應: {response_text}") return {"error": f"Upload failed: {response.status} - {response_text}"} return json.loads(response_text) except Exception as e: print(f"上傳過程發生例外: {e}") return {"error": str(e)} # --- 3. 對話請求函式 (改用 file_id) --- async def send_chat_message(LLM_URL, LLM_API, category, file_id): """ 使用 file_id 發送對話請求 """ payload = { "inputs": {}, "query": category, "conversation_id": "", "user": USER_ID, "response_mode": "streaming", "files": [ { "type": "image", "transfer_method": "local_file", # 注意:使用 ID 時這裡通常是 local_file "upload_file_id": file_id } ] } print(f"發送請求中... (File ID: {file_id})") answer = "" try: async with aiohttp.ClientSession() as session: async with session.post( f"{LLM_URL}/chat-messages", headers={ "Authorization": f"Bearer {LLM_API}", "Content-Type": "application/json" }, json=payload ) as response: if response.status != 200: error_text = await response.text() return f"Chat Error {response.status}: {error_text}" async for line_bytes in response.content: line = line_bytes.decode("utf-8").strip() if line.startswith("data: "): try: data = json.loads(line[6:]) if "answer" in data: answer += data["answer"] if "error" in data: return f"Stream Error: {data}" except: continue except Exception as e: return f"Request Exception: {str(e)}" return answer or "No answer returned." # --- 4. 主處理邏輯 --- async def handle_input(file_path, category): if not file_path: return "請先上傳圖片" # 步驟 1: 上傳檔案 upload_result = await upload_file(LLM_URL, LLM_API, file_path, USER_ID) # 檢查上傳是否成功 if "error" in upload_result: return f"上傳錯誤: {upload_result['error']}" file_id = upload_result.get("id") if not file_id: return f"錯誤: 上傳成功但未回傳 ID。回應: {upload_result}" # 步驟 2: 發送對話 return await send_chat_message(LLM_URL, LLM_API, category, file_id) # UI 元件 & 資料 examples = [ ['DEMO/Medical1.jpg', '診斷證明書'], ['DEMO/Medical2.jpg', '診斷證明書'], ['DEMO/passport.png', '護照'], ['DEMO/residence.png', '居留證'], ['DEMO/boarding-pass.png', '機票'], ['DEMO/taxi.jpg', '計程車乘車證明'], ['DEMO/etag.jpg', '通行明細 (etag)'], ["DEMO/qrcode.jpg", 'QRCODE發票'], ['DEMO/mthsr.JPG', '超商高鐵車票'], ['DEMO/thsr.jpg', '高鐵車票'], ['DEMO/mtra.jpg', '超商台鐵車票'], ['DEMO/tra.JPG', '台鐵車票'], ['DEMO/ID-back.png', '身份證背面'], ['DEMO/ID.png', '身份證正面'], ['DEMO/health.png', '健保卡'], ] TITLE = """