docker_gent / app.py
nakas's picture
Update app.py
d489acb verified
import gradio as gr
import subprocess
import os
import sys
import threading
from openai import OpenAI
import re
import traceback
import signal
class GradioAgentBuilder:
def __init__(self):
self.api_key = None
self.client = None
self.current_code = ""
self.generated_process = None
def initialize_client(self, api_key):
self.api_key = api_key
self.client = OpenAI(api_key=api_key)
return "OpenAI client initialized successfully!"
def clean_code(self, code):
code = re.sub(r'^```\w*\n', '', code)
code = re.sub(r'\n```$', '', code)
code = code.strip()
return code
def get_completion(self, prompt, error_context=None):
if not self.client:
return "Please set your OpenAI API key first!"
messages = [
{"role": "system", "content": """You are an AI agent specialized in creating Gradio applications.
Generate complete, working Python code for Gradio apps based on user prompts using Gradio 3.x.
CRITICAL REQUIREMENTS:
1. The app must run on port 7861
2. All imports must be at the top
3. The app must be a complete, runnable program
4. Must include demo.launch(server_name="0.0.0.0", server_port=7861) at the end
Example:
import gradio as gr
import numpy as np
with gr.Blocks() as demo:
gr.Markdown("# Calculator")
with gr.Row():
num1 = gr.Number(label="First Number")
num2 = gr.Number(label="Second Number")
with gr.Row():
add_btn = gr.Button("Add")
result = gr.Number(label="Result")
def add(a, b):
return float(a) + float(b)
add_btn.click(
fn=add,
inputs=[num1, num2],
outputs=result
)
if __name__ == "__main__":
demo.launch(server_name="0.0.0.0", server_port=7861)"""}
]
if error_context:
messages.append({"role": "user", "content": f"The previous code generated an error: {error_context}. Please fix the code following the requirements above."})
else:
messages.append({"role": "user", "content": f"Create a Gradio app that: {prompt}. Follow the requirements above."})
try:
response = self.client.chat.completions.create(
model="gpt-4-turbo-preview",
messages=messages,
temperature=0.7,
max_tokens=2000
)
code = response.choices[0].message.content
return self.clean_code(code)
except Exception as e:
return f"Error generating code: {str(e)}"
def stop_generated_app(self):
if self.generated_process:
try:
os.killpg(os.getpgid(self.generated_process.pid), signal.SIGTERM)
self.generated_process = None
except:
pass
def execute_code(self, code):
if not code or isinstance(code, str) and code.startswith("Error"):
return "", "No valid code to execute"
output = ""
error = None
try:
cleaned_code = self.clean_code(code)
# Use the generated directory with proper permissions
generated_file = os.path.join("/code/generated", "generated_app.py")
# Stop any existing generated app
self.stop_generated_app()
# Save the code to the file
with open(generated_file, "w") as f:
f.write(cleaned_code)
# Start the generated app in a new process
self.generated_process = subprocess.Popen(
[sys.executable, generated_file],
preexec_fn=os.setsid,
stderr=subprocess.PIPE,
cwd="/code/generated" # Set working directory
)
# Check for immediate startup errors
error_output = self.generated_process.stderr.readline().decode()
if error_output:
error = error_output
self.stop_generated_app()
else:
output = "Generated app is running at: /proxy/7861\n"
output += "You can access it in a new browser tab."
self.current_code = cleaned_code
except Exception as e:
error = traceback.format_exc()
self.stop_generated_app()
return output, error
def iterate_on_code(self, code, error):
if not error:
return code, "Code executed successfully!", None
try:
error_context = f"""
The code generated this error: {error}
Original code:
{code}
Please fix the issues and return a complete, working version.
"""
fixed_code = self.get_completion(None, error_context)
output, new_error = self.execute_code(fixed_code)
if new_error:
output += "\nStill encountering errors. You may want to try 'Fix & Iterate' again."
return fixed_code, output, new_error
except Exception as e:
return code, f"Error during iteration: {str(e)}", str(e)
# Create the main interface
agent = GradioAgentBuilder()
with gr.Blocks() as demo:
gr.Markdown("# Gradio App Builder Agent")
gr.Markdown("""Enter your OpenAI API key and describe the Gradio app you want to create.
The generated app will be available at /proxy/7861 (open in a new tab)""")
with gr.Row():
api_key_input = gr.Textbox(
label="OpenAI API Key",
placeholder="Enter your OpenAI API key...",
type="password"
)
api_status = gr.Textbox(
label="API Status",
placeholder="API status will appear here...",
interactive=False
)
set_key_btn = gr.Button("Set API Key")
with gr.Row():
with gr.Column():
prompt_input = gr.Textbox(
label="What kind of Gradio app would you like to create?",
placeholder="Describe the Gradio app you want...",
lines=3
)
generate_btn = gr.Button("Generate App", interactive=False)
iterate_btn = gr.Button("Fix & Iterate", interactive=False)
with gr.Column():
code_output = gr.Code(
label="Generated Code",
language="python"
)
with gr.Row():
execution_output = gr.Textbox(
label="Execution Output",
lines=5,
placeholder="Execution results will appear here..."
)
error_output = gr.Textbox(
label="Error Output",
lines=5,
placeholder="Any errors will appear here..."
)
def set_api_key(key):
try:
status = agent.initialize_client(key)
return (
status,
gr.update(interactive=True),
gr.update(interactive=True)
)
except Exception as e:
return (
f"Error initializing API client: {str(e)}",
gr.update(interactive=False),
gr.update(interactive=False)
)
def generate_app(prompt):
initial_code = agent.get_completion(prompt)
output, error = agent.execute_code(initial_code)
return initial_code, output, error
def iterate_app(code, error):
if not code:
return "", "No code to iterate on", "Please generate code first"
new_code, output, new_error = agent.iterate_on_code(code, error)
return new_code, output, new_error
set_key_btn.click(
fn=set_api_key,
inputs=api_key_input,
outputs=[api_status, generate_btn, iterate_btn]
)
generate_btn.click(
fn=generate_app,
inputs=prompt_input,
outputs=[code_output, execution_output, error_output]
)
iterate_btn.click(
fn=iterate_app,
inputs=[code_output, error_output],
outputs=[code_output, execution_output, error_output]
)
if __name__ == "__main__":
# Clean up any existing processes on startup
agent.stop_generated_app()
demo.queue().launch(server_name="0.0.0.0", server_port=7860)