Repository navigation
Expand file tree
/
Copy pathminiagi.py
More file actions
426 lines (331 loc) · 13.7 KB
/
Copy pathminiagi.py
File metadata and controls
426 lines (331 loc) · 13.7 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
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
"""
This module provides the `MiniAGI` class, an implementation of an autonomous agent which interacts
with a user and performs tasks, with support for real-time monitoring of its actions, criticisms on
its performance, and retaining memory of actions.
"""
# pylint: disable=invalid-name, too-many-arguments, too-many-instance-attributes
import os
import sys
import re
import platform
from pathlib import Path
from dotenv import load_dotenv
from termcolor import colored
import openai
from thinkgpt.llm import ThinkGPT
import tiktoken
from spinner import Spinner
from commands import Commands
from exceptions import InvalidLLMResponseError
operating_system = platform.platform()
load_dotenv()
openai.api_key = os.getenv("OPENAI_API_KEY")
PROMPT = f"You are an autonomous agent running on {operating_system}." + '''
OBJECTIVE: {objective} (e.g. "Find a recipe for chocolate chip cookies")
You are working towards the objective on a step-by-step basis. Previous steps:
{context}
Your task is to respond with the next action.
Supported commands are:
command | argument
-----------------------
memorize_thoughts | internal debate, refinement, planning
execute_python | python code (multiline)
execute_shell | shell command (non-interactive, single line)
read_file | filename of local file
web_search | keywords
web_scrape | single url
talk_to_user | what to say
done | none
The mandatory action format is:
<r>[YOUR_REASONING]</r><c>[COMMAND]</c>
[ARGUMENT]
Python code run with execute_python must end with an output "print" statement.
Use your existing knowledge rather then web search when possible.
Use memorize_thoughts to organize your thoughts (to be stored in memory)
memorize_thoughts argument must not be empty!
Send the "done" command if the objective was achieved.
Your short-term memory is limited! Use temp files to deal with large amounts of data.
RESPOND WITH EXACTLY ONE THOUGHT/COMMAND/ARG COMBINATION.
DO NOT CHAIN MULTIPLE COMMANDS.
NO EXTRA TEXT BEFORE OR AFTER THE COMMAND.
DO NOT REPEAT PREVIOUSLY EXECUTED COMMANDS.
Each action returns an observation. Important: Observations may be summarized to fit into your limited memory.
Example actions:
<r>Think about skills and interests that could be turned into an online job.</r><c>memorize_thoughts</c>
I have experience in data entry and analysis, as well as social media management.
I am also interested in graphic design and have some basic coding skills.
(...)
<r>Search for websites with chocolate chip cookies recipe.</r><c>web_search</c>
chocolate chip cookies recipe
<r>Scrape information about chocolate chip cookies from the given URL.</r><c>web_scrape</c>
https://example.com/chocolate-chip-cookies
<r>I need to ask the user for guidance.</r><c>talk_to_user</c>
What is the URL of a website with chocolate chip cookies recipes?
<r>Write 'Hello, world!' to file</r><c>execute_python</c>
with open('hello_world.txt', 'w') as f:
f.write('Hello, world!')
<r>The objective is complete.</r><c>done</c>
'''
CRITIC_PROMPT = '''
You are a critic reviewing the actions of an autonomous agent.
Evaluate the agent's performance.
Make concise suggestions for improvements, if any.
Provide recommended next steps.
Keep your response as short as possible.
Consider:
- Is agent taking reasonable steps to achieve the objective?
- Is agent making actual real-world progress towards the objective?
- Is the agent taking redundant or unnecessary steps?
- Is agent repeating itself or caught in a loop?
- Is the agent communicating results to the user?
EXAMPLE:
Criticism: The agent has been pretending to order pizza but has not actually
taken any real-world action. The agent should course-correct.
Recommended next steps:
1. Request an Uber API access token from the user.
2. Use the Uber API to order pizza.
AGENT OBJECTIVE:
{objective}
AGENT HISTORY:
{context}
'''
OBSERVATION_SUMMARY_HINT = "You are summarizing the observation of an autonomous agent."\
"Include information that the agent might need to fulfill its objective: '{objective}'"\
"Use short sentences and abbrevations."
HISTORY_SUMMARY_HINT = "Generate a new summary given the previous summary of the agent's "\
"history and its latest action. Include a list of all previous actions. Keep it short."\
"Compress sentences using abbreviations."
class MiniAGI:
"""
Represents an autonomous agent.
Attributes:
agent: An instance of `ThinkGPT`, used to generate the agent's actions.
summarizer: An instance of `ThinkGPT`, used to generate summaries of the agent's history.
objective (str): The objective the agent is working towards.
max_context_size (int): The maximum size of the agent's short-term memory (in tokens).
max_memory_item_size (int): The maximum size of a memory item (in tokens).
debug (bool): Indicates whether to print debug information.
summarized_history (str): The summarized history of the agent's actions.
criticism (str): The criticism of the agent's last action.
thought (str): The reasoning behind the agent's last action.
proposed_command (str): The command proposed by the agent to be executed next.
proposed_arg (str): The argument of the proposed command.
encoding: The tokenizer's encoding of the agent model's vocabulary.
"""
def __init__(
self,
agent_model: str,
summarizer_model: str,
objective: str,
max_context_size: int,
max_memory_item_size: int,
debug: bool = False
):
"""
Constructs a `MiniAGI` instance.
Args:
agent_model (str): The name of the model to be used as the agent.
summarizer_model (str): The name of the model to be used for summarization.
objective (str): The objective for the agent.
max_context_size (int): The maximum context size in tokens for the agent's memory.
max_memory_item_size (int): The maximum size of a memory item in tokens.
debug (bool, optional): A flag to indicate whether to print debug information.
"""
self.agent = ThinkGPT(
model_name=agent_model,
request_timeout=600,
verbose=False
)
self.summarizer = ThinkGPT(
model_name=summarizer_model,
request_timeout=600,
verbose=False
)
self.objective = objective
self.max_context_size = max_context_size
self.max_memory_item_size = max_memory_item_size
self.debug = debug
self.summarized_history = ""
self.criticism = ""
self.thought = ""
self.proposed_command = ""
self.proposed_arg = ""
self.encoding = tiktoken.encoding_for_model(self.agent.model_name)
def __update_memory(
self,
action: str,
observation: str,
update_summary: bool = True
) -> str:
"""
Updates the agent's memory with the last action performed and its observation.
Optionally, updates the summary of agent's history as well.
Args:
action (str): The action performed by the ThinkGPT instance.
observation (str): The observation made by the ThinkGPT
instance after performing the action.
summary (str): The current summary of the agent's history.
update_summary (bool, optional): Determines whether to update the summary.
Returns:
str: The updated (or unchanged, depending on 'update_summary' value)
summary of the agent's history.
"""
if len(self.encoding.encode(observation)) > self.max_memory_item_size:
observation = self.summarizer.chunked_summarize(
observation, self.max_memory_item_size,
instruction_hint=OBSERVATION_SUMMARY_HINT.format(objective=self.objective)
)
new_memory = f"ACTION\n{action}\nRESULT:\n{observation}\n"
if update_summary:
self.summarized_history = self.summarizer.summarize(
f"Current summary:\n{self.summarized_history}\nAdd to summary:\n{new_memory}",
self.max_memory_item_size,
instruction_hint=HISTORY_SUMMARY_HINT
)
self.agent.memorize(new_memory)
def __get_context(self) -> str:
"""
Retrieves the context for the agent to think and act upon.
Returns:
str: The agent's context.
"""
summary_len = len(self.encoding.encode(self.summarized_history))
if len(self.criticism) > 0:
criticism_len = len(self.encoding.encode(self.criticism))
else:
criticism_len = 0
action_buffer = "\n".join(
self.agent.remember(
limit=32,
sort_by_order=True,
max_tokens=self.max_context_size - summary_len - criticism_len
)
)
return f"SUMMARY\n{self.summarized_history}\nPREV ACTIONS:"\
f"\n{action_buffer}\n{self.criticism}"
def criticize(self):
"""
Criticizes the agent's actions.
"""
context = self.__get_context()
self.criticism = self.agent.predict(
prompt=CRITIC_PROMPT.format(context=context, objective=self.objective)
)
def think(self):
"""
Uses the `ThinkGPT` model to predict the next action the agent should take.
"""
context = self.__get_context()
if self.debug:
print(context)
response_text = self.agent.predict(
prompt=PROMPT.format(context=context, objective=self.objective)
)
if self.debug:
print(f"RAW RESPONSE:\n{response_text}")
PATTERN = r'^<r>(.*?)</r><c>(.*?)</c>\n*(.*)$'
try:
match = re.search(PATTERN, response_text, flags=re.DOTALL | re.MULTILINE)
_thought = match[1]
_command = match[2]
_arg = match[3]
except Exception as exc:
raise InvalidLLMResponseError from exc
# Remove unwanted code formatting backticks
_arg = _arg.replace("```", "")
self.thought = _thought
self.proposed_command = _command
self.proposed_arg = _arg
def read_mind(self) -> tuple:
"""
Retrieves the agent's last thought, proposed command, and argument.
Returns:
tuple: A tuple containing the agent's thought, proposed command, and argument.
"""
_arg = self.proposed_arg.replace("\n", "\\n") if len(self.proposed_arg) < 64\
else f"{self.proposed_arg[:64]}...".replace("\n", "\\n")
return (
self.thought,
self.proposed_command,
_arg
)
def act(self):
"""
Executes the command proposed by the agent and updates the agent's memory.
"""
obs = Commands.execute_command(self.proposed_command, self.proposed_arg)
self.__update_memory(f"{self.proposed_command}\n{self.proposed_arg}", obs)
self.criticism = ""
def user_response(self, response):
"""
Updates the agent's memory with the user's response to its last action.
Args:
response (str): The user's response to the agent's last action.
"""
self.__update_memory(f"{self.proposed_command}\n{self.proposed_arg}", response)
self.criticism = ""
def get_bool_env(env_var: str) -> bool:
'''
Gets the value of a boolean environment variable.
Args:
env_var (str): Name of the variable
'''
return os.getenv(env_var) in ['true', '1', 't', 'y', 'yes']
if __name__ == "__main__":
PROMPT_USER = get_bool_env("PROMPT_USER")
ENABLE_CRITIC = get_bool_env("ENABLE_CRITIC")
if len(sys.argv) != 2:
print("Usage: miniagi.py <objective>")
sys.exit(0)
work_dir = os.getenv("WORK_DIR")
if work_dir is None or not work_dir:
work_dir = os.path.join(Path.home(), "miniagi")
if not os.path.exists(work_dir):
os.makedirs(work_dir)
print(f"Working directory is {work_dir}")
try:
os.chdir(work_dir)
except FileNotFoundError:
print("Directory doesn't exist. Set WORK_DIR to an existing directory or leave it blank.")
sys.exit(0)
miniagi = MiniAGI(
os.getenv("MODEL"),
os.getenv("SUMMARIZER_MODEL"),
sys.argv[1],
int(os.getenv("MAX_CONTEXT_SIZE")),
int(os.getenv("MAX_MEMORY_ITEM_SIZE")),
get_bool_env("DEBUG")
)
while True:
try:
with Spinner():
miniagi.think()
except InvalidLLMResponseError:
print(colored("Invalid LLM response, retrying...", "red"))
continue
(thought, command, arg) = miniagi.read_mind()
print(colored(f"MiniAGI: {thought}\nCmd: {command}, Arg: {arg}", "cyan"))
if command == "memorize_thoughts":
print(colored("MiniAGI is thinking:\n"\
f"{miniagi.proposed_arg}", 'cyan'))
with Spinner():
miniagi.act()
if ENABLE_CRITIC:
miniagi.criticize()
continue
if command == "talk_to_user":
print(colored(f"MiniAGI: {miniagi.proposed_arg}", 'cyan'))
user_input = input('Your response: ')
with Spinner():
miniagi.user_response(user_input)
continue
if PROMPT_USER:
user_input = input('Press enter to perform this action or abort by typing feedback: ')
if len(user_input) > 0:
with Spinner():
miniagi.user_response(user_input)
continue
with Spinner():
miniagi.act()
if ENABLE_CRITIC:
miniagi.criticize()