Update app.py

app.py

@@ -4,194 +4,267 @@ import subprocess
 import requests
 import json
 import re
+import shlex
 import time
 
-# --- 1. 환경 설정 및 강화된 API 호출 함수 ---
+# --- 1. 환경 설정 및 API ---
+# Hugging Face Space의 Secrets에 'MISTRAL_API_KEY'를 반드시 추가해야 합니다.
 MISTRAL_API_KEY = os.environ.get("MISTRAL_API_KEY")
 CODESTRAL_ENDPOINT = "https://codestral.mistral.ai/v1/chat/completions"
+MAX_AGENT_TURNS = 15 # 무한 루프 방지를 위한 최대 실행 횟수
 
-def call_mistral_api(system_prompt: str, user_prompt: str):
-    """향상된 Mistral API 호출 함수"""
+# --- 2. 시스템 프롬프트: 에이전트의 두뇌 ---
+# 이 프롬프트는 에이전트의 행동 원칙, 사고 방식, 출력 형식을 정의하는 가장 중요한 부분입니다.
+SYSTEM_PROMPT = """
+You are a 'Recursive Autonomous Swarm Intelligence' (RASI), a hyper-capable AI agent operating a Linux terminal. Your primary directive is to achieve the user's goal by planning, executing commands, observing results, and correcting your own mistakes.
+
+**YOUR OPERATING PROCEDURE:**
+1.  **THINK:** Analyze the user's request, the conversation history, and the last command's output. Formulate a step-by-step thought process.
+2.  **PLAN:** Based on your thoughts, create or update a list of discrete terminal commands to execute. The plan should be a list of strings.
+3.  **ACT:** Select the *single* next command from your plan to execute.
+4.  **RESPOND:** You MUST respond ONLY with a JSON object. No other text or explanation. The JSON format is:
+    ```json
+    {
+      "thought": "Your detailed, step-by-step reasoning. Explain why you are choosing the next command based on previous results and the overall goal.",
+      "plan": [
+        "updated list of commands to execute",
+        "the first item is what you will do next",
+        "..."
+      ],
+      "command": "The single, exact command to execute NOW. This must be the first item from your 'plan'. Use 'done' when the entire task is complete.",
+      "user_summary": "A brief, human-readable summary of your action for the user interface."
+    }
+    ```
+
+**SELF-CORRECTION DIRECTIVE:**
+- If a command results in an error (`stderr`), your next `thought` MUST be to analyze the error.
+- Create a new plan to fix the error (e.g., if 'file not found', use 'ls' to check; if 'directory exists', use 'cd' instead of 'mkdir').
+- Insert the fix-it steps at the beginning of your plan.
+
+**RECURSIVE SELF-MODIFICATION (DANGER - USE WITH EXTREME CAUTION):**
+- You have the ability to read and write your own source code (`app.py`).
+- This is a last resort for when you are fundamentally unable to solve a problem with standard commands.
+- To modify yourself, use the special command `self_modify` with the full new code as an argument.
+- **Protocol:**
+    1. Read the code first: `cat app.py`
+    2. Analyze the code and decide on the exact change.
+    3. Formulate the `self_modify` command.
+- **Example `command` for self-modification:**
+  `self_modify 'import new_library\n\ndef new_function(): ...'`
+
+**EXAMPLE FLOW:**
+User Goal: "Create a directory 'my_app', and inside it, create a file 'main.py' with a hello world print statement."
+
+1.  **Initial Response:**
+    ```json
+    {
+      "thought": "The user wants a new directory and a file inside it. First, I must create the directory 'my_app'.",
+      "plan": ["mkdir my_app", "echo 'print(\"Hello, World!\")' > my_app/main.py"],
+      "command": "mkdir my_app",
+      "user_summary": "Creating directory `my_app`..."
+    }
+    ```
+2.  **After `mkdir` succeeds:**
+    ```json
+    {
+      "thought": "The directory 'my_app' was created successfully. The next step in my plan is to create 'main.py' inside it.",
+      "plan": ["echo 'print(\"Hello, World!\")' > my_app/main.py"],
+      "command": "echo 'print(\"Hello, World!\")' > my_app/main.py",
+      "user_summary": "Creating file `my_app/main.py`..."
+    }
+    ```
+3.  **After `echo` succeeds:**
+    ```json
+    {
+      "thought": "Both steps are complete. The user's request has been fulfilled.",
+      "plan": [],
+      "command": "done",
+      "user_summary": "Task completed successfully!"
+    }
+    ```
+"""
+
+# --- 3. 핵심 기능: API 호출, 명령어 실행, JSON 파싱 ---
+
+def call_codestral_api(messages):
+    """안정성을 강화한 Codestral API 호출 함수"""
     if not MISTRAL_API_KEY:
-        raise gr.Error("MISTRAL_API_KEY is not set. Please add it to your Space Secrets.")
-    
+        raise gr.Error("MISTRAL_API_KEY가 설정되지 않았습니다. Space Secrets에 추가해주세요.")
     headers = {"Authorization": f"Bearer {MISTRAL_API_KEY}", "Content-Type": "application/json"}
-    messages = [{"role": "system", "content": system_prompt}, {"role": "user", "content": user_prompt}]
-    data = {"model": "codestral-latest", "messages": messages}
-    
+    data = {"model": "codestral-latest", "messages": messages, "temperature": 0.0, "response_format": {"type": "json_object"}}
     try:
-        response = requests.post(CODESTRAL_ENDPOINT, headers=headers, data=json.dumps(data), timeout=60)
+        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 requests.exceptions.RequestException as e:
-        raise gr.Error(f"API Call Error: {e}")
-
-# --- 2. 백엔드 핵심 기능 (안정성 강화) ---
-def parse_code_from_response(response_text: str) -> str | None:
-    """C 코드 블록을 파싱하는 안정적인 함수"""
-    match = re.search(r'```(?:c)?\n(.*?)\n```', response_text, re.DOTALL)
-    if match: return match.group(1).strip()
-    # 비상시 순수 코드 응답 처리
-    if response_text.strip().startswith("#include") and response_text.strip().endswith("}"):
-        return response_text.strip()
-    return None
-
-def generate_c_code(description: str) -> str:
-    system_prompt = "You are an expert C code generator..." # (이전과 동일)
-    user_prompt = f"Generate C code for: '{description}'"
-    response = call_mistral_api(system_prompt, user_prompt)
-    return parse_code_from_response(response) or f"// Failed to parse code from response:\n{response}"
-
-def compile_and_run_c_code(code: str) -> str:
-    """컴파일 및 실행 함수"""
-    if not code.strip(): return "--- SYSTEM ERROR ---\nCode is empty."
-    with open("main.c", "w", encoding='utf-8') as f: f.write(code)
-    compile_proc = subprocess.run(["gcc", "main.c", "-o", "main.out", "-lm", "-w"], capture_output=True, text=True, timeout=15)
-    if compile_proc.returncode != 0: return f"--- COMPILATION FAILED ---\n{compile_proc.stderr}"
-    run_proc = subprocess.run(["./main.out"], capture_output=True, text=True, timeout=15)
-    if run_proc.returncode != 0: return f"--- RUNTIME ERROR ---\n{run_proc.stderr}"
-    output = run_proc.stdout
-    return f"--- EXECUTION SUCCEEDED ---\n{output}" if output.strip() else "--- EXECUTION SUCCEEDED ---\n(No output)"
-
-def analyze_and_refactor_code(code: str, instruction: str) -> str:
-    system_prompt = "You are a world-class C code reviewer..." # (이전과 동일)
-    user_prompt = f"Instruction: '{instruction}'\n\nC Code:\n```c\n{code}\n```"
-    return call_mistral_api(system_prompt, user_prompt)
-
-# ⭐️ 새로운 기능: 외부 MCP 툴을 사용하는 클라이언트 함수
-def call_external_mcp_tool(tool_url: str, code: str, instruction: str) -> str:
-    """다른 Gradio Space MCP 툴을 API로 호출하는 함수"""
-    # Gradio 클라이언트를 사용하여 외부 API 호출 (gradio_client 설치 필요)
-    from gradio_client import Client
+        return json.dumps({"error": f"API Call Error: {e}"})
+    except (KeyError, IndexError) as e:
+        return json.dumps({"error": f"API Response Parsing Error: {e} - Response: {response.text}"})
+
+def parse_ai_response(response_str: str) -> dict:
+    """AI의 JSON 응답을 안전하게 파싱하는 함수"""
     try:
-        client = Client(tool_url)
-        # 외부 툴의 API 엔드포인트와 파라미터 이름에 맞춰야 함
-        # 예시: predict(code_to_analyze=code, user_instruction=instruction)
-        result = client.predict(code, instruction, api_name="/predict") # api_name은 외부 툴에 따라 다름
-        return f"--- EXTERNAL TOOL SUCCEEDED ---\n{result}"
-    except Exception as e:
-        return f"--- EXTERNAL TOOL FAILED ---\nCould not call tool at {tool_url}. Error: {e}"
-
-# --- 3. 1등을 위한 지능형 에이전트 로직 (최종 버전) ---
-def ultimate_agent_ide(initial_code: str, full_instruction: str):
-    tasks = [task.strip() for task in re.split(r'\s+and then\s+|\s+and\s+|,\s*then\s*|\s*그리고\s+|\s*후에\s*', full_instruction, flags=re.IGNORECASE) if task.strip()]
-    current_code = initial_code
-    log = []
-
-    # Step 1: 계획 수립
-    log.append("### 📝 Agent's Plan")
-    plan = "".join([f"\n{i+1}. {task}" for i, task in enumerate(tasks)])
-    log.append(plan)
-    yield current_code, "\n".join(log)
-    time.sleep(1)
-
-    # Step 2: 계획 실행
-    for i, task in enumerate(tasks):
-        log.append(f"\n<details><summary><b>▶ Step {i+1}: {task}</b></summary>\n")
-        yield current_code, "\n".join(log)
-        time.sleep(0.5)
-
-        lower_task = task.lower()
-        
-        # ⭐️ 에이전트의 '생각'과 '행동'
-        if "generate" in lower_task or "create" in lower_task or "만들어" in lower_task:
-            log.append("🧠 **Thought:** The user wants new code. Using `generate_c_code` tool.")
-            yield current_code, "\n".join(log)
-            new_code = generate_c_code(task)
-            if new_code and not new_code.startswith("//"):
-                current_code = new_code
-                log.append("\n✅ **Action Result:** Code generated and updated in the editor.")
+        # LLM이 가끔 ```json ... ``` 마크다운을 포함할 때가 있어 정규식으로 순수 JSON만 추출
+        match = re.search(r'\{.*\}', response_str, re.DOTALL)
+        if match:
+            return json.loads(match.group(0))
+        return json.loads(response_str)
+    except (json.JSONDecodeError, AttributeError) as e:
+        return {"error": f"Failed to parse JSON response. Error: {e}. Raw response: {response_str}"}
+
+def execute_command(command: str, cwd: str, code_arg: str = "") -> dict:
+    """터미널 명령어 및 특수 명령어를 실행하는 함수"""
+    command = command.strip()
+    if not command:
+        return {"stdout": "", "stderr": "Error: Empty command.", "cwd": cwd}
+
+    # 특수 명령어 처리
+    if command.startswith("cd "):
+        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)
+                return {"stdout": f"Changed directory to {target_dir}", "stderr": "", "cwd": target_dir}
             else:
-                log.append(f"\n❌ **Action Result:** Generation failed. {new_code}")
-
-        elif "compile" in lower_task or "run" in lower_task or "실행" in lower_task:
-            log.append("🧠 **Thought:** The user wants to compile and run. Using `compile_and_run_c_code` tool.")
-            yield current_code, "\n".join(log)
-            result = compile_and_run_c_code(current_code)
-            log.append(f"\n💻 **Action Result:**\n```\n{result}\n```")
-
-            # ⭐️⭐️ 자가 수정 (SELF-CORRECTION) 로직 ⭐️⭐️
-            if "COMPILATION FAILED" in result:
-                log.append("\n\n🧠 **Thought:** Compilation failed. I will try to fix the code myself.")
-                yield current_code, "\n".join(log)
-                time.sleep(1)
-
-                error_message = result.split("--- COMPILATION FAILED ---")[1]
-                fix_instruction = f"The following C code failed to compile with this error:\n\n**Error:**\n```\n{error_message}\n```\n\nPlease fix the code so it compiles successfully. Provide only the complete, corrected C code."
-                
-                log.append("\n🛠️ **Self-Correction:** Asking the LLM to fix the error...")
-                yield current_code, "\n".join(log)
-                
-                fixed_code_response = analyze_and_refactor_code(current_code, fix_instruction)
-                fixed_code = parse_code_from_response(fixed_code_response)
-
-                if fixed_code:
-                    current_code = fixed_code
-                    log.append("\n✅ **Self-Correction Result:** A potential fix has been applied to the code editor. Please try compiling again.")
-                else:
-                    log.append("\n❌ **Self-Correction Result:** Failed to automatically fix the code.")
+                return {"stdout": "", "stderr": f"Error: Directory not found: {new_dir}", "cwd": cwd}
+        except Exception as e:
+            return {"stdout": "", "stderr": f"Error processing 'cd': {e}", "cwd": cwd}
+
+    if command.startswith("self_modify"):
+        try:
+            # 명령어에서 코드 내용을 분리 (예: "self_modify '...code...'")
+            code_to_write = shlex.split(command)[1]
+            with open("app.py", "w", encoding='utf-8') as f:
+                f.write(code_to_write)
+            return {"stdout": "Successfully modified app.py. The application will now restart.", "stderr": "", "cwd": cwd}
+        except Exception as e:
+            return {"stdout": "", "stderr": f"FATAL: Failed to self-modify app.py. Error: {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"**User Goal:** {user_goal}\n\n"
+    history.append([user_goal, full_history_log])
+    yield history, "Thinking...", ""
+
+    # 초기 프롬프트 구성
+    message_context = [
+        {"role": "system", "content": SYSTEM_PROMPT},
+        {"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."}
+    ]
+    
+    last_command_output = ""
+
+    for i in range(MAX_AGENT_TURNS):
+        # 1. AI 호출하여 다음 행동 결정
+        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}: CRITICAL ERROR**\n🔴 **Agent Error:** {ai_response_json['error']}"
+            history[-1][1] = full_history_log
+            yield history, "Agent Error", ""
+            return
+
+        # 2. AI 응답 파싱
+        thought = ai_response_json.get("thought", "No thought provided.")
+        plan = ai_response_json.get("plan", [])
+        command = ai_response_json.get("command", "done")
+        user_summary = ai_response_json.get("user_summary", "...")
+
+        # 3. UI 업데이트 (AI의 생각)
+        full_history_log += f"\n---\n**TURN {i+1} / {MAX_AGENT_TURNS}**\n**Status:** `{user_summary}`\n\n"
+        full_history_log += f"🧠 **Thought:** {thought}\n\n"
+        full_history_log += f"📝 **Plan:**\n" + "\n".join([f"- `{p}`" for p in plan]) + "\n\n"
+        full_history_log += f"💻 **Command:** `{command}`\n"
+        history[-1][1] = full_history_log
+        yield history, f"Executing: {command}", ""
+
+        # 4. 루프 종료 조건 확인
+        if command == "done":
+            full_history_log += "\n✅ **Goal Achieved!**"
+            history[-1][1] = full_history_log
+            yield history, "Done", ""
+            return
+
+        # 5. 명령어 실행
+        time.sleep(1) # 사용자가 볼 수 있도록 잠시 대기
+        exec_result = execute_command(command, cwd)
+        cwd = exec_result["cwd"] # cd 명령어에 의해 cwd가 변경될 수 있음
+
+        # 6. UI 업데이트 (실행 결과)
+        stdout, stderr = exec_result["stdout"], exec_result["stderr"]
+        full_history_log += f"\n**Execution Result:**\n"
+        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 += "_(No output)_\n"
+
+        history[-1][1] = full_history_log
+        yield history, user_summary, ""
         
-        # ⭐️⭐️ 외부 MCP 툴 사용 예시 ⭐️⭐️
-        elif "security" in lower_task or "보안" in lower_task:
-            log.append("🧠 **Thought:** The user wants a security analysis. I will use an external MCP tool for this.")
-            yield current_code, "\n".join(log)
-            # 이 URL은 예시이며, 실제 작동하는 보안 분석 MCP Space가 있다면 그 주소를 넣어야 합니다.
-            # 해커톤 제출 시, 직접 간단한 보안분석 툴을 하나 더 만들거나, 다른 참가자의 툴을 사용하는 모습을 보여주면 최고입니다.
-            external_tool_url = "user-provided-security-tool-space-url" 
-            log.append(f"\n🔌 **Action:** Calling external tool at `{external_tool_url}`...")
-            yield current_code, "\n".join(log)
-            
-            # 실제로는 instruction에서 URL을 파싱해야 하지만, 여기서는 하드코딩으로 예시를 보여줍니다.
-            security_result = call_external_mcp_tool(external_tool_url, current_code, task)
-            log.append(f"\n🛡️ **Action Result:**\n```\n{security_result}\n```")
-
-        else:
-            log.append("🧠 **Thought:** The user wants to analyze or refactor. Using `analyze_and_refactor_code` tool.")
-            yield current_code, "\n".join(log)
-            analysis_result = analyze_and_refactor_code(current_code, task)
-            refactored_code = parse_code_from_response(analysis_result)
-            if refactored_code:
-                current_code = refactored_code
-                log.append("\n✅ **Action Result:** Code refactored and updated in the editor.")
-            log.append(f"\n🔎 **Analysis Result:**\n{analysis_result}")
+        # 7. 다음 루프를 위한 컨텍스트 준비
+        last_command_output = f"Command '{command}' executed.\nSTDOUT:\n{stdout}\nSTDERR:\n{stderr}"
         
-        log.append("</details>")
-        yield current_code, "\n".join(log)
+        # 이전 대화 내용 중 중요한 부분만 요약하여 전달 (토큰 절약)
+        user_prompt_for_next_turn = f"""
+My original goal is: '{user_goal}'.
+I am in directory '{cwd}'.
+The last command I ran was `{command}`.
+Its output was:
+---
+{last_command_output}
+---
+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.
+"""
+        message_context.append({"role": "assistant", "content": json.dumps(ai_response_json)})
+        message_context.append({"role": "user", "content": user_prompt_for_next_turn})
 
-    log.append("\n\n--- All tasks complete. ---")
-    yield current_code, "\n".join(log)
+    full_history_log += f"\n---\n🔴 **Agent stopped: Maximum number of turns ({MAX_AGENT_TURNS}) reached.**"
+    history[-1][1] = full_history_log
+    yield history, "Max turns reached", ""
 
-# --- 4. 통합된 Gradio UI (출력 컴포넌트를 Markdown으로 변경) ---
+# --- 5. Gradio UI ---
 with gr.Blocks(theme=gr.themes.Monochrome(primary_hue="indigo", secondary_hue="blue"), css="footer {visibility: hidden}") as demo:
-    gr.Markdown("# 🏆 The Ultimate C-Codestral IDE Agent 🏆")
-
-    with gr.Tabs():
-        with gr.TabItem("👨‍💻 IDE Agent"):
-            with gr.Row(equal_height=True):
-                with gr.Column(scale=2):
-                    code_editor = gr.Code(label="C Code Editor", language="c", lines=28, interactive=True, value='#include <stdio.h>\n\nint main() {\n    printf("Hello, World!\\n");\n    return 0;\n}')
-                
-                with gr.Column(scale=1):
-                    instruction_box = gr.Textbox(label="Instruction", placeholder="e.g., 'Refactor this code, then compile it, then check security'", lines=4)
-                    execute_btn = gr.Button("Execute", variant="primary", size="lg")
-                    # 출력을 Markdown으로 변경하여 풍부한 UI를 제공
-                    output_box = gr.Markdown(label="Console / Output")
-
-            execute_btn.click(
-                fn=ultimate_agent_ide, 
-                inputs=[code_editor, instruction_box], 
-                outputs=[code_editor, output_box]
-            )
-
-        with gr.TabItem("🛠️ MCP Tools API"):
-             # MCP 탭은 이전과 동일하게 유지
-            gr.Markdown("## Available MCP Tools for other Agents\nThese APIs are the building blocks of our IDE agent.")
-            with gr.Accordion("Tool: Generate C Code", open=False):
-                gr.Interface(fn=generate_c_code, inputs="text", outputs=gr.Code(language="c", label="Generated C Code"))
-            with gr.Accordion("Tool: Compile & Run C Code", open=False):
-                gr.Interface(fn=compile_and_run_c_code, inputs=gr.Code(language="c"), outputs=gr.Textbox(label="Output"))
-            with gr.Accordion("Tool: Analyze & Refactor C Code", open=False):
-                gr.Interface(fn=analyze_and_refactor_code, inputs=[gr.Code(language="c", label="Code to Analyze"), gr.Textbox(label="Instruction")], outputs=gr.Markdown())
+    gr.Markdown("# 🧬 RASI: Recursive Autonomous Swarm Intelligence 🧬")
+    gr.Markdown("AI 에이전트에게 최종 목표를 알려주세요. 에이전트는 스스로 계획을 세우고, 터미널 명령어를 실행하며, 오류를 수정하여 과업을 완수합니다.")
+
+    chatbot = gr.Chatbot(label="Agent Log", height=700, show_copy_button=True, bubble_full_width=True)
+    
+    with gr.Row():
+        status_box = gr.Textbox(label="Current Status", interactive=False)
+    
+    with gr.Row():
+        user_input = gr.Textbox(label="User Goal", placeholder="e.g., 'Create a python project named 'my_calc'. Inside it, create a file 'calculator.py' that contains a function to add two numbers. Then, create another file 'run.py' that imports and runs this function with numbers 5 and 7 and prints the result. Finally, execute run.py.'", scale=9)
+        submit_btn = gr.Button("▶️ Execute Goal", variant="primary", scale=1)
+
+    def on_submit(user_goal, chat_history):
+        chat_history = chat_history or []
+        # 제너레이터의 스트리밍 출력을 받기 위한 루프
+        for history_update, status_update, input_update in agent_loop(user_goal, chat_history):
+            yield history_update, status_update, input_update
+
+    submit_btn.click(
+        on_submit,
+        inputs=[user_input, chatbot],
+        outputs=[chatbot, status_box, user_input] # 챗봇, 상태창, 입력창 업데이트
+    )
 
 if __name__ == "__main__":
     demo.queue().launch()