StateFlow - 使用自定义发言人选择在 GroupChat 中构建基于状态的工作流程

**TL;DR：**介绍 StateFlow，一种将复杂任务解决过程概念化为状态机的任务解决范式，以 LLM 为支撑。介绍如何使用 GroupChat 实现这一想法，并具备自定义发言人选择功能。

引言

使用大型语言模型（LLM）来解决复杂任务，例如需要一系列操作和与工具及外部环境的动态交互的任务，是一个值得关注的趋势。 在本文中，我们提出了一种新颖的基于LLM的任务解决范式，称为StateFlow，它将复杂任务解决过程概念化为状态机。 在StateFlow中，我们区分了“过程基础”（通过状态和状态转换）和“子任务解决”（通过状态内的操作），增强了任务解决过程的控制性和可解释性。 状态代表了正在运行的过程的状态。状态之间的转换由启发式规则或LLM做出的决策控制，从而实现动态和自适应的进展。 进入一个状态后，执行一系列操作，不仅涉及调用由不同提示引导的LLM，还可能利用外部工具。

StateFlow

有限状态机（FSM）被用作控制系统来监控实际应用，例如交通信号灯控制。 定义的状态机是一种行为模型，根据当前状态决定要做什么。状态代表了FSM可能处于的一种情况。 借鉴这个概念，我们希望使用FSM来对LLM的任务解决过程建模。当使用LLM解决具有多个步骤的任务时，任务解决过程的每个步骤可以映射到一个状态。

让我们以一个SQL任务为例（见下图）。 对于这个任务，期望的步骤是：

1. 收集关于数据库中表和列的信息，
2. 构建一个查询以检索所需信息，
3. 最后验证任务是否解决并结束过程。

对于每个步骤，我们创建一个相应的状态。同时，我们定义一个错误状态来处理失败情况。 在图中，红色箭头表示执行结果失败，绿色箭头表示成功。 根据特定规则进行状态转换。例如，在成功的“提交”命令下，模型转移到结束状态。 到达一个状态时，执行定义的一系列输出函数（例如，M_i -> E 表示先调用模型，然后执行SQL命令）。

实验

**InterCode：**我们在 InterCode 基准测试中使用 GTP-3.5-Turbo 和 GPT-4-Turbo 对 SQL 任务和 Bash 任务评估 StateFlow。 我们记录了不同的指标进行综合比较。'SR'（成功率）衡量了性能， 'Turns' 表示与环境的交互次数，'Error Rate' 表示执行命令时的错误率。 我们还记录了 LLM 使用的成本。

我们与以下基准进行比较： (1) ReAct：一种少样本提示方法，提示模型生成思考和行动。 (2) Plan & Solve：一种两步提示策略，首先要求模型提出一个计划，然后执行该计划。

Bash 任务的结果如下所示：

ALFWorld： 我们还在 ALFWorld 基准测试中进行了实验，这是一个在 TextWorld 环境中实现的合成文本游戏。 我们使用 GPT-3.5-Turbo 进行测试，并进行了平均 3 次尝试。

我们进行了以下评估： (1) ReAct：我们使用 ReAct 的两次提示。注意，每种类型的任务都有一个特定的提示。 (2) ALFChat (2 个代理)：AutoGen 中的两个代理系统设置，包括一个助理代理和一个执行代理。ALFChat 基于 ReAct，修改了 ReAct 的提示以符合对话方式。 (3) ALFChat (3 个代理)：在 2 个代理系统的基础上，引入了一个基准代理，每当助理连续三次输出相同的动作时，基准代理会提供常识事实。

对于这两个任务，StateFlow 在成本最低的情况下实现了最佳性能。更多详细信息，请参阅我们的论文。

使用 GroupChat 实现 StateFlow

我们将演示如何使用 GroupChat 构建 StateFlow。之前的博客 FSM Group Chat 介绍了 GroupChat 的一个新功能，允许我们输入一个转换图来约束代理的转换。 它要求我们使用自然语言来描述代理的 FSM 的转换条件，并使用 LLM 来接收描述并为下一个代理做出决策。 在本博客中，我们利用传递给 GroupChat 对象的 speaker_selection_method 的自定义发言者选择函数。 这个函数允许我们自定义代理之间的转换逻辑，并可以与 FSM Group Chat 中介绍的转换图一起使用。当前的 StateFlow 实现还允许用户覆盖转换图。 这些转换可以基于当前发言者和对话历史的静态检查（例如，检查最后一条消息中是否包含 'Error'）。

我们提供了一个使用 GroupChat 构建状态导向工作流的示例。 我们定义了一个自定义的发言者选择函数，将其传递给 GroupChat 的 speaker_selection_method 参数。 在这里，任务是根据给定的主题检索相关的研究论文，并为这些论文创建一个 markdown 表格。

我们定义了以下代理：

• Initializer：通过发送一个任务来启动工作流程。
• 编码者：通过编写代码从互联网上获取论文。
• 执行者：执行代码。
• 科学家：阅读论文并撰写摘要。

# 定义代理，此代码仅用于说明目的，无法执行。
initializer = autogen.UserProxyAgent(
   name="Init"
)
coder = autogen.AssistantAgent(
   name="Coder",
   system_message="""你是编码者。编写Python代码从arxiv获取论文。"""
)
executor = autogen.UserProxyAgent(
   name="Executor",
   system_message="执行者。执行编码者编写的代码并报告结果。"
)
scientist = autogen.AssistantAgent(
   name="Scientist",
   system_message="""你是科学家。在打印的摘要中查看论文后，请对论文进行分类，并创建一个包含领域、标题、作者、摘要和链接的Markdown表格。最后返回“TERMINATE”."""
)

在图中，我们定义了一个简单的研究工作流程，包括4个状态：Init（初始化）、Retrieve（获取）、Research（研究）和End（结束）。在每个状态下，我们将调用不同的代理来执行任务。

• Init（初始化）：我们使用initializer来启动工作流程。
• Retrieve（获取）：我们首先调用coder来编写代码，然后调用executor来执行代码。
• Research（研究）：我们将调用scientist来阅读论文并撰写摘要。
• End（结束）：我们将结束工作流程。

然后，我们定义一个自定义函数来控制状态之间的转换：

def state_transition(last_speaker, groupchat):
   messages = groupchat.messages

   if last_speaker is initializer:
       # init -> retrieve
       return coder
   elif last_speaker is coder:
       # retrieve: action 1 -> action 2
       return executor
   elif last_speaker is executor:
       if messages[-1]["content"] == "exitcode: 1":
           # retrieve --(execution failed)--> retrieve
           return coder
       else:
           # retrieve --(execution success)--> research
           return scientist
   elif last_speaker == "Scientist":
       # research -> end
       return None


groupchat = autogen.GroupChat(
   agents=[initializer, coder, executor, scientist],
   messages=[],
   max_round=20,
   speaker_selection_method=state_transition,
)

我们建议在自定义函数中为每个发言者实现状态转换逻辑。类似于状态机，状态转换函数根据当前状态和输入确定下一个状态。 除了返回表示下一个发言者的Agent类之外，我们还可以从['auto', 'manual', 'random', 'round_robin']中返回一个字符串，以选择要使用的默认方法。 例如，我们可以始终默认使用内置的auto方法，以使用基于LLM的群聊管理器选择下一个发言者。 当返回None时，群聊将终止。请注意，某些转换，例如 "initializer" -> "coder" 可以使用转换图来定义。

进一步阅读

• StateFlow 论文 [20]
• StateFlow 笔记本
• 具有自定义发言者选择的 GroupChat 笔记本
• FSM Group Chat
• autogen 的文档