LangChain入门：24.通过Baby AGI实现自动生成和执行任务

news2024/12/21 17:05:06

随着 ChatGPT 的崭露头角，我们迎来了一种新型的代理——Autonomous Agents（自治代理或自主代理）。

这些代理的设计初衷就是能够独立地执行任务，并持续地追求长期目标。

在 LangChain 的代理、工具和记忆这些组件的支持下，它们能够在无需外部干预的情况下自主运行，这在真实世界的应用中具有巨大的价值。

目前，GitHub 上已有好几个备受关注的“网红”项目，如 AutoGPT、BabyAGI 和 HuggingGPT，它们都代表了自治代理的初步尝试。尽管这些代理仍处于实验阶段，但潜力十分巨大。它们都是基于 LangChain 框架构建的。

通过 LangChain，你可以在这些开源项目中轻松地切换和测试多种 LLM、使用多种向量存储作为记忆，以及充分利用 LangChain 的丰富工具集。今天我就带着大家尝试通过 LangChain 完成一个 BabyAGI 的实现。

Baby AGI

BabyAGI 是由中岛洋平（Yohei Nakajima）于 2023 年 3 月 28 日开发的自主任务驱动 AI 系统。核心在于，它可以根据设定的目标生成、组织、确定优先级以及执行任务。它也使用 OpenAI 的 GPT-4 语言模型来理解和创建任务，利用 Pinecone 向量搜索来存储和检索特定任务的结果，提供执行任务的上下文，并采用 LangChain 框架进行决策。

BabyAGI 尝试使用预定义的目标进行自我驱动，自动化个人任务管理。它不仅可以自动生成和执行任务，而且还可以根据完成的任务结果生成新任务，并且可以实时确定任务的优先级。

与传统的 AI 工具（如 ChatGPT）不同，BabyAGI 不仅仅是解释查询和提供响应，而且能够根据目标生成任务列表，连续执行它们，并根据先前任务的输出适应性地创建更多任务。

和 Auto-GPT 一样，该系统发布后广受关注，也被某些人誉为完全自主人工智能的起点。

在 BabyAGI 中，你向系统提出一个目标之后，它将不断优先考虑需要实现或完成的任务，以实现该目标。具体来说，系统将形成任务列表，从任务列表中拉出优先级最高的第一个任务，使用 OpenAI API 根据上下文将任务发送到执行代理并完成任务，一旦这些任务完成，它们就会被存储在内存（或者 Pinecone 这类向量数据库）中，然后，根据目标和上一个任务的结果创建新任务并确定优先级。

整个过程如下图所示：
在这里插入图片描述
在这个过程中，驱动任务的是三个不同作用的代理。分别是执行代理 execution_agent，任务创建代理 task_creation_agent，以及优先级设置代理 prioritization_agent。

执行代理，是系统的核心，利用 OpenAI 的 API 来处理任务。这个代理的实现函数有两个参数，目标和任务，用于向 OpenAI 的 API 发送提示，并以字符串形式返回任务结果。
任务创建代理，通过 OpenAI 的 API根据当前对象和先前任务的结果创建新任务。这个代理的实现函数有四个参数，目标、上一个任务的结果、任务描述和当前任务列表。这个代理会向 OpenAI 的 API 发送一条提示，该 API将以字符串形式返回新任务列表。然后，该函数将以字典列表的形式返回这些新任务，其中每个字典都包含任务的名称。
优先级设置代理，负责任务列表的排序和优先级，仍然是通过调用 OpenAI 的 API来重新确定任务列表的优先级。这个代理的实现函数有一个参数，即当前任务的 ID。这个代理会向 OpenAI 的 API发送提示，并返回已重新优先排序为编号列表的新任务列表。

接下来，我就用这个 BabyAGI 的框架来开发一个能够根据气候变化自动制定鲜花存储策略的 AI 智能代理。

根据气候变化自动制定鲜花存储策略

下面，我们就解析一下 LangChain 中 BabyAGI 的具体实现。这里的 “BabyAGI” 是一个简化版的实现，其核心功能是自动创建、优先级排序和执行任务。

首先，我们导入相关的库。

# 设置API Key
import os
os.environ["OPENAI_API_KEY"] = 'Your OpenAI API Key

# 导入所需的库和模块
from collections import deque
from typing import Dict, List, Optional, Any
from langchain.chains import LLMChain
from langchain.prompts import PromptTemplate
from langchain_openai import OpenAIEmbeddings
from langchain_openai import ChatOpenAI
from langchain.llms import BaseLLM
from langchain.vectorstores.base import VectorStore
from pydantic import BaseModel, Field
from langchain.chains.base import Chain
from langchain_community.vectorstores import FAISS
import faiss
from langchain_community.docstore import InMemoryDocstore

然后，我们初始化 OpenAIEmbedding 作为嵌入模型，并使用 Faiss 作为向量数据库存储任务信息。

# 定义嵌入模型
embeddings_model = OpenAIEmbeddings(
        openai_api_key='你的api key',
        base_url="https://api.chatanywhere.tech/v1",
    )
# 初始化向量存储
embedding_size = 1536
index = faiss.IndexFlatL2(embedding_size)
vectorstore = FAISS(embeddings_model.embed_query, index, InMemoryDocstore({
   }), {
   })

下面是定义任务生成链，基于给定的条件，这个链可以创建新任务。例如，它可以根据最后一个完成的任务的结果来生成新任务。

# 任务生成链
class TaskCreationChain(LLMChain):
    """负责生成任务的链"""
    @classmethod
    def from_llm(cls, llm: BaseLLM, verbose: bool = True) -> LLMChain:
        """从LLM获取响应解析器"""
        task_creation_template = (
            "You are a 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 = PromptTemplate(
            template=task_creation_template,
            input_variables=[
                "result",
                "task_description",
                "incomplete_tasks",
                "objective",
            ],
        )
        return cls(prompt=prompt, llm=llm, verbose=verbose)

下面是定义任务优先级链，这个链负责重新排序任务的优先级。给定一个任务列表，它会返回一个新的按优先级排序的任务列表。

# 任务优先级链
class TaskPrioritizationChain(LLMChain):
    """负责任务优先级排序的链"""
    @classmethod
    def from_llm(cls, llm: BaseLLM, verbose: bool = True) -> LLMChain:
        """从LLM获取响应解析器"""
        task_prioritization_template = (
            "You are a 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}."
        )
        prompt = PromptTemplate(
            template=task_prioritization_template,
            input_variables=["task_names", "next_task_id", "objective"],
        )
        return cls(prompt=prompt, llm=llm, verbose=verbose)

下面是定义任务执行链，这个链负责执行具体的任务，并返回结果。

# 任务执行链
class ExecutionChain(LLMChain):
    """负责执行任务的链"""

    @classmethod
    def from_llm(cls, llm: BaseLLM, verbose: bool = True) -> LLMChain:
        """从LLM获取响应解析器"""
        execution_template = (
            "You are an AI who performs one task based on the following objective: {objective}."