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
HEA_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"
HfFolder.save_token(HF_API_TOKEN)
DATASET_NAME = os.environ.get("DATASET_NAME")
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 thought 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):
feedback = {
"response": response,
"feedback_type": feedback_type,
"improvement": improvement
}
save_feedback(response, feedback_type, improvement)
return "Thank you for your feedback!"
def handle_user_input(user_input):
print(f"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 = """"""
SUBTITLE = """"""
LINKS = """那些語音處理 (Speech Processing) 踩的坑 | 那些自然語言處理 (Natural Language Processing, NLP) 踩的坑 | 那些ASR和TTS可能會踩的坑 | 那些大模型開發會踩的坑 | 什麼是大語言模型，它是什麼？想要嗎？
用PaddleOCR的PPOCRLabel來微調醫療診斷書和收據 | 基於機器閱讀理解和指令微調的統一信息抽取框架之診斷書醫囑資訊擷取分析
「高熵合金」(High-entropy alloys) 參考論文
"""
iface = gr.Blocks()
with iface:
gr.HTML(TITLE)
gr.HTML(SUBTITLE)
gr.HTML(LINKS)
with gr.Row():
user_input = gr.Textbox(label='歡迎問我關於「高熵合金」(High-entropy alloys) 的各種疑難雜症', lines=2, placeholder="在此輸入問題...")
submit_button = gr.Button("提交")
with gr.Row():
response_output = gr.Textbox(label='模型回應', interactive=False)
with gr.Row():
like_button = gr.Button("👍")
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='所有反饋')
submit_button.click(fn=handle_user_input, inputs=user_input, outputs=response_output)
like_button.click(
fn=lambda response, improvement: handle_feedback(response, "like", improvement),
inputs=[response_output, improvement_input],
outputs=feedback_output
)
dislike_button.click(
fn=lambda response, improvement: handle_feedback(response, "dislike", improvement),
inputs=[response_output, improvement_input],
outputs=feedback_output
)
show_feedback_button.click(fn=show_feedback, outputs=feedback_display)
iface.launch()