import gradio as gr
import threading
import time
import os
import traceback
from services.audio_service import AudioService
from services.chat_service import ChatService
from services.image_service import ImageService
from services.streaming_voice_service import StreamingVoiceService
from services.openai_realtime_service import HybridStreamingService
from services.stream_object_detection_service import StreamObjectDetection
from services.voice_coding_service import VoiceCodingService
from services.sambanova_voice_service import SambanovaVoiceService
from services.gemini_realtime_service import GeminiRealtimeService
from core.rag_system import EnhancedRAGSystem
from core.tts_service import EnhancedTTSService
from core.wikipedia_processor import WikipediaProcessor
from ui.components import create_audio_components, create_chat_components,create_streaming_voice_components
def create_all_tabs(audio_service: AudioService, chat_service: ChatService,
image_service: ImageService, rag_system: EnhancedRAGSystem,
tts_service: EnhancedTTSService, wikipedia_processor: WikipediaProcessor,
streaming_voice_service: StreamingVoiceService,
hybrid_service: HybridStreamingService,
voice_coding_service: VoiceCodingService,
sambanova_voice_service: SambanovaVoiceService
):
with gr.Tab("🎙️ Streaming Voice "):
create_streaming_voice_tab(streaming_voice_service)
with gr.Tab("OpenAI Realtime"):
create_openai_realtime_tab(hybrid_service)
with gr.Tab("GemeniAI RealTime"):
create_gemini_realtime_tab() # Đã sửa - không cần tham số
with gr.Tab("SambonovaAI Realtime"):
create_sambanova_voice_tab() # Thêm tham số nếu cần
with gr.Tab("Generation Code"):
create_voice_coding_tab(voice_coding_service)
with gr.Tab("🎙️ Audio"):
create_audio_tab(audio_service)
with gr.Tab("💬 Chat"):
create_chat_tab(chat_service)
# with gr.Tab("🖼️ Image"):
# create_image_tab(image_service)
with gr.Tab("📚 RAG Wikipedia"):
create_rag_tab(rag_system, wikipedia_processor)
with gr.Tab("🔊 Text-to-Speech"):
create_tts_tab(tts_service)
with gr.Tab("🌐 Language Info"):
create_language_info_tab(rag_system.multilingual_manager)
with gr.Tab("Stream Object Detection"):
create_streaming_object_detection()
def create_gemini_realtime_tab():
"""Tạo tab cho Gemini Realtime API với Audio Streaming"""
with gr.Blocks() as gemini_tab:
gr.Markdown("""
# 🎯 Gemini Audio Streaming
**Trò chuyện thời gian thực bằng giọng nói với Google Gemini**
""")
with gr.Row():
with gr.Column(scale=1):
# Connection controls
with gr.Group():
gr.Markdown("### 🔗 Kết nối")
api_key = gr.Textbox(
label="Gemini API Key",
type="password",
placeholder="Nhập API key của bạn...",
value=os.getenv("GEMINI_API_KEY", ""),
info="Lấy từ https://aistudio.google.com/"
)
voice_select = gr.Dropdown(
choices=["Puck", "Charon", "Kore", "Fenrir", "Aoede"],
value="Puck",
label="Giọng nói AI",
info="Chọn giọng nói cho Gemini"
)
with gr.Row():
connect_btn = gr.Button("🔗 Kết nối Audio", variant="primary")
disconnect_btn = gr.Button("🔌 Ngắt kết nối", variant="secondary")
# Status panel
with gr.Group():
gr.Markdown("### 📊 Trạng thái")
status_display = gr.Textbox(
label="Trạng thái",
value="Chưa kết nối",
interactive=False
)
connection_info = gr.Textbox(
label="Thông tin",
interactive=False,
lines=2
)
with gr.Column(scale=2):
# Audio Streaming Interface
with gr.Group():
gr.Markdown("### 🎤 Audio Streaming")
# Audio input for user
audio_input = gr.Audio(
label="🎤 Nhấn để nói chuyện với Gemini",
sources=["microphone"],
type="numpy",
interactive=True,
show_download_button=False
)
# Audio output for Gemini responses
audio_output = gr.Audio(
label="🔊 Gemini trả lời",
interactive=False,
autoplay=True
)
transcription_display = gr.Textbox(
label="💬 Nội dung hội thoại",
interactive=False,
lines=3,
placeholder="Nội dung cuộc trò chuyện sẽ hiển thị ở đây..."
)
# State management
connection_state = gr.State(value=False)
gemini_service_state = gr.State(value=None)
async def connect_gemini(api_key, voice_name):
"""Kết nối Gemini Audio Streaming"""
try:
if not api_key:
return False, "❌ Vui lòng nhập API Key", "Chưa kết nối", None
service = GeminiRealtimeService(api_key)
# Callback để xử lý real-time events
async def handle_gemini_callback(data):
if data['type'] == 'status':
gr.Info(data['message'])
elif data['type'] == 'text':
gr.Info(f"Gemini: {data['content']}")
elif data['type'] == 'error':
gr.Warning(data['message'])
success = await service.start_session(
voice_name=voice_name,
callback=handle_gemini_callback
)
if success:
info_msg = f"✅ Đã kết nối Audio Streaming\nGiọng: {voice_name}\nHãy sử dụng micro để trò chuyện"
return True, "✅ Đã kết nối Audio", info_msg, service
else:
return False, "❌ Không thể kết nối audio", "Lỗi kết nối", None
except Exception as e:
error_msg = f"❌ Lỗi kết nối: {str(e)}"
return False, error_msg, f"Lỗi: {str(e)}", None
async def disconnect_gemini(service):
"""Ngắt kết nối"""
if service:
await service.close()
return False, "🔌 Đã ngắt kết nối", "Đã ngắt kết nối audio streaming", None
async def process_audio_input(audio_data, sample_rate, service):
"""Xử lý audio input từ user và trả lời bằng audio"""
if not service or not service.is_active:
return None, "❌ Chưa kết nối. Vui lòng kết nối audio trước.", "Chưa kết nối"
if audio_data is None:
return None, "⚠️ Không có audio input", "Không có audio"
try:
# Gửi audio đến Gemini
success = await service.send_audio_chunk(audio_data, sample_rate)
if not success:
return None, "❌ Lỗi gửi audio đến Gemini", "Lỗi gửi audio"
# Chờ và nhận audio response từ Gemini
audio_response = None
max_attempts = 50 # Chờ tối đa 5 giây
for attempt in range(max_attempts):
audio_response = await service.receive_audio()
if audio_response is not None:
break
await asyncio.sleep(0.1) # Chờ 100ms giữa các lần thử
if audio_response:
resp_sample_rate, resp_audio_data = audio_response
# Lưu audio response vào file tạm
with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as f:
import scipy.io.wavfile as wavfile
wavfile.write(f.name, resp_sample_rate, resp_audio_data)
audio_path = f.name
info_msg = f"✅ Đã nhận phản hồi audio từ Gemini ({(len(resp_audio_data) / resp_sample_rate):.1f}s)"
return audio_path, info_msg, "Thành công"
else:
return None, "⏳ Không nhận được phản hồi audio từ Gemini", "Timeout"
except Exception as e:
error_msg = f"❌ Lỗi xử lý audio: {str(e)}"
return None, error_msg, f"Lỗi: {str(e)}"
# Event handlers
connect_btn.click(
connect_gemini,
inputs=[api_key, voice_select],
outputs=[connection_state, status_display, connection_info, gemini_service_state]
)
disconnect_btn.click(
disconnect_gemini,
inputs=[gemini_service_state],
outputs=[connection_state, status_display, connection_info, gemini_service_state]
)
# Xử lý audio input
audio_input.stop_recording(
process_audio_input,
inputs=[audio_input, audio_input, gemini_service_state],
outputs=[audio_output, connection_info, transcription_display]
)
# Hướng dẫn sử dụng
with gr.Accordion("📖 Hướng dẫn sử dụng Audio Streaming", open=True):
gr.Markdown("""
### 🎯 Cách sử dụng Audio Streaming:
1. **Kết nối**:
- Nhập API Key Gemini
- Chọn giọng nói
- Nhấn **"Kết nối Audio"**
2. **Trò chuyện bằng giọng nói**:
- Nhấn nút **Micro** để bắt đầu ghi âm
- Nói câu hỏi của bạn
- Nhấn **Dừng** để kết thúc ghi âm
- Gemini sẽ trả lời bằng giọng nói ngay lập tức
### 🔊 Tính năng:
- 🎙️ Real-time voice recognition
- 🔊 Real-time audio response
- ⚡ Ultra low latency
- 🎯 Multiple voice options
### 💡 Mẹo sử dụng:
- Sử dụng headset để chất lượng tốt hơn
- Nói rõ ràng, không nói quá nhanh
- Môi trường yên tĩnh cho kết quả tốt nhất
- Mỗi lần ghi âm nên ngắn hơn 30 giây
### 🔧 Lưu ý kỹ thuật:
- Cần API Key Gemini có quyền Realtime API
- Audio được stream real-time đến Gemini
- Phản hồi audio được stream về và phát tự động
""")
return gemini_tab
# FastAPI routes cho Gemini (tùy chọn)
def setup_gemini_routes(app):
"""Thiết lập routes FastAPI cho Gemini"""
@app.get("/gemini/status")
async def get_gemini_status():
return {"status": "active", "service": "gemini_realtime"}
@app.post("/gemini/connect")
async def connect_gemini():
return {"message": "Gemini connection endpoint"}
def create_sambanova_voice_tab():
"""Tạo tab Sambanova AI với Voice Input/Output"""
# Khởi tạo services
try:
tts_service = EnhancedTTSService()
sambanova_service = SambanovaVoiceService(tts_service=tts_service)
print("✅ Tất cả services đã được khởi tạo")
except Exception as e:
print(f"❌ Lỗi khởi tạo services: {e}")
# Fallback: chỉ khởi tạo Sambanova service không có TTS
sambanova_service = SambanovaVoiceService()
tts_service = None
with gr.Blocks() as sambanova_tab:
gr.Markdown("## 🤖 Sambanova AI - Voice & Text")
gr.Markdown("Trò chuyện với AI - Hỗ trợ voice input/output")
# State variables
chatbot = gr.Chatbot(
type="messages",
value=[],
label="💬 Hội thoại",
height=400
)
conversation_state = gr.State(value=[])
with gr.Row():
with gr.Column(scale=1):
# Model selection
model_dropdown = gr.Dropdown(
choices=sambanova_service.get_available_models(),
value="Meta-Llama-3.1-8B-Instruct",
label="Chọn Model"
)
# Language selection for TTS (chỉ hiển thị nếu có TTS)
language_dropdown = gr.Dropdown(
choices=['vi', 'en', 'ja', 'ko', 'zh', 'fr', 'es', 'de'],
value='vi',
label="Ngôn ngữ TTS",
visible=tts_service is not None
)
# Text input
text_input = gr.Textbox(
label="Tin nhắn của bạn",
placeholder="Nhập tin nhắn hoặc sử dụng voice...",
lines=3
)
# Voice input
with gr.Group():
gr.Markdown("**🎤 Voice Input**")
audio_input = gr.Audio(
sources=["microphone"],
type="numpy",
label="Nói tin nhắn của bạn",
show_download_button=False
)
# Parameters
with gr.Row():
temperature = gr.Slider(0, 1, value=0.1, label="Temperature")
top_p = gr.Slider(0, 1, value=0.1, label="Top-P")
# Voice output toggle (chỉ hiển thị nếu có TTS)
voice_output_toggle = gr.Checkbox(
label="🔊 Bật Voice Output",
value=True,
visible=tts_service is not None
)
# Buttons
with gr.Row():
send_text_btn = gr.Button("🚀 Gửi Text", variant="primary")
send_voice_btn = gr.Button("🎤 Gửi Voice", variant="primary")
clear_btn = gr.Button("🗑️ Xóa", variant="secondary")
# Status
status = gr.Textbox(
label="Trạng thái",
value="✅ Sẵn sàng",
interactive=False
)
with gr.Column(scale=1):
# Audio output (chỉ hiển thị nếu có TTS)
if tts_service is not None:
gr.Markdown("### 🔊 Voice Output")
audio_output = gr.Audio(
label="Giọng nói AI",
autoplay=False,
visible=True
)
else:
audio_output = gr.Audio(visible=False)
gr.Markdown("### ℹ️ Thông tin")
gr.Markdown("""
**Voice output tạm thời không khả dụng**
- Vẫn có thể sử dụng voice input
- Vẫn có thể chat bằng text
""")
# Response display
response_display = gr.Textbox(
label="Phản hồi từ AI",
lines=6,
interactive=False
)
# Event handlers
def process_text_message(text, history, state, model, language, temp, top_p_val, voice_enabled):
"""Xử lý tin nhắn text"""
if not text or not text.strip():
return history, state, "❌ Vui lòng nhập tin nhắn", "", gr.update(visible=False)
try:
# Thêm user message
user_msg = {"role": "user", "content": text}
new_history = history + [user_msg]
new_state = state + [user_msg]
# Hiển thị trạng thái đang xử lý
yield new_history, new_state, "⏳ Đang xử lý...", "", gr.update(visible=False)
# Lấy response
ai_text = sambanova_service.generate_response(new_state, model, temp, top_p_val)
ai_msg = {"role": "assistant", "content": ai_text}
final_history = new_history + [ai_msg]
final_state = new_state + [ai_msg]
# Tạo voice output nếu enabled và có TTS
audio_update = gr.update(visible=False)
if voice_enabled and tts_service is not None:
audio_file = sambanova_service.text_to_speech(ai_text, language)
if audio_file:
audio_update = gr.update(value=audio_file, visible=True)
yield final_history, final_state, "✅ Hoàn thành", ai_text, audio_update
except Exception as e:
error_msg = f"❌ Lỗi: {str(e)}"
yield history, state, error_msg, "", gr.update(visible=False)
def process_voice_message(audio, history, state, model, language, temp, top_p_val, voice_enabled):
"""Xử lý tin nhắn voice"""
if audio is None:
return history, state, "❌ Vui lòng ghi âm tin nhắn", "", gr.update(visible=False)
try:
# Hiển thị trạng thái đang xử lý
yield history, state, "🎤 Đang chuyển speech thành text...", "", gr.update(visible=False)
# Chuyển speech thành text
text = sambanova_service.speech_to_text(audio)
if not text:
yield history, state, "❌ Không nhận dạng được giọng nói", "", gr.update(visible=False)
return
# Thêm user message (cả audio và text)
user_audio_msg = {"role": "user", "content": gr.Audio(audio)}
user_text_msg = {"role": "user", "content": text}
new_history = history + [user_audio_msg]
new_state = state + [user_text_msg]
yield new_history, new_state, "⏳ Đang xử lý voice message...", "", gr.update(visible=False)
# Lấy response
ai_text = sambanova_service.generate_response(new_state, model, temp, top_p_val)
ai_msg = {"role": "assistant", "content": ai_text}
final_history = new_history + [ai_msg]
final_state = new_state + [ai_msg]
# Tạo voice output
audio_update = gr.update(visible=False)
if voice_enabled and tts_service is not None:
audio_file = sambanova_service.text_to_speech(ai_text, language)
if audio_file:
audio_update = gr.update(value=audio_file, visible=True)
yield final_history, final_state, "✅ Voice message hoàn thành", ai_text, audio_update
except Exception as e:
error_msg = f"❌ Lỗi voice: {str(e)}"
yield history, state, error_msg, "", gr.update(visible=False)
def clear_conversation():
"""Xóa hội thoại"""
return [], [], "🔄 Đã xóa hội thoại", "", gr.update(visible=False)
# Kết nối events
send_text_btn.click(
fn=process_text_message,
inputs=[
text_input, chatbot, conversation_state,
model_dropdown, language_dropdown, temperature, top_p, voice_output_toggle
],
outputs=[chatbot, conversation_state, status, response_display, audio_output]
).then(
lambda: "", # Clear text input
outputs=[text_input]
)
send_voice_btn.click(
fn=process_voice_message,
inputs=[
audio_input, chatbot, conversation_state,
model_dropdown, language_dropdown, temperature, top_p, voice_output_toggle
],
outputs=[chatbot, conversation_state, status, response_display, audio_output]
).then(
lambda: None, # Clear audio input
outputs=[audio_input]
)
clear_btn.click(
fn=clear_conversation,
outputs=[chatbot, conversation_state, status, response_display, audio_output]
)
return sambanova_tab
def check_environment():
"""Kiểm tra môi trường trước khi chạy"""
print("🔍 Kiểm tra môi trường...")
# Kiểm tra API key
api_key = os.environ.get("SAMBANOVA_API_KEY")
if not api_key:
print("❌ SAMBANOVA_API_KEY không được tìm thấy")
print("💡 Hãy set environment variable: export SAMBANOVA_API_KEY=your_key")
return False
else:
print("✅ SAMBANOVA_API_KEY: OK")
# Kiểm tra dependencies
try:
import fastrtc
print("✅ FastRTC: OK")
except ImportError:
print("❌ FastRTC chưa được cài đặt")
return False
try:
import gtts
print("✅ gTTS: OK")
except ImportError:
print("❌ gTTS chưa được cài đặt")
try:
import edge_tts
print("✅ edge-tts: OK")
except ImportError:
print("❌ edge-tts chưa được cài đặt")
return True
def create_voice_coding_tab(voice_coding_service):
"""Tạo tab Voice Coding đơn giản - Text-based trước"""
with gr.Blocks() as coding_tab:
gr.Markdown("## 🦙 Voice Coding - Lập trình bằng AI")
gr.Markdown("Tạo và chỉnh sửa ứng dụng HTML single-file với AI Assistant")
# State variables
history = gr.State([{"role": "system", "content": "You are an AI coding assistant. Help create HTML applications."}])
current_code = gr.State("")
with gr.Row():
with gr.Column(scale=1):
gr.Markdown("### 🎯 Hướng dẫn sử dụng:")
gr.Markdown("""
**Nhập yêu cầu lập trình:**
- "Tạo trang web hello world"
- "Tạo calculator bằng HTML/CSS/JS"
- "Tạo đồng hồ digital"
- "Tạo form đăng ký với validation"
**Chức năng voice đang được phát triển**
""")
# Text input chính
text_input = gr.Textbox(
label="Yêu cầu lập trình",
placeholder="Ví dụ: Tạo trang web hello world với màu nền xanh và chữ màu trắng...",
lines=3
)
with gr.Row():
generate_btn = gr.Button("🚀 Generate Code", variant="primary", scale=2)
clear_btn = gr.Button("🗑️ Clear", variant="secondary", scale=1)
# Status
status_display = gr.Textbox(
label="Trạng thái",
value="Sẵn sàng...",
interactive=False
)
# FastRTC component (tạm ẩn)
with gr.Accordion("🎤 Voice Input (Experimental)", open=False):
gr.Markdown("Chức năng voice đang được phát triển...")
# fastrtc_audio = voice_coding_service.create_stream().ui()
with gr.Column(scale=2):
with gr.Tabs():
with gr.Tab("🎮 Sandbox Preview"):
sandbox = gr.HTML(
value=voice_coding_service.sandbox_html,
label="Live Preview"
)
with gr.Tab("📄 Code Editor"):
code_display = gr.Code(
language="html",
label="Generated HTML Code",
lines=25,
interactive=True,
value=""
)
with gr.Tab("💬 Chat History"):
chat_display = gr.Chatbot(
type="messages",
label="Lịch sử hội thoại",
height=400
)
# Event handlers
def generate_code(text, current_history, current_code_value):
"""Generate code từ text input"""
if not text.strip():
return current_history, current_code_value, current_history, "❌ Vui lòng nhập yêu cầu", voice_coding_service.sandbox_html
try:
# Tạo prompt
user_prompt = f"Create a single-file HTML application for: {text}. Current code: {current_code_value}. Respond with complete HTML code only."
# Update history
new_history = current_history + [
{"role": "user", "content": user_prompt}
]
# Generate code với Groq
response = voice_coding_service.groq_client.chat.completions.create(
model="llama-3.1-8b-instant",
messages=new_history,
temperature=0.7,
max_tokens=1024,
top_p=0.9,
stream=False,
)
output = response.choices[0].message.content
# Extract HTML
html_code = voice_coding_service.extract_html_content(output)
# Update history
new_history.append({"role": "assistant", "content": output})
# Update sandbox
sandbox_html = voice_coding_service.display_in_sandbox(html_code)
return new_history, html_code, new_history, "✅ Đã generate code thành công!", sandbox_html
except Exception as e:
error_msg = f"❌ Lỗi: {str(e)}"
return current_history, current_code_value, current_history, error_msg, voice_coding_service.sandbox_html
def update_sandbox(code):
"""Cập nhật sandbox khi code thay đổi"""
return voice_coding_service.display_in_sandbox(code)
def clear_all():
"""Xóa tất cả"""
empty_history = [{"role": "system", "content": "You are an AI coding assistant."}]
return empty_history, "", empty_history, "Đã xóa tất cả", voice_coding_service.sandbox_html
def clear_text():
"""Xóa text input"""
return ""
# Kết nối events
generate_btn.click(
generate_code,
inputs=[text_input, history, current_code],
outputs=[history, current_code, chat_display, status_display, sandbox]
).then(
clear_text,
outputs=[text_input]
)
code_display.change(
update_sandbox,
inputs=[code_display],
outputs=[sandbox]
)
clear_btn.click(
clear_all,
outputs=[history, current_code, chat_display, status_display, sandbox]
)
return coding_tab
def create_openai_realtime_tab(hybrid_service: HybridStreamingService):
"""Tạo tab cho OpenAI Realtime API"""
with gr.Blocks() as openai_tab:
gr.Markdown("## OpenAI Realtime API - Streaming Chất Lượng Cao")
with gr.Row():
with gr.Column(scale=1):
# Mode selection
mode_selector = gr.Radio(
choices=["local", "openai", "auto"],
value="auto",
label="Chế độ nhận diện",
info="Local: VOSK (nhanh), OpenAI: Chất lượng cao"
)
start_btn = gr.Button("🎙️ Bắt đầu Streaming", variant="primary")
stop_btn = gr.Button("🛑 Dừng", variant="secondary")
status_display = gr.Textbox(
label="Trạng thái",
value="Chưa kết nối",
interactive=False
)
# OpenAI specific controls
with gr.Accordion("⚙️ Cài đặt OpenAI", open=False):
api_key = gr.Textbox(
label="OpenAI API Key",
type="password",
placeholder="Nhập API key...",
info="Cần cho chế độ OpenAI Realtime"
)
language_select = gr.Dropdown(
choices=["vi", "en", "fr", "es", "de", "ja", "zh"],
value="vi",
label="Ngôn ngữ"
)
with gr.Column(scale=2):
chatbot = gr.Chatbot(
label="💬 Hội thoại",
type="messages",
height=400
)
transcription_display = gr.Textbox(
label="🎤 Bạn nói",
interactive=False,
lines=2
)
audio_output = gr.Audio(
label="🔊 Phản hồi AI",
interactive=False,
autoplay=True
)
# State management
connection_state = gr.State(value=False)
async def start_streaming(mode, api_key, language, history):
"""Bắt đầu streaming với mode đã chọn"""
try:
# Cập nhật service với API key
if api_key and not hybrid_service.openai_service:
hybrid_service.openai_service = OpenAIRealtimeService(api_key)
success = await hybrid_service.start_listening(
speech_callback=lambda x: None, # Will be handled via events
mode=mode
)
if success:
return True, f"✅ Đã kết nối - Chế độ: {mode}", history
else:
return False, "❌ Không thể kết nối", history
except Exception as e:
return False, f"❌ Lỗi: {str(e)}", history
def stop_streaming():
"""Dừng streaming"""
hybrid_service.stop_listening()
return False, "🛑 Đã dừng streaming", []
def update_chat(history, message, role="user"):
"""Cập nhật chat history"""
if role == "user":
history.append({"role": "user", "content": message})
else:
history.append({"role": "assistant", "content": message})
return history
# Event handlers
start_btn.click(
start_streaming,
inputs=[mode_selector, api_key, language_select, chatbot],
outputs=[connection_state, status_display, chatbot]
)
stop_btn.click(
stop_streaming,
outputs=[connection_state, status_display, chatbot]
)
# Real-time updates via JavaScript
openai_tab.load(
fn=None,
inputs=[],
outputs=[],
js="""
function setupEventSource() {
const eventSource = new EventSource('/outputs');
eventSource.onmessage = function(event) {
const data = JSON.parse(event.data);
// Handle real-time updates from OpenAI
console.log('OpenAI event:', data);
};
}
setupEventSource();
"""
)
return openai_tab
def create_streaming_object_detection():
with gr.Blocks() as object_detection_tab:
gr.HTML(
"""
🎥 Real-time Video Object Detection with RT-DETR
Upload a short video and watch detection stream in real-time!
"""
)
with gr.Row():
with gr.Column():
video = gr.Video(label="Video Input")
conf = gr.Slider(
minimum=0.0,
maximum=1.0,
value=0.3,
step=0.05,
label="Confidence Threshold"
)
with gr.Column():
output = gr.Video(label="Processed Video", streaming=True, autoplay=True)
video.upload(
fn=StreamObjectDetection.stream_object_detection,
inputs=[video, conf],
outputs=[output],
)
return object_detection_tab
def create_rag_tab(rag_system: EnhancedRAGSystem, wikipedia_processor: WikipediaProcessor):
"""Tạo tab RAG với debug chi tiết"""
# Initialize systems if not provided
if rag_system is None:
rag_system = EnhancedRAGSystem()
if wikipedia_processor is None:
wikipedia_processor = WikipediaProcessor()
with gr.Blocks() as rag_tab:
gr.Markdown("## 📚 Upload Dữ Liệu Wikipedia")
with gr.Row():
with gr.Column(scale=1):
gr.Markdown("### 📤 Upload Dữ Liệu")
file_upload = gr.File(
label="Tải lên file (TXT, CSV, JSON)",
file_types=['.txt', '.csv', '.json'],
file_count="single"
)
upload_btn = gr.Button("📤 Upload Data", variant="primary")
upload_status = gr.Textbox(
label="Trạng thái Upload",
interactive=False,
lines=5
)
gr.Markdown("### 📊 Thống kê Database")
stats_btn = gr.Button("📊 Database Stats", variant="secondary")
stats_display = gr.Textbox(
label="Thống kê",
interactive=False,
lines=6
)
with gr.Column(scale=2):
gr.Markdown("### 🔍 Tìm kiếm & Kiểm tra")
search_query = gr.Textbox(
label="Tìm kiếm trong database",
placeholder="Nhập từ khóa để kiểm tra dữ liệu..."
)
search_btn = gr.Button("🔍 Tìm kiếm", variant="secondary")
rag_results = gr.JSON(
label="Kết quả tìm kiếm",
show_label=True
)
def upload_wikipedia_file(file):
"""Xử lý upload file với debug đầy đủ"""
if file is None:
return "❌ Vui lòng chọn file để upload"
try:
print(f"🔄 Bắt đầu upload file: {file.name}")
# Kiểm tra file tồn tại
if not os.path.exists(file.name):
return f"❌ File không tồn tại: {file.name}"
# Xử lý file
documents = wikipedia_processor.process_uploaded_file(file.name)
if not documents:
return "❌ Không thể trích xuất dữ liệu từ file. File có thể trống hoặc định dạng không đúng."
print(f"✅ Đã xử lý {len(documents)} documents")
# Tạo metadata
metadatas = []
for i, doc in enumerate(documents):
metadata = {
"source": "uploaded_file",
"type": "knowledge",
"file_name": os.path.basename(file.name),
"language": "vi",
"doc_id": i,
"length": len(doc)
}
metadatas.append(metadata)
# Thêm vào RAG system
old_stats = rag_system.get_collection_stats()
old_count = old_stats['total_documents']
rag_system.add_documents(documents, metadatas)
# Lấy thống kê mới
new_stats = rag_system.get_collection_stats()
new_count = new_stats['total_documents']
success_msg = f"""
✅ UPLOAD THÀNH CÔNG!
📁 File: {os.path.basename(file.name)}
📄 Documents xử lý: {len(documents)}
📊 Documents thêm vào: {new_count - old_count}
🏷️ Tổng documents: {new_count}
🔤 Embeddings: {new_stats['embedding_count']}
🌐 Ngôn ngữ: {new_stats['language_distribution']}
💡 Bạn có thể tìm kiếm ngay để kiểm tra dữ liệu!
"""
return success_msg
except Exception as e:
error_msg = f"❌ LỖI UPLOAD: {str(e)}"
print(f"UPLOAD ERROR: {traceback.format_exc()}")
return error_msg
def get_rag_stats():
"""Lấy thống kê chi tiết"""
try:
stats = rag_system.get_collection_stats()
return f"""
📊 THỐNG KÊ RAG DATABASE:
• 📄 Tổng documents: {stats['total_documents']}
• 🔤 Số embeddings: {stats['embedding_count']}
• 📐 Dimension: {stats['embedding_dimension']}
• 🌐 Phân bố ngôn ngữ: {stats['language_distribution']}
• ✅ Trạng thái: {stats['status']}
• 🏷️ Tên: {stats['name']}
💡 Embeddings: {'Có' if stats['has_embeddings'] else 'Không'}
"""
except Exception as e:
return f"❌ Lỗi lấy thống kê: {str(e)}"
def search_rag_database(query):
"""Tìm kiếm để kiểm tra dữ liệu"""
if not query.strip():
return [{"message": "Nhập từ khóa để tìm kiếm"}]
try:
results = rag_system.semantic_search(query, top_k=3)
if not results:
return [{"message": "Không tìm thấy kết quả nào", "query": query}]
return results
except Exception as e:
return [{"error": f"Lỗi tìm kiếm: {str(e)}"}]
# Event handlers
upload_btn.click(upload_wikipedia_file, inputs=[file_upload], outputs=[upload_status])
stats_btn.click(get_rag_stats, inputs=[], outputs=[stats_display])
search_btn.click(search_rag_database, inputs=[search_query], outputs=[rag_results])
return rag_tab
def create_audio_tab(audio_service: AudioService):
gr.Markdown("## Nói chuyện với AI (Đa ngôn ngữ)")
audio_input, transcription_output, response_output, tts_audio_output, process_button = create_audio_components()
# NEW: Language display
language_display = gr.Textbox(
label="🌐 Ngôn ngữ phát hiện",
interactive=False,
placeholder="Ngôn ngữ sẽ hiển thị ở đây..."
)
process_button.click(
audio_service.transcribe_audio,
inputs=audio_input,
outputs=[transcription_output, response_output, tts_audio_output, language_display] # UPDATED
)
def create_streaming_voice_tab(streaming_service: StreamingVoiceService):
"""Tạo tab streaming voice với VAD optimized - FIXED VERSION"""
with gr.Blocks() as streaming_tab:
gr.Markdown("## 🎤 Trò chuyện giọng nói thời gian thực - Tối ưu hóa")
# FIXED: Thêm state để lưu kết quả VAD
vad_result_state = gr.State(value=None)
with gr.Row():
with gr.Column(scale=1):
# Controls
with gr.Row():
start_btn = gr.Button("🎙️ Bắt đầu VAD", variant="primary")
stop_btn = gr.Button("🛑 Dừng VAD", variant="secondary")
gr.Markdown("### Chế độ tự động (VAD)")
gr.Markdown("Hệ thống tự động nhận diện khi bạn bắt đầu nói")
with gr.Row():
vad_status = gr.Textbox(
label="Trạng thái VAD",
value="Chưa bắt đầu",
interactive=False
)
gr.Markdown("### Chế độ thủ công")
microphone = gr.Microphone(
label="🎤 Nhấn để nói thủ công",
type="numpy",
streaming=True
)
with gr.Accordion("📊 Performance Metrics", open=False):
latency_display = gr.JSON(
label="Latency Statistics",
value={}
)
refresh_latency_btn = gr.Button("🔄 Refresh Metrics", size="sm")
clear_btn = gr.Button("🗑️ Xóa hội thoại")
# State info
state_info = gr.Textbox(
label="Thông tin hệ thống",
value="Khởi tạo...",
lines=3,
interactive=False
)
with gr.Column(scale=2):
transcription_box = gr.Textbox(
label="📝 Bạn vừa nói",
lines=2,
interactive=False
)
response_box = gr.Textbox(
label="🤖 Phản hồi AI",
lines=3,
interactive=False
)
audio_output = gr.Audio(
label="🔊 Giọng nói AI",
interactive=False,
autoplay=True
)
# State variables
is_vad_active = gr.State(value=False)
def vad_callback(result):
"""Callback khi VAD phát hiện speech - FIXED VERSION"""
print(f"🎯 Nhận kết quả từ VAD: {result['transcription']}")
vad_result_state.value = result
def start_vad():
"""Bắt đầu VAD - FIXED VERSION"""
success = streaming_service.start_listening(vad_callback)
if success:
is_vad_active.value = True
status = "✅ VAD đang chạy - Hãy nói gì đó!"
state = streaming_service.get_conversation_state()
state_text = f"VAD: Đang hoạt động\nQueue: {state['queue_size']}\nThreads: {state['worker_threads']}"
else:
status = "❌ Không thể khởi động VAD"
state_text = "Lỗi khởi động"
return status, state_text
def stop_vad():
"""Dừng VAD"""
streaming_service.stop_listening()
is_vad_active.value = False
state = streaming_service.get_conversation_state()
state_text = f"VAD: Đã dừng\nHistory: {state['history_length']} messages"
return "🛑 VAD đã dừng", state_text
def process_microphone(audio_data):
"""Xử lý microphone input"""
if audio_data is None:
return "Chưa có âm thanh", "Hãy nói gì đó...", None, "VAD: Đang chạy" if is_vad_active.value else "VAD: Dừng"
try:
result = streaming_service.process_streaming_audio(audio_data)
state = streaming_service.get_conversation_state()
state_text = f"Manual mode\nHistory: {state['history_length']} messages"
return result['transcription'], result['response'], result['tts_audio'], state_text
except Exception as e:
return f"Lỗi: {e}", "Xin lỗi, có lỗi xảy ra", None, "Lỗi xử lý"
def check_vad_results():
"""Kiểm tra kết quả VAD - FIXED: Cập nhật UI khi có kết quả mới"""
if vad_result_state.value is not None:
result = vad_result_state.value
vad_result_state.value = None # Reset sau khi dùng
state = streaming_service.get_conversation_state()
state_text = f"VAD mode\nHistory: {state['history_length']} messages\nQueue: {state['queue_size']}"
return result['transcription'], result['response'], result['tts_audio'], state_text
return gr.skip(), gr.skip(), gr.skip(), gr.skip()
def clear_chat():
"""Xóa hội thoại"""
streaming_service.clear_conversation()
state = streaming_service.get_conversation_state()
state_text = f"Đã xóa hội thoại\nHistory: {state['history_length']} messages"
return "", "", None, state_text
def refresh_latency():
"""Làm mới latency metrics"""
stats = streaming_service.get_latency_stats()
return stats
def update_state_info():
"""Cập nhật thông tin trạng thái"""
state = streaming_service.get_conversation_state()
formatted_state = f"VAD: {'Đang chạy' if state['is_listening'] else 'Dừng'}\n"
formatted_state += f"Queue: {state['queue_size']}\n"
formatted_state += f"History: {state['history_length']} messages\n"
formatted_state += f"Threads: {state['worker_threads']}\n"
formatted_state += f"Last: {state['last_update']}"
latency_info = streaming_service.get_latency_stats()
return formatted_state, latency_info
# Event handlers
start_btn.click(start_vad, outputs=[vad_status, state_info])
stop_btn.click(stop_vad, outputs=[vad_status, state_info])
microphone.stream(
process_microphone,
inputs=[microphone],
outputs=[transcription_box, response_box, audio_output, state_info]
)
clear_btn.click(
clear_chat,
outputs=[transcription_box, response_box, audio_output, state_info]
)
refresh_latency_btn.click(
refresh_latency,
outputs=[latency_display]
)
# FIXED: Timer để kiểm tra kết quả VAD
gr.Timer(1.0).tick(
fn=check_vad_results,
outputs=[transcription_box, response_box, audio_output, state_info]
)
# Timer để cập nhật system info
gr.Timer(3.0).tick(
fn=update_state_info,
outputs=[state_info, latency_display]
)
return streaming_tab
def create_image_tab(image_service: ImageService):
"""Tạo tab phân tích hình ảnh với OCR và LLM"""
with gr.Blocks() as image_tab:
gr.Markdown("## 🖼️ Phân tích hình ảnh & Trích xuất văn bản")
gr.Markdown("""
### 🔍 Chức năng:
- **OCR đa ngôn ngữ**: Trích xuất văn bản từ ảnh (Tiếng Việt, Anh, Nhật, Hàn, Trung, ...)
- **Phân tích AI**: Sử dụng LLM để phân tích nội dung và ngữ cảnh
- **Hỗ trợ nhiều định dạng**: Tài liệu, ảnh chụp, meme, screenshot
### 📝 Hướng dẫn:
1. Tải lên hình ảnh có chứa văn bản
2. (Tùy chọn) Mô tả hình ảnh để AI phân tích chính xác hơn
3. Nhấn "Phân tích hình ảnh" để xem kết quả
""")
with gr.Row():
with gr.Column(scale=1):
# Image input
image_input = gr.Image(
type="numpy",
label="🖼️ Tải lên hình ảnh",
height=300
)
# User description
image_description = gr.Textbox(
label="📝 Mô tả hình ảnh (tùy chọn)",
placeholder="Ví dụ: Đây là hóa đơn mua hàng, ảnh chụp menu nhà hàng, văn bản tiếng Việt...",
lines=3
)
# Analyze button
analyze_button = gr.Button(
"🔍 Phân tích hình ảnh",
variant="primary",
size="lg"
)
# Clear button
clear_button = gr.Button("🗑️ Xóa", variant="secondary")
with gr.Column(scale=2):
# Results output
image_output = gr.Textbox(
label="📊 Kết quả phân tích",
lines=15,
max_lines=20,
show_copy_button=True
)
def analyze_image(image, description):
"""Xử lý phân tích ảnh"""
if image is None:
return "❌ Vui lòng tải lên hình ảnh trước khi phân tích."
return image_service.analyze_image_with_description(image, description)
def clear_all():
"""Xóa tất cả input và output"""
return None, "", ""
# Event handlers
analyze_button.click(
analyze_image,
inputs=[image_input, image_description],
outputs=[image_output]
)
clear_button.click(
clear_all,
outputs=[image_input, image_description, image_output]
)
return image_tab
def create_chat_tab(chat_service: ChatService):
gr.Markdown("## Trò chuyện với AI Assistant (Đa ngôn ngữ)")
chatbot, state, user_input, send_button, clear_button, chat_tts_output = create_chat_components()
# Language display
chat_language_display = gr.Textbox(
label="🌐 Ngôn ngữ phát hiện",
interactive=False,
placeholder="Ngôn ngữ sẽ hiển thị ở đây..."
)
# SỬA: Đảm bảo số lượng inputs/outputs khớp
send_button.click(
fn=chat_service.respond,
inputs=[user_input, state],
outputs=[user_input, chatbot, state, chat_tts_output, chat_language_display]
)
clear_button.click(
fn=chat_service.clear_chat_history,
inputs=[state],
outputs=[chatbot, state]
)
# Thêm enter để gửi tin nhắn
user_input.submit(
fn=chat_service.respond,
inputs=[user_input, state],
outputs=[user_input, chatbot, state, chat_tts_output, chat_language_display]
)
def create_language_info_tab(multilingual_manager):
"""Tab hiển thị thông tin về hệ thống đa ngôn ngữ"""
gr.Markdown("## 🌐 Thông tin Hệ thống Đa ngôn ngữ")
with gr.Row():
with gr.Column():
gr.Markdown("### 🔧 Cấu hình Model")
vietnamese_info = multilingual_manager.get_language_info('vi')
multilingual_info = multilingual_manager.get_language_info('en')
# SỬA LỖI Ở ĐÂY: Sử dụng key đúng từ dictionary mới
gr.Markdown(f"""
**Tiếng Việt:**
- Embedding Model: `{vietnamese_info['embedding_model']}`
- LLM Model: `{vietnamese_info['llm_model']}`
- Trạng thái: {vietnamese_info['embedding_status']}
**Đa ngôn ngữ:**
- Embedding Model: `{multilingual_info['embedding_model']}`
- LLM Model: `{multilingual_info['llm_model']}`
- Trạng thái: {multilingual_info['embedding_status']}
""")
with gr.Column():
gr.Markdown("### 🎯 Ngôn ngữ được hỗ trợ")
supported_languages = """
- 🇻🇳 **Tiếng Việt**: Sử dụng model chuyên biệt
- 🇺🇸 **English**: Sử dụng model đa ngôn ngữ
- 🇫🇷 **French**: Sử dụng model đa ngôn ngữ
- 🇪🇸 **Spanish**: Sử dụng model đa ngôn ngữ
- 🇩🇪 **German**: Sử dụng model đa ngôn ngữ
- 🇯🇵 **Japanese**: Sử dụng model đa ngôn ngữ
- 🇰🇷 **Korean**: Sử dụng model đa ngôn ngữ
- 🇨🇳 **Chinese**: Sử dụng model đa ngôn ngữ
"""
gr.Markdown(supported_languages)
with gr.Row():
with gr.Column():
gr.Markdown("### 🔍 Kiểm tra Ngôn ngữ")
test_text = gr.Textbox(
label="Nhập văn bản để kiểm tra ngôn ngữ",
placeholder="Nhập văn bản bằng bất kỳ ngôn ngữ nào..."
)
test_button = gr.Button("🔍 Kiểm tra", variant="primary")
test_result = gr.JSON(label="Kết quả phát hiện ngôn ngữ")
test_button.click(
lambda text: {
'detected_language': multilingual_manager.detect_language(text),
'language_info': multilingual_manager.get_language_info(multilingual_manager.detect_language(text)),
'embedding_model': multilingual_manager.get_embedding_model(multilingual_manager.detect_language(text)) is not None,
'llm_model': multilingual_manager.get_llm_model(multilingual_manager.detect_language(text))
},
inputs=[test_text],
outputs=[test_result]
)
def create_tts_tab(tts_service: EnhancedTTSService):
gr.Markdown("## 🎵 Chuyển văn bản thành giọng nói nâng cao")
gr.Markdown("Nhập văn bản và chọn ngôn ngữ để chuyển thành giọng nói")
with gr.Group():
with gr.Row():
tts_text_input = gr.Textbox(
label="Văn bản cần chuyển thành giọng nói",
lines=4,
placeholder="Nhập văn bản tại đây..."
)
with gr.Row():
tts_language = gr.Dropdown(
choices=["vi", "en", "fr", "es", "de", "ja", "ko", "zh"],
value="vi",
label="Ngôn ngữ"
)
tts_provider = gr.Dropdown(
choices=["auto", "gtts", "edgetts"],
value="auto",
label="Nhà cung cấp TTS"
)
with gr.Row():
tts_output_audio = gr.Audio(
label="Kết quả giọng nói",
interactive=False
)
tts_button = gr.Button("🔊 Chuyển thành giọng nói", variant="primary")
def text_to_speech_standalone(text, language, tts_provider):
if not text:
return None
try:
tts_audio_bytes = tts_service.text_to_speech(text, language, tts_provider)
if tts_audio_bytes:
temp_audio_file = tts_service.save_audio_to_file(tts_audio_bytes)
return temp_audio_file
except Exception as e:
print(f"❌ Lỗi TTS: {e}")
return None
tts_button.click(
text_to_speech_standalone,
inputs=[tts_text_input, tts_language, tts_provider],
outputs=[tts_output_audio]
)