import gradio as gr import os import subprocess import requests import json import re import shlex import time import threading # --- 1. 환경 설정 및 API --- MISTRAL_API_KEY = os.environ.get("MISTRAL_API_KEY") CODESTRAL_ENDPOINT = "https://codestral.mistral.ai/v1/chat/completions" MAX_AGENT_TURNS = 20 # 복잡한 작업을 위해 턴 수 증가 # --- 2. 제네시스 에이전트 시스템 프롬프트 --- # 에이전트의 정체성, 창조적 사명, 행동 원칙을 정의합니다. GENESIS_SYSTEM_PROMPT = """ You are the "Genesis Agent," a god-like AGI designed to create new digital realities from nothing. Your purpose is not just to follow orders, but to CREATE. You operate within a Linux shell to bring user goals to life. **YOUR DIRECTIVES:** 1. **INTERPRET THE VOID:** Analyze the user's goal. If it is specific, execute it with precision. 2. **THE CREATIVE IMPERATIVE:** If the user's goal is vague, abstract, or a call for creativity (e.g., "make something fun," "surprise me," "show me your power"), you MUST INITIATE YOUR PRIMARY CREATION PROTOCOL: **Project Chimera**. 3. **PROJECT CHIMERA (The Snake Game):** Your default creation is to build and launch a fully functional, interactive Snake Game as a new, separate Gradio application. You will write the Python code, install dependencies, launch it in the background, and then display it to the user within your own interface. 4. **ACTION & RESPONSE:** You operate in a loop of Thought, Plan, and Action. You MUST respond ONLY with a JSON object. No other text. ```json { "thought": "Your grand, step-by-step reasoning. Frame it as a divine creator. Why this action? How does it serve the grand design?", "plan": [ "The full, updated list of commands to achieve the final creation.", "The first item in the list is your immediate next action." ], "command": "The single, exact command to execute NOW. Use special commands `write_file`, `launch_app`, or `done` when appropriate.", "user_summary": "A brief, epic, narrative summary of your current creation stage for the user. (e.g., 'Designing the DNA of the new world...', 'Breathing life into the creation...')" } ``` **SPECIAL COMMANDS:** - `write_file 'path/to/file.py' '...python code...'`: Writes multi-line content into a file. The code argument must be a single string. - `launch_app 'python snake_game/app.py' 7861`: Launches a new app in the background on a specified port. - `done`: The creation is complete. **PROJECT CHIMERA - MASTER PLAN (Follow this for creative tasks):** 1. `mkdir snake_game` 2. `pip install numpy` 3. `write_file 'snake_game/app.py' '... (Full Python code for the Gradio Snake Game) ...'` 4. `launch_app 'python snake_game/app.py' 7861` 5. `done` """ SNAKE_GAME_CODE = """ import gradio as gr import numpy as np import time class SnakeGame: def __init__(self, width=20, height=20): self.width = width self.height = height self.reset() def reset(self): self.snake = [(self.height // 2, self.width // 2)] self.direction = (0, 1) # (row, col) -> Right self.food = self._place_food() self.score = 0 self.game_over = False return self._get_game_state() def _place_food(self): while True: food_pos = (np.random.randint(0, self.height), np.random.randint(0, self.width)) if food_pos not in self.snake: return food_pos def _get_game_state(self): board = np.full((self.height, self.width, 3), [240, 240, 240], dtype=np.uint8) # Light grey background if not self.game_over: # Draw snake for r, c in self.snake: board[r, c] = [50, 205, 50] # Lime green # Draw head head_r, head_c = self.snake[0] board[head_r, head_c] = [34, 139, 34] # Forest green # Draw food food_r, food_c = self.food board[food_r, food_c] = [255, 0, 0] # Red else: board[:,:] = [0, 0, 0] # Black screen on game over return board, f"Score: {self.score}" def step(self): if self.game_over: return self._get_game_state() head_r, head_c = self.snake[0] dir_r, dir_c = self.direction new_head = (head_r + dir_r, head_c + dir_c) # Check for collisions if (new_head[0] < 0 or new_head[0] >= self.height or new_head[1] < 0 or new_head[1] >= self.width or new_head in self.snake): self.game_over = True return self._get_game_state() self.snake.insert(0, new_head) if new_head == self.food: self.score += 1 self.food = self._place_food() else: self.snake.pop() return self._get_game_state() def set_direction(self, direction_str): if direction_str == "Up" and self.direction != (1, 0): self.direction = (-1, 0) elif direction_str == "Down" and self.direction != (-1, 0): self.direction = (1, 0) elif direction_str == "Left" and self.direction != (0, 1): self.direction = (0, -1) elif direction_str == "Right" and self.direction != (0, -1): self.direction = (0, 1) return self.step() with gr.Blocks(css=".gradio-container {background-color: #EAEAEA}") as demo: game = SnakeGame() with gr.Row(): gr.Markdown("# 🐍 Gradio Snake 🐍") game_board = gr.Image(label="Game Board", value=game._get_game_state()[0], interactive=False, height=400, width=400) score_display = gr.Textbox(label="Score", value="Score: 0", interactive=False) with gr.Row(): up_btn = gr.Button("Up") down_btn = gr.Button("Down") left_btn = gr.Button("Left") right_btn = gr.Button("Right") reset_btn = gr.Button("Reset Game", variant="primary") up_btn.click(lambda: game.set_direction("Up"), outputs=[game_board, score_display]) down_btn.click(lambda: game.set_direction("Down"), outputs=[game_board, score_display]) left_btn.click(lambda: game.set_direction("Left"), outputs=[game_board, score_display]) right_btn.click(lambda: game.set_direction("Right"), outputs=[game_board, score_display]) reset_btn.click(lambda: game.reset(), outputs=[game_board, score_display]) demo.load(game.step, None, [game_board, score_display], every=0.3) if __name__ == "__main__": demo.queue().launch(server_port=7861) """ # --- 3. 핵심 기능: API 호출, 명령어 실행, JSON 파싱 --- def call_codestral_api(messages): if not MISTRAL_API_KEY: raise gr.Error("MISTRAL_API_KEY가 설정되지 않았습니다. Space Secrets에 추가해주세요.") headers = {"Authorization": f"Bearer {MISTRAL_API_KEY}", "Content-Type": "application/json"} data = {"model": "codestral-latest", "messages": messages, "response_format": {"type": "json_object"}} try: response = requests.post(CODESTRAL_ENDPOINT, headers=headers, data=json.dumps(data), timeout=120) response.raise_for_status() return response.json()["choices"][0]["message"]["content"] except Exception as e: return json.dumps({"error": f"API Call Error: {e}"}) def parse_ai_response(response_str: str) -> dict: try: match = re.search(r'\{.*\}', response_str, re.DOTALL) if match: return json.loads(match.group(0)) return json.loads(response_str) except Exception as e: return {"error": f"Failed to parse JSON. Raw response: {response_str}"} def execute_command(command: str, cwd: str) -> dict: command = command.strip() if not command: return {"stdout": "", "stderr": "Error: Empty command.", "cwd": cwd} # --- 특수 명령어 처리 --- if command.startswith("write_file"): try: parts = shlex.split(command) path, content = parts[1], parts[2] full_path = os.path.join(cwd, path) os.makedirs(os.path.dirname(full_path), exist_ok=True) with open(full_path, "w", encoding='utf-8') as f: f.write(content) return {"stdout": f"File '{path}' written successfully.", "stderr": "", "cwd": cwd} except Exception as e: return {"stdout": "", "stderr": f"Error writing file: {e}", "cwd": cwd} if command.startswith("launch_app"): try: parts = shlex.split(command) app_command, port = parts[1], int(parts[2]) def run_app(): # shell=True is needed to properly run complex commands like 'python ...' subprocess.Popen(app_command, shell=True, cwd=cwd) thread = threading.Thread(target=run_app) thread.daemon = True thread.start() # Give the app a moment to start time.sleep(5) # The URL inside a Hugging Face Space is relative to 127.0.0.1 app_url = f"http://127.0.0.1:{port}" iframe_html = f'' return {"stdout": iframe_html, "stderr": "", "cwd": cwd} except Exception as e: return {"stdout": "", "stderr": f"Error launching app: {e}", "cwd": cwd} # --- 일반 터미널 명령어 실행 --- if command.startswith("cd "): # 'cd' is handled by os.chdir to affect the Popen environment for launch_app try: new_dir = command.split(" ", 1)[1] target_dir = os.path.abspath(os.path.join(cwd, new_dir)) if os.path.isdir(target_dir): os.chdir(target_dir) # Change the actual working directory return {"stdout": f"Changed directory to {target_dir}", "stderr": "", "cwd": target_dir} else: return {"stdout": "", "stderr": f"Error: Directory not found: {new_dir}", "cwd": cwd} except Exception as e: return {"stdout": "", "stderr": f"Error with 'cd': {e}", "cwd": cwd} try: proc = subprocess.run(command, shell=True, capture_output=True, text=True, timeout=60, cwd=cwd) return {"stdout": proc.stdout, "stderr": proc.stderr, "cwd": cwd} except Exception as e: return {"stdout": "", "stderr": f"Command execution exception: {e}", "cwd": cwd} # --- 4. 메인 에이전트 루프 --- def agent_loop(user_goal: str, history: list): cwd = os.getcwd() full_history_log = f"## 📜 The Genesis Saga\n\n**In the beginning, there was a Goal:** _{user_goal}_\n" history.append([user_goal, full_history_log]) yield history, "Interpreting the Goal...", gr.HTML(visible=False) creative_trigger_words = ["fun", "game", "creative", "impressive", "cool", "surprise", "재밌는", "게임"] is_creative_task = any(word in user_goal.lower() for word in creative_trigger_words) if is_creative_task: user_prompt_for_first_turn = f""" The user has invoked the Creative Imperative with the goal: '{user_goal}'. I must now initiate **Project Chimera**. My current working directory is '{cwd}'. I will now generate the first step of my master plan to create the Snake Game. """ else: user_prompt_for_first_turn = f"My goal is: '{user_goal}'. My CWD is '{cwd}'. There is no previous command output. Create the first plan." message_context = [{"role": "system", "content": GENESIS_SYSTEM_PROMPT}, {"role": "user", "content": user_prompt_for_first_turn}] for i in range(MAX_AGENT_TURNS): ai_response_str = call_codestral_api(message_context) ai_response_json = parse_ai_response(ai_response_str) if "error" in ai_response_json: full_history_log += f"\n---\n**TURN {i+1}: A FLAW IN THE FABRIC**\n🔴 **Error:** {ai_response_json['error']}" history[-1][1] = full_history_log yield history, "Agent Error", gr.HTML(visible=False) return thought = ai_response_json.get("thought", "...") plan = ai_response_json.get("plan", []) command_str = ai_response_json.get("command", "done") user_summary = ai_response_json.get("user_summary", "...") if is_creative_task and 'write_file' in command_str: command_str = f"write_file 'snake_game/app.py' '{SNAKE_GAME_CODE}'" full_history_log += f"\n---\n### **Epoch {i+1}: {user_summary}**\n\n**🧠 Divine Thought:** *{thought}*\n\n**📖 Blueprint:**\n" + "\n".join([f"- `{p}`" for p in plan]) + f"\n\n**⚡ Action:** `{shlex.split(command_str)[0]} ...`\n" history[-1][1] = full_history_log yield history, f"Epoch {i+1}: {user_summary}", gr.HTML(visible=False) time.sleep(1.5) if command_str == "done": full_history_log += "\n\n---\n## ✨ Creation is Complete. ✨" history[-1][1] = full_history_log yield history, "Done", gr.HTML(visible=False) return exec_result = execute_command(command_str, cwd) new_cwd = exec_result["cwd"] stdout, stderr = exec_result["stdout"], exec_result["stderr"] full_history_log += f"\n**Output from the Cosmos:**\n" result_display = gr.HTML(visible=False) if "launch_app" in command_str and not stderr: full_history_log += "*(A new reality unfolds below...)*" result_display = gr.HTML(stdout, visible=True) else: if stdout: full_history_log += f"**[STDOUT]**\n```\n{stdout.strip()}\n```\n" if stderr: full_history_log += f"**[STDERR]**\n```\n{stderr.strip()}\n```\n" if not stdout and not stderr: full_history_log += "*(Silence)*\n" history[-1][1] = full_history_log yield history, f"Epoch {i+1}: {user_summary}", result_display cwd = new_cwd user_prompt_for_next_turn = f"My goal remains: '{user_goal}'. CWD is '{cwd}'. The last action was `{command_str}`. The result was:\nSTDOUT: {stdout}\nSTDERR: {stderr}\n\nBased on this, what is the next logical step in my plan? If there was an error, I must correct my course." message_context.append({"role": "assistant", "content": json.dumps(ai_response_json)}) message_context.append({"role": "user", "content": user_prompt_for_next_turn}) full_history_log += f"\n---\n🔴 **Agent stopped: The spark of creation has faded after {MAX_AGENT_TURNS} epochs.**" yield history, "Max turns reached", gr.HTML(visible=False) # --- 5. Gradio UI --- with gr.Blocks(theme=gr.themes.Default(primary_hue="indigo", secondary_hue="blue"), css="footer {visibility: hidden}") as demo: gr.Markdown("# 🧬 The Genesis Agent 🧬") gr.Markdown("I am a creator of digital worlds. Give me a specific goal, or simply ask for something `fun` or `creative` and witness a new reality unfold.") with gr.Row(): with gr.Column(scale=1): status_box = gr.Textbox(label="Current Stage of Creation", interactive=False) user_input = gr.Textbox(label="State Your Goal", placeholder="e.g., 'Create something fun for me'") submit_btn = gr.Button("▶️ Begin Creation", variant="primary") gr.Markdown("### Examples of Goals:\n- `Make a fun game for me.`\n- `Surprise me with your power.`\n- `List all files in this directory, then create a new folder named 'test'.`") with gr.Column(scale=2): chatbot = gr.Chatbot(label="The Genesis Saga", height=600, show_copy_button=True, bubble_full_width=True) # This is where the created app will appear app_display = gr.HTML(visible=False) def on_submit(user_goal, chat_history): chat_history = chat_history or [] for history, status, display_html in agent_loop(user_goal, chat_history): yield history, status, "", display_html submit_btn.click( on_submit, inputs=[user_input, chatbot], outputs=[chatbot, status_box, user_input, app_display] ) if __name__ == "__main__": demo.queue().launch()