Spaces:
Running
Running
import gradio as gr | |
import os | |
import subprocess | |
import requests | |
import json | |
import re | |
import time | |
# --- 1. ํ๊ฒฝ ์ค์ ๋ฐ ๊ฐํ๋ API ํธ์ถ ํจ์ --- | |
MISTRAL_API_KEY = os.environ.get("MISTRAL_API_KEY") | |
CODESTRAL_ENDPOINT = "https://codestral.mistral.ai/v1/chat/completions" | |
def call_mistral_api(system_prompt: str, user_prompt: str): | |
"""ํฅ์๋ Mistral API ํธ์ถ ํจ์""" | |
if not MISTRAL_API_KEY: | |
raise gr.Error("MISTRAL_API_KEY is not set. Please add it to your 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} | |
try: | |
response = requests.post(CODESTRAL_ENDPOINT, headers=headers, data=json.dumps(data), timeout=60) | |
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 | |
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.") | |
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.") | |
# โญ๏ธโญ๏ธ ์ธ๋ถ 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}") | |
log.append("</details>") | |
yield current_code, "\n".join(log) | |
log.append("\n\n--- All tasks complete. ---") | |
yield current_code, "\n".join(log) | |
# --- 4. ํตํฉ๋ Gradio UI (์ถ๋ ฅ ์ปดํฌ๋ํธ๋ฅผ Markdown์ผ๋ก ๋ณ๊ฒฝ) --- | |
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()) | |
if __name__ == "__main__": | |
demo.queue().launch() |