import gradio as gr from datasets import load_dataset import numpy as np from sentence_transformers import SentenceTransformer import time from datetime import datetime import json import os # =================================================================== # 1. НАСТРОЙКИ И ЗАГРУЗКА ДАННЫХ # =================================================================== MODEL_NAME = "BAAI/bge-large-en-v1.5" DATASET_NAME = "wikimedia/wikipedia" LANGUAGE = "20231101.en" ARTICLE_LIMIT = 1000 ADMIN_USER = "admin" ADMIN_PASS = "hfpassword21" # Файл для хранения логов LOG_FILE = "query_logs.json" # Загружаем логи или создаём новый файл if os.path.exists(LOG_FILE): with open(LOG_FILE, "r") as f: query_logs = json.load(f) else: query_logs = [] @gr.cache_resource def load_wikipedia(): """Загружает датасет Википедии""" dataset = load_dataset(DATASET_NAME, LANGUAGE, split="train", streaming=True) articles = [] for i, row in enumerate(dataset): if i >= ARTICLE_LIMIT: break articles.append({ "id": row.get("id", i), "title": row.get("title", "Без названия"), "text": row.get("text", "")[:3000], "url": row.get("url", "") }) return articles @gr.cache_resource def load_embedder(): """Загружает модель для эмбеддингов""" return SentenceTransformer(MODEL_NAME) articles = load_wikipedia() embedder = load_embedder() # Создаём эмбеддинги статей заранее texts = [f"{a['title']}\n\n{a['text']}" for a in articles] embeddings = embedder.encode(texts, normalize_embeddings=True) # =================================================================== # 2. ОСНОВНЫЕ ФУНКЦИИ # =================================================================== def search_wikipedia(query, history): """Ищет ответ на вопрос в Википедии и логирует запрос""" if not query: return "❓ Введите вопрос", history start_time = time.time() # Поиск query_vector = embedder.encode([query], normalize_embeddings=True)[0] scores = embeddings @ query_vector top_indices = np.argsort(-scores)[:3] results = [] for idx in top_indices: score = float(scores[int(idx)]) if score > 0.3: article = articles[int(idx)] results.append( f"### 📄 {article['title']} (сходство: {score:.2f})\n" f"{article['text'][:1000]}...\n" f"🔗 [Читать на Википедии]({article['url']})\n" ) response = "\n---\n".join(results) if results else "😕 Не нашёл подходящей статьи. Попробуй уточнить вопрос." # Логируем запрос log_entry = { "timestamp": datetime.now().isoformat(), "query": query, "results_count": len(results), "response_time": round(time.time() - start_time, 2) } query_logs.append(log_entry) # Сохраняем логи в файл with open(LOG_FILE, "w") as f: json.dump(query_logs[-100:], f) # Храним последние 100 запросов # Обновляем историю для админ-панели history = query_logs[-20:] # Последние 20 запросов return response, history def login(username, password): if username == ADMIN_USER and password == ADMIN_PASS: return gr.update(visible=True), "✅ Доступ разрешён", gr.update(visible=True) return gr.update(visible=False), "❌ Неверный логин или пароль", gr.update(visible=False) def get_admin_stats(): """Собирает статистику для админ-панели""" total_queries = len(query_logs) if total_queries > 0: avg_time = sum(q["response_time"] for q in query_logs) / total_queries popular_queries = sorted(query_logs, key=lambda x: x["results_count"], reverse=True)[:5] else: avg_time = 0 popular_queries = [] # Формируем отчёт stats = f""" ## 📊 Статистика - **Всего запросов:** {total_queries} - **Среднее время ответа:** {avg_time:.2f} сек. - **Загружено статей:** {len(articles)} - **Модель:** {MODEL_NAME} ## 🔥 Популярные запросы """ for i, q in enumerate(popular_queries, 1): stats += f"{i}. {q['query']} (найдено: {q['results_count']})\n" return stats def get_recent_logs(): """Показывает последние 10 запросов""" if not query_logs: return "📭 Логов пока нет" logs = "## 📋 Последние запросы\n\n" for q in query_logs[-10:]: logs += f"**{q['timestamp']}**\n" logs += f"Вопрос: {q['query']}\n" logs += f"Результатов: {q['results_count']}, Время: {q['response_time']}с\n\n" return logs def clear_logs(): """Очищает логи""" global query_logs query_logs = [] with open(LOG_FILE, "w") as f: json.dump(query_logs, f) return "✅ Логи очищены", "Логов пока нет" # =================================================================== # 3. ИНТЕРФЕЙС GRADIO # =================================================================== with gr.Blocks(theme=gr.themes.Soft(), title="Wikipedia Assistant") as demo: gr.Markdown("# 🌍 Wikipedia Assistant") gr.Markdown("Задай вопрос — я найду ответ в Википедии!") # Состояние для истории запросов history_state = gr.State([]) with gr.Row(): with gr.Column(scale=4): query_input = gr.Textbox( label="🔍 Что хочешь узнать?", placeholder="Например: Как измеряют расстояние до галактик?", lines=2 ) with gr.Row(): search_btn = gr.Button("🔎 Найти", variant="primary") clear_btn = gr.Button("🗑️ Очистить", variant="secondary") output = gr.Markdown("💡 Напиши вопрос и нажми 'Найти'") with gr.Column(scale=1): gr.Markdown("### 👤 Вход в админку") username = gr.Textbox(label="Логин", placeholder="admin") password = gr.Textbox(label="Пароль", type="password", placeholder="hfpassword21") login_btn = gr.Button("🔑 Войти", variant="primary") status = gr.Textbox(label="Статус", interactive=False) # Админ-панель (скрыта по умолчанию) with gr.Tab("👑 Админ-панель", visible=False) as admin_tab: with gr.Row(): with gr.Column(scale=2): stats_output = gr.Markdown("## 📊 Загрузка статистики...") refresh_stats_btn = gr.Button("🔄 Обновить статистику") with gr.Column(scale=3): logs_output = gr.Markdown("📭 Загрузка логов...") with gr.Row(): refresh_logs_btn = gr.Button("🔄 Обновить логи") clear_logs_btn = gr.Button("🗑️ Очистить логи", variant="stop") # =================================================================== # 4. ОБРАБОТЧИКИ СОБЫТИЙ # =================================================================== # Поиск search_btn.click( search_wikipedia, inputs=[query_input, history_state], outputs=[output, history_state] ) query_input.submit( search_wikipedia, inputs=[query_input, history_state], outputs=[output, history_state] ) # Очистка clear_btn.click(lambda: ("", "💡 Напиши вопрос и нажми 'Найти'"), None, [query_input, output]) # Вход в админку login_btn.click( login, [username, password], [admin_tab, status, gr.update(visible=True)] ) # Обновление статистики refresh_stats_btn.click(get_admin_stats, None, stats_output) # Обновление логов refresh_logs_btn.click(get_recent_logs, None, logs_output) # Очистка логов clear_logs_btn.click(clear_logs, None, [logs_output, logs_output]) # Автообновление при входе admin_tab.select( lambda: (get_admin_stats(), get_recent_logs()), None, [stats_output, logs_output] ) # =================================================================== # 5. ЗАПУСК # =================================================================== if __name__ == "__main__": demo.launch(share=True)