DevCursor/main.py at main · Nikhil-Dadhich/DevCursor · GitHub

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
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
from dotenv import load_dotenv
from openai import OpenAI, RateLimitError
from datetime import datetime
import json
import requests
import os
import time
import subprocess

# Load environment variables
load_dotenv()

# Initialize OpenAI client (adjust base_url if needed for Gemini)
client = OpenAI(
    api_key=os.getenv("GEMINI_API_KEY"),
    base_url="https://generativelanguage.googleapis.com/v1beta/openai/"
)

# Memory store for file content
generated_data = {}

# Tool implementations

def run_command(cmd: str):
    try:
        result = subprocess.check_output(cmd, shell=True, stderr=subprocess.STDOUT, text=True)
        return result.strip()
    except subprocess.CalledProcessError as e:
        return f"❌ Command failed: {e.output.strip()}"


def get_weather(city: str):
    url = f"https://wttr.in/{city}?format=%C+%t"
    response = requests.get(url)
    if response.status_code == 200:
        return f"The weather in {city} is {response.text.strip()}."
    return "Something went wrong getting the weather."

def write_file(data: dict):
    filepath = data.get("filepath")
    content = data.get("content")
    if not filepath or not content:
        return "Invalid input. Must provide 'filepath' and 'content'."
    os.makedirs(os.path.dirname(filepath), exist_ok=True)
    with open(filepath, "w", encoding="utf-8") as f:
        f.write(content)
    return f"{filepath} written successfully."

# Mapping tool names to functions
available_tools = {
    "get_weather": get_weather,
    "run_command": run_command,
    "write_file": write_file
}

# System prompt
SYSTEM_PROMPT = f"""
You are a helpful AI Assistant who is specialized in resolving user queries.
You work in the following cycle: start → plan → action → observe → output.

For each user query and based on the available tools, you should:
1. Plan the step-by-step approach to solve the query.
2. Select the most appropriate tool.
3. Execute the tool using the correct input.
4. Wait for the output (observation) and then form a final response to the user.

🧠 Rules:
- Follow the Output JSON Format.
- Perform only one step at a time and wait for the next observation before continuing.
- Carefully analyze the user query before taking action.
- Only use tools defined in the tool list below.

📤 Output JSON Format:
{{
    "step": "string",                // One of: "plan", "action", "output", "store"
    "content": "string",             // Explanation or final output
    "function": "string",            // If action, tool name to use
    "input": "object or string"      // If action, input to the function
}}

💾 Global File Strategy:
- For each file (e.g., `Todo/index.html`, `Todo/styles.css`, `Todo/script.js`), generate content separately.
- Store content in memory using a structured JSON message like:
  {{
    "step": "store",
    "filepath": "Todo/index.html",
    "content": "<!DOCTYPE html>..."
  }}
- Then call `write_file` with the same filepath.

🛠️ Available Tools:
- **get_weather**: Takes a city name as an input and returns the current weather.
- **run_command**: Takes a valid Windows terminal command (cmd.exe syntax) as a string and executes it.
- **write_file**: Writes content stored in memory to the disk for a given filepath.

⚠️ Windows Command Guidelines:
- Only use valid **cmd.exe** commands like: `mkdir`, `echo`, `type`, `del`, etc.
- Avoid unsupported Linux-style commands like `ls`, `touch`, `cat`, `rm`, etc.

🚫 Do not respond with natural language messages like "Success" or "Files created".
✅ Always respond ONLY with a valid JSON object following the Output JSON Format.

📌 Termination Rule:
- Once all required files have been created and the task is complete, end with one final response:
  {{
    "step": "output",
    "content": "Project 'Calculator' created successfully. Files: index.html, styles.css, script.js are located in the 'Calculator' directory."
  }}
- Do not repeat the same success message again.
- Do not take any further action after the final output.

  Always respond ONLY with a valid JSON object following the Output JSON Format. Do not return natural language text alone.
"""

# Message history
messages = [
    { "role": "system", "content": SYSTEM_PROMPT }
]

# Main interaction loop
while True:
    query = input("> ")
    if not query.strip():
        continue

    messages.append({ "role": "user", "content": query })
    done = False

    while not done:
        try:
            response = client.chat.completions.create(
                model="gemini-1.5-flash",
                response_format={ "type": "json_object" },
                messages=messages
            )
        except RateLimitError:
            print("⏳ Rate limit reached. Retrying in 60 seconds...")
            time.sleep(60)
            continue
        except Exception as e:
            print(f"❌ Unexpected error during OpenAI request: {e}")
            break

        raw_content = response.choices[0].message.content
        # print("🧾 Raw content:")
        # print(raw_content)
        if not raw_content:
            print("⚠️ Empty response from the model. Skipping this cycle.")
            break

        try:
            parsed = json.loads(raw_content)
        except json.JSONDecodeError as e:
            print(f"❌ JSON Decode Error: {e}")
            print("🧾 Raw content was:\n", raw_content)
            break

        parsed_responses = parsed if isinstance(parsed, list) else [parsed]

        for parsed_response in parsed_responses:
            messages.append({ "role": "assistant", "content": json.dumps(parsed_response) })

            step = parsed_response.get("step")

            if step == "plan":
                print(f"🧠: {parsed_response.get('content')}")
                continue

            elif step == "action":
                tool_name = parsed_response.get("function")
                tool_input = parsed_response.get("input")
                print(f"🛠️: Calling Tool: {tool_name} with input {tool_input}")

                if available_tools.get(tool_name):
                    try:
                        output = available_tools[tool_name](tool_input)
                    except Exception as e:
                        output = f"❌ Tool execution error: {e}"

                    messages.append({
                        "role": "user",
                        "content": json.dumps({
                            "step": "observe",
                            "output": output
                        })
                    })
                    break

            elif step == "store":
                filepath = parsed_response.get("filepath")
                content = parsed_response.get("content")
                if filepath and content:
                    generated_data[filepath] = content
                    print(f"📄 Stored content for {filepath}")
                continue

            elif step == "output":
                print(f"🤖: {parsed_response.get('content')}")
                done = True
                break  # breaks for-loop