大模型测试前端：基于gradio，单文件，点赞点踩记录到SQLite。

gradio-chat.py

import gradio as gr
import time
import json
import os
import requests
import sqlite3
from threading import Lock
from typing import List, Dict, Any, Optional
from openai import OpenAI

# 模型选项配置：URL、model名称、API KEY
MODEL_OPTIONS = {
    "DeepSeek V3": ("https://api.deepseek.com/", "deepseek-chat", os.environ['DS_API_KEY']),
    "DeepSeek R1": ("https://api.deepseek.com/", "deepseek-reasoner", os.environ['DS_API_KEY']), 
}

# 预设提示选项
PROMPT_TEMPLATES = {
    "复读机Prompt": "你是一个复读机。",
    "default": "You are a helpful assistant."
}

# 服务参数
SERVER_NAME = '0.0.0.0'
SERVER_PORT = 7860
AUTH = [
    ('xx', 'xx123')
]

# SQLite数据库配置
DB_FILE = "feedback.db"
db_lock = Lock()

# 初始化数据库
def init_db():
    with sqlite3.connect(DB_FILE) as conn:
        conn.execute("""
        CREATE TABLE IF NOT EXISTS feedback (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            username TEXT NOT NULL,
            feedback_type TEXT NOT NULL,
            model_name TEXT NOT NULL,
            system_prompt TEXT NOT NULL,
            timestamp REAL NOT NULL,
            conversation_history TEXT NOT NULL
        )
        """)
        # 创建索引
        conn.execute("CREATE INDEX IF NOT EXISTS idx_username ON feedback(username)")
        conn.execute("CREATE INDEX IF NOT EXISTS idx_timestamp ON feedback(timestamp)")

# 初始化数据库
init_db()

def get_dict_caller_with_default(d: dict, default: str | None = None):
    def caller(key):
        return d.get(key, d[default] if default else None)
    return caller

# 存储反馈的函数
def save_feedback(feedback_type, message_history, model_name, system_prompt, request: gr.Request):
    """保存反馈数据到SQLite数据库
    
    Args:
        feedback_type: "like" 或 "dislike"
        message_history: 对话历史
        model_name: 模型名称
        system_prompt: 使用的系统提示词
        request: Gradio请求对象
    """
    username = request.username if request else "anonymous"
    
    feedback = {
        "username": username,
        "feedback_type": feedback_type,
        "model": model_name,
        "system_prompt": system_prompt,
        "timestamp": time.time(),
        "conversation_history": json.dumps(message_history, ensure_ascii=False)
    }
    
    try:
        with db_lock:
            with sqlite3.connect(DB_FILE) as conn:
                # 设置WAL模式提高并发性能
                conn.execute("PRAGMA journal_mode=WAL")
                conn.execute("PRAGMA busy_timeout=3000")
                
                cursor = conn.cursor()
                cursor.execute("""
                INSERT INTO feedback 
                (username, feedback_type, model_name, system_prompt, timestamp, conversation_history)
                VALUES (?, ?, ?, ?, ?, ?)
                """, (
                    feedback["username"],
                    feedback["feedback_type"],
                    feedback["model"],
                    feedback["system_prompt"],
                    feedback["timestamp"],
                    feedback["conversation_history"]
                ))
                conn.commit()
        
        return f"已记录{feedback_type}反馈，用户: {username}, 模型: {model_name}, 提示词: {system_prompt[:10]}..."
    
    except Exception as e:
        return f"保存反馈失败: {str(e)}"

def goodcase(message_history, model_name, system_prompt, request: gr.Request):
    return save_feedback('good', message_history, model_name, system_prompt, request)

def badcase(message_history, model_name, system_prompt, request: gr.Request):
    return save_feedback('bad', message_history, model_name, system_prompt, request)

# 发送请求到模型API (保持不变)
def query_model(message: str, history: list[gr.ChatMessage], model_name: str, system_prompt: str):
    """向指定模型的API发送请求
    
    Args:
        message: 用户输入消息
        history: 对话历史
        model_name: 选择的模型名称
        system_prompt: 当前提示词
    
    Returns:
        模型的响应和会话元数据
    """
    if model_name not in MODEL_OPTIONS:
        return "错误: 未知的模型选项"
    
    # 获取模型API URL
    api_url, model, api_key = MODEL_OPTIONS[model_name]

    def _serialize_history(history):
        result = []
        for m in history:
            if isinstance(m, dict):
                result.append(m)
            elif hasattr(m, "role") and hasattr(m, "content"):
                result.append({"role": m.role, "content": m.content})
            else:
                raise TypeError(f"无法序列化消息类型：{type(m)}")
        return result
    
    try:
        # 准备请求数据
        messages = [gr.ChatMessage(system_prompt, role='system')] + history + [gr.ChatMessage(message, role='user')]
        client = OpenAI(base_url=api_url, api_key=api_key)
        
        # 发送请求到模型API
        # 注意: 实际项目中可能需要添加错误处理、身份验证等
        response = client.chat.completions.create(
            model=model,
            messages=_serialize_history(messages),
            temperature=.7,
            max_tokens=256,
            timeout=30,
        )
        model_response = response.choices[0].message.content
        return model_response
        
    except Exception as e:
        return f"请求错误: {str(e)}"

# 主界面 (保持不变)
with gr.Blocks(title="高级聊天界面") as demo:
    gr.Markdown("# 智能对话系统\n选择模型和提示模板进行对话")
    
    with gr.Row():
        with gr.Column(scale=3):
            # 聊天界面
            chatbot = gr.Chatbot(height=500, label="对话记录", type='messages')
            
            # 消息输入区
            with gr.Row():
                msg = gr.Textbox(
                    label="输入消息", 
                    placeholder="请输入您的问题...",
                    lines=2,
                    show_label=False
                )
            
            # 控制按钮区
            with gr.Row():
                submit_btn = gr.Button("发送", variant="primary")
                retry_btn = gr.Button("重试")
                undo_btn = gr.Button("撤销")
                clear_btn = gr.Button("清空对话")
            
            # 反馈按钮区
            with gr.Row():
                like_btn = gr.Button("👍 好")
                dislike_btn = gr.Button("👎 差")
            
        with gr.Column(scale=1):
            # 模型选择
            model_dropdown = gr.Dropdown(
                choices=list(MODEL_OPTIONS.keys()),
                value=list(MODEL_OPTIONS.keys())[0],
                label="选择模型"
            )
            
            # 提示模板选择
            prompt_dropdown = gr.Dropdown(
                choices=list(PROMPT_TEMPLATES.keys()),
                value=list(PROMPT_TEMPLATES.keys())[0],
                label="选择提示模板"
            )

            system_prompt = gr.Textbox(label="System Prompt", value=PROMPT_TEMPLATES[prompt_dropdown.value])
            
            # 当前会话信息
            session_info = gr.Textbox(label="当前对话信息", interactive=False)

            # 反馈状态显示
            feedback_status = gr.Textbox(label="操作执行反馈", interactive=False, visible=True)
        
    
    # 添加状态管理
    message_metadata_state = gr.State([])  # 每条消息的元数据
    
    # 处理发送消息 (保持不变)
    def handle_message(message: str, history: list[gr.ChatMessage], metadata_history: list[dict], model_name: str, system_prompt: str):
        if not message.strip():
            return "", history, metadata_history, "", ""
        
        # 保存对话历史
        updated_history = history.copy() if history else []
        updated_metadata = metadata_history.copy() if metadata_history else []
        
        # 记录会话元数据
        metadata = {
            "model": model_name,
            "system_prompt": system_prompt,
            "timestamp": time.time(),
        }
        
        # 获取模型响应和元数据
        response_text = query_model(message, history, model_name, system_prompt)

        metadata['completion_timestamp'] = time.time()

        # 更新对话历史
        updated_history.append(gr.ChatMessage(message, role='user'))
        updated_history.append(gr.ChatMessage(response_text, role='assistant'))
        updated_metadata.append(metadata)
        
        # 更新会话信息
        session_status = f"使用模型: {model_name} | 提示模板: {system_prompt[:10]}... | 消息总数: {len(updated_history)}"
        # 更新操作反馈
        func_status = f"模型输出花费 {metadata['completion_timestamp'] - metadata['timestamp']} 秒"
        
        return "", updated_history, updated_metadata, session_status, func_status
    
    # 重试最后一次对话 - 使用当前选择的模型和prompt (保持不变)
    def retry_last_message(history: list[gr.ChatMessage], metadata_history: list[dict], current_model: str, current_prompt: str):
        if not history or len(history) <= 0:
            return history, metadata_history, "没有可重试的消息"
        
        # 获取最后一条用户消息
        last_user_message: str = history[-2]['content']

        # 移除最后一次对话和元数据
        updated_history = history[:-2]
        updated_metadata = metadata_history[:-1] if len(metadata_history) > 0 else []

        # 使用当前模型和提示模板重新发送请求
        res  = handle_message(last_user_message, updated_history, updated_metadata, current_model, current_prompt)
        updated_history, updated_metadata = res[1:3]
        metadata = updated_metadata[-1]
        func_status = f"已使用当前模型({current_model})和提示模板({current_prompt[:10]}...)重试最后一条消息，花费 {metadata['completion_timestamp'] - metadata['timestamp']} 秒"
        
        return updated_history, updated_metadata, func_status
    
    # 撤销最后一次对话 (保持不变)
    def undo_last_message(history, metadata_history):
        if not history or len(history) <= 0:
            return history, metadata_history, "没有可撤销的消息"
        
        # 移除最后一次对话和元数据
        updated_history = history[:-2]
        updated_metadata = metadata_history[:-1] if len(metadata_history) > 0 else []
        
        return updated_history, updated_metadata, "已撤销最后一对QA"
    
    # 绑定消息发送事件 (保持不变)
    submit_click_event = submit_btn.click(
        handle_message, 
        [msg, chatbot, message_metadata_state, model_dropdown, system_prompt], 
        [msg, chatbot, message_metadata_state, session_info, feedback_status]
    )

    prompt_dropdown.change(get_dict_caller_with_default(PROMPT_TEMPLATES, 'default'), [prompt_dropdown], system_prompt)
    
    # 绑定撤销按钮事件 (保持不变)
    undo_btn.click(
        undo_last_message,
        [chatbot, message_metadata_state],
        [chatbot, message_metadata_state, feedback_status]
    )
    
    # 绑定重试按钮事件 - 使用当前模型和提示模板 (保持不变)
    retry_btn.click(
        retry_last_message,
        [chatbot, message_metadata_state, model_dropdown, system_prompt],
        [chatbot, message_metadata_state, feedback_status]
    )
    
    # 绑定清空按钮事件 (保持不变)
    def clear_chat():
        return [], [], "对话已清空"
    
    clear_btn.click(
        clear_chat,
        [],
        [chatbot, message_metadata_state, feedback_status]
    )
    
    # 绑定点赞和点踩按钮事件 - 更新为传递request对象
    like_btn.click(
        goodcase,
        [chatbot, model_dropdown, system_prompt], 
        [feedback_status]
    )
    
    dislike_btn.click(
        badcase,
        [chatbot, model_dropdown, system_prompt], 
        [feedback_status]
    )

if __name__ == "__main__":
    demo.launch(server_name=SERVER_NAME, server_port=SERVER_PORT, auth=AUTH,)