forked from yoheinakajima/babyagi
-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathrun.py
executable file
·203 lines (160 loc) · 6.89 KB
/
run.py
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
#!/opt/homebrew/Caskroom/miniconda/base/envs/babyagi311/bin/python
from dotenv import load_dotenv
import os
import chromadb
import tiktoken as tiktoken
# import re
# from baby_agi import BabyAGI
from tasks import TaskListStore, TaskNode
# default opt out of chromadb telemetry.
from chromadb.config import Settings
from utils import get_completion
os.chdir(os.path.dirname(os.path.abspath(__file__)))
cwd = os.getcwd()
print(cwd)
def read_settings_from_env():
load_dotenv()
agent_settings = {}
agent_settings["vectordb_client"] = chromadb.Client(
Settings(anonymized_telemetry=False)
)
agent_settings["LLM_MODEL"] = os.getenv("LLM_MODEL", "").lower()
agent_settings["LLAMA_API_PATH"] = os.getenv("LLAMA_API_PATH", "")
agent_settings["RESULTS_STORE_NAME"] = os.getenv("RESULTS_STORE_NAME", "")
agent_settings["AGENT_NAME"].AGENT_NAME = os.getenv("AGENT_NAME", "")
agent_settings["OBJECTIVE"] = os.getenv("OBJECTIVE", "")
agent_settings["INITIAL_TASK"] = os.getenv("INITIAL_TASK", "")
return agent_settings
# def prioritization_agent():
# task_names = tasks_storage.get_task_names()
# bullet_string = "\n"
# prompt = f""""""
# print(f"\n****TASK PRIORITIZATION AGENT PROMPT****\n{prompt}\n")
# response = openai_call(prompt, max_tokens=2000)
# print(f"\n****TASK PRIORITIZATION AGENT RESPONSE****\n{response}\n")
# if not response:
# print(
# "Received empty response from prioritization agent. Keeping task list unchanged."
# )
# return
# new_tasks = response.split("\n") if "\n" in response else [response]
# new_tasks_list = []
# for task_string in new_tasks:
# task_parts = task_string.strip().split(".", 1)
# if len(task_parts) == 2:
# task_id = "".join(s for s in task_parts[0] if s.isnumeric())
# task_name = re.sub(r"[^\w\s_]+", "", task_parts[1]).strip()
# if task_name.strip():
# new_tasks_list.append({"task_id": task_id, "task_name": task_name})
# return new_tasks_list
# Execute a task based on the objective and five previous tasks
# def execution_agent(objective: str, task: str) -> str:
# """
# Executes a task based on the given objective and previous context.
# Args:
# objective (str): The objective or goal for the AI to perform the task.
# task (str): The task to be executed by the AI.
# Returns:
# str: The response generated by the AI for the given task.
# """
# context = context_agent(query=objective, top_results_num=5)
# # print("\n****RELEVANT CONTEXT****\n")
# # print(context)
# # print('')
# prompt = f"Perform one task based on the following objective: {objective}.\n"
# if context:
# prompt += "Take into account these previously completed tasks:" + "\n".join(
# context
# )
# prompt += f"\nYour task: {task}\nResponse:"
# return openai_call(prompt, max_tokens=2000)
# Get the top n completed tasks for the objective
# def context_agent(query: str, top_results_num: int):
# """
# Retrieves context for a given query from an index of tasks.
# Args:
# query (str): The query or objective for retrieving context.
# top_results_num (int): The number of top results to retrieve.
# Returns:
# list: A list of tasks as context for the given query, sorted by relevance.
# """
# results = results_storage.query(query=query, top_results_num=top_results_num)
# # print("****RESULTS****")
# # print(results)
# return results
def describe_task(task: TaskNode, **kwargs) -> dict:
prompt = open("prompts/describe/describe_task.md", "r").read()
return get_completion(prompt.format(TASK=task.name), **kwargs)
def describe_task_detail(task: TaskNode, **kwargs) -> dict:
question_list = (
open("prompts/describe/describe_questions.csv", "r").read().split("\n")
)
response = {}
for question in question_list:
prompt = open("prompts/describe/describe_task_question.md", "r").read()
response[question] = get_completion(
prompt.format(
TASK=task.name,
TASK_DESCRIPTION=task.description,
TASK_QUESTION=question,
),
**kwargs
)
return response
def generate_faq_task(task: TaskNode, **kwargs) -> dict:
prompt = open("prompts/generate_faq.md", "r").read()
return get_completion(
prompt.format(TASK_NAME=task.name, TASK_DESCRIPTION=task.description), **kwargs
)
def generate_tasks(goal: str, description: str, **kwargs):
# loads `./prompts/create_task.txt` and generates a task
prompt = open("prompts/create_task.md", "r").read()
prompt.format(OBJECTIVE=goal)
def main():
load_dotenv()
loop = True
# Make a Fried Egg
initial_task = TaskNode("Make a Fried Egg")
task_description = describe_task(initial_task)
print(describe_task_detail(initial_task))
print(generate_faq_task(initial_task))
# while loop:
# # As long as there are tasks in the storage...
# if not tasks_storage.is_empty():
# # Print the task list
# print("\033[95m\033[1m" + "\n*****TASK LIST*****\n" + "\033[0m\033[0m")
# for t in tasks_storage.get_task_names():
# print(" • " + str(t))
# # Send to execution function to complete the task based on the context
# result = execution_agent(OBJECTIVE, str(task["task_name"]))
# print("\033[93m\033[1m" + "\n*****TASK RESULT*****\n" + "\033[0m\033[0m")
# print(result)
# # Step 2: Enrich result and store in the results storage
# # This is where you should enrich the result if needed
# enriched_result = {"data": result}
# # extract the actual result from the dictionary
# # since we don't do enrichment currently
# # vector = enriched_result["data"]
# result_id = f"result_{task['task_id']}"
# results_storage.add(task, result, result_id)
# # Step 3: Create new tasks and re-prioritize task list
# # only the main instance in cooperative mode does that
# new_tasks = task_creation_agent(
# OBJECTIVE,
# enriched_result,
# task["task_name"],
# tasks_storage.get_task_names(),
# )
# print("Adding new tasks to task_storage")
# for new_task in new_tasks:
# new_task.update({"task_id": tasks_storage.next_task_id()})
# print(str(new_task))
# tasks_storage.append(new_task)
# prioritized_tasks = prioritization_agent()
# if prioritized_tasks:
# tasks_storage.replace(prioritized_tasks)
# else:
# print("Done.")
# loop = False
if __name__ == "__main__":
main()