本文将利用LangChain实现BabyAGI。通过本文内容，读者可以更加直观地看到每一步骤的运行情况，并且也可以在自己的环境中进行实验。

BabyAGI介绍

    BabyAGI是由Yohei Nakajima在2023年5月发布的一个自治的AI代理程序代码。这种自治的AI代理旨在根据给定的目标生成和执行任务。它利用OpenAI、Pinecone、LangChain和Chroma来自动化任务并实现代理特定目标。在Agent模块中，可以把AgentExecutor看作一个项目经理，其实BabyAGI也可以被看作一个项目经理管理项目。BabyAGI通过创建、优先处理和执行任务列表来实现代理特定的目标。它还适应变化，并进行必要的调整以确保达到目标。与项目经理一样，BabyAGI具有从以前的经验中学习并做出明智决策的能力。

    我们也可以认为BabyAGI是计算机中AI驱动的个人助手。通过解释给定的目标，它创建了一个所需任务的列表，然后执行它们。每完成一个任务后，BabyAGI都会评估结果并相应地调整其方法。BabyAGI的独特之处在于它能够从试验和错误反馈中学习，做出类似人类的认知决策。它还可以编写和运行代码来实现特定的目标。

    使用BabyAGI的好处是，可以让我们有更多的时间专注于更高价值的任务，如决策和创意项目。在原BabyAGI项目中，BabyAGI按照以下步骤来创建Agent,承担不同的任务，开展自动化任务并联合这些Aget以实现目标。下面依然遵照这样的步骤，使用LangChain内部的模块功能，创建Agent,实现与BabyAGI相同能力的Agent。下面先了解一下原来BabyAGI的实施步骤（如图所示是作者绘制的流程图)。

1. 设置明确的目标：首先，用户设置BabyAGI将完成的目标。
2. 任务生成（Agent):接下来，BabyAGI将使用诸如GPT-4之类的大语言模型，将目标细分为一系列潜在任务。然后将任务列表存储在长期内存（向量数据库)中供将来参考。
3. 任务优先级(Agent):有了任务列表后，BabyAGI将使用其推理能力评估任务并根据它们的重要性和依赖性对任务进行优先排序，以达到最终的结果。BabyAGI将决定首先执行哪个任务。
4. 任务执行(Agent):然后，BabyAGI将执行并完成任务。执行的结果和收集到的信息也将被保存在长期记忆中供将来使用。
5. 评估和创建新任务：执行任务后，BabyAGI将使用其推理能力评估剩余的任务和先前执行的结果。基于评估，它将创建要完成的新任务，以达到最终的目标。
6. 重复：重复这些步骤，直到BabyAGI实现原始目标或用户干预为止。BabyAGI将不断评估目标的进展，并相应地调整任务列表和优先级，以有效地达到期望的结果。

环境与工具

    对于此次实验，需要两个主要工具：OpenAI及一个搜索引擎API。这两者将会协同完成BabyAGI的构建。

!pip -q install langchain huggingface_hub openai google-search-results
tiktoken cohere faiss-cpu
import os
os.environ["OPENAI API KEY"]="填人你的OPENAI密钥"
os,environ["SERPAPI API KEY"]="填人你的SERPAPI密钥"

    导入工具：

import os
from collections import deque
from typing import Dict,List,Optional,Any
from langchain import LLMChain,OpenAI,PromptTemplate
from langchain.embeddings import OpenAIEmbeddings
from langchain.11ms import BaseLLM
from langchain.vectorstores.base import VectorStore
from pydantic import BaseModel,Field
from langchain.chains.base import Chain

向量存储

    在此示例中使用了FAISS向量存储。这是一种内存存储技术，无须进行任何外部调用，例如向Pinecone请求。但如果你愿意，完全可以改变其中的一些设定，将其连接到Pinecone。向量存储是利用OpenAI的嵌人模型进行的。先导人FAISS向量库：

from langchain.vectorstores import FAISS
from langchain.docstore import InMemoryDocstore

    在构建一个特定的嵌人模型，生成向量索引并存储这些向量时，可以按照以下步骤来操作。首先，需要选择一个适当的嵌人模型。这种模型可以是词嵌人模型，如Word2Vec或GloVe,也可以是句子嵌人模型，如BERT或者Doc2Vec。这些模型通过将词或句子映射到高维度的向量空间，实现了对词或句子语义的捕捉。选择哪种嵌入模型主要取决于处理的任务特性和数据的特点。这里使用的是OpenAI的文本嵌人模型。OpenAI的文本嵌人模型可以精确地嵌入大段文本，输出1536维的向量列表。

# Define your embedding model
embeddings_model = OpenAIEmbeddings (）

    其次，对文本数据进行处理，生成相应的嵌人向量。在生成向量后，需要构建一个索引，以便能够高效地查询和比较向量。

# Initialize the vectorstore as empty
import faiss
embedding_size = 1536
index_faiss.IndexFlatL2 (embedding_size)

    最后，需要将生成的向量和构建的索引进行存储。

vectorstore = FAISS (embeddings_model.embed_query,index,InMemoryDocstore(()),())

构建任务链

    LangChain的好处在于，可以让我们清楚地看到链组件在执行哪些操作，以及它们的提示是什么。其中有3个主要链组件：创建任务链、任务优先级链和执行链。这些链组件都在为达成整体目标而工作，它们会生成一系列任务。通过定义TaskCreationChain类实现创建任务链的功能，这个类定义了一个名为TaskCreationChain的类，它的主要职责是基于已有的任务结果和目标，自动生成新的任务。这个类是LLMChain的子类，专门用于生成任务。

    该类有一个类方法from llm,这个方法接受一个BaseLLM类型的对象和一个可选的布尔参数verbose.from_llm方法中定义了一个模板字符串task_creation_template,这个模板用于描述如何从已有任务的结果和描述，以及未完成任务的列表中，生成新的任务。

class TaskCreationChain (LLMChain):"""Chain to generates tasks."""@classmethoddef from_llm(cls,llm:BaseLLM,verbose:bool True)->LLMChain:"""Get the response parser.""task_creation_template =("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:(incomplete tasks).""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."prompt PromptTemplatetemplate=task creation_template,input_variables=["result","task description","incomplete tasks","objective"],return cls (prompt=prompt,llm=llm,verbose=verbose)

    这些步骤看起来很简单，但这里就是你可以进行修改，从而使AI更符合你需求的地方。这个链组件的主要职责是将传人的任务进行清理，重新设置它们的优先级，以便按照你的最终目标进行排序。任务优先级链组件不会删除任何任务，而是将任务以编号列表的形式返回。

class TaskPrioritizationChain (LLMChain):"""Chain to prioritize tasks."""@classmethoddef from 11m(cls,11m:BaseLLM,verbose:bool True)->LLMChain:"""Get the response parser."""task_prioritization template ="You are an task prioritization AI tasked with cleaning theformatting 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)."prompt = PromptTemplate(template=task prioritization template,input_variables=["task_names","next task_id","objective"],)return cls(prompt=prompt,llm=llm,verbose=verbose)

    在这个过程中，定义了一个执行代理，并传递了一些工具给它。这个执行代理是一个计划者，能够为给定的目标制定一个待办事项清单。传递搜索和待办事项这两种工具给它，是为了让它能够在需要的时候进行搜索或者制定待办事项清单。

from langchain.agents import ZeroShotAgent,Tool,AgentExecutor
from langchain import OpenAI,SerpAPIWrapper,LLMChain
todo prompt PromptTemplate.from template("You are a planner
who is an expert at coming up with a todo list for a given objective.
Come up with a todo list for this objective:(objective)")
todo_chain LLMChain(llm=OpenAI(temperature=0),prompt=todo_prompt)
search SerpAPIWrapper()
tools=Tool(name "Search",func=search.run,description="useful for when you need to answer questions aboutcurrent events"）,Tool(name "TODO",func=todo chain.run,description="useful for when you need to come up with todo lists.Input:an objective to create a todo list for.Output:a todo listfor that objective.Please be very clear what the objective is!")]
prefix = """You are an AI who performs one task based on the following
objective:(objective}.Take into account these previously completed tasks:(context}.
"""
suffix="""Question:(task)(agent scratchpad)
"""
prompt = ZeroShotAgent.create_prompt(
tools,
prefix=prefix,
suffix=suffix,
input variables=["objective","task","context","agent scratchpad"]
)

    可以看到，这个执行器使用ZeroShotAgent将提示词、前缀/后缀及输人变量一并输人。通过这种方式，可以让我们更清楚地看到在程序执行过程中，这些部分是如何组合在一起工作的。下面定义一组函数，主要负责任务的创建（get next task)、优先级排序(prioritize tasks和get top tasks)和执行（execute task)。它使用了LLMChain对象来运行不同的任务管理逻辑，并使用vectorstore来进行相似度搜索。这一系列功能合在一起为任务管理提供了一套完整的解决方案。

def get_next_task(task_creation_chain:LLMChain,result:Dict,task description:str, task list:List(str],objective:str)->List[Dict]:"""Get the next task."""incomplete tasks "".join(task_list)response task_creation_chain.run(result=result,task_description=task description,incomplete_tasks=incomplete_tasks,objective=objectivenew_tasks response.split ('\n')return [("task name":task name)for task name in new tasks if task name.strip()]
def prioritize_tasks(task prioritization chain:LLMChain,this task id:int,task list:List [Dict],objective:str)->List[Dict]:"""Prioritize tasks."""task_names = [t["task name"]for t in task list]next_task_id = int(this_task_id)+1response_task_prioritization chain.run(task_names=task_names,next_task_id=next_task_id,objective=objective)new_tasks = response.split ('\n')prioritized_task_list [for task_string in new_tasks:if not task_string.strip():continuetask_parts = task_string.strip().split(".",1)if len(task_parts)==2:task_id = task_parts[0].strip()task_name = task_parts[1].strip()prioritized_task_list.append ("task_id":task_id,"task_name":task_name))return prioritized_task_list
def get top tasks(vectorstore,query:str,k:int)->List[str]:"""Get the top k tasks based on the query.""results = vectorstore.similarity_search_with_score(query,k=k)if not results:return [sorted_results,_=zip(*sorted(results,key=lambda x:x[1],reverse=True))return [str(item.metadata['task'])for item in sorted results]
def execute_task(vectorstore,execution chain:LLMChain,objective:str,task:str,k:int 5)->str:"""Execute a task."context =_get_top_tasks(vectorstore,query=objective,k=k)return execution_chain.run(objective=objective,context=context,task=task)