Replit Agent实现全自动AI编程开发

import jsonimport subprocessimport refrom dataclasses import dataclass, fieldfrom typing import List, Dict, Optional, Tuple @dataclassclass ProjectFile:    """项目文件: 存储文件路径和内容"""    path: str    content: str    file_type: str = "text" @dataclassclass BugReport:    """Bug报告: 记录错误信息和修复状态"""    error_message: str    error_type: str    stack_trace: str    file_path: str    fix_applied: str = ""    is_fixed: bool = False @dataclassclass DeployConfig:    """部署配置: 环境变量和构建参数"""    port: int = 3000    environment_vars: Dict[str, str] = field(default_factory=dict)    build_command: str = "npm install && npm run build"    start_command: str = "npm start"    database_url: str = "" class ReplitAgentSimulator:    """模拟Replit Agent的自主开发全流程: 需求理解-代码生成-自动调试-部署上线"""     def __init__(self, project_name: str):        self.project_name = project_name        self.files: List[ProjectFile] = []        self.bug_reports: List[BugReport] = []        self.deploy_config: Optional[DeployConfig] = None        self.iteration_count: int = 0        self.max_debug_iterations: int = 5     def understand_requirement(self, description: str) -> Dict:        """需求理解: 从自然语言描述中提取项目结构和技术栈"""        tech_stack = self._infer_tech_stack(description)        project_structure = self._plan_project_structure(description, tech_stack)        print(f"[Agent] 需求理解完成:")        print(f"  技术栈: {tech_stack}")        print(f"  项目结构: {project_structure}")        return {"tech_stack": tech_stack, "structure": project_structure}     def _infer_tech_stack(self, description: str) -> List[str]:        """根据需求描述推断技术栈"""        stack = []        desc_lower = description.lower()        if any(kw in desc_lower for kw in ["web", "网站", "页面", "应用"]):            stack.extend(["React", "Node.js"])        if any(kw in desc_lower for kw in ["数据库", "存储", "data", "db"]):            stack.extend(["PostgreSQL", "Prisma"])        if any(kw in desc_lower for kw in ["api", "接口", "后端"]):            stack.append("Express")        if any(kw in desc_lower for kw in ["实时", "chat", "聊天", "消息"]):            stack.append("WebSocket")        return stack if stack else ["React", "Node.js"]     def _plan_project_structure(self, description: str, tech_stack: List[str]) -> List[str]:        """规划项目文件结构"""        structure = [            "package.json",            "src/index.js",            "src/App.jsx",            "src/components/",            ".env",            "README.md"        ]        if "PostgreSQL" in tech_stack:            structure.append("prisma/schema.prisma")        if "Express" in tech_stack:            structure.append("src/server.js")            structure.append("src/routes/api.js")        return structure     def generate_code(self, plan: Dict) -> List[ProjectFile]:        """代码生成: 根据规划生成多文件协调的完整项目代码"""        tech_stack = plan["tech_stack"]        self.files = []         if "Express" in tech_stack:            self.files.append(ProjectFile(                "src/server.js",                'const express = require("express");\n'                'const app = express();\n'                'app.use(express.json());\n'                'app.listen(3000, () => console.log("Server running on port 3000"));'            ))            self.files.append(ProjectFile(                "src/routes/api.js",                'const router = require("express").Router();\n'                'router.get("/items", (req, res) => res.json({items: []}));\n'                'module.exports = router;'            ))         self.files.append(ProjectFile(            "src/App.jsx",            'import React, { useState, useEffect } from "react";\n'            'export default function App() {\n'            '  const [items, setItems] = useState([]);\n'            '  useEffect(() => {\n'            '    fetch("/api/items").then(r => r.json()).then(d => setItems(d.items));\n'            '  }, []);\n'            '  return <div><h1>{items.length} Items</h1></div>;\n'            '}'        ))         self.files.append(ProjectFile(            "package.json",            json.dumps({                "name": self.project_name,                "scripts": {"start": "node src/server.js", "build": "react-scripts build"},                "dependencies": {"express": "^4.18.0", "react": "^18.2.0"}            }, indent=2)        ))         self.files.append(ProjectFile(            ".env",            "PORT=3000\nDATABASE_URL=postgresql://localhost:5432/myapp\n"        ))         print(f"[Agent] 代码生成完成: {len(self.files)}个文件")        for f in self.files:            print(f"  - {f.path}")        return self.files     def run_and_debug(self) -> bool:        """自动调试: 运行代码检测错误分析原因并修复"""        self.iteration_count = 0        while self.iteration_count < self.max_debug_iterations:            self.iteration_count += 1            print(f"\n[Agent] 调试迭代 #{self.iteration_count}")            errors = self._simulate_run()            if not errors:                print("[Agent] 运行成功, 无错误!")                return True            for error in errors:                print(f"[Agent] 检测到错误: {error.error_type} in {error.file_path}")                self._analyze_and_fix(error)                self.bug_reports.append(error)            print(f"[Agent] 已修复 {len(errors)} 个错误, 重新运行验证...")        print(f"[Agent] 达到最大调试次数 {self.max_debug_iterations}")        return False     def _simulate_run(self) -> List[BugReport]:        """模拟运行代码并检测错误"""        errors = []        if self.iteration_count == 1:            errors.append(BugReport(                error_message="Error: listen EADDRINUSE: address already in use :::3000",                error_type="PortConflict",                stack_trace="at Server.listen (net.js:1234)",                file_path="src/server.js"            ))        if self.iteration_count == 1:            errors.append(BugReport(                error_message="Module not found: Can't resolve 'express'",                error_type="MissingDependency",                stack_trace="at require (module.js:18)",                file_path="src/server.js"            ))        return errors     def _analyze_and_fix(self, bug: BugReport) -> None:        """分析错误原因并自主修复"""        if bug.error_type == "PortConflict":            bug.fix_applied = "将端口从3000改为3001, 更新.env配置"            for f in self.files:                if f.path == ".env":                    f.content = "PORT=3001\nDATABASE_URL=postgresql://localhost:5432/myapp\n"            print(f"  [修复] {bug.fix_applied}")         elif bug.error_type == "MissingDependency":            bug.fix_applied = "运行 npm install express 安装缺失依赖"            print(f"  [修复] {bug.fix_applied}")         elif bug.error_type == "CORS":            bug.fix_applied = "添加 cors 中间件: app.use(require('cors')())"            print(f"  [修复] {bug.fix_applied}")         bug.is_fixed = True     def deploy(self) -> Dict:        """自动部署: 配置环境并发布应用"""        self.deploy_config = DeployConfig(            port=3001,            environment_vars={"NODE_ENV": "production"},            build_command="npm install && npm run build",            start_command="npm start",            database_url="postgresql://prod-db:5432/myapp"        )        deploy_url = f"https://{self.project_name}.replit.app"        print(f"\n[Agent] 部署配置完成:")        print(f"  端口: {self.deploy_config.port}")        print(f"  环境: {self.deploy_config.environment_vars}")        print(f"  构建命令: {self.deploy_config.build_command}")        print(f"  应用URL: {deploy_url}")        return {"status": "deployed", "url": deploy_url}     def run_full_pipeline(self, requirement: str) -> Dict:        """执行完整的自主开发流程"""        print(f"{'='*60}")        print(f"Replit Agent 自主开发: {self.project_name}")        print(f"{'='*60}")        plan = self.understand_requirement(requirement)        self.generate_code(plan)        success = self.run_and_debug()        deploy_result = self.deploy()        return {            "project": self.project_name,            "files_generated": len(self.files),            "bugs_fixed": len(self.bug_reports),            "debug_iterations": self.iteration_count,            "deploy_status": deploy_result["status"],            "app_url": deploy_result["url"]        } agent = ReplitAgentSimulator("todo-app")result = agent.run_full_pipeline(    "创建一个待办事项Web应用, 需要数据库存储, 提供REST API接口")print(f"\n{'='*60}")print("开发完成!")for k, v in result.items():    print(f"  {k}: {v}")