import os import gradio as gr import aiohttp import asyncio import json from datasets import Dataset, DatasetDict, load_dataset, load_from_disk from huggingface_hub import HfApi, 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() # 定義 CSS 樣式 custom_css = """ #title { font-size: 2em; font-weight: bold; text-align: center; margin-bottom: 20px; } #subtitle { font-size: 1.5em; text-align: center; margin-bottom: 20px; } #links { text-align: center; margin-bottom: 30px; } #user_input, #improvement_input, #feedback_output { width: 100%; margin-bottom: 10px; } #chatbot { height: 300px; overflow-y: auto; background: #f7f7f7; border-radius: 10px; padding: 20px; border: 1px solid #ccc; } #feedback_display { background: #f7f7f7; border-radius: 10px; padding: 20px; border: 1px solid #ccc; } .gr-button { width: 100%; margin-bottom: 10px; background-color: #4CAF50; color: white; border: none; padding: 10px 20px; text-align: center; text-decoration: none; display: inline-block; font-size: 16px; cursor: pointer; border-radius: 5px; } .gr-button:hover { background-color: #45a049; } """ # 從環境變量中獲取 Hugging Face API 令牌和其他配置 HF_API_TOKEN = os.environ.get("HF_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(user_input): payload = { "inputs": {}, "query": user_input, "response_mode": "streaming", "conversation_id": "", "user": USER_ID, } print("Sending chat message payload:", payload) 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: print(f"Error: {response.status}") return f"Error: {response.status}" full_response = [] async for line in response.content: line = line.decode('utf-8').strip() if not line: continue if "data: " not in line: continue try: data = json.loads(line.split("data: ")[1]) if "answer" in data: full_response.append(data["answer"]) except (IndexError, json.JSONDecodeError) as e: print(f"Error parsing line: {line}, error: {e}") continue if full_response: return ''.join(full_response).strip() else: return "Error: No response found in the response" except Exception as e: print(f"Exception: {e}") return f"Exception: {e}" async def handle_input(user_input): print(f"Handling input: {user_input}") chat_response = await send_chat_message(user_input) print("Chat response:", chat_response) return chat_response def run_sync(user_input): print(f"Running sync with input: {user_input}") return asyncio.run(handle_input(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["feedback"].add_item(new_data) 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 handle_user_input(user_input): print(f"User input: {user_input}") global last_user_input last_user_input = user_input # 保存最新的用戶輸入 return run_sync(user_input) # 讀取並顯示反饋內容的函數 def show_feedback(): try: feedbacks = dataset["feedback"].to_pandas().to_dict(orient="records") return feedbacks except Exception as e: return f"Error: {e}" TITLE = """

Large Language Model (LLM) Playground 💬 High-Entropy-Alloys-FAQ

""" SUBTITLE = """

TonTon Huang Ph.D. @ 2024/04

""" LINKS = """ Deep Learning 101 Github | Deep Learning 101 | 台灣人工智慧社團 FB | YouTube
手把手帶你一起踩AI坑 | ComfyUI + Stable Diffuision
白話文手把手帶你科普 GenAI | 大型語言模型直接就打完收工？
什麼是大語言模型，它是什麼？想要嗎？ | 那些檢索增強生成要踩的坑
那些語音處理 (Speech Processing) 踩的坑 | 那些自然語言處理 (Natural Language Processing, NLP) 踩的坑
那些ASR和TTS可能會踩的坑 | 那些大模型開發會踩的坑
用PPOCRLabel來幫PaddleOCR做OCR的微調和標註 | 基於機器閱讀理解和指令微調的統一信息抽取框架之診斷書醫囑資訊擷取分析
""" # 添加示例 examples = [ ["請問high entropy nitride coatings的形成，主要可透過那些元素來讓這個材料形成熱穩定？"], ["AlCoCrFeNi HEA coating 可用怎樣的實驗方法做到 ?"], ["高熵合金與傳統合金在成分和特性的差異？"], ["高熵塗層材料的種類有哪些?"], ["如何提高LWHEAs延展性？"], ["如何優化HEA的性能？"], ["查詢 https://doi.org/10.3390/ma17020453"] ] with gr.Blocks(css=custom_css) as iface: gr.HTML(TITLE) gr.HTML(SUBTITLE) gr.HTML(LINKS) with gr.Row(): chatbot = gr.Chatbot(elem_id="chatbot") with gr.Row(): user_input = gr.Textbox(label='輸入您的問題', placeholder="在此輸入問題...", elem_id="user_input") submit_button = gr.Button("問題輸入好，請點我送出", elem_id="submit_button") gr.Examples(examples=examples, inputs=user_input) with gr.Row(): dislike_button = gr.Button("👎 覺得答案待改善，請輸入改進建議，再按我送出保存", elem_id="dislike_button") improvement_input = gr.Textbox(label='請輸入改進建議', placeholder='請輸入如何改進模型回應的建議', elem_id="improvement_input") with gr.Row(): feedback_output = gr.Textbox(label='反饋結果執行狀態', interactive=False, elem_id="feedback_output") with gr.Row(): show_feedback_button = gr.Button("查看目前所有反饋記錄", elem_id="show_feedback_button") feedback_display = gr.JSON(label='所有反饋記錄', elem_id="feedback_display") def chat(user_input, history): response = handle_user_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()