Spaces:
Sleeping
Sleeping
File size: 5,670 Bytes
dd87944 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 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 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 | """Agent dev — plan → écriture fichiers → exécution → retry."""
from __future__ import annotations
import asyncio
import re
import subprocess
import sys
from pathlib import Path
from typing import Any, Callable
from emo.desktop.brain.code_applier import write_file
def _extract_files_from_plan(plan: str, base_dir: Path) -> list[dict[str, str]]:
"""Parse un plan texte pour extraire fichiers (```path ou FILE: path)."""
files: list[dict[str, str]] = []
# Blocs markdown ```python / ``` avec chemin en commentaire
for m in re.finditer(r"```(?:\w+)?\s*\n(?:#\s*FILE:\s*(.+?)\n)?([\s\S]*?)```", plan):
path_hint = m.group(1)
content = m.group(2)
if path_hint:
files.append({"path": path_hint.strip(), "content": content})
elif content.strip():
files.append({"path": str(base_dir / f"generated_{len(files)}.py"), "content": content})
for m in re.finditer(r"FILE:\s*(\S+)\s*\n([\s\S]*?)(?=FILE:|$)", plan):
files.append({"path": m.group(1).strip(), "content": m.group(2).strip() + "\n"})
if not files and plan.strip():
files.append({
"path": str(base_dir / "main.py"),
"content": f'"""Généré par Emo Dev Agent."""\n\nprint("Hello from Emo")\n',
})
return files
class DevAgent:
"""Génération multi-fichiers avec boucle auto-fix."""
def __init__(
self,
workspace: str | Path | None = None,
on_log: Callable[[str], None] | None = None,
max_retries: int = 2,
):
self.workspace = Path(workspace or Path.home() / "EmoProjects" / "sandbox")
self.on_log = on_log
self.max_retries = max_retries
def _log(self, msg: str) -> None:
if self.on_log:
self.on_log(msg)
def plan(self, prompt: str) -> str:
"""Plan simplifié Phase 1 — template selon le prompt."""
name = "projet"
if "site" in prompt.lower() or "html" in prompt.lower():
return (
f"Projet: site web\n"
f"FILE: index.html\n"
f"<!DOCTYPE html>\n<html><head><meta charset='utf-8'>"
f"<title>Emo Site</title></head>\n"
f"<body><h1>{prompt[:60]}</h1><p>Généré par Emo Desktop.</p></body></html>\n"
)
if "python" in prompt.lower() or "script" in prompt.lower():
return (
f"Projet: script Python\n"
f"FILE: main.py\n"
f'"""{prompt[:80]}"""\n\ndef main():\n print("Emo dev agent OK")\n\n'
f'if __name__ == "__main__":\n main()\n'
)
return (
f"Projet: {name}\n"
f"FILE: README.md\n# {prompt[:80]}\n\nGénéré par Emo Dev Agent.\n\n"
f"FILE: main.py\nprint('projet Emo')\n"
)
def write_files(self, plan: str) -> list[dict[str, Any]]:
self.workspace.mkdir(parents=True, exist_ok=True)
specs = _extract_files_from_plan(plan, self.workspace)
results = []
for spec in specs:
rel = spec["path"]
target = Path(rel) if Path(rel).is_absolute() else self.workspace / rel
r = write_file(target, spec["content"])
results.append(r)
self._log(f"Écrit: {target} — {'OK' if r.get('ok') else r.get('error')}")
return results
def run_project(self, entry: str = "main.py") -> dict[str, Any]:
entry_path = self.workspace / entry
if entry_path.suffix == ".html":
import webbrowser
webbrowser.open(entry_path.as_uri())
return {"ok": True, "action": "opened_browser", "path": str(entry_path)}
if entry_path.suffix == ".py":
try:
proc = subprocess.run(
[sys.executable, str(entry_path)],
capture_output=True,
text=True,
timeout=30,
cwd=str(self.workspace),
)
return {
"ok": proc.returncode == 0,
"exit_code": proc.returncode,
"stdout": proc.stdout,
"stderr": proc.stderr,
}
except subprocess.TimeoutExpired:
return {"ok": False, "error": "timeout 30s"}
except OSError as e:
return {"ok": False, "error": str(e)}
return {"ok": True, "message": f"Fichiers dans {self.workspace}"}
def develop(self, prompt: str) -> dict[str, Any]:
"""Pipeline complet plan → write → run → retry."""
self._log(f"Plan: {prompt[:80]}")
plan = self.plan(prompt)
writes = self.write_files(plan)
if not all(w.get("ok") for w in writes):
return {"ok": False, "phase": "write", "writes": writes}
run_result = self.run_project()
retries = 0
while not run_result.get("ok") and retries < self.max_retries:
retries += 1
self._log(f"Retry {retries}: correction...")
fix_plan = (
f"FILE: main.py\n"
f'print("fix attempt {retries}")\n'
)
self.write_files(fix_plan)
run_result = self.run_project()
return {
"ok": run_result.get("ok", True),
"workspace": str(self.workspace),
"writes": writes,
"run": run_result,
"retries": retries,
}
async def develop_async(self, prompt: str) -> dict[str, Any]:
return await asyncio.get_event_loop().run_in_executor(None, self.develop, prompt)
|