深入解析BabyAGI：原理、实战与源码拆解

1

docker-compose up

1

注意：BabyAGI是一个任务管理系统，被设计为连续运行。这可能导致API的高使用率、内存和资金。所以，请务必谨慎使用，同时监控系统的行为以及确保应用程序的有效性非常重要。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

#设置表的名称,任务结果存储在这里。
RESULTS_STORE_NAME=baby-agi-interest-table

# BabyAGI实例的名称
INSTANCE_NAME=BabyInterestAGI
COOPERATIVE_MODE=none # local

# 定义目标
OBJECTIVE=培养孩子的兴趣
#OBJECTIVE=Solve world hunger
# 初始任务
#INITIAL_TASK=Develop a task list
INITIAL_TASK=制定任务清单

 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

def main():
    loop = True
    while loop:
        #只要存储中存在任务，则往下执行.
        if not tasks_storage.is_empty():
            # 打印任务列表
            print("\033[95m\033[1m" + "\n*****TASK LIST*****\n" + "\033[0m\033[0m")
            for t in tasks_storage.get_task_names():
                print(" • " + str(t))

            # Step 1: 取出第一个未完成的任务
            task = tasks_storage.popleft()
            #打印接下来要执行的任务
            print("\033[92m\033[1m" + "\n*****NEXT TASK*****\n" + "\033[0m\033[0m")
            print(str(task["task_name"]))

            # 发送给任务执行代理去完成任务，传入两个参数：目标、任务名称。该执行代理向OpenAI的API发送提			   # 示，提示包括AI系统任务、目标和任务本身的描述，返回字符串（string ）形式任务结果。
            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: 丰富结果并存储任务、及其执行结果、ID 在 results_storage 中.
            # 如果需要，在这里丰富结果内容.
            enriched_result = {
                "data": result
            }
            # 如果有丰富结果内容，可以从字典中提取丰富的实际结果，这里没有，已注释。
            # vector = enriched_result["data"]

            result_id = f"result_{task['task_id']}"

			# 任务执行后，会将任务、及其响应的结果、ID保存到results_storage（结果存储）中。
            results_storage.add(task, result, result_id)

            # Step 3: 调用任务创建代理task_creation_agent，使用OpenAI的API基于当前对象和之前任务             # 执行的结果创建新任务，有四个参数：目标、上一个任务的结果、任务描述和当前任务列表。返回新的             # 任务列表
            new_tasks = task_creation_agent(
                OBJECTIVE,
                enriched_result,
                task["task_name"],
                tasks_storage.get_task_names(),
            )

			#把新的任务添加到任务存储(tasks_storage)中保存下来
            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)

			# JOIN_EXISTING_OBJECTIVE(加入已存在的目标)变量控制是否在生产新任务时重新优化排序，该值			 # 默认是false,也即需要重新优化排序.
            if not JOIN_EXISTING_OBJECTIVE:
            	#调用任务排序代理、重新排列任务列表优先级。向OpenAI的API发送提示，返回已重新排序的新任                 #务列表。
                prioritized_tasks = prioritization_agent()
                #将返回的新任务列表替换旧任务列表
                if prioritized_tasks:
                    tasks_storage.replace(prioritized_tasks)

            # 休眠一会儿
            time.sleep(5)
        else:
            print('Done.')
            loop = False


if __name__ == "__main__":
    main()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

def execution_agent(objective: str, task: str) -> str:
    """
    根据给定的目标和先前的上下文执行任务.传入两个参数.
    
    Args:
        objective (str): 执行任务的目的或目标
        task (str): 要执行的任务

    Returns:
        str: AI为给定任务生成字符串格式的响应.

    """
	#获取目标的关联上下文信息，从results_storage中查询之前执行的任务及其响应结果信息。这是在执行任务完成     #后，将任务及其结果保存到results_storage。
	#results_storage的默认实现是使用chroma(向量数据库)。
    context = context_agent(query=objective, top_results_num=5)
    # print("\n****RELEVANT CONTEXT****\n")
    # print(context)
    # print('')
    #构建提示词,传入objective(目标)和context(关联上下文)、task(任务)组装提示词。
    prompt = f'根据以下目标执行一项任务: {objective}.\n'
    if context:
        prompt += '考虑到这些之前完成的任务:' + '\n'.join(context)
    prompt += f'\nYour task: {task}\nResponse:'
    return openai_call(prompt, max_tokens=2000)

 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

def task_creation_agent(
        objective: str, result: Dict, task_description: str, task_list: List[str]
):
    #构建提示词,使用参数"目标、上一个任务的结果、任务描述和当前未完成的任务列表"组装提示词.
    prompt = f"""
您将使用执行代理的结果来创建具有以下目标的新任务: {objective}.
最后完成的任务结果: \n{result["data"]}
这个结果是基于这个任务描述: {task_description}.\n"""

    if task_list:
        prompt += f"未完成的任务: {', '.join(task_list)}\n"
    prompt += "根据结果，返回需要完成的任务列表以实现目标. "
    if task_list:
        prompt += "这些新任务不能与未完成的任务重叠. "

    prompt += """
在响应中每行返回一个任务。结果必须是以下格式的编号列表: 

#. First task
#. Second task

每个条目的编号后面必须跟一个句号。如果你的清单是空的，那就写
"There are no tasks to add at this time."
除非您的列表为空，否则不要在编号列表之前包含任何标题，或者在编号列表之后使用任何其他输出."""

    print(f'\n*****TASK CREATION AGENT PROMPT****\n{prompt}\n')
    #发送给openAI 模型
    response = openai_call(prompt, max_tokens=2000)
    print(f'\n****TASK CREATION AGENT RESPONSE****\n{response}\n')
    #获取响应的新任务列表
    new_tasks = response.split('\n')
    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() and task_id.isnumeric():
                new_tasks_list.append(task_name)
            # print('New task created: ' + task_name)

    out = [{"task_name": task_name} for task_name in new_tasks_list]
    return out

 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

def prioritization_agent():
    //获取当前存储的未完成任务列表
    task_names = tasks_storage.get_task_names()
    bullet_string = '\n'

    prompt = f"""
你需要对以下任务进行排序：: {bullet_string + bullet_string.join(task_names)}
考虑团队的最终目标: {OBJECTIVE}.
任务应该从最高优先级到最低优先级进行排序，其中高优先级的任务是那些作为先决条件或对实现目标更重要的任务。
不要删除任何任务。以下列格式返回排序任务的编号列表:

#. First task
#. Second task

条目必须连续编号，从1开始。每个条目的编号后面必须跟一个句号。
在排名列表之前不包含任何标题，或者在列表之后使用任何其他输出."""

    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 priotritization 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

1
2
3
4

            if not JOIN_EXISTING_OBJECTIVE:
                prioritized_tasks = prioritization_agent()
                if prioritized_tasks:
                    tasks_storage.replace(prioritized_tasks)

 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

class DefaultResultsStorage:
    def __init__(self):
        logging.getLogger('chromadb').setLevel(logging.ERROR)
        # 创建Chroma集合
        chroma_persist_dir = "chroma"
        chroma_client = chromadb.PersistentClient(
            settings=chromadb.config.Settings(
                persist_directory=chroma_persist_dir,
            )
        )

		#配置嵌入(Embedding)模型
        metric = "cosine"
        if LLM_MODEL.startswith("llama"):
            embedding_function = LlamaEmbeddingFunction()
        else:
            embedding_function = OpenAIEmbeddingFunction(api_key=OPENAI_API_KEY)
        self.collection = chroma_client.get_or_create_collection(
            name=RESULTS_STORE_NAME,
            metadata={"hnsw:space": metric},
            embedding_function=embedding_function,
        )

	#添加任务及其响应结果到 collection中.
    def add(self, task: Dict, result: str, result_id: str):

        # Break the function if LLM_MODEL starts with "human" (case-insensitive)
        if LLM_MODEL.startswith("human"):
            return
        # Continue with the rest of the function

        embeddings = llm_embed.embed(result) if LLM_MODEL.startswith("llama") else None
        if (
                len(self.collection.get(ids=[result_id], include=[])["ids"]) > 0
        ):  # Check if the result already exists
            self.collection.update(
                ids=result_id,
                embeddings=embeddings,
                documents=result,
                metadatas={"task": task["task_name"], "result": result},
            )
        else:
            self.collection.add(
                ids=result_id,
                embeddings=embeddings,
                documents=result,
                metadatas={"task": task["task_name"], "result": result},
            )

    #检索相关信息
    def query(self, query: str, top_results_num: int) -> List[dict]:
        count: int = self.collection.count()
        if count == 0:
            return []
        results = self.collection.query(
            query_texts=query,
            n_results=min(top_results_num, count),
            include=["metadatas"]
        )
        return [item["task"] for item in results["metadatas"][0]]

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

# 任务存储，只有一个实例，数据保存在一个队列中。
class SingleTaskListStorage:
    def __init__(self):
        self.tasks = deque([])
        self.task_id_counter = 0

    def append(self, task: Dict):
        self.tasks.append(task)

    def replace(self, tasks: List[Dict]):
        self.tasks = deque(tasks)

    def popleft(self):
        return self.tasks.popleft()

    def is_empty(self):
        return False if self.tasks else True

    def next_task_id(self):
        self.task_id_counter += 1
        return self.task_id_counter

    def get_task_names(self):
        return [t["task_name"] for t in self.tasks]