AIAgent框架——multi-agent技术详解

关于multi-agent

随着LLM的涌现，以LLM为中枢构建的Agent系统在近期受到了广泛的关注。Agent系统旨在利用LLM的归纳推理能力，通过为不同的Agent分配角色与任务信息，并配备相应的工具插件，从而完成复杂的任务。

目前更常见的框架大多聚焦于single-agent的场景。single-agent的核心在于LLM与工具的配合。LLM通过理解用户的任务，推理出需要调用的工具，并基于调用结果给用户反馈。在完成任务的过程中，Agent可能与用户有多轮交互。下图即展示了一个主流的Agent执行流程。

与此同时，也有越来越多的Agent框架开始聚焦于multi-agent场景。为了完成任务，multi-agent会为不同的Agent赋予不同的角色定位，通过Agent之间的协同合作来完成复杂的任务。而在完成任务的过程中，相比于single-agent来说，与用户的交互会更少一些。

multi-agent的主要组件

为了构建一个multi-agent框架，我们需要思考相比于single-agent，框架中多了哪些组件。

• • environment：所有的agent应该处于同一个环境中。环境中包含了全局的状态信息，agent与环境之间存在信息的交互与更新。
• • stage：要完成一个复杂的任务，现有multi-agent框架往往采用SOP的思想，把复杂的任务分解成若干个子任务。对应到软件公司这个场景，”编写2048游戏”这个任务可以被分解为：编写prd，设计框架、写code，code review等子任务。
• • controller：controller可以是LLM，也可以是预先定义好的规则。它主要负责环境在不同agent和stage之间的切换。
• • memory：在single-agent中，记忆只包括了用户、LLM回应和工具调用结果这几个部分。而在multi-agent框架中，一方面由于agent数量的增多使得消息数量增多，另一方面，在每条消息中可能还需要对发送方、接收方等字段进行记录。

multi-agent的核心流程

multi-agent框架的核心交互流程可以概括如下：

• • controller更新当前环境的状态，选择下一时刻行动的agentA。
• • agentA t与环境交互，更新自身的memory信息。
• • agentA调用LLM，基于指令执行动作，获取输出message。
• • 将输出message更新到公共环境中。

下面的部分我们将简单介绍现有的三个multi-agent框架，并对它们进行简单的比较。

multi-agent框架分析

MetaGPT

相关资料

代码repo：https://github.com/geekan/MetaGPT

论文：https://arxiv.org/pdf/2308.00352.pdf

核心模块

MetaGPT论文中给出的架构图如上所示。

• • Role：可以理解为不同角色信息的agent（如Engineer，Architect）

class Role:

    """Role/Agent"""



    def __init__(self, name="", profile="", goal="", constraints="", desc=""):

        self._llm = LLM() #llm

        self._setting = RoleSetting(name=name, profile=profile, goal=goal, constraints=constraints, desc=desc)

        self._states = []

        self._actions = [] # 对应的action/ stage

        self._role_id = str(self._setting) 

        self._rc = RoleContext()

在Role中，又有两个主要组件：RoleSetting和RoleContext。其中RoleSetting存储了角色的名字、目标等信息。RoleContext则包含了运行中的状态信息，如记忆，需要执行的动作等。

class RoleSetting(BaseModel):

    """Role Settings"""

    name: str

    profile: str

    goal: str

    constraints: str

    desc: str



class RoleContext(BaseModel):

    """Role Runtime Context"""

    # 维护运行过程中的一些状态信息

    env: 'Environment' = Field(default=None) # environment，所有角色共享

    memory: Memory = Field(default_factory=Memory) # 记忆信息

    state: int = Field(default=0) # 下一时刻要执行的动作

    todo: Action = Field(default=None)

    watch: set[Type[Action]] = Field(default_factory=set) # 关注/订阅的信息

    news: list[Type[Message]] = Field(default=[]) # 本轮新增的信息

• • Environment：环境，包括了所有的角色信息和一个全局memory。
• • Action：可以理解为前面所说的stage。不同的Action会重载run()函数，并根据逻辑调用LLM或者工具，得到执行结果。

核心流程

下面简单介绍一下MetaGPT的流程。

首先，在environment.run()函数中，会依次调用每个角色的role.run()

async def run(self, k=1):

    """处理一次所有信息的运行

    Process all Role runs at once

    """

    for _ in range(k):

        futures = []

        # 执行

        for role in self.roles.values():

            future = role.run()

            futures.append(future)



        await asyncio.gather(*futures)

在role.run()函数中，与前面所述流程类似分为三步：a) _observe()函数观察环境、更新memory。b) _react()函数思考并执行动作。c) _publish_message()函数将执行结果更新到环境中。

async def run(self, message=None):

    """Observe, and think and act based on the results of the observation"""

    if message:

        ...

    # 观察环境、更新memory

    elif not await self._observe():

        # If there is no new information, suspend and wait

        return

    # 执行动作

    rsp = await self._react()

    # 把结果传给environment

    self._publish_message(rsp)

    return rsp

_react()函数可以进一步分为_think()和_act()两个阶段。_think()负责决定SOP的状态转移，_act()通过调用action.run()完成具体执行。两个阶段可能都需要调用LLM。

自定义扩展

最后在作者给出的examples中，用户如果想要实现一个自定义场景，一般情况下只需要实现自己的Action和Role类及对应的关键函数即可。

AgentVerse

相关资料

代码repo：https://github.com/OpenBMB/AgentVerse

论文：https://arxiv.org/pdf/2308.10848.pdf

除此之外，清华大学的开源框架AgentVerse也提供了一个基于LLMs来搭建多智能体交互的框架。

核心模块

论文中，提供了如上图所示的流程图，它的设计流程模拟了人类的决策过程，包含了四个阶段，分别是：专家招聘、协作决策、行动执行和评估。

• • 专家招聘阶段根据当前问题解决的进展来确定智能体的专家成员都有哪些。
• • 协作决策阶段，各个智能体参与讨论和制定策略，并达成共识，这里会有多轮执行
• • 行动执行阶段，各个智能体根据决策进行执行
• • 评估阶段，提供关于当前状态的进行与目标的对比，如果没有达到预期就根据奖励反馈重新回到专家招聘阶段进行新的一轮交互。

根据包括软件开发，咨询和游戏的几个不同领域的案例表明，这种多智能体合作的方式优于单一智能体，并且优势明显。

AgentVerse 在代码实现上，包含了两个基础元素agent和environment， agent不用过多介绍，他的工作行为依赖特定的prompt以及配合大模型llm进行结果生成。

这里的environment就是代表着任务，在environment中通过定义agents以及rules等信息，来确保多智能体的交互流程能够按照上述进行。

值得注意的是，environment 的rules中包含了5个组建，这些组件分别是： 描述器，顺序，选择器，更新器，以及可见性。他们的作用具体如下，（参考https://mp.weixin.qq.com/s/jkW2JRnbfsK81ClhwsCxqA）

• • 描述器（Describer），它为每个agent提供每次环境的描述。通过自定义该组件，你就能根据你自己的特定要求来构建环境。
• • 顺序（Order），它定义agent在环境中采取操作的顺序，可以采用几个默认选项，比如random（随机）、sequential（按顺序）和concurrent（所有agent在每个回合中都采取行动），也可以自定义。
• • 选择器（Selector），有时agent会生成一些无效信息，它就用于过滤这些信息，选择有效内容。
• • 更新器（Updater），用于更新每个agent的内存。这是因为有时某个agent的响应不应被所有agent看到（比如不在一个房间），它的作用就是在每个响应发生后，仅更新每个agent应看到内容。
• • Visibility（可见性），用于维护agent列表，每当有agent移动到另一个房间或者产生其他变化时，它会向所有agent更新列表。

具体Environment定义的代码如下：

@EnvironmentRegistry.register("basic")

class BasicEnvironment(BaseEnvironment):

    """

    A basic environment implementing the logic of conversation.



    Args:

        agents: List of agents

        rule: Rule for the environment

        max_turns: Maximum number of turns

        cnt_turn: Current turn number

        last_messages: Messages from last turn

        rule_params: Variables set by the rule

    """



    agents: List[BaseAgent]

    rule: Rule

    max_turns: int = 10

    cnt_turn: int = 0

    last_messages: List[Message] = []

    rule_params: Dict = {}



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

        rule_config = rule

        order_config = rule_config.get("order", {"type": "sequential"})

        visibility_config = rule_config.get("visibility", {"type": "all"})

        selector_config = rule_config.get("selector", {"type": "basic"})

        updater_config = rule_config.get("updater", {"type": "basic"})

        describer_config = rule_config.get("describer", {"type": "basic"})

        rule = Rule(

            order_config,

            visibility_config,

            selector_config,

            updater_config,

            describer_config,

        )

        super().__init__(rule=rule, **kwargs)

对应Agent的定义如下：

class BaseAgent(BaseModel):

    name: str # 名字

    llm: BaseLLM # llm

    output_parser: OutputParser # 输出解析

    prompt_template: str # 模板

    role_description: str = Field(default="") # 角色信息

    memory: BaseMemory = Field(default_factory=ChatHistoryMemory) # 记忆

    max_retry: int = Field(default=3)

    receiver: Set[str] = Field(default=set({"all"})) # 信息的接收方

    async_mode: bool = Field(default=True)

核心流程

用户初始化，并调用 agentVerse.run()作为程序入口

class AgentVerse:

    def __init__(self, agents: List[BaseAgent], environment: BaseEnvironment):

        self.agents = agents

        self.environment = environment



  def run(self):

        """Run the environment from scratch until it is done."""

        self.environment.reset()

        # 循环执行

        while not self.environment.is_done():

            asyncio.run(self.environment.step())

相关environment流程逻辑如下：

async def step(self) -> List[Message]:

    """Run one step of the environment"""



    # 选择下一时刻的行动的agent(s)

    agent_ids = self.rule.get_next_agent_idx(self)



    # 环境描述信息（每个agents不一定一样）

    env_descriptions = self.rule.get_env_description(self)



    # agent行动，返回结果

    messages = await asyncio.gather(

        *[self.agents[i].astep(env_descriptions[i]) for i in agent_ids]

    )



    # 选择过滤message信息

    selected_messages = self.rule.select_message(self, messages)





    # 更新mmory

    self.rule.update_memory(self)



    # 更新agent之间的可见性

    self.rule.update_visible_agents(self)



    self.cnt_turn += 1



    return selected_messages

相关 agent 流程逻辑如下：

def step(self, env_description: str = "") -> Message:

    parsed_response = None

    tool_observation = [self.tool_memory.to_string()]

    while True:

        # 拼接prompt

        prompt = self._fill_prompt_template(env_description, tool_observation)



        try:

            # 调用LLM

            response = self.llm.generate_response(prompt)

            # 解析结果

            parsed_response = self.output_parser.parse(response)

            if isinstance(parsed_response, AgentAction):

                # 调用工具

                observation = self._call_tool(parsed_response)

                tool_observation.append(

                    parsed_response.log.strip()

                    + f"\nObservation: {observation.strip()}"

                )

            break

        except BaseException as e:

            logging.error(e)

            logging.warning("Retrying...")

            continue

        if parsed_response is None or isinstance(parsed_response, AgentFinish):

            break



    self._update_tool_memory(tool_observation)



    message = Message(

        content=""

        if parsed_response is None

        else parsed_response.return_values["output"],

        sender=self.name,

        receiver=self.get_receiver(),

    )

    return message

拿软件开发的任务举例子，用户需要提供一个json配置文件，包含environment信息所需的信息，并且定义若干个不同的agent，如code_writer，code_reviwer，unit_test_generator，以及他们之间沟通时候的rule，最终初始化environment， agents以及agentverse入口方法，并调用 agentVerse.run()即可。

Agents

相关资料

代码repo：https://github.com/aiwaves-cn/agents

论文：https://arxiv.org/pdf/2309.07870.pdf

最后，波形智能联合浙大和苏黎世联邦理工大学，提供了另外一种支持mutli-agent的设计方案。

核心模块

该方案围绕SOP模块来确定agent之间交互的推进方式，通过SOP来管理状态的变换，并将相关状态信息变换记录到环境中，以便不同的agent进行各自任务的推进。 整体流程如下图所示：

Agents有三个核心模块，除了上述SOP，还有Agent以及Environment。

（参考：https://mp.weixin.qq.com/s/toblMJJkpFKtv0dfJFfHKA）

• • Agent这里和其他开源框架中的Agent概念基本一致，代表着智能体，并且可以进行基于LLM的推理和生成，tool的调用，长短期记忆等能力。
• • Environment与agentverse中的 environment略有不同，这里的environment主要负责记录智能体之间的对话历史和环境本身。
• • 这里再详述一下最重要的SOP类，它是由状态和状态之间的连接来确下一步计划。SOP 的每个状态节点由 State 类定义，State 类中涵盖了 Agent 在这个状态内特有模块化的 Prompt 和可以使用的各种工具 / API 等，由用户在配置文件中定义。每次行动时，Agent 会将这些模块化 prompt 和工具 / API 的输出组装成完整的 prompt，然后调用 LLM 决定如何行动。SOP 中还包括了一个控制器函数，利用LLM的推理能力来动态决定状态的转移和下一个行动的 Agent 是哪个。

核心流程

入口代码如下所示，用户通过配置文件来初始化agent，environment以及sop，并通过如下入口程序开始agent交互流程。

def run(agents,sop,environment):

    while True:

        # 更新状态，决定下一个行动的agent

        current_state,current_agent= sop.next(environment,agents)

        if sop.finished:

            os.environ.clear()

            break

      # agent执行

        action = current_agent.step(current_state)

        # 更新memory

        memory = process(action)

        environment.update_memory(memory,current_state)

sop的流程方法如下，通过调用transit方法进行状态转移的操作，以及用route进行agent的选择。

def next(self, environment, agents):

    """

    Determine the next state and the agent that needs action based on the current situation

    """

  # 一些初始化流程：获取记忆、相关信息等

  ...

    

    

    # 下一时刻的状态

    next_state = self.transit(

        chat_history=environment.shared_memory["long_term_memory"][

            environment.current_chat_history_idx :

        ],

        relevant_history=relevant_history,

        environment=environment,

    )



    # 如果进入终止节点，则直接终止

    if next_state.name == self.finish_state_name:

        self.finished = True

        return None, None



  # 更新状态

    self.current_state = next_state

    

  # 决定下一时刻要行动的agent

    next_agent = self.route(

        chat_history=environment.shared_memory["long_term_memory"][

            environment.current_chat_history_idx :

        ],

        agents = agents,

        relevant_history=relevant_history,

    )



    return self.current_state, next_agent

agent的执行方法如下，主要是根据当前环境信息，来进行prompt生成，并调用llm进行生成

def step(self, current_state,input=""):

    """

    return actions by current state and environment

    Return: action(Action)

    """

    

    current_state.chat_nums +=1

    state_begin = current_state.is_begin

    agent_begin = self.begins[current_state.name]["is_begin"]

    self.begins[current_state.name]["is_begin"] = False

    current_state.is_begin = False

    environment = self.environment

    

    self.current_state = current_state

    

    # 先根据当前环境更新信息    

    if len(environment.shared_memory["long_term_memory"])>0:

        current_history = self.observe()

        self.long_term_memory.append(current_history)



    response,res_dict = self.act()

    

    

    action_dict =  {

        "response": response,

        "res_dict": res_dict,

        "role": self.state_roles[current_state.name],

        "name": self.name,

        "state_begin" : state_begin,

        "agent_begin" : agent_begin,

        "is_user" : self.is_user

    }

    return  Action(**action_dict)



def act(self):

    """

    return actions by the current state

    """

    current_state = self.current_state

    chat_history = self.long_term_memory

    current_LLM = self.LLMs[current_state.name]



    # 拼接prompt

    system_prompt, last_prompt, res_dict = self.compile()



    # 调用LLM

    response = current_LLM.get_response(

        chat_history, system_prompt, last_prompt, stream=True

    )

    return response,res_dict

最后，agent执行结束后，environment会更新本轮环境信息用于下一轮

class Memory:

    def __init__(self,role,name,content) -> None:

        self.send_role = role

        self.send_name = name

        self.content = content





def update_memory(self, memory, current_state):

    """

    更新环境的memory信息

    """

    MAX_CHAT_HISTORY = eval(os.environ["MAX_CHAT_HISTORY"])

    self.shared_memory["long_term_memory"].append(memory)

    current_embedding = get_embedding(memory.content)

  # 对过去几轮信息作summary

    if len(self.shared_memory["long_term_memory"]) % MAX_CHAT_HISTORY == 0:

        summary = self.summary(current_state)

        self.shared_memory["short_term_memory"] = summary



    self.agents[memory.send_name].update_memory(memory)

目前，Agents 除了独特的SOP系统以外，相对于其他框架不同的点还有一个重要功能是支持人类使用者扮演multi-agent系统中的一个或多个智能体的功能，可以方便地支持各种人 – 智能体交互的应用场景，如人和智能体一起玩游戏、辩论等。

总结与比较

最后，我们针对上述几个市面上比较主流的agents框架进行简单的总结与比较。部分相关特性如下所示：

综上，multi-agent框架的核心还是在agent之间的交互与状态的转移。而LLM对工具的调用往往只是作为其中一个子模块，且大部分agent都只调用单个tool。因此，在multi-agent的场景中，使用多个单tool的agent还是一个多tool的agent，也是一个值得考虑的问题。

本文章转载微信公众号@青稞AI