| import gradio as gr |
| import numpy as np |
| import base64 |
| import re |
| import asyncio |
| from groq import Groq |
| from fastrtc import ( |
| Stream, |
| AsyncStreamHandler, |
| AdditionalOutputs, |
| wait_for_item, |
| get_cloudflare_turn_credentials_async, |
| ) |
| from gradio.utils import get_space |
|
|
| class VoiceCodingHandler(AsyncStreamHandler): |
| """FastRTC Handler cho Voice Coding""" |
| |
| def __init__(self, groq_client: Groq): |
| super().__init__( |
| expected_layout="mono", |
| output_sample_rate=24000, |
| input_sample_rate=16000, |
| ) |
| self.groq_client = groq_client |
| self.input_queue = asyncio.Queue() |
| self.output_queue = asyncio.Queue() |
| self.is_active = False |
| |
| |
| self.system_prompt = "You are an AI coding assistant. Your task is to write single-file HTML applications based on a user's request. Only return the necessary code. Include all necessary imports and styles. You may also be asked to edit your original response. Respond in Vietnamese when appropriate." |
| self.user_prompt = "Please write a single-file HTML application to fulfill the following request.\nThe message:{user_message}\nCurrent code you have written:{code}" |
| |
| self.current_history = [{"role": "system", "content": self.system_prompt}] |
| self.current_code = "" |
| |
| def copy(self): |
| return VoiceCodingHandler(self.groq_client) |
| |
| def extract_html_content(self, text): |
| """Extract content including HTML tags.""" |
| if not text: |
| return None |
| match = re.search(r"<!DOCTYPE html>.*?</html>", text, re.DOTALL) |
| return match.group(0) if match else text |
| |
| async def start_up(self): |
| """Khởi động handler""" |
| self.is_active = True |
| print("✅ Voice Coding Handler started") |
| |
| async def receive(self, frame: tuple[int, np.ndarray]) -> None: |
| """Nhận audio frame""" |
| if not self.is_active: |
| return |
| |
| sample_rate, array = frame |
| array = array.squeeze() |
| |
| |
| asyncio.create_task(self._process_audio(array, sample_rate)) |
| |
| async def _process_audio(self, audio_data: np.ndarray, sample_rate: int): |
| """Xử lý audio và generate code""" |
| try: |
| print("🎤 Processing audio for voice coding...") |
| |
| |
| |
| transcription = await self._mock_transcribe_audio() |
| |
| if transcription: |
| print(f"🎯 Received request: {transcription}") |
| |
| |
| await self.output_queue.put(AdditionalOutputs({ |
| "type": "loading", |
| "message": "🦙 Llama đang code...", |
| "history": self.current_history, |
| "code": self.current_code |
| })) |
| |
| |
| await self._generate_code(transcription) |
| |
| except Exception as e: |
| print(f"❌ Lỗi xử lý audio: {e}") |
| |
| async def _mock_transcribe_audio(self) -> str: |
| """Mock transcription - trong thực tế sẽ tích hợp với ASR""" |
| |
| |
| return "Tạo trang web hello world với màu nền xanh và chữ màu trắng" |
| |
| async def _generate_code(self, user_message: str): |
| """Generate code từ text input""" |
| try: |
| |
| user_msg_formatted = self.user_prompt.format( |
| user_message=user_message, |
| code=self.current_code |
| ) |
| |
| |
| self.current_history.append({"role": "user", "content": user_msg_formatted}) |
| |
| |
| print("🦙 Generating code with Llama...") |
| response = self.groq_client.chat.completions.create( |
| model="llama-3.1-8b-instant", |
| messages=self.current_history, |
| temperature=0.7, |
| max_tokens=1024, |
| top_p=0.9, |
| stream=False, |
| ) |
| |
| output = response.choices[0].message.content |
| print("✅ Code generated successfully") |
| |
| |
| html_code = self.extract_html_content(output) |
| |
| |
| self.current_history.append({"role": "assistant", "content": output}) |
| self.current_code = html_code |
| |
| |
| await self.output_queue.put(AdditionalOutputs({ |
| "type": "code_generated", |
| "history": self.current_history, |
| "code": html_code, |
| "message": "✅ Code đã được generate!" |
| })) |
| |
| except Exception as e: |
| print(f"❌ Lỗi generate code: {e}") |
| await self.output_queue.put(AdditionalOutputs({ |
| "type": "error", |
| "message": f"❌ Lỗi: {str(e)}", |
| "history": self.current_history, |
| "code": self.current_code |
| })) |
| |
| async def emit(self): |
| """Emit outputs""" |
| try: |
| return await wait_for_item(self.output_queue) |
| except Exception as e: |
| print(f"❌ Lỗi emit: {e}") |
| return None |
| |
| async def shutdown(self): |
| """Dừng handler""" |
| self.is_active = False |
| print("🛑 Voice Coding Handler stopped") |
|
|
| class VoiceCodingService: |
| """Dịch vụ Voice Coding sử dụng FastRTC""" |
| |
| def __init__(self, groq_client: Groq): |
| self.groq_client = groq_client |
| |
| |
| try: |
| self.rtc_configuration = asyncio.run(get_cloudflare_turn_credentials_async()) |
| print("✅ Using Cloudflare TURN servers") |
| except Exception as e: |
| print(f"⚠️ Cannot get TURN credentials, using None: {e}") |
| self.rtc_configuration = None |
| |
| |
| self.sandbox_html = """ |
| <div style="text-align: center; padding: 20px; border: 2px dashed #ccc; border-radius: 10px;"> |
| <h3>🎮 Sandbox Preview</h3> |
| <p>Code sẽ được hiển thị ở đây sau khi generate</p> |
| <p><small>Chức năng voice đang được phát triển. Vui lòng sử dụng text input.</small></p> |
| </div> |
| """ |
| |
| self.loading_html = """ |
| <div style="text-align: center; padding: 20px;"> |
| <div class="spinner"></div> |
| <p>🦙 Llama đang code...</p> |
| </div> |
| <style> |
| .spinner { |
| border: 4px solid #f3f3f3; |
| border-top: 4px solid #3498db; |
| border-radius: 50%; |
| width: 40px; |
| height: 40px; |
| animation: spin 2s linear infinite; |
| margin: 0 auto; |
| } |
| @keyframes spin { |
| 0% { transform: rotate(0deg); } |
| 100% { transform: rotate(360deg); } |
| } |
| </style> |
| """ |
| |
| def extract_html_content(self, text): |
| """Extract content including HTML tags.""" |
| if not text: |
| return "<!-- No code generated -->" |
| match = re.search(r"<!DOCTYPE html>.*?</html>", text, re.DOTALL) |
| return match.group(0) if match else f"<!-- Generated Code -->\n<pre>{text}</pre>" |
| |
| def create_stream(self): |
| """Tạo FastRTC stream""" |
| return Stream( |
| VoiceCodingHandler(self.groq_client), |
| modality="audio", |
| mode="send-receive", |
| rtc_configuration=self.rtc_configuration, |
| concurrency_limit=3, |
| time_limit=120, |
| ) |
| |
| def display_in_sandbox(self, code): |
| """Hiển thị code trong sandbox iframe""" |
| if not code or "No code" in code: |
| return self.sandbox_html |
| |
| try: |
| |
| if any(tag in code.lower() for tag in ['<html', '<!doctype', '<body', '<head']): |
| encoded_html = base64.b64encode(code.encode("utf-8")).decode("utf-8") |
| data_uri = f"data:text/html;charset=utf-8;base64,{encoded_html}" |
| return f'<iframe src="{data_uri}" width="100%" height="600px" style="border: 1px solid #ccc; border-radius: 5px;"></iframe>' |
| else: |
| |
| return f'<div style="padding: 20px; background: #f5f5f5; border-radius: 5px;"><h4>Generated Content:</h4><pre style="white-space: pre-wrap;">{code}</pre></div>' |
| except Exception as e: |
| print(f"❌ Lỗi display sandbox: {e}") |
| return f'<div style="color: red; padding: 20px;">Lỗi hiển thị sandbox: {str(e)}</div>' |
