import os import gradio as gr import aiohttp import asyncio import json from datasets import Dataset, DatasetDict, load_dataset from huggingface_hub import HfFolder import subprocess def upgrade_pip(): try: subprocess.check_call([os.sys.executable, "-m", "pip", "install", "--upgrade", "pip"]) print("pip 升級成功") except subprocess.CalledProcessError: print("pip 升級失敗") # 呼叫升級函數 upgrade_pip() # 從環境變量中獲取 Hugging Face API 令牌和其他配置 HF_API_TOKEN = os.environ.get("Feedback_API_TOKEN") LLM_API = os.environ.get("LLM_API") LLM_URL = os.environ.get("LLM_URL") USER_ID = "HuggingFace Space" DATASET_NAME = os.environ.get("DATASET_NAME") # 確保令牌不為空 if HF_API_TOKEN is None: raise ValueError("HF_API_TOKEN 環境變量未設置。請在 Hugging Face Space 的設置中添加該環境變量。") # 設置 Hugging Face API 令牌 HfFolder.save_token(HF_API_TOKEN) # 定義數據集特徵 features = { "user_input": "string", "response": "string", "feedback_type": "string", "improvement": "string" } # 加載或創建數據集 try: dataset = load_dataset(DATASET_NAME) except: dataset = DatasetDict({ "feedback": Dataset.from_dict({ "user_input": [], "response": [], "feedback_type": [], "improvement": [] }) }) async def send_chat_message(LLM_URL, LLM_API, user_input): payload = { "inputs": {}, "query": user_input, "response_mode": "streaming", "conversation_id": "", "user": USER_ID, } print("Sending chat message payload:", payload) full_response = [] buffer = b"" # 使用字節緩衝區來處理不完整的行 async with aiohttp.ClientSession() as session: try: async with session.post( url=f"{LLM_URL}/chat-messages", headers={"Authorization": f"Bearer {LLM_API}"}, json=payload, timeout=aiohttp.ClientTimeout(total=60) ) as response: if response.status != 200: error_text = await response.text() print(f"Error: {response.status}, Body: {error_text}") return f"Error: {response.status}" # 手動迭代數據塊 (chunk) async for chunk in response.content.iter_any(): buffer += chunk # 嘗試按行分割緩衝區中的數據 while b'\n' in buffer: line, buffer = buffer.split(b'\n', 1) line_str = line.decode('utf-8').strip() if not line_str or "data: " not in line_str: continue try: # 檢查是否是我們要的答案，而不是工作流日誌 # 這些日誌訊息 (e.g., event: "node_started") 就是造成問題的元兇 # 我們只關心包含 "answer" 的 message_chunk if '"event": "message_chunk"' in line_str or '"answer"' in line_str: print("Received line:", line_str) # 只打印有用的行 data_part = line_str.split("data: ", 1)[1] data = json.loads(data_part) if "answer" in data: full_response.append(data["answer"]) # 如果是 Dify API，答案可能在 data['data']['answer'] elif "data" in data and "answer" in data["data"]: full_response.append(data["data"]["answer"]) except (IndexError, json.JSONDecodeError) as e: # 忽略無法解析的行，它們很可能是日誌 # print(f"Skipping unparsable line: {line_str}, error: {e}") pass if full_response: return ''.join(full_response).strip() else: # 如果循環結束都沒有收到答案，返回提示 return "抱歉，我無法從 API 取得有效的回覆。" except Exception as e: print(f"Exception: {e}") return f"Exception: {e}" async def handle_input_async(user_input): print(f"Handling input: {user_input}") chat_response = await send_chat_message(LLM_URL, LLM_API, user_input) print("Chat response:", chat_response) # Debug information return chat_response def handle_input(user_input): print(f"Handling input synchronously: {user_input}") return asyncio.run(handle_input_async(user_input)) def save_feedback(user_input, response, feedback_type, improvement): feedback = { "user_input": user_input, "response": response, "feedback_type": feedback_type, "improvement": improvement } print(f"Saving feedback: {feedback}") # Append to the dataset new_data = { "user_input": [user_input], "response": [response], "feedback_type": [feedback_type], "improvement": [improvement] } global dataset dataset["feedback"] = Dataset.from_dict({ "user_input": dataset["feedback"]["user_input"] + [user_input], "response": dataset["feedback"]["response"] + [response], "feedback_type": dataset["feedback"]["feedback_type"] + [feedback_type], "improvement": dataset["feedback"]["improvement"] + [improvement] }) dataset.push_to_hub(DATASET_NAME) def handle_feedback(response, feedback_type, improvement): global last_user_input save_feedback(last_user_input, response, feedback_type, improvement) return "感謝您的反饋！" # 讀取並顯示反饋內容的函數 def show_feedback(): try: feedbacks = dataset["feedback"].to_pandas().to_dict(orient="records") print(f"Feedbacks: {feedbacks}") # Debug information return feedbacks except Exception as e: print(f"Error: {e}") # Debug information return {"error": str(e)} TITLE = """

Large Language Model (LLM) Playground 💬 Insurance FAQ

""" SUBTITLE = """

deep-learning-101.github.io | AI | TonTon Huang Ph.D. | 手把手帶你一起踩AI坑

""" LINKS = """ Deep Learning 101 Github | Deep Learning 101 | 台灣人工智慧社團 FB | YouTube
AI 陪伴機器人：2025 趨勢分析技術突破、市場潛力與未來展望 | 金融科技新浪潮：生成式 AI (GenAI) 應用場景、效益與導入挑戰
避開 AI Agent 開發陷阱：常見問題、挑戰與解決方案 (實戰經驗)：探討多種 AI 代理人工具的應用經驗與挑戰，分享實用經驗與工具推薦。
白話文手把手帶你科普 GenAI：淺顯介紹生成式人工智慧核心概念，強調硬體資源和數據的重要性。
大型語言模型直接就打完收工？：回顧 LLM 領域探索歷程，討論硬體升級對 AI 開發的重要性。
檢索增強生成不是萬靈丹：挑戰與優化技巧：探討 RAG 技術應用與挑戰，提供實用經驗分享和工具建議。
大型語言模型 (LLM) 入門完整指南：原理、應用與未來 (2025 版)：探討多種 LLM 工具的應用與挑戰，強調硬體資源的重要性。
解析探索大型語言模型：模型發展歷史、訓練及微調技術的 VRAM 估算：探討 LLM 的發展與應用，強調硬體資源在開發中的關鍵作用。。
Diffusion Model 完全解析：從原理、應用到實作 (AI 圖像生成)：深入探討影像生成與分割技術的應用，強調硬體資源的重要性。
ASR/TTS 開發避坑指南：語音辨識與合成的常見挑戰與對策：探討 ASR 和 TTS 技術應用中的問題，強調數據質量的重要性。
那些自然語言處理 (Natural Language Processing, NLP) 踩的坑：分享 NLP 領域的實踐經驗，強調數據質量對模型效果的影響。
那些語音處理 (Speech Processing) 踩的坑：分享語音處理領域的實務經驗，強調資料品質對模型效果的影響。
用PPOCRLabel來幫PaddleOCR做OCR的微調和標註
基於機器閱讀理解和指令微調的統一信息抽取框架之診斷書醫囑資訊擷取分析
""" # 添加示例 examples = [ ["什麼是實支實付？"], ["我要查房貸利率"], ["保險天數如何計算"], ["CaaS是什麼？"], ["介紹機車強制險？"], ["汽車強制險怎保"], ["汽車第三人責任險與強制汽/機車責任險有什麼差別？"], ["青壯年生涯保險？中年生涯保險？高齡生涯保險??"], ["微型保險是什麼？"] ] with gr.Blocks() as iface: gr.HTML(TITLE) gr.HTML(SUBTITLE) gr.HTML(LINKS) with gr.Row(): chatbot = gr.Chatbot() with gr.Row(): user_input = gr.Textbox(label='輸入您的問題', placeholder="在此輸入問題...") submit_button = gr.Button("問題輸入好，請點我送出") gr.Examples(examples=examples, inputs=user_input) with gr.Row(): dislike_button = gr.Button(" 👎 覺得答案待改善，請輸入改進建議，再按我送出保存") improvement_input = gr.Textbox(label='請輸入改進建議', placeholder='請輸入如何改進模型回應的建議') with gr.Row(): feedback_output = gr.Textbox(label='反饋結果執行狀態', interactive=False) with gr.Row(): show_feedback_button = gr.Button("查看目前所有反饋記錄") feedback_display = gr.JSON(label='所有反饋記錄') def chat(user_input, history): response = handle_input(user_input) history.append((user_input, response)) return history, history submit_button.click(fn=chat, inputs=[user_input, chatbot], outputs=[chatbot, chatbot]) dislike_button.click( fn=lambda response, improvement: handle_feedback(response, "dislike", improvement), inputs=[chatbot, improvement_input], outputs=feedback_output ) show_feedback_button.click(fn=show_feedback, outputs=feedback_display) iface.launch()