Pipeline 和 MsgHub

Pipeline和Message Hub主要用于描绘应用中信息的交换和传播过程，它们极大简化了Multi-Agent应用流程的编排工作。在本教程中，我们将详细的介绍Pipeline和Message Hub的原理和使用方式。

Pipeline

在AgentScope中，消息的交换、传播构成了Multi-Agent应用。但是对复杂应用来说，细致的描绘每一次信息交流对开发者来说是非常困难的。 Pipeline主要用于简化“描述消息传播”的编程工作。

Pipeline中接收的对象是Operator，即信息的加工和传播单元（例如智能体Agent是Operator 的一个子类），而Pipeline自身也是Operator的子类。以下是所有Pipeline的基类：

class PipelineBase(Operator):
   """所有pipelines的基础接口."""
    # ... [为简洁起见省略代码]
    @abstractmethod
    def __call__(self, x: Optional[dict] = None) -> dict:
        """在这定义pipeline采取的操作。

        Args:
            x (Optional[`dict`], optional):
                对话历史以及一些环境信息。

        Returns:
            `dict`: 经过Pipeline处理后的返回消息。
        """

类别

为了方便开发者的使用，对于同一功能的Pipeline，AgentScope提供了两种不同的实现策略：

对象类型Pipeline
- 这些Pipeline是面向对象的，继承自 PipelineBase。它们本身是Operator，可以与其他运算符组合以创建复杂的交互模式，并且可以复用。
```
# 实例化并调用
pipeline = ClsPipeline([agent1, agent2, agent3])
x = pipeline(x)
```
函数式Pipeline
- 函数式Pipeline是独立的函数实现，在不需要复用的一次性使用场景中很有用。
```
# 只需要调用
x = funcpipeline([agent1, agent2, agent3], x)
```

Pipeline根据其功能被分类成以下的类型。下表概述了 AgentScope 中可用的不同 Pipeline：

运算符类型Pipeline	函数式Pipeline	描述
`SequentialPipeline`	`sequentialpipeline`	按顺序执行一系列运算符，将一个运算符的输出作为下一个运算符的输入。
`IfElsePipeline`	`ifelsepipeline`	实现条件逻辑，如果条件为真，则执行一个运算符；如果条件为假，则执行另一个运算符。
`SwitchPipeline`	`switchpipeline`	实现分支选择，根据条件的结果从映射集中执行一个运算符。
`ForLoopPipeline`	`forlooppipeline`	重复执行一个运算符，要么达到设定的迭代次数，要么直到满足指定的中止条件。
`WhileLoopPipeline`	`whilelooppipeline`	只要给定条件保持为真，就持续执行一个运算符。
-	`placeholder`	在流控制中不需要任何操作的分支，如 if-else/switch 中充当占位符。

使用说明

本节通过比较有无 Pipeline 的情况下多智能体应用程序中逻辑实现的方式，来阐释 Pipeline 如何简化逻辑实现。 注意： 请注意，在下面提供的示例中，我们使用术语 agent 来代表任何可以作为 Operator 的实例。这是为了便于理解，并说明 Pipeline 是如何协调不同操作之间的交互的。您可以将 agent 替换为任何 Operator，从而在实践中允许 agent 和 pipeline 的混合使用。

`SequentialPipeline`

不使用 pipeline:

x = agent1(x)
x = agent2(x)
x = agent3(x)

使用 pipeline:

from agentscope.pipelines import SequentialPipeline

pipe = SequentialPipeline([agent1, agent2, agent3])
x = pipe(x)

使用函数式 pipeline:

from agentscope.pipelines import sequentialpipeline

x = sequentialpipeline([agent1, agent2, agent3], x)

`IfElsePipeline`

不使用 pipeline:

if condition(x):
    x = agent1(x)
else:
    x = agent2(x)

使用 pipeline:

from agentscope.pipelines import IfElsePipeline

pipe = IfElsePipeline(condition, agent1, agent2)
x = pipe(x)

使用函数式 pipeline:

from agentscope.functional import ifelsepipeline

x = ifelsepipeline(condition, agent1, agent2, x)

`SwitchPipeline`

不使用 pipeline:

switch_result = condition(x)
if switch_result == case1:
    x = agent1(x)
elif switch_result == case2:
    x = agent2(x)
else:
    x = default_agent(x)

使用 pipeline:

from agentscope.pipelines import SwitchPipeline

case_operators = {case1: agent1, case2: agent2}
pipe = SwitchPipeline(condition, case_operators, default_agent)
x = pipe(x)

使用函数式 pipeline:

from agentscope.functional import switchpipeline

case_operators = {case1: agent1, case2: agent2}
x = switchpipeline(condition, case_operators, default_agent, x)

`ForLoopPipeline`

不使用 pipeline:

for i in range(max_iterations):
    x = agent(x)
    if break_condition(x):
        break

使用 pipeline:

from agentscope.pipelines import ForLoopPipeline

pipe = ForLoopPipeline(agent, max_iterations, break_condition)
x = pipe(x)

使用函数式 pipeline:

from agentscope.functional import forlooppipeline

x = forlooppipeline(agent, max_iterations, break_condition, x)

`WhileLoopPipeline`

不使用 pipeline:
```
while condition(x):
    x = agent(x)
```

使用 pipeline:

from agentscope.pipelines import WhileLoopPipeline

pipe = WhileLoopPipeline(agent, condition)
x = pipe(x)

使用函数式 pipeline:

from agentscope.functional import whilelooppipeline

x = whilelooppipeline(agent, condition, x)

Pipeline 组合

值得注意的是，AgentScope 支持组合 Pipeline 来创建复杂的交互。例如，我们可以创建一个 Pipeline，按顺序执行一系列智能体，然后执行另一个 Pipeline，根据条件执行一系列智能体。

from agentscope.pipelines import SequentialPipeline, IfElsePipeline
# 创建一个按顺序执行智能体的 Pipeline
pipe1 = SequentialPipeline([agent1, agent2, agent3])
# 创建一个条件执行智能体的 Pipeline
pipe2 = IfElsePipeline(condition, agent4, agent5)
# 创建一个按顺序执行 pipe1 和 pipe2 的 Pipeline
pipe3 = SequentialPipeline([pipe1, pipe2])
# 调用 Pipeline
x = pipe3(x)

MsgHub

MsgHub 旨在管理一组智能体之间的对话/群聊，其中允许共享消息。通过 MsgHub，智能体可以使用 broadcast 向群组中的所有其他智能体广播消息。

以下是 MsgHub 的核心类：

class MsgHubManager:
    """MsgHub 管理类，用于在一组智能体之间共享对话。"""
    # ... [为简洁起见省略代码]

    def broadcast(self, msg: Union[dict, list[dict]]) -> None:
        """将消息广播给所有参与者。"""
        for agent in self.participants:
            agent.observe(msg)

    def add(self, new_participant: Union[Sequence[AgentBase], AgentBase]) -> None:
       """将新参与者加入此 hub"""
        # ... [为简洁起见省略代码]

    def delete(self, participant: Union[Sequence[AgentBase], AgentBase]) -> None:
       """从参与者中删除智能体。"""
        # ... [为简洁起见省略代码]

使用说明

创建一个 MsgHub

要创建一个 MsgHub，请通过调用 msghub 辅助函数并传入参与智能体列表来实例化一个 MsgHubManager。此外，您可以提供一个可选的初始声明announcement，如果提供，将在初始化时广播给所有参与者。

from agentscope.msg_hub import msghub

# Initialize MsgHub with participating agents
hub_manager = msghub(
    participants=[agent1, agent2, agent3], announcement=initial_announcement
)

在 MsgHub 中广播消息

MsgHubManager 可以与上下文管理器一起使用，以处理MsgHub环境的搭建和关闭：

with msghub(
    participants=[agent1, agent2, agent3], announcement=initial_announcement
) as hub:
    # 智能体现在可以在这个块中广播和接收消息
    agent1()
    agent2()

    # 或者手动广播一条消息
    hub.broadcast(some_message)

退出上下文块时，MsgHubManager 会确保每个智能体的听众被清空，防止在中心环境之外的任何意外消息共享。

添加和删除参与者

你可以动态地从 MsgHub 中添加或移除智能体：

# 添加一个新参与者
hub.add(new_agent)

# 移除一个现有的参与者
hub.delete(existing_agent)

[返回顶部]

Pipeline 和 MsgHub

Pipeline

类别

使用说明

SequentialPipeline

IfElsePipeline

SwitchPipeline

ForLoopPipeline

WhileLoopPipeline

Pipeline 组合

MsgHub

使用说明

创建一个 MsgHub

在 MsgHub 中广播消息

添加和删除参与者

`SequentialPipeline`

`IfElsePipeline`

`SwitchPipeline`

`ForLoopPipeline`

`WhileLoopPipeline`