Re-Organised project and added benchmark code

benchmark/server.py

@@ -0,0 +1,121 @@
+import io
+import json
+import logging
+import shutil
+from pathlib import Path
+from random import randint
+from typing import Annotated, Any, Dict, List
+
+from fastapi import FastAPI, File, Form, HTTPException, UploadFile
+from fastapi.responses import StreamingResponse
+from pydantic import BaseModel
+
+# Set up logging
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+app = FastAPI()
+artifacts: List[Dict[str, Any]] = []
+
+with open("agent/gpt-engineer/agbenchmark/config.json", "r") as file:
+    config = json.load(file)
+    logger.info("Loaded configuration")
+
+
+class Task(BaseModel):
+    input: str
+
+
+@app.post("/agent/tasks/{task_id}/artifacts")
+async def upload_file(
+    task_id: str, file: Annotated[UploadFile, File()], relative_path: str = Form("")
+) -> Dict[str, Any]:
+    logger.info(
+        "Uploading file for task_id: %s with relative path: %s", task_id, relative_path
+    )
+    absolute_directory_path = Path(__file__).parent.absolute()
+    save_path = absolute_directory_path / "agent/gpt-engineer" / config["workspace"]
+
+    random_string = str(randint(0, 100000))
+    while random_string in artifacts:
+        random_string = str(randint(0, 100000))
+
+    artifact_data = await file.read()
+    artifacts.append(
+        {
+            "binary": artifact_data,
+            "relative_path": relative_path,
+            "file_name": file.filename,
+            "artifact_id": random_string,
+        }
+    )
+
+    print(artifacts)
+    return {
+        "artifact_id": random_string,
+        "file_name": "file_name",
+        "relative_path": "relative_path",
+    }
+
+
+@app.get("/agent/tasks/{task_id}/artifacts")
+async def get_files() -> List[Dict[str, Any]]:
+    logger.info("Fetching list of files for task")
+    return artifacts
+
+
+@app.get("/agent/tasks/{task_id}/artifacts/{artifact_id}")
+async def get_file(artifact_id: str):
+    for artifact in artifacts:
+        if artifact["artifact_id"] == artifact_id:
+            break
+    else:
+        logger.error("Attempt to access nonexistent artifact with ID: %s", artifact_id)
+        raise HTTPException(status_code=404, detail="Artifact not found")
+
+    logger.info("Fetching artifact with ID: %s", artifact_id)
+    # find aritifact where artifact_id = artifact_id
+
+    for artifact in artifacts:
+        if artifact["artifact_id"] == artifact_id:
+            return StreamingResponse(
+                io.BytesIO(artifact["binary"]),
+                media_type="application/octet-stream",
+                headers={"Content-Disposition": f"attachment; filename=test.txt"},
+            )
+    # return 404
+    return HTTPException(status_code=404, detail="Artifact not found")
+
+
+@app.post("/agent/tasks/{task_id}/steps")
+async def create_steps(task_id: str):
+    logger.info("Creating step for task_id: %s", task_id)
+    return {
+        "input": "random",
+        "additional_input": {},
+        "task_id": task_id,
+        "step_id": "random_step",
+        "name": "random",
+        "status": "created",
+        "output": "random",
+        "additional_output": {},
+        "artifacts": [],
+        "is_last": True,
+    }
+
+
+@app.post("/agent/tasks")
+async def create_tasks(task: Task):
+    artifacts.clear()
+    return {
+        "input": "random",
+        "additional_input": {},
+        "task_id": "static_task_id",
+        "artifacts": [],
+    }
+
+
+if __name__ == "__main__":
+    import uvicorn
+
+    uvicorn.run(app, host="0.0.0.0", port=8000)