import os
import gradio as gr
import aiohttp
import asyncio
import json
import urllib.parse
import traceback # 新增此行以便捕捉完整的異常堆疊
LLM_API = os.environ.get("LLM_API")
LLM_URL = os.environ.get("LLM_URL")
USER_ID = "HuggingFace Space" # Placeholder user ID
import traceback
import aiohttp
import asyncio
import urllib.parse
# 設置重試次數和延遲
MAX_RETRIES = 3
RETRY_DELAY = 2 # 每次重試之間的延遲秒數
TIMEOUT_DURATION = 120 # 超時時間設定為 120 秒
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
for attempt in range(MAX_RETRIES):
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=TIMEOUT_DURATION) # 增加超時時間
) as response:
if response.status != 200:
print(f"Error: {response.status}")
return f"Error: {response.status}"
full_response = []
async for line in response.content.iter_chunked(1024):
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:
decoded_answer = urllib.parse.unquote(data["answer"])
full_response.append(decoded_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 (asyncio.TimeoutError, aiohttp.ClientError) as e:
print(f"Attempt {attempt + 1} failed with error: {e}")
print(traceback.format_exc()) # 顯示詳細的錯誤堆疊
if attempt < MAX_RETRIES - 1:
await asyncio.sleep(RETRY_DELAY) # 延遲後重試
else:
return f"Exception: {e}"
async def handle_input(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 run_sync(func, *args):
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
result = loop.run_until_complete(func(*args))
loop.close()
return result
# 定義 Gradio 介面
user_input = gr.Textbox(label='請輸入您想查詢的關鍵公司名稱')
examples = [
["加密貨幣"],
# ["國泰金控"],
["中華電信"],
# ["台灣大哥大"],
["台積電"],
# ["BlockTempo"]
]
TITLE = """Social Media Trends 💬 分析社群相關資訊，並判斷其正、負、中立等評價及趨勢
"""
SUBTITLE = """"""
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的微調和標註 | 基於機器閱讀理解和指令微調的統一信息抽取框架之診斷書醫囑資訊擷取分析
"""
# 使用 Gradio Blocks 設定頁面內容
with gr.Blocks() as iface:
gr.HTML(TITLE)
gr.HTML(SUBTITLE)
gr.HTML(LINKS)
gr.Interface(
fn=lambda x: run_sync(handle_input, x),
inputs=user_input,
outputs="text",
examples=examples,
allow_flagging="never"
)
iface.launch()