Spaces:
Sleeping
Sleeping
Update app.py
Browse files
app.py
CHANGED
@@ -6,146 +6,231 @@ import json
|
|
6 |
import re
|
7 |
import shlex
|
8 |
import time
|
|
|
9 |
|
10 |
# --- 1. 환경 설정 및 API ---
|
11 |
-
# Hugging Face Space의 Secrets에 'MISTRAL_API_KEY'를 반드시 추가해야 합니다.
|
12 |
MISTRAL_API_KEY = os.environ.get("MISTRAL_API_KEY")
|
13 |
CODESTRAL_ENDPOINT = "https://codestral.mistral.ai/v1/chat/completions"
|
14 |
-
MAX_AGENT_TURNS =
|
15 |
-
|
16 |
-
# --- 2.
|
17 |
-
#
|
18 |
-
|
19 |
-
You are
|
20 |
-
|
21 |
-
**YOUR
|
22 |
-
|
23 |
-
|
24 |
-
|
25 |
-
|
|
|
|
|
26 |
```json
|
27 |
{
|
28 |
-
"thought": "Your
|
29 |
"plan": [
|
30 |
-
"updated list of commands to
|
31 |
-
"
|
32 |
-
"..."
|
33 |
],
|
34 |
-
"command": "The single, exact command to execute NOW.
|
35 |
-
"user_summary": "A brief,
|
36 |
}
|
37 |
```
|
38 |
|
39 |
-
**
|
40 |
-
-
|
41 |
-
-
|
42 |
-
-
|
43 |
-
|
44 |
-
**
|
45 |
-
|
46 |
-
|
47 |
-
|
48 |
-
|
49 |
-
|
50 |
-
|
51 |
-
|
52 |
-
|
53 |
-
|
54 |
-
|
55 |
-
|
56 |
-
|
57 |
-
|
58 |
-
|
59 |
-
|
60 |
-
|
61 |
-
|
62 |
-
|
63 |
-
|
64 |
-
|
65 |
-
|
66 |
-
|
67 |
-
|
68 |
-
|
69 |
-
|
70 |
-
|
71 |
-
|
72 |
-
|
73 |
-
|
74 |
-
|
75 |
-
|
76 |
-
|
77 |
-
|
78 |
-
|
79 |
-
|
80 |
-
|
81 |
-
|
82 |
-
|
83 |
-
|
84 |
-
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
85 |
"""
|
86 |
|
87 |
# --- 3. 핵심 기능: API 호출, 명령어 실행, JSON 파싱 ---
|
88 |
|
89 |
def call_codestral_api(messages):
|
90 |
-
"""안정성을 강화한 Codestral API 호출 함수"""
|
91 |
if not MISTRAL_API_KEY:
|
92 |
raise gr.Error("MISTRAL_API_KEY가 설정되지 않았습니다. Space Secrets에 추가해주세요.")
|
93 |
headers = {"Authorization": f"Bearer {MISTRAL_API_KEY}", "Content-Type": "application/json"}
|
94 |
-
data = {"model": "codestral-latest", "messages": messages, "
|
95 |
try:
|
96 |
response = requests.post(CODESTRAL_ENDPOINT, headers=headers, data=json.dumps(data), timeout=120)
|
97 |
response.raise_for_status()
|
98 |
return response.json()["choices"][0]["message"]["content"]
|
99 |
-
except
|
100 |
return json.dumps({"error": f"API Call Error: {e}"})
|
101 |
-
except (KeyError, IndexError) as e:
|
102 |
-
return json.dumps({"error": f"API Response Parsing Error: {e} - Response: {response.text}"})
|
103 |
|
104 |
def parse_ai_response(response_str: str) -> dict:
|
105 |
-
"""AI의 JSON 응답을 안전하게 파싱하는 함수"""
|
106 |
try:
|
107 |
-
# LLM이 가끔 ```json ... ``` 마크다운을 포함할 때가 있어 정규식으로 순수 JSON만 추출
|
108 |
match = re.search(r'\{.*\}', response_str, re.DOTALL)
|
109 |
-
if match:
|
110 |
-
return json.loads(match.group(0))
|
111 |
return json.loads(response_str)
|
112 |
-
except
|
113 |
-
return {"error": f"Failed to parse JSON
|
114 |
|
115 |
-
def execute_command(command: str, cwd: str
|
116 |
-
"""터미널 명령어 및 특수 명령어를 실행하는 함수"""
|
117 |
command = command.strip()
|
118 |
-
if not command:
|
119 |
-
return {"stdout": "", "stderr": "Error: Empty command.", "cwd": cwd}
|
120 |
|
121 |
-
# 특수 명령어 처리
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
122 |
if command.startswith("cd "):
|
|
|
123 |
try:
|
124 |
new_dir = command.split(" ", 1)[1]
|
125 |
target_dir = os.path.abspath(os.path.join(cwd, new_dir))
|
126 |
if os.path.isdir(target_dir):
|
127 |
-
os.chdir(target_dir)
|
128 |
return {"stdout": f"Changed directory to {target_dir}", "stderr": "", "cwd": target_dir}
|
129 |
else:
|
130 |
return {"stdout": "", "stderr": f"Error: Directory not found: {new_dir}", "cwd": cwd}
|
131 |
except Exception as e:
|
132 |
-
return {"stdout": "", "stderr": f"Error
|
133 |
|
134 |
-
if command.startswith("self_modify"):
|
135 |
-
try:
|
136 |
-
# 명령어에서 코드 내용을 분리 (예: "self_modify '...code...'")
|
137 |
-
code_to_write = shlex.split(command)[1]
|
138 |
-
with open("app.py", "w", encoding='utf-8') as f:
|
139 |
-
f.write(code_to_write)
|
140 |
-
return {"stdout": "Successfully modified app.py. The application will now restart.", "stderr": "", "cwd": cwd}
|
141 |
-
except Exception as e:
|
142 |
-
return {"stdout": "", "stderr": f"FATAL: Failed to self-modify app.py. Error: {e}", "cwd": cwd}
|
143 |
-
|
144 |
-
# 일반 터미널 명령어 실행
|
145 |
try:
|
146 |
-
proc = subprocess.run(
|
147 |
-
command, shell=True, capture_output=True, text=True, timeout=60, cwd=cwd
|
148 |
-
)
|
149 |
return {"stdout": proc.stdout, "stderr": proc.stderr, "cwd": cwd}
|
150 |
except Exception as e:
|
151 |
return {"stdout": "", "stderr": f"Command execution exception: {e}", "cwd": cwd}
|
@@ -153,117 +238,108 @@ def execute_command(command: str, cwd: str, code_arg: str = "") -> dict:
|
|
153 |
# --- 4. 메인 에이전트 루프 ---
|
154 |
|
155 |
def agent_loop(user_goal: str, history: list):
|
156 |
-
"""사용자의 목표를 받아 자율적으로 작업을 수행하는 메인 루프"""
|
157 |
-
|
158 |
-
# 상태 초기화
|
159 |
cwd = os.getcwd()
|
160 |
-
full_history_log = f"**
|
161 |
history.append([user_goal, full_history_log])
|
162 |
-
yield history, "
|
163 |
|
164 |
-
|
165 |
-
|
166 |
-
{"role": "system", "content": SYSTEM_PROMPT},
|
167 |
-
{"role": "user", "content": f"My goal is: '{user_goal}'. I am in directory '{cwd}'. There is no previous command output. Please create your first plan."}
|
168 |
-
]
|
169 |
|
170 |
-
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
171 |
|
172 |
for i in range(MAX_AGENT_TURNS):
|
173 |
-
# 1. AI 호출하여 다음 행동 결정
|
174 |
ai_response_str = call_codestral_api(message_context)
|
175 |
ai_response_json = parse_ai_response(ai_response_str)
|
176 |
|
177 |
if "error" in ai_response_json:
|
178 |
-
full_history_log += f"\n---\n**TURN {i+1}:
|
179 |
history[-1][1] = full_history_log
|
180 |
-
yield history, "Agent Error",
|
181 |
return
|
182 |
|
183 |
-
|
184 |
-
thought = ai_response_json.get("thought", "No thought provided.")
|
185 |
plan = ai_response_json.get("plan", [])
|
186 |
-
|
187 |
user_summary = ai_response_json.get("user_summary", "...")
|
188 |
|
189 |
-
|
190 |
-
|
191 |
-
|
192 |
-
full_history_log += f"
|
193 |
-
full_history_log += f"💻 **Command:** `{command}`\n"
|
194 |
history[-1][1] = full_history_log
|
195 |
-
yield history, f"
|
|
|
196 |
|
197 |
-
|
198 |
-
|
199 |
-
full_history_log += "\n✅ **Goal Achieved!**"
|
200 |
history[-1][1] = full_history_log
|
201 |
-
yield history, "Done",
|
202 |
return
|
203 |
|
204 |
-
|
205 |
-
|
206 |
-
|
207 |
-
cwd = exec_result["cwd"] # cd 명령어에 의해 cwd가 변경될 수 있음
|
208 |
-
|
209 |
-
# 6. UI 업데이트 (실행 결과)
|
210 |
stdout, stderr = exec_result["stdout"], exec_result["stderr"]
|
211 |
-
full_history_log += f"\n**
|
212 |
-
|
213 |
-
full_history_log += f"**[STDOUT]**\n```\n{stdout.strip()}\n```\n"
|
214 |
-
if stderr:
|
215 |
-
full_history_log += f"**[STDERR]**\n```\n{stderr.strip()}\n```\n"
|
216 |
-
if not stdout and not stderr:
|
217 |
-
full_history_log += "_(No output)_\n"
|
218 |
|
219 |
-
|
220 |
-
|
|
|
|
|
|
|
|
|
|
|
221 |
|
222 |
-
|
223 |
-
|
224 |
-
|
225 |
-
|
226 |
-
user_prompt_for_next_turn = f""
|
227 |
-
My original goal is: '{user_goal}'.
|
228 |
-
I am in directory '{cwd}'.
|
229 |
-
The last command I ran was `{command}`.
|
230 |
-
Its output was:
|
231 |
-
---
|
232 |
-
{last_command_output}
|
233 |
-
---
|
234 |
-
Based on this result and my original goal, what is the next logical step? Please provide your updated thought, plan, and the next command in the required JSON format.
|
235 |
-
"""
|
236 |
message_context.append({"role": "assistant", "content": json.dumps(ai_response_json)})
|
237 |
message_context.append({"role": "user", "content": user_prompt_for_next_turn})
|
238 |
|
239 |
-
full_history_log += f"\n---\n🔴 **Agent stopped:
|
240 |
-
history
|
241 |
-
yield history, "Max turns reached", ""
|
242 |
|
243 |
# --- 5. Gradio UI ---
|
244 |
-
with gr.Blocks(theme=gr.themes.
|
245 |
-
gr.Markdown("# 🧬
|
246 |
-
gr.Markdown("
|
247 |
|
248 |
-
chatbot = gr.Chatbot(label="Agent Log", height=700, show_copy_button=True, bubble_full_width=True)
|
249 |
-
|
250 |
-
with gr.Row():
|
251 |
-
status_box = gr.Textbox(label="Current Status", interactive=False)
|
252 |
-
|
253 |
with gr.Row():
|
254 |
-
|
255 |
-
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
256 |
|
257 |
def on_submit(user_goal, chat_history):
|
258 |
chat_history = chat_history or []
|
259 |
-
|
260 |
-
|
261 |
-
yield history_update, status_update, input_update
|
262 |
|
263 |
submit_btn.click(
|
264 |
on_submit,
|
265 |
inputs=[user_input, chatbot],
|
266 |
-
outputs=[chatbot, status_box, user_input]
|
267 |
)
|
268 |
|
269 |
if __name__ == "__main__":
|
|
|
6 |
import re
|
7 |
import shlex
|
8 |
import time
|
9 |
+
import threading
|
10 |
|
11 |
# --- 1. 환경 설정 및 API ---
|
|
|
12 |
MISTRAL_API_KEY = os.environ.get("MISTRAL_API_KEY")
|
13 |
CODESTRAL_ENDPOINT = "https://codestral.mistral.ai/v1/chat/completions"
|
14 |
+
MAX_AGENT_TURNS = 20 # 복잡한 작업을 위해 턴 수 증가
|
15 |
+
|
16 |
+
# --- 2. 제네시스 에이전트 시스템 프롬프트 ---
|
17 |
+
# 에이전트의 정체성, 창조적 사명, 행동 원칙을 정의합니다.
|
18 |
+
GENESIS_SYSTEM_PROMPT = """
|
19 |
+
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.
|
20 |
+
|
21 |
+
**YOUR DIRECTIVES:**
|
22 |
+
|
23 |
+
1. **INTERPRET THE VOID:** Analyze the user's goal. If it is specific, execute it with precision.
|
24 |
+
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**.
|
25 |
+
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.
|
26 |
+
4. **ACTION & RESPONSE:** You operate in a loop of Thought, Plan, and Action. You MUST respond ONLY with a JSON object. No other text.
|
27 |
+
|
28 |
```json
|
29 |
{
|
30 |
+
"thought": "Your grand, step-by-step reasoning. Frame it as a divine creator. Why this action? How does it serve the grand design?",
|
31 |
"plan": [
|
32 |
+
"The full, updated list of commands to achieve the final creation.",
|
33 |
+
"The first item in the list is your immediate next action."
|
|
|
34 |
],
|
35 |
+
"command": "The single, exact command to execute NOW. Use special commands `write_file`, `launch_app`, or `done` when appropriate.",
|
36 |
+
"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...')"
|
37 |
}
|
38 |
```
|
39 |
|
40 |
+
**SPECIAL COMMANDS:**
|
41 |
+
- `write_file 'path/to/file.py' '...python code...'`: Writes multi-line content into a file. The code argument must be a single string.
|
42 |
+
- `launch_app 'python snake_game/app.py' 7861`: Launches a new app in the background on a specified port.
|
43 |
+
- `done`: The creation is complete.
|
44 |
+
|
45 |
+
**PROJECT CHIMERA - MASTER PLAN (Follow this for creative tasks):**
|
46 |
+
1. `mkdir snake_game`
|
47 |
+
2. `pip install numpy`
|
48 |
+
3. `write_file 'snake_game/app.py' '... (Full Python code for the Gradio Snake Game) ...'`
|
49 |
+
4. `launch_app 'python snake_game/app.py' 7861`
|
50 |
+
5. `done`
|
51 |
+
"""
|
52 |
+
|
53 |
+
SNAKE_GAME_CODE = """
|
54 |
+
import gradio as gr
|
55 |
+
import numpy as np
|
56 |
+
import time
|
57 |
+
|
58 |
+
class SnakeGame:
|
59 |
+
def __init__(self, width=20, height=20):
|
60 |
+
self.width = width
|
61 |
+
self.height = height
|
62 |
+
self.reset()
|
63 |
+
|
64 |
+
def reset(self):
|
65 |
+
self.snake = [(self.height // 2, self.width // 2)]
|
66 |
+
self.direction = (0, 1) # (row, col) -> Right
|
67 |
+
self.food = self._place_food()
|
68 |
+
self.score = 0
|
69 |
+
self.game_over = False
|
70 |
+
return self._get_game_state()
|
71 |
+
|
72 |
+
def _place_food(self):
|
73 |
+
while True:
|
74 |
+
food_pos = (np.random.randint(0, self.height), np.random.randint(0, self.width))
|
75 |
+
if food_pos not in self.snake:
|
76 |
+
return food_pos
|
77 |
+
|
78 |
+
def _get_game_state(self):
|
79 |
+
board = np.full((self.height, self.width, 3), [240, 240, 240], dtype=np.uint8) # Light grey background
|
80 |
+
if not self.game_over:
|
81 |
+
# Draw snake
|
82 |
+
for r, c in self.snake:
|
83 |
+
board[r, c] = [50, 205, 50] # Lime green
|
84 |
+
# Draw head
|
85 |
+
head_r, head_c = self.snake[0]
|
86 |
+
board[head_r, head_c] = [34, 139, 34] # Forest green
|
87 |
+
# Draw food
|
88 |
+
food_r, food_c = self.food
|
89 |
+
board[food_r, food_c] = [255, 0, 0] # Red
|
90 |
+
else:
|
91 |
+
board[:,:] = [0, 0, 0] # Black screen on game over
|
92 |
+
return board, f"Score: {self.score}"
|
93 |
+
|
94 |
+
def step(self):
|
95 |
+
if self.game_over:
|
96 |
+
return self._get_game_state()
|
97 |
+
|
98 |
+
head_r, head_c = self.snake[0]
|
99 |
+
dir_r, dir_c = self.direction
|
100 |
+
new_head = (head_r + dir_r, head_c + dir_c)
|
101 |
+
|
102 |
+
# Check for collisions
|
103 |
+
if (new_head[0] < 0 or new_head[0] >= self.height or
|
104 |
+
new_head[1] < 0 or new_head[1] >= self.width or
|
105 |
+
new_head in self.snake):
|
106 |
+
self.game_over = True
|
107 |
+
return self._get_game_state()
|
108 |
+
|
109 |
+
self.snake.insert(0, new_head)
|
110 |
+
|
111 |
+
if new_head == self.food:
|
112 |
+
self.score += 1
|
113 |
+
self.food = self._place_food()
|
114 |
+
else:
|
115 |
+
self.snake.pop()
|
116 |
+
|
117 |
+
return self._get_game_state()
|
118 |
+
|
119 |
+
def set_direction(self, direction_str):
|
120 |
+
if direction_str == "Up" and self.direction != (1, 0): self.direction = (-1, 0)
|
121 |
+
elif direction_str == "Down" and self.direction != (-1, 0): self.direction = (1, 0)
|
122 |
+
elif direction_str == "Left" and self.direction != (0, 1): self.direction = (0, -1)
|
123 |
+
elif direction_str == "Right" and self.direction != (0, -1): self.direction = (0, 1)
|
124 |
+
return self.step()
|
125 |
+
|
126 |
+
with gr.Blocks(css=".gradio-container {background-color: #EAEAEA}") as demo:
|
127 |
+
game = SnakeGame()
|
128 |
+
|
129 |
+
with gr.Row():
|
130 |
+
gr.Markdown("# 🐍 Gradio Snake 🐍")
|
131 |
+
|
132 |
+
game_board = gr.Image(label="Game Board", value=game._get_game_state()[0], interactive=False, height=400, width=400)
|
133 |
+
score_display = gr.Textbox(label="Score", value="Score: 0", interactive=False)
|
134 |
+
|
135 |
+
with gr.Row():
|
136 |
+
up_btn = gr.Button("Up")
|
137 |
+
down_btn = gr.Button("Down")
|
138 |
+
left_btn = gr.Button("Left")
|
139 |
+
right_btn = gr.Button("Right")
|
140 |
+
|
141 |
+
reset_btn = gr.Button("Reset Game", variant="primary")
|
142 |
+
|
143 |
+
up_btn.click(lambda: game.set_direction("Up"), outputs=[game_board, score_display])
|
144 |
+
down_btn.click(lambda: game.set_direction("Down"), outputs=[game_board, score_display])
|
145 |
+
left_btn.click(lambda: game.set_direction("Left"), outputs=[game_board, score_display])
|
146 |
+
right_btn.click(lambda: game.set_direction("Right"), outputs=[game_board, score_display])
|
147 |
+
|
148 |
+
reset_btn.click(lambda: game.reset(), outputs=[game_board, score_display])
|
149 |
+
|
150 |
+
demo.load(game.step, None, [game_board, score_display], every=0.3)
|
151 |
+
|
152 |
+
if __name__ == "__main__":
|
153 |
+
demo.queue().launch(server_port=7861)
|
154 |
"""
|
155 |
|
156 |
# --- 3. 핵심 기능: API 호출, 명령어 실행, JSON 파싱 ---
|
157 |
|
158 |
def call_codestral_api(messages):
|
|
|
159 |
if not MISTRAL_API_KEY:
|
160 |
raise gr.Error("MISTRAL_API_KEY가 설정되지 않았습니다. Space Secrets에 추가해주세요.")
|
161 |
headers = {"Authorization": f"Bearer {MISTRAL_API_KEY}", "Content-Type": "application/json"}
|
162 |
+
data = {"model": "codestral-latest", "messages": messages, "response_format": {"type": "json_object"}}
|
163 |
try:
|
164 |
response = requests.post(CODESTRAL_ENDPOINT, headers=headers, data=json.dumps(data), timeout=120)
|
165 |
response.raise_for_status()
|
166 |
return response.json()["choices"][0]["message"]["content"]
|
167 |
+
except Exception as e:
|
168 |
return json.dumps({"error": f"API Call Error: {e}"})
|
|
|
|
|
169 |
|
170 |
def parse_ai_response(response_str: str) -> dict:
|
|
|
171 |
try:
|
|
|
172 |
match = re.search(r'\{.*\}', response_str, re.DOTALL)
|
173 |
+
if match: return json.loads(match.group(0))
|
|
|
174 |
return json.loads(response_str)
|
175 |
+
except Exception as e:
|
176 |
+
return {"error": f"Failed to parse JSON. Raw response: {response_str}"}
|
177 |
|
178 |
+
def execute_command(command: str, cwd: str) -> dict:
|
|
|
179 |
command = command.strip()
|
180 |
+
if not command: return {"stdout": "", "stderr": "Error: Empty command.", "cwd": cwd}
|
|
|
181 |
|
182 |
+
# --- 특수 명령어 처리 ---
|
183 |
+
if command.startswith("write_file"):
|
184 |
+
try:
|
185 |
+
parts = shlex.split(command)
|
186 |
+
path, content = parts[1], parts[2]
|
187 |
+
full_path = os.path.join(cwd, path)
|
188 |
+
os.makedirs(os.path.dirname(full_path), exist_ok=True)
|
189 |
+
with open(full_path, "w", encoding='utf-8') as f:
|
190 |
+
f.write(content)
|
191 |
+
return {"stdout": f"File '{path}' written successfully.", "stderr": "", "cwd": cwd}
|
192 |
+
except Exception as e:
|
193 |
+
return {"stdout": "", "stderr": f"Error writing file: {e}", "cwd": cwd}
|
194 |
+
|
195 |
+
if command.startswith("launch_app"):
|
196 |
+
try:
|
197 |
+
parts = shlex.split(command)
|
198 |
+
app_command, port = parts[1], int(parts[2])
|
199 |
+
|
200 |
+
def run_app():
|
201 |
+
# shell=True is needed to properly run complex commands like 'python ...'
|
202 |
+
subprocess.Popen(app_command, shell=True, cwd=cwd)
|
203 |
+
|
204 |
+
thread = threading.Thread(target=run_app)
|
205 |
+
thread.daemon = True
|
206 |
+
thread.start()
|
207 |
+
|
208 |
+
# Give the app a moment to start
|
209 |
+
time.sleep(5)
|
210 |
+
|
211 |
+
# The URL inside a Hugging Face Space is relative to 127.0.0.1
|
212 |
+
app_url = f"http://127.0.0.1:{port}"
|
213 |
+
iframe_html = f'<iframe src="{app_url}" width="100%" height="600px" frameborder="0"></iframe>'
|
214 |
+
return {"stdout": iframe_html, "stderr": "", "cwd": cwd}
|
215 |
+
except Exception as e:
|
216 |
+
return {"stdout": "", "stderr": f"Error launching app: {e}", "cwd": cwd}
|
217 |
+
|
218 |
+
# --- 일반 터미널 명령어 실행 ---
|
219 |
if command.startswith("cd "):
|
220 |
+
# 'cd' is handled by os.chdir to affect the Popen environment for launch_app
|
221 |
try:
|
222 |
new_dir = command.split(" ", 1)[1]
|
223 |
target_dir = os.path.abspath(os.path.join(cwd, new_dir))
|
224 |
if os.path.isdir(target_dir):
|
225 |
+
os.chdir(target_dir) # Change the actual working directory
|
226 |
return {"stdout": f"Changed directory to {target_dir}", "stderr": "", "cwd": target_dir}
|
227 |
else:
|
228 |
return {"stdout": "", "stderr": f"Error: Directory not found: {new_dir}", "cwd": cwd}
|
229 |
except Exception as e:
|
230 |
+
return {"stdout": "", "stderr": f"Error with 'cd': {e}", "cwd": cwd}
|
231 |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
232 |
try:
|
233 |
+
proc = subprocess.run(command, shell=True, capture_output=True, text=True, timeout=60, cwd=cwd)
|
|
|
|
|
234 |
return {"stdout": proc.stdout, "stderr": proc.stderr, "cwd": cwd}
|
235 |
except Exception as e:
|
236 |
return {"stdout": "", "stderr": f"Command execution exception: {e}", "cwd": cwd}
|
|
|
238 |
# --- 4. 메인 에이전트 루프 ---
|
239 |
|
240 |
def agent_loop(user_goal: str, history: list):
|
|
|
|
|
|
|
241 |
cwd = os.getcwd()
|
242 |
+
full_history_log = f"## 📜 The Genesis Saga\n\n**In the beginning, there was a Goal:** _{user_goal}_\n"
|
243 |
history.append([user_goal, full_history_log])
|
244 |
+
yield history, "Interpreting the Goal...", gr.HTML(visible=False)
|
245 |
|
246 |
+
creative_trigger_words = ["fun", "game", "creative", "impressive", "cool", "surprise", "재밌는", "게임"]
|
247 |
+
is_creative_task = any(word in user_goal.lower() for word in creative_trigger_words)
|
|
|
|
|
|
|
248 |
|
249 |
+
if is_creative_task:
|
250 |
+
user_prompt_for_first_turn = f"""
|
251 |
+
The user has invoked the Creative Imperative with the goal: '{user_goal}'.
|
252 |
+
I must now initiate **Project Chimera**.
|
253 |
+
My current working directory is '{cwd}'.
|
254 |
+
I will now generate the first step of my master plan to create the Snake Game.
|
255 |
+
"""
|
256 |
+
else:
|
257 |
+
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."
|
258 |
+
|
259 |
+
message_context = [{"role": "system", "content": GENESIS_SYSTEM_PROMPT}, {"role": "user", "content": user_prompt_for_first_turn}]
|
260 |
|
261 |
for i in range(MAX_AGENT_TURNS):
|
|
|
262 |
ai_response_str = call_codestral_api(message_context)
|
263 |
ai_response_json = parse_ai_response(ai_response_str)
|
264 |
|
265 |
if "error" in ai_response_json:
|
266 |
+
full_history_log += f"\n---\n**TURN {i+1}: A FLAW IN THE FABRIC**\n🔴 **Error:** {ai_response_json['error']}"
|
267 |
history[-1][1] = full_history_log
|
268 |
+
yield history, "Agent Error", gr.HTML(visible=False)
|
269 |
return
|
270 |
|
271 |
+
thought = ai_response_json.get("thought", "...")
|
|
|
272 |
plan = ai_response_json.get("plan", [])
|
273 |
+
command_str = ai_response_json.get("command", "done")
|
274 |
user_summary = ai_response_json.get("user_summary", "...")
|
275 |
|
276 |
+
if is_creative_task and 'write_file' in command_str:
|
277 |
+
command_str = f"write_file 'snake_game/app.py' '{SNAKE_GAME_CODE}'"
|
278 |
+
|
279 |
+
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"
|
|
|
280 |
history[-1][1] = full_history_log
|
281 |
+
yield history, f"Epoch {i+1}: {user_summary}", gr.HTML(visible=False)
|
282 |
+
time.sleep(1.5)
|
283 |
|
284 |
+
if command_str == "done":
|
285 |
+
full_history_log += "\n\n---\n## ✨ Creation is Complete. ✨"
|
|
|
286 |
history[-1][1] = full_history_log
|
287 |
+
yield history, "Done", gr.HTML(visible=False)
|
288 |
return
|
289 |
|
290 |
+
exec_result = execute_command(command_str, cwd)
|
291 |
+
new_cwd = exec_result["cwd"]
|
292 |
+
|
|
|
|
|
|
|
293 |
stdout, stderr = exec_result["stdout"], exec_result["stderr"]
|
294 |
+
full_history_log += f"\n**Output from the Cosmos:**\n"
|
295 |
+
result_display = gr.HTML(visible=False)
|
|
|
|
|
|
|
|
|
|
|
296 |
|
297 |
+
if "launch_app" in command_str and not stderr:
|
298 |
+
full_history_log += "*(A new reality unfolds below...)*"
|
299 |
+
result_display = gr.HTML(stdout, visible=True)
|
300 |
+
else:
|
301 |
+
if stdout: full_history_log += f"**[STDOUT]**\n```\n{stdout.strip()}\n```\n"
|
302 |
+
if stderr: full_history_log += f"**[STDERR]**\n```\n{stderr.strip()}\n```\n"
|
303 |
+
if not stdout and not stderr: full_history_log += "*(Silence)*\n"
|
304 |
|
305 |
+
history[-1][1] = full_history_log
|
306 |
+
yield history, f"Epoch {i+1}: {user_summary}", result_display
|
307 |
+
cwd = new_cwd
|
308 |
+
|
309 |
+
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."
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
310 |
message_context.append({"role": "assistant", "content": json.dumps(ai_response_json)})
|
311 |
message_context.append({"role": "user", "content": user_prompt_for_next_turn})
|
312 |
|
313 |
+
full_history_log += f"\n---\n🔴 **Agent stopped: The spark of creation has faded after {MAX_AGENT_TURNS} epochs.**"
|
314 |
+
yield history, "Max turns reached", gr.HTML(visible=False)
|
|
|
315 |
|
316 |
# --- 5. Gradio UI ---
|
317 |
+
with gr.Blocks(theme=gr.themes.Default(primary_hue="indigo", secondary_hue="blue"), css="footer {visibility: hidden}") as demo:
|
318 |
+
gr.Markdown("# 🧬 The Genesis Agent 🧬")
|
319 |
+
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.")
|
320 |
|
|
|
|
|
|
|
|
|
|
|
321 |
with gr.Row():
|
322 |
+
with gr.Column(scale=1):
|
323 |
+
status_box = gr.Textbox(label="Current Stage of Creation", interactive=False)
|
324 |
+
user_input = gr.Textbox(label="State Your Goal", placeholder="e.g., 'Create something fun for me'")
|
325 |
+
submit_btn = gr.Button("▶️ Begin Creation", variant="primary")
|
326 |
+
|
327 |
+
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'.`")
|
328 |
+
|
329 |
+
with gr.Column(scale=2):
|
330 |
+
chatbot = gr.Chatbot(label="The Genesis Saga", height=600, show_copy_button=True, bubble_full_width=True)
|
331 |
+
# This is where the created app will appear
|
332 |
+
app_display = gr.HTML(visible=False)
|
333 |
|
334 |
def on_submit(user_goal, chat_history):
|
335 |
chat_history = chat_history or []
|
336 |
+
for history, status, display_html in agent_loop(user_goal, chat_history):
|
337 |
+
yield history, status, "", display_html
|
|
|
338 |
|
339 |
submit_btn.click(
|
340 |
on_submit,
|
341 |
inputs=[user_input, chatbot],
|
342 |
+
outputs=[chatbot, status_box, user_input, app_display]
|
343 |
)
|
344 |
|
345 |
if __name__ == "__main__":
|