LangChain Agent：创建你自己的Multi-action Agent

1. 回顾

1.1 什么是LangChain Agent？

自主Agent是无需人工干预即可做出决策并采取行动的系统，但它们可能会表现出不可预测的行为。为了解决这个问题，Agent了解他们可以做什么（操作）非常重要。在LangChain的ReAct型Agent中，Agent可以采取的行动被写在提示中，并根据提示做出决策和行动。这使您可以对Agent的行为进行一定程度的控制。

1.2 任务的生成

用户提出的问题并不总是简单的，但通常很复杂，需要拆分成多项任务。上篇文章，我们解释了如何将用户的问题传递给 LLM、解释它们并将其转换为特定的任务。在这个任务生成过程中，Agent在生成任务的同时了解自己可能采取的行动非常重要。

2. Multi-action Agent

任务生成后，生成的任务（单个或多个）实际上被放入Agent中，但通常使用的LangChain的LLMSingleActionAgent无法处理多个任务。必须创建自己的Multi-action Agent来处理多个任务并将它们组合成一个答案。接下来就讲解一下如何使用LangChain创建自己的Multi-action Agent！

2.1 创建Multi-action Agent

现在，让我们看看如何创建Multi-action Agent。这次，我们将参考LangChain的BaseMultiActionAgent类来创建一个自定义代理。

from typing import List, Tuple, Any, Union, Dict, Optional

from langchain.chat_models import ChatOpenAI

from langchain.schema import AgentAction, AgentFinish

from langchain.agents import Tool, BaseMultiActionAgent, AgentOutputParser

from langchain.chains import LLMChain

from langchain.callbacks.manager import Callbacks

from TaskListCreation import TaskListCreation

class LLMMultiActionAgent(BaseMultiActionAgent):

    """Base class for multi action agents using LLMChain."""

    llm_chain: LLMChain

    output_parser: AgentOutputParser

    stop: List[str]

    def __init__(self, *args, **kwargs):

        super().__init__(*args, **kwargs)

        self.task_list_creation = TaskListCreation()

    @property

    def input_keys(self) -> List[str]:

        """Return the input keys."""

        return list(set(self.llm_chain.input_keys) - {"intermediate_steps"})

    def dict(self, **kwargs: Any) -> Dict:

        """Return dictionary representation of agent."""

        _dict = super().dict()

        del _dict["output_parser"]

        return _dict

    def plan(

        self, 

        intermediate_steps: List[Tuple[AgentAction, str]], 

        callbacks: Callbacks = None,

        **kwargs: Any

    ) -> Union[List[AgentAction], AgentFinish]:

        """Given input, decide what to do.

        Args:

            intermediate_steps: Steps taken to date, along with observations

            callbacks: Callbacks to run.

            **kwargs: User inputs.

        Returns:

            Action specifying what tool to use or finish action.

        """

        latest_user_message = kwargs.get("input", "")

        if len(intermediate_steps) == 0:

            task_list = self.task_list_creation.create_task_list(latest_user_message)

        else:  

            task_list = []

        actions = []

        final_outputs = [obs for _, obs in intermediate_steps]

        kwargs.pop("input", None)

        for task in task_list:

            output = self.llm_chain.run(

                intermediate_steps=intermediate_steps,

                stop=self.stop,

                input=task,

                **kwargs

            )

            response = self.output_parser.parse(output)

            if isinstance(response, AgentAction):

                for tool in ALLOWED_TOOLS:

                    if tool in response.tool:

                        response.tool = tool  

                        break                

                action = AgentAction(tool=response.tool, tool_input=response.tool_input, log=response.log)

                actions.append(action)

            elif not isinstance(response, AgentFinish):

                return AgentFinish(return_values={"output": "Unexpected response type."}, log="")

        if actions:

            return actions

        return AgentFinish(return_values={"output": final_outputs}, log="\n\nAll questions answered.")

    def tool_run_logging_kwargs(self) -> Dict:

        return {

            "llm_prefix": "",

            "observation_prefix": "" if len(self.stop) == 0 else self.stop[0],

        }

请关注plan方法。这定义了实际的Agent执行流程，返回值是AgentAction 的列表或AgentFinish

让我们按顺序看一下：

1. 通过kwargs接收用户的问题

2. intermediate_steps == 0时，即ReAct思维过程进入第一个循环时，生成执行任务。这里，我们暂且用TaskListCreation命名生成任务的类，并调用它。

3. 对于每个生成的任务，运行 llm_chain 并将输出传递给 output_parser 以获取响应。

4. 根据此响应定义并返回 AgentAction 或 AgentFinish

这就是整个流程。 响应包含[Question/Thought/Action/Action Input]一系列思考结果。 LangChain的agent本质上是循环思考，直到得到最终答案，并且intermediate_steps每次都会递增。 但这一次，我尝试在第一轮之后停止思考循环。

2.2 AgentExecutor

简单解释一下 AgentExecutor 的作用，_call在方法内部，_take_next_step被调用、AgentAction执行并Observation获取任务的答案。多任务代理需要多加小心，因为有多个任务。如果所有生成的任务都能独立处理是没有问题的，但是如果任务之间存在依赖关系怎么办？

例如，如果用户提出诸如“比较 A 公司的产品 X 和 B 公司的产品 Y 的功能”之类的问题，则会生成以下任务。

1. 调查A公司产品X的特点

2. 调查B公司产品Y的特点

3. 比较产品 X 和产品 Y 的功能

如果我们独立地对待这些，第三个任务会发生什么？在不了解产品 X 和产品 Y 的情况下，您无法比较它们。换句话说，只有在收到前两个任务的答案后，第三个任务才变得有意义。AgentExecutor 最初的设计并不是基于这样的假设，因此在这种情况下，有必要将任务之间的答案正确地传递到下一个任务。您需要将此类“内存接管”添加到您的 AgentExecutor 中。

例如，您可以通过执行以下操作来解决此问题：

        for agent_action in actions:

            current_memory = self.local_memory.load_memory_variables({})

            if current_memory['history']:

                memory_msg = "执行Action时请参考以下对话历史记录。\n对话历史记录：\n" + current_memory['history']

                agent_action.tool_input += memory_msg

2.3 aplan方法

如果任务之间不存在依赖关系，我们建议并行处理任务以加快进程。这种情况下，就需要定义前面提到的LLMtMultiActionAgent的aplan。

例如，您可以包含如下代码片段：

import asyncio

async def run_task(task):

    output = await self.llm_chain.arun(

        intermediate_steps=intermediate_steps,

        stop=self.stop,

        input=task,

        **kwargs

    )

    return self.output_parser.parse(output)

tasks = [run_task(task) for task in task_list]

results = await asyncio.gather(*tasks)

使用 arun 而不是 run 来获取所有任务的一组思考结果，然后使用 asyncio.gather 同时运行多个异步任务。 通过这样进行并行处理，即使有多个任务，也可以像只有一个任务时一样快速获得任务的答案。

2.4 Agent实例

最后，如下所示定义代理的实例并run使用命令运行代理。当发生特定于代理的解析错误时，我认为最好适当地使用 try/ except 进行响应。

llm_chain = LLMChain(llm=self.LLM, prompt=self.prompt_agent)

tool_names = [tool.name for tool in tools]

agent = LLMMultiActionAgent(

    llm_chain=llm_chain,

    output_parser=output_parser,

    stop=["\nObservation:"],

    allowed_tools=tool_names

)        

self.agent_executor = CustomAgentExecutor.from_agent_and_tools(

    agent=agent, 

    tools=tools, 

    verbose=True, 

    memory=self.memory,

    handle_parsing_errors=False

)

result = self.agent_executor.run(query)

3. 总结

在本文中，我们详细解释了如何使用 LangChain 创建Multi-action Agent。通过开发自己的代理，您将能够理解复杂的用户查询，将其分解为适当的任务，并有效地处理它们。