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) # Debug information 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: print("Received line:", line) # Debug information 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}") # Debug information 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_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 = """

TonTon Huang Ph.D. @ 2024/06

""" LINKS = """那些語音處理 (Speech Processing) 踩的坑 | 那些自然語言處理 (Natural Language Processing, NLP) 踩的坑 | 那些ASR和TTS可能會踩的坑 | 那些大模型開發會踩的坑 | 什麼是大語言模型，它是什麼？想要嗎？
用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()