Spaces:
Build error
Build error
| import os | |
| import re | |
| import shutil | |
| import subprocess | |
| import tempfile | |
| import gradio as gr | |
| # Ensure workspace folder exists | |
| WORKSPACE_DIR = os.path.abspath("./workspace") | |
| os.makedirs(WORKSPACE_DIR, exist_ok=True) | |
| # Helper to strip ANSI escape codes from terminal outputs | |
| def strip_ansi(text): | |
| ansi_escape = re.compile(r'\x1B(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~])') | |
| return ansi_escape.sub('', text) | |
| # Helper to get files list | |
| def list_workspace_files(): | |
| files_list = [] | |
| for root, dirs, files in os.walk(WORKSPACE_DIR): | |
| # Ignore hidden files/directories like .git or .qwen | |
| if ".git" in root or ".qwen" in root: | |
| continue | |
| for file in files: | |
| full_path = os.path.join(root, file) | |
| rel_path = os.path.relpath(full_path, WORKSPACE_DIR) | |
| files_list.append(rel_path) | |
| return sorted(files_list) | |
| # Load file content | |
| def load_file_content(filepath): | |
| if not filepath: | |
| return "No file selected." | |
| full_path = os.path.join(WORKSPACE_DIR, filepath) | |
| if not os.path.exists(full_path) or os.path.isdir(full_path): | |
| return f"File {filepath} not found or is a directory." | |
| try: | |
| with open(full_path, "r", encoding="utf-8", errors="ignore") as f: | |
| return f.read() | |
| except Exception as e: | |
| return f"Error reading file: {str(e)}" | |
| # Save file content | |
| def save_file_content(filepath, content): | |
| if not filepath: | |
| return "No file selected.", list_workspace_files() | |
| full_path = os.path.join(WORKSPACE_DIR, filepath) | |
| try: | |
| os.makedirs(os.path.dirname(full_path), exist_ok=True) | |
| with open(full_path, "w", encoding="utf-8") as f: | |
| f.write(content) | |
| return f"Successfully saved {filepath}!", gr.update(choices=list_workspace_files(), value=filepath) | |
| except Exception as e: | |
| return f"Error saving file: {str(e)}", list_workspace_files() | |
| # Initialize templates | |
| def load_template(template_name): | |
| # Clear existing workspace | |
| for item in os.listdir(WORKSPACE_DIR): | |
| item_path = os.path.join(WORKSPACE_DIR, item) | |
| if item == ".qwen": | |
| continue | |
| if os.path.isdir(item_path): | |
| shutil.rmtree(item_path) | |
| else: | |
| os.remove(item_path) | |
| if template_name == "Python Math Library": | |
| os.makedirs(os.path.join(WORKSPACE_DIR, "pymath"), exist_ok=True) | |
| with open(os.path.join(WORKSPACE_DIR, "pymath", "math.py"), "w") as f: | |
| f.write("def add(a, b):\n return a + b\n\ndef multiply(a, b):\n return a * b\n") | |
| with open(os.path.join(WORKSPACE_DIR, "README.md"), "w") as f: | |
| f.write("# Python Math Library\n\nThis is a simple math library. Ask Qwen-Code to add features or tests!\n") | |
| elif template_name == "NodeJS Web App": | |
| with open(os.path.join(WORKSPACE_DIR, "package.json"), "w") as f: | |
| f.write('{\n "name": "simple-web-app",\n "version": "1.0.0",\n "main": "index.js",\n "dependencies": {}\n}\n') | |
| with open(os.path.join(WORKSPACE_DIR, "index.js"), "w") as f: | |
| f.write("console.log('Hello from NodeJS web app!');\n") | |
| elif template_name == "HTML Landing Page": | |
| with open(os.path.join(WORKSPACE_DIR, "index.html"), "w") as f: | |
| f.write("<!DOCTYPE html>\n<html>\n<head>\n <title>My Landing Page</title>\n</head>\n<body>\n <h1>Welcome to my Qwen-Code Generated Page!</h1>\n</body>\n</html>\n") | |
| return f"Initialized workspace with '{template_name}' template.", gr.update(choices=list_workspace_files()) | |
| # Zip workspace for download | |
| def download_workspace(): | |
| temp_dir = tempfile.gettempdir() | |
| zip_path = os.path.join(temp_dir, "workspace_archive") | |
| shutil.make_archive(zip_path, 'zip', WORKSPACE_DIR) | |
| return zip_path + ".zip" | |
| # Execute Qwen-Code agent | |
| def run_agent(provider, api_key, model_name, custom_model, prompt, history): | |
| if not api_key: | |
| yield history, "⚠️ Please enter an API key under API settings first!", gr.update() | |
| return | |
| selected_model = custom_model if model_name == "custom" else model_name | |
| if not selected_model: | |
| yield history, "⚠️ Please select or specify a model first!", gr.update() | |
| return | |
| # Prepare command and env | |
| cmd = ["qwen", "-p", prompt, "--yolo"] | |
| env = os.environ.copy() | |
| env["QWEN_SANDBOX"] = "false" # Run inside the container's environment natively | |
| # Configure env variables based on provider | |
| if provider == "DashScope (Qwen-compatible)": | |
| env["DASHSCOPE_API_KEY"] = api_key | |
| env["OPENAI_API_KEY"] = api_key | |
| env["OPENAI_BASE_URL"] = "https://dashscope.aliyuncs.com/compatible-mode/v1" | |
| env["OPENAI_MODEL"] = selected_model | |
| elif provider == "Alibaba Cloud Coding Plan (Beijing)": | |
| env["BAILIAN_CODING_PLAN_API_KEY"] = api_key | |
| env["OPENAI_BASE_URL"] = "https://coding.dashscope.aliyuncs.com/v1" | |
| env["OPENAI_MODEL"] = selected_model | |
| elif provider == "Alibaba Cloud Coding Plan (International)": | |
| env["BAILIAN_CODING_PLAN_API_KEY"] = api_key | |
| env["OPENAI_BASE_URL"] = "https://coding-intl.dashscope.aliyuncs.com/v1" | |
| env["OPENAI_MODEL"] = selected_model | |
| elif provider == "OpenAI": | |
| env["OPENAI_API_KEY"] = api_key | |
| env["OPENAI_MODEL"] = selected_model | |
| elif provider == "Anthropic Claude": | |
| env["ANTHROPIC_API_KEY"] = api_key | |
| env["ANTHROPIC_MODEL"] = selected_model | |
| elif provider == "Google Gemini": | |
| env["GEMINI_API_KEY"] = api_key | |
| env["GEMINI_MODEL"] = selected_model | |
| history.append((prompt, "... Starting Qwen Code Agent ...")) | |
| yield history, "Initializing Qwen Code CLI process...", gr.update() | |
| # Launch subprocess and read stdout in real-time | |
| console_output = "" | |
| try: | |
| proc = subprocess.Popen( | |
| cmd, | |
| cwd=WORKSPACE_DIR, | |
| env=env, | |
| stdout=subprocess.PIPE, | |
| stderr=subprocess.STDOUT, | |
| text=True, | |
| bufsize=1 | |
| ) | |
| for line in iter(proc.stdout.readline, ""): | |
| cleaned_line = strip_ansi(line) | |
| console_output += cleaned_line | |
| # Update chat history and terminal output dynamically | |
| history[-1] = (prompt, f"🤖 Agent is working...\n\n```\n{console_output[-4000:]}\n```") | |
| yield history, console_output, gr.update() | |
| proc.stdout.close() | |
| return_code = proc.wait() | |
| if return_code == 0: | |
| status_msg = "✅ Qwen Code Agent finished successfully!" | |
| else: | |
| status_msg = f"❌ Qwen Code Agent exited with return code {return_code}." | |
| history[-1] = (prompt, f"{status_msg}\n\n### Terminal Log Summary:\n```\n{console_output[-4000:]}\n```") | |
| yield history, console_output, gr.update(choices=list_workspace_files()) | |
| except Exception as e: | |
| error_msg = f"Failed to execute command: {str(e)}" | |
| history[-1] = (prompt, f"❌ Error: {error_msg}") | |
| yield history, error_msg, gr.update() | |
| # UI Layout | |
| with gr.Blocks(theme=gr.themes.Soft(primary_hue="purple", secondary_hue="blue")) as demo: | |
| gr.HTML(""" | |
| <div style='text-align: center; margin-bottom: 20px;'> | |
| <h1 style='color: #8A2BE2; font-size: 2.5rem; margin-bottom: 5px;'>🤖 Qwen-Code3 AI Coding Workspace</h1> | |
| <p style='font-size: 1.1rem; color: #555;'>The ultimate terminal coding agent powered by Qwen. Connect your API key and let Qwen-Code code for you!</p> | |
| </div> | |
| """) | |
| with gr.Accordion("⚙️ API Configuration (Credentials are kept in-memory and never saved to disk)", open=True): | |
| with gr.Row(): | |
| provider = gr.Dropdown( | |
| label="API Provider", | |
| choices=[ | |
| "DashScope (Qwen-compatible)", | |
| "Alibaba Cloud Coding Plan (Beijing)", | |
| "Alibaba Cloud Coding Plan (International)", | |
| "OpenAI", | |
| "Anthropic Claude", | |
| "Google Gemini" | |
| ], | |
| value="DashScope (Qwen-compatible)" | |
| ) | |
| api_key = gr.Textbox( | |
| label="API Key", | |
| placeholder="Enter your sk-... or other provider credentials", | |
| type="password" | |
| ) | |
| model_name = gr.Dropdown( | |
| label="Model", | |
| choices=["qwen3-coder-plus", "qwen3.5-plus", "gpt-4o", "claude-3-5-sonnet-latest", "gemini-1.5-pro", "custom"], | |
| value="qwen3-coder-plus" | |
| ) | |
| custom_model = gr.Textbox( | |
| label="Custom Model ID (if selected 'custom')", | |
| placeholder="e.g. qwen-max", | |
| visible=False | |
| ) | |
| # Automatically show/hide custom model input | |
| def update_model_visibility(model_choice): | |
| return gr.update(visible=(model_choice == "custom")) | |
| model_name.change(update_model_visibility, inputs=[model_name], outputs=[custom_model]) | |
| with gr.Tabs(): | |
| with gr.Tab("💬 Chat & Agent Agentic Terminal"): | |
| with gr.Row(): | |
| with gr.Column(scale=3): | |
| chatbot = gr.Chatbot(label="Agent Workspace Log", bubble_full_width=False, height=500) | |
| prompt_input = gr.Textbox( | |
| label="What should the Qwen Code Agent build or fix?", | |
| placeholder="e.g. 'Build a python server in main.py that returns health status'", | |
| lines=2 | |
| ) | |
| run_btn = gr.Button("🚀 Run Qwen Code Agent", variant="primary") | |
| with gr.Column(scale=2): | |
| console_log = gr.Textbox( | |
| label="📟 Live Terminal Stdout (Scrolls with Agent progress)", | |
| placeholder="Stdout will stream here...", | |
| interactive=False, | |
| lines=25, | |
| max_lines=30 | |
| ) | |
| with gr.Tab("📂 Workspace File Explorer"): | |
| with gr.Row(): | |
| with gr.Column(scale=1): | |
| file_list = gr.Dropdown( | |
| label="Workspace Files", | |
| choices=list_workspace_files(), | |
| interactive=True | |
| ) | |
| refresh_files_btn = gr.Button("🔄 Refresh Files") | |
| download_btn = gr.Button("📥 Download Workspace (.ZIP)", variant="secondary") | |
| download_file_output = gr.File(label="Workspace Download Link") | |
| with gr.Column(scale=3): | |
| file_path_display = gr.Textbox(label="Editing File Path", interactive=False) | |
| file_content_editor = gr.Code(label="File Content Editor / Viewer", language="python", lines=20) | |
| save_file_btn = gr.Button("💾 Save Changes", variant="primary") | |
| save_status = gr.Markdown() | |
| # Explorer actions | |
| def on_file_selected(filepath): | |
| if not filepath: | |
| return "", "" | |
| return filepath, load_file_content(filepath) | |
| file_list.change(on_file_selected, inputs=[file_list], outputs=[file_path_display, file_content_editor]) | |
| refresh_files_btn.click( | |
| lambda: gr.update(choices=list_workspace_files()), | |
| outputs=[file_list] | |
| ) | |
| save_file_btn.click( | |
| save_file_content, | |
| inputs=[file_path_display, file_content_editor], | |
| outputs=[save_status, file_list] | |
| ) | |
| download_btn.click( | |
| download_workspace, | |
| outputs=[download_file_output] | |
| ) | |
| with gr.Tab("📋 Workspace Templates"): | |
| gr.Markdown("### Initialize your Workspace with a sample project template:") | |
| with gr.Row(): | |
| py_template_btn = gr.Button("🐍 Python Math Library Template", variant="secondary") | |
| node_template_btn = gr.Button("🟢 NodeJS Web App Template", variant="secondary") | |
| html_template_btn = gr.Button("🎨 HTML Landing Page Template", variant="secondary") | |
| template_status = gr.Markdown() | |
| py_template_btn.click( | |
| lambda: load_template("Python Math Library"), | |
| outputs=[template_status, file_list] | |
| ) | |
| node_template_btn.click( | |
| lambda: load_template("NodeJS Web App"), | |
| outputs=[template_status, file_list] | |
| ) | |
| html_template_btn.click( | |
| lambda: load_template("HTML Landing Page"), | |
| outputs=[template_status, file_list] | |
| ) | |
| # Connect Run Button | |
| run_btn.click( | |
| run_agent, | |
| inputs=[provider, api_key, model_name, custom_model, prompt_input, chatbot], | |
| outputs=[chatbot, console_log, file_list] | |
| ) | |
| if __name__ == "__main__": | |
| demo.queue() | |
| demo.launch(server_name="0.0.0.0", server_port=7860) | |