上一节中我们简单的介绍了一个BabyAgi的主要功能，以及主要功能。没有看过的朋友可以直接点击babyagi: 人工智能任务管理系统前去查看
在本篇中，我们将深入探讨通过prompt和context约束GPT-4实现的BabyAGI系统的源代码解析与应用。文章将详细介绍BabyAGI的核心组件，如任务执行、任务创建与任务优先级等功能，以及如何结合GPT-4、Pinecone和LangChain等技术实现这一高效的任务驱动智能代理。通过对源代码的分析，我们将揭示这一系统在处理任务、生成新任务以及实时优先级调整方面的关键技术细节，并探讨潜在的改进和风险。本文旨在帮助读者更好地理解和利用GPT-4的强大功能，以实现多样化的应用场景。
源代码
为了尽量节省篇幅，我这里就不直接贴源代码了。大家可以去github上babyagi 查看源代码，其实也就只有一个python文件，200行代码左右，因为目前项目更新的比较快，本文采用的是2023年04月07日的之前的版本进行解读。
主要流程
先给大家看一下我自己通过阅读源码，整理出来的babyagi执行流程图。这里大家先看一下全部流程，先有个印象
主要流程代码如下所示：
# Add the first task
first_task = {
"task_id": 1,
"task_name": YOUR_FIRST_TASK
}
add_task(first_task)
# Main loop
task_id_counter = 1
while True:
if task_list:
# Print the task list
print("33[95m33[1m"+"n*****TASK LIST*****n"+"33[0m33[0m")
for t in task_list:
print(str(t['task_id'])+": "+t['task_name'])
# Step 1: Pull the first task
task = task_list.popleft()
print("33[92m33[1m"+"n*****NEXT TASK*****n"+"33[0m33[0m")
print(str(task['task_id'])+": "+task['task_name'])
# Send to execution function to complete the task based on the context
result = execution_agent(OBJECTIVE,task["task_name"])
this_task_id = int(task["task_id"])
print("33[93m33[1m"+"n*****TASK RESULT*****n"+"33[0m33[0m")
print(result)
# Step 2: Enrich result and store in Pinecone
enriched_result = {'data': result} # This is where you should enrich the result if needed
result_id = f"result_{task['task_id']}"
# print(f"任务执行结果 {enriched_result} 任务执行ID {result_id}")
vector = enriched_result['data'] # extract the actual result from the dictionary
index.upsert([(result_id, get_ada_embedding(vector),{"task":task['task_name'],"result":result})])
# Step 3: Create new tasks and reprioritize task list
new_tasks = task_creation_agent(OBJECTIVE,enriched_result, task["task_name"], [t["task_name"] for t in task_list])
for new_task in new_tasks:
task_id_counter += 1
new_task.update({"task_id": task_id_counter})
add_task(new_task)
prioritization_agent(this_task_id)
- 根据.env文件中读取的FIRST_TASK创建第一个任务并设置任务id,初始化任务计数器
task_id_counter - 进入循环，判断任务列表当中是否有任务，如果有任务那么就将当前任务列表当中的任务打印出来
- 从deque队列的尾部取出第一个元素，也就是第一个任务，打印任务id和任务名称
- 将第一个任务放入执行代理
execution_agent并获取执行结果，并将当期先任务的id保存在this_task_id当中 - 将任务结果使用
get_ada_embedding变为高维向量，作为向量对象的值，再任务名称和任务结果封装为一个map作为向量对象的meta元数据。最后将向量对象插入到向量索引当中 - 将用户设定的目标以及刚刚执行的任务，任务执行结果，当前任务列表，放入到任务创建代理函数
task_creation_agent创建出新的任务，并将新的任务添加到任务列表当中， - 调用
prioritization_agent函数，根据当前任务列表和GPT-4的指导，重新调整任务的优先级。
处理函数
在介绍完了大概得执行流程之后，下面我们来分别讲解下各个处理的函数的实现方法以及原理。
GPT处理函数
该函数用于处理传入的prompt，在使用对应的模型进行处理之后，返回对应的complete。
def openai_call(prompt: str, model: str = OPENAI_API_MODEL, temperature: float = 0.5, max_tokens: int = 100):
if not model.startswith('gpt-'):
# Use completion API
response = openai.Completion.create(
engine=model,
prompt=prompt,
temperature=temperature,
max_tokens=max_tokens,
top_p=1,
frequency_penalty=0,
presence_penalty=0
)
return response.choices[0].text.strip()
else:
# Use chat completion API
messages=[{"role": "user", "content": prompt}]
response = openai.ChatCompletion.create(
model=model,
messages=messages,
temperature=temperature,
max_tokens=max_tokens,
n=1,
stop=None,
)
return response.choices[0].message.content.strip()
熟悉openai的api的小伙伴应该知道，openai的对话模型需要我们传入一个prompt，然后openai根据使用的model以及相关参数的设置(具体看openai开发者官方网站介绍)。它就会返回给我们一个complete也就是我们需要的结果。这里opanai_call函数只需要接收一个参数prompt即可，返回值为complete
向量处理函数
get_ada_embedding
顾名思义该函数的用处是将text文本转化为向量的。
def get_ada_embedding(text):
text = text.replace("n", " ")
return openai.Embedding.create(input=[text], model="text-embedding-ada-002")["data"][0]["embedding"]
这里使用的是openai向量转化模型text-embedding-ada-002，该模型的使用方法如下
这样一来我们就可以按照openai的规则将文本转化为其支持的向量。
context_agent
向量索引查询函数，接受传入的查询条件，以及查询结果限制数量n，返回与查询条件相关度最高的任务。
def context_agent(query: str, n: int):
query_embedding = get_ada_embedding(query)
results = index.query(query_embedding, top_k=n, include_metadata=True)
#print("***** RESULTS *****")
#print(results)
sorted_results = sorted(results.matches, key=lambda x: x.score, reverse=True)
return [(str(item.metadata['task'])) for item in sorted_results]
执行代理
该函数的主要逻辑是传入用户设置的目标objective,和需要执行的任务名称，返回执行的任务结果，当然执行代理在执行过程当中，也会通过context_agent函数获取与目标最相关的任务。
def execution_agent(objective: str, task: str) -> str:
context=context_agent(query=objective, n=5)
# print("n*******RELEVANT CONTEXT******n")
# print(context)
prompt =f"You are an AI who performs one task based on the following objective: {objective}.nTake into account these previously completed tasks: {context}nYour task: {task}nResponse:"
return openai_call(prompt, temperature=0.7, max_tokens=2000)
核心prompt
为了方便大家理解prompt对模型的约束，这里提供prompt的中英文对照版
原始prompt
You are an AI who performs one task based on the following objective:{objective}
Take into account these previously completed tasks:{context}
Your task:{task}
Response:
中文版
你是一个基于以下目标执行任务的AI: objective
考虑这些之前完成的任务:context
你的任务:task
执行结果:也就是AI的回答
通过这些prompt我们成功的给ai设置的任务，这个任务主要切合我们设置的目标，还有考虑之前做的任务，避免相似的任务重复执行，让AI有了上下文关联避免了无效的任务执行
任务创建代理
该函数的主要任务是根据给定的任务目标、已完成任务的结果、已完成任务的描述以及未完成任务列表，使用 GPT-4 生成一组新任务
def task_creation_agent(objective: str, result: Dict, task_description: str, task_list: List[str]):
prompt = f"You are an task creation AI that uses the result of an execution agent to create new tasks with the following objective: {objective}, The last completed task has the result: {result}. This result was based on this task description: {task_description}. These are incomplete tasks: {', '.join(task_list)}. Based on the result, create new tasks to be completed by the AI system that do not overlap with incomplete tasks. Return the tasks as an array."
response = openai_call(prompt)
new_tasks = response.split('n')
return [{"task_name": task_name} for task_name in new_tasks]
接收以下四个参数：
objective：当前任务的目标。result：已完成任务的结果。task_description：已完成任务的描述。task_list：未完成任务的列表。
核心prompt
//prompt1
You are an task creation AI that uses the result of an execution agent to create new tasks with the following objective: {objective},
//prompt2
The last completed task has the result: {result}.
//prompt3
This result was based on this task description: {task_description}.
//prompt4
These are incomplete tasks: {', '.join(task_list)}.
//prompt5
Based on the result, create new tasks to be completed by the AI system that do not overlap with incomplete tasks. Return the tasks as an array.
中文版
//prompt1
你是一个任务创建AI，使用执行代理的结果来创建新的任务，目标如下:{objective}
//prompt2
最近一个完成的任务的结果是:{task_description}.
//prompt3
最近一个完成的任务名称是:{task_description}
//prompt4
以下是未完成的任务: {', '.join(task_list)}.
//prompt5
根据结果，创建新的任务，由AI系统完成，不与未完成的任务重叠。以数组的形式返回任务。
通过这些prompt我们可以规定AI在创建新的任务的时候，避免当前任务列表当中已经存在的任务，并参考最近一次任务执行结果，以及任务名称。来实时生成新的不重复的任务下一次循环执行。
优先级代理
主要作用是重新对任务列表进行优先级排序。它根据提供的当前任务ID（this_task_id）和全局任务列表（task_list），使用GPT-4为任务生成一个新的优先级顺序
def prioritization_agent(this_task_id: int):
global task_list
task_names = [t["task_name"] for t in task_list]
next_task_id = int(this_task_id)+1
prompt = f"""You are an task prioritization AI tasked with cleaning the formatting of and reprioritizing the following tasks: {task_names}. Consider the ultimate objective of your team:{OBJECTIVE}. Do not remove any tasks. Return the result as a numbered list, like:
#. First task
#. Second task
Start the task list with number {next_task_id}."""
response = openai_call(prompt)
new_tasks = response.split('n')
task_list = deque()
for task_string in new_tasks:
task_parts = task_string.strip().split(".", 1)
if len(task_parts) == 2:
task_id = task_parts[0].strip()
task_name = task_parts[1].strip()
task_list.append({"task_id": task_id, "task_name": task_name})
核心prompt
原prompt
//prompt1
You are an task prioritization AI tasked with cleaning the formatting of and reprioritizing the following tasks: {task_names}.
//prompt2
Consider the ultimate objective of your team:{OBJECTIVE}.
//prompt3
Do not remove any tasks. Return the result as a numbered list, like:
#. First task
#. Second task
Start the task list with number {next_task_id}.
中文版本
//prompt1
你是一个任务优先级AI，任务是清理以下任务的格式和重新排序:{task_names}
//prompt2
需要考虑团队的最终目标:{OBJECTIVE}
//prompt3
不要删除任何任务。返回结果为一个编号的列表，
如:
#。第一个任务
#。第二个任务
使用数字{next_task_id}启动任务列表。
该prompt构建一个包含任务名称、团队的最终目标（OBJECTIVE）以及要求按顺序排列任务的字符串提示（prompt）将提示传递给 GPT-4，以获取新的任务优先级顺序。
当然GPT4返回给我们的结果肯定是一个字符串，如下形式，所以我们需要使用分隔符 n ,将字符串转化为结果数组new_tasks，然后在将new_tasks中的每一个任务字符串(1.xxxxx) 切割开来，分为task_id部分和task内容部分，存入到全局task_list当中。
"1.xxxn2.xxxn3.xxxxxn....."
结尾
本文探讨了如何在GPT-4模型中运用prompt和context对AI系统施加约束，以实现任务自动生成与自主执行的能力。这些方法有助于提高AI系统的针对性、可控性和可靠性，为实现真正的人工智能（AGI）提供了重要基础。
我们阐述了通过精心设计的prompt和充分利用上下文信息，如何确保AI在生成新任务时遵循预定规则，避免与现有任务重复。同时，AI系统能够根据最近一次任务执行结果和任务描述，实时生成不重复的任务并进行下一轮的自主执行。这种自主性有助于提高AI系统的工作效率，降低资源消耗和冗余操作。
将prompt与GPT-4模型相结合，有望更好地应对实际应用场景中的挑战，为构建更智能、更自主的AI系统奠定基础。随着技术的不断发展，我们预期未来的AGI将能够更深入地理解人类需求，自主地创建和执行任务，为我们的生活和工作带来更多的便利与价值。这种进步将推动人工智能研究领域的发展，为未来AI系统的应用提供更广阔的前景。
🔥最后欢迎大家留言与我一起探讨问题。喜欢的话也可以给我点个赞，顺便关注我一波🔥
