在上一篇文章中，我们深入探讨了 AgentScope 框架的 agent 模块，并成功搭建了两个智能体实例运行出结果。今天，我们基于之前的代码，稍作调整，将框架中的 Pipeline 模块引入进来，看看如何用最简洁的步骤实现多智能体之间的对话流程——这也算是正式步入 Pipeline 模块的学习阶段。
0. 回顾：之前的代码实现
代码详解请看我之前的文章：【AI Agent系列】【阿里AgentScope框架】0. 快速上手：AgentScope框架简介与你的第一个AgentScope程序
import agentscope
import os
openai_api_key = os.getenv('OPENAI_API_KEY')
# 一次性初始化多个模型配置
openai_cfg_dict = {
"config_name": "openai_cfg",
"model_type": "openai",
"model_name": "gpt-3.5-turbo",
"api_key": openai_api_key,
}
agentscope.init(model_configs=[openai_cfg_dict])
from agentscope.agents import DialogAgent, UserAgent
dialogAgent = DialogAgent(name="assistant", model_config_name="openai_cfg", sys_prompt="You are a helpful ai assistant")
userAgent = UserAgent()
x = None
x = dialogAgent(x)
print("diaglogAgent: \n", x)
x = userAgent(x)
print("userAgent: \n", x)
这段代码的末尾部分如下：
x = None
x = dialogAgent(x)
print("diaglogAgent: \n", x)
x = userAgent(x)
print("userAgent: \n", x)
可以看出，dialogAgent 的输出传递给了 userAgent，但 userAgent 并未将消息再回传给 dialogAgent。两个智能体之间仅存在单向的数据流动，距离真正的“对话”还差关键一步。
1. 实现智能体之间的双向交互
要让两个智能体真正实现对话，方法非常简单——只需添加一个 while 循环：
x = None
while True:
x = dialogAgent(x)
x = userAgent(x)
# 如果用户输入"exit"，则终止对话
if x.content == "exit":
print("Exiting the conversation.")
break
这样一来，dialogAgent 的输出会作为输入喂给 userAgent，而 userAgent 的输出又会回流到 dialogAgent，形成完整的循环。根据上一篇文章对 agent 实现机制的分析，每次 x 被传递给智能体时，都会自动存入其 memory 中——于是双方各自拥有自己及对方的回复作为上下文，对话自然能够持续进行。
if self.memory:
self.memory.add(x)
2. 使用 Pipeline 模块编排智能体交互
2.1 Pipeline 是什么？——个人理解
AgentScope 团队为了帮助开发者更便捷地编排智能体之间的交互逻辑，专门封装了 Pipeline 模块。该模块包含一系列“管道”，本质上类似于编程语言中的控制结构：顺序执行、条件分支、循环等。借助这些 Pipeline，你可以灵活控制多个智能体之间的信息流转——而且代码比手动编写循环更加清晰优雅。
2.2 快速上手 Pipeline
引入 Pipeline 后，前面的循环代码可以改写为：
from agentscope.pipelines.functional import sequentialpipeline
# 在Pipeline结构中执行对话循环
x = None
while x is None or x.content != "exit":
x = sequentialpipeline([dialogAgent, userAgent], x)
这里我们使用了 sequentialpipeline，它的作用就是让信息按照智能体列表的顺序依次传递。利用它，两个智能体之间的交互逻辑一目了然，代码的可读性也显著提升。
2.3 sequentialpipeline 源码解读
那么 sequentialpipeline 内部究竟是如何运作的？直接查看源码：
def sequentialpipeline(
operators: Sequence[Operator],
x: Optional[dict] = None,
) -> dict:
"""Functional version of SequentialPipeline.
Args:
operators (`Sequence[Operator]`):
Participating operators.
x (`Optional[dict]`, defaults to `None`):
The input dictionary.
Returns:
`dict`: the output dictionary.
"""
if len(operators) == 0:
raise ValueError("No operators provided.")
msg = operators[0](x)
for operator in operators[1:]:
msg = operator(msg)
return msg
函数接收两个参数：
operators：按顺序排列的 agent 列表。x：可选的初始输入；若为None，则多个 agent 之间各自独立运行，不会进行信息交互。
内部逻辑实际上与第1节中手动编写的传递过程完全一致：
x = dialogAgent(x)
x = userAgent(x)
只不过它通过循环将列表中的每个 agent 串接起来，更加通用和简洁。
3. 最终代码与实现效果
3.1 完整代码
import agentscope
openai_cfg_dict = {
"config_name": "openai_cfg",
"model_type": "openai",
"model_name": "gpt-3.5-turbo",
}
agentscope.init(model_configs=[openai_cfg_dict])
from agentscope.agents import DialogAgent, UserAgent
dialogAgent = DialogAgent(name="assistant", model_config_name="openai_cfg", sys_prompt="You are a helpful ai assistant")
userAgent = UserAgent()
from agentscope.pipelines.functional import sequentialpipeline
x = None
while x is None or x.content != "exit":
x = sequentialpipeline([dialogAgent, userAgent], x)
3.2 运行效果