Repository navigation
Expand file tree
/
Copy pathdevika.py
More file actions
199 lines (152 loc) · 5.54 KB
/
Copy pathdevika.py
File metadata and controls
199 lines (152 loc) · 5.54 KB
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
"""
DO NOT REARRANGE THE ORDER OF THE FUNCTION CALLS AND VARIABLE DECLARATIONS
AS IT MAY CAUSE IMPORT ERRORS AND OTHER ISSUES
"""
import eventlet
eventlet.monkey_patch()
from src.init import init_devika
init_devika()
from flask import Flask, request, jsonify, send_file
from flask_cors import CORS
from src.socket_instance import socketio, emit_agent
import os
import logging
from threading import Thread
import tiktoken
from src.apis.project import project_bp
from src.config import Config
from src.logger import Logger, route_logger
from src.project import ProjectManager
from src.state import AgentState
from src.agents import Agent
from src.llm import LLM
app = Flask(__name__)
CORS(app)
app.register_blueprint(project_bp)
socketio.init_app(app)
log = logging.getLogger("werkzeug")
log.disabled = True
TIKTOKEN_ENC = tiktoken.get_encoding("cl100k_base")
os.environ["TOKENIZERS_PARALLELISM"] = "false"
manager = ProjectManager()
AgentState = AgentState()
config = Config()
logger = Logger()
# initial socket
@socketio.on('socket_connect')
def test_connect(data):
print("Socket connected :: ", data)
emit_agent("socket_response", {"data": "Server Connected"})
@app.route("/api/data", methods=["GET"])
@route_logger(logger)
def data():
project = manager.get_project_list()
models = LLM().list_models()
search_engines = ["Bing", "Google", "DuckDuckGo"]
return jsonify({"projects": project, "models": models, "search_engines": search_engines})
@app.route("/api/messages", methods=["POST"])
def get_messages():
data = request.json
project_name = data.get("project_name")
messages = manager.get_messages(project_name)
return jsonify({"messages": messages})
# Main socket
@socketio.on('user-message')
def handle_message(data):
action = data.get('action')
message = data.get('message')
base_model = data.get('base_model')
project_name = data.get('project_name')
search_engine = data.get('search_engine').lower()
agent = Agent(base_model=base_model, search_engine=search_engine)
if action == 'continue':
new_message = manager.new_message()
new_message['message'] = message
new_message['from_devika'] = False
manager.add_message_to_project(project_name, new_message)
if AgentState.is_agent_completed(project_name):
thread = Thread(target=lambda: agent.subsequent_execute(message, project_name))
thread.start()
if action == 'execute_agent':
thread = Thread(target=lambda: agent.execute(message, project_name, search_engine))
thread.start()
@app.route("/api/is-agent-active", methods=["POST"])
@route_logger(logger)
def is_agent_active():
data = request.json
project_name = data.get("project_name")
is_active = AgentState.is_agent_active(project_name)
return jsonify({"is_active": is_active})
@app.route("/api/get-agent-state", methods=["POST"])
@route_logger(logger)
def get_agent_state():
data = request.json
project_name = data.get("project_name")
agent_state = AgentState.get_latest_state(project_name)
return jsonify({"state": agent_state})
@app.route("/api/get-browser-snapshot", methods=["GET"])
@route_logger(logger)
def browser_snapshot():
snapshot_path = request.args.get("snapshot_path")
return send_file(snapshot_path, as_attachment=True)
@app.route("/api/get-browser-session", methods=["GET"])
@route_logger(logger)
def get_browser_session():
project_name = request.args.get("project_name")
agent_state = AgentState.get_latest_state(project_name)
if not agent_state:
return jsonify({"session": None})
else:
browser_session = agent_state["browser_session"]
return jsonify({"session": browser_session})
@app.route("/api/get-terminal-session", methods=["GET"])
@route_logger(logger)
def get_terminal_session():
project_name = request.args.get("project_name")
agent_state = AgentState.get_latest_state(project_name)
if not agent_state:
return jsonify({"terminal_state": None})
else:
terminal_state = agent_state["terminal_session"]
return jsonify({"terminal_state": terminal_state})
@app.route("/api/run-code", methods=["POST"])
@route_logger(logger)
def run_code():
data = request.json
project_name = data.get("project_name")
code = data.get("code")
# TODO: Implement code execution logic
return jsonify({"message": "Code execution started"})
@app.route("/api/calculate-tokens", methods=["POST"])
@route_logger(logger)
def calculate_tokens():
data = request.json
prompt = data.get("prompt")
tokens = len(TIKTOKEN_ENC.encode(prompt))
return jsonify({"token_usage": tokens})
@app.route("/api/token-usage", methods=["GET"])
@route_logger(logger)
def token_usage():
project_name = request.args.get("project_name")
token_count = AgentState.get_latest_token_usage(project_name)
return jsonify({"token_usage": token_count})
@app.route("/api/logs", methods=["GET"])
def real_time_logs():
log_file = logger.read_log_file()
return jsonify({"logs": log_file})
@app.route("/api/settings", methods=["POST"])
@route_logger(logger)
def set_settings():
data = request.json
print("Data: ", data)
config.config.update(data)
config.save_config()
return jsonify({"message": "Settings updated"})
@app.route("/api/settings", methods=["GET"])
@route_logger(logger)
def get_settings():
configs = config.get_config()
return jsonify({"settings": configs})
if __name__ == "__main__":
logger.info("Devika is up and running!")
socketio.run(app, debug=False, port=1337, host="0.0.0.0")