Spaces:
Sleeping
Sleeping
File size: 7,934 Bytes
dd10f90 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 |
import gradio as gr
import json
import time
from typing import Dict, Any, List, Optional
class EvaluationUI:
"""处理评估服务的Gradio界面"""
def __init__(self, queue_manager):
"""初始化UI
Args:
queue_manager: 队列管理器实例
"""
self.queue_manager = queue_manager
self.app = None
def submit_evaluation(self, input_data):
"""提交评估请求
Args:
input_data: 输入JSON数据
Returns:
dict: 包含请求ID和状态的字典
"""
try:
# 解析JSON
if isinstance(input_data, str):
input_data = json.loads(input_data)
# 提交到队列
request_id = self.queue_manager.enqueue(input_data)
# 返回请求ID
return {"request_id": request_id, "status": "queued"}
except Exception as e:
return {"status": "error", "message": str(e)}
def get_evaluation_result(self, request_id):
"""获取评估结果
Args:
request_id: 请求ID
Returns:
dict: 评估结果
"""
return self.queue_manager.get_result(request_id)
def get_queue_status(self):
"""获取队列状态
Returns:
dict: 队列状态信息
"""
return self.queue_manager.get_queue_status()
def render_queue_status(self):
"""渲染队列状态HTML
Returns:
str: 队列状态的HTML表示
"""
status = self.queue_manager.get_queue_status()
stats = status["stats"]
queue_items = status["queue_items"]
html = f"""
<div style="padding: 10px; background-color: #f5f5f5; border-radius: 5px; margin-bottom: 20px;">
<h3>队列状态</h3>
<div style="display: flex; justify-content: space-between; margin-bottom: 15px;">
<div style="text-align: center; padding: 5px; background-color: #e0f7fa; border-radius: 5px; flex: 1; margin-right: 5px;">
<div style="font-weight: bold; font-size: 24px;">{stats['queued']}</div>
<div>等待中</div>
</div>
<div style="text-align: center; padding: 5px; background-color: #fff9c4; border-radius: 5px; flex: 1; margin-right: 5px;">
<div style="font-weight: bold; font-size: 24px;">{stats['processing']}</div>
<div>处理中</div>
</div>
<div style="text-align: center; padding: 5px; background-color: #c8e6c9; border-radius: 5px; flex: 1; margin-right: 5px;">
<div style="font-weight: bold; font-size: 24px;">{stats['completed']}</div>
<div>已完成</div>
</div>
<div style="text-align: center; padding: 5px; background-color: #ffcdd2; border-radius: 5px; flex: 1;">
<div style="font-weight: bold; font-size: 24px;">{stats['error']}</div>
<div>错误</div>
</div>
</div>
"""
if queue_items:
html += """
<h4>当前队列</h4>
<table style="width: 100%; border-collapse: collapse;">
<tr style="background-color: #e0e0e0;">
<th style="padding: 8px; text-align: left; border: 1px solid #ddd;">ID</th>
<th style="padding: 8px; text-align: left; border: 1px solid #ddd;">状态</th>
<th style="padding: 8px; text-align: left; border: 1px solid #ddd;">等待时间</th>
</tr>
"""
current_time = time.time()
for item in queue_items:
status_color = "#fff9c4" if item["status"] == "processing" else "#e0f7fa"
wait_time = current_time - item["created_at"]
wait_time_str = f"{int(wait_time // 60)}分{int(wait_time % 60)}秒"
html += f"""
<tr>
<td style="padding: 8px; text-align: left; border: 1px solid #ddd;">{item['id'][:8]}...</td>
<td style="padding: 8px; text-align: left; border: 1px solid #ddd; background-color: {status_color};">
{"处理中" if item["status"] == "processing" else "等待中"}
</td>
<td style="padding: 8px; text-align: left; border: 1px solid #ddd;">{wait_time_str}</td>
</tr>
"""
html += "</table>"
else:
html += "<p>队列为空</p>"
html += "</div>"
return html
def create_evaluation_interface(self):
"""创建评估接口
Returns:
gr.Interface: Gradio接口
"""
# 创建用于直接提交JSON的接口
with gr.Blocks(title="代码评估服务") as app:
with gr.Row():
with gr.Column(scale=3):
gr.Markdown("# 代码评估服务")
gr.Markdown("支持多种编程语言的代码评估服务，使用消息队列处理并发请求。")
# 输入JSON
json_input = gr.JSON(
label="输入JSON",
value=[{"language": "python", "processed_completions": ["def add(a, b):\n return a + b"], "prompt": "", "tests": "assert add(1, 2) == 3"}]
)
# 提交按钮
submit_btn = gr.Button("提交评估")
# 结果输出
result_output = gr.JSON(label="评估结果")
with gr.Column(scale=2):
# 队列状态
queue_status = gr.HTML(label="队列状态")
refresh_btn = gr.Button("刷新队列状态")
# 获取特定结果
with gr.Row():
request_id_input = gr.Textbox(label="请求ID")
get_result_btn = gr.Button("获取结果")
# 特定结果输出
specific_result = gr.JSON(label="请求结果")
# 事件处理
submit_btn.click(
fn=self.submit_evaluation,
inputs=[json_input],
outputs=[result_output]
)
refresh_btn.click(
fn=self.render_queue_status,
inputs=[],
outputs=[queue_status]
)
get_result_btn.click(
fn=self.get_evaluation_result,
inputs=[request_id_input],
outputs=[specific_result]
)
# 定时更新队列状态
app.load(fn=self.render_queue_status, inputs=None, outputs=queue_status, every=2)
self.app = app
return app
def create_api_interface(self):
"""创建API接口
Returns:
gr.Interface: Gradio接口
"""
return gr.Interface(
fn=self.submit_evaluation,
inputs=gr.JSON(),
outputs=gr.JSON(),
title="代码评估API",
description="API接口用于提交代码评估请求"
)
def get_app(self):
"""获取Gradio应用
Returns:
gr.Blocks: Gradio应用
"""
if self.app is None:
self.create_evaluation_interface()
return self.app |