从零开始用Python玩转ROS话题通信：手把手教你搭建聊天机器人原型

你是否曾想过，让两个机器人像老朋友一样，通过“聊天”来交换信息、协同工作？这听起来像是科幻电影里的场景，但借助ROS（机器人操作系统）和Python，你完全可以在自己的电脑上构建出这样一个有趣的原型。对于许多刚接触ROS的创客和机器人爱好者来说，那些抽象的概念——节点、话题、消息——常常让人望而却步。与其一头扎进枯燥的理论，不如从一个看得见、摸得着的项目入手。本文将带你亲手搭建一个简易的“聊天机器人”对话系统，通过模拟两个机器人互相打招呼的场景，把ROS话题通信的核心机制——发布者（Publisher）和订阅者（Subscriber）——拆解得明明白白。你会发现，用Python实现这一切，远比想象中更直观、更有趣。

1. 项目蓝图：构思一个会“对话”的机器人系统

在开始敲代码之前，我们先来描绘一下这个项目的全貌。我们的目标不是创造一个能理解自然语言的AI，而是构建一个通信框架，让两个独立的程序（在ROS中称为“节点”）能够像两个机器人一样，通过约定的“频道”交换文本信息。

想象这样一个场景：机器人A和机器人B在同一个网络中。机器人A定期“说话”，发布一条包含当前时间的问候语。机器人B则一直“倾听”，一旦听到A的发言，就将其内容打印出来，并可以据此做出响应（比如也发布一条回复）。这就是ROS话题通信最经典的“发布-订阅”模型的生动体现。

• 发布者 (Publisher)： 好比一个广播电台。它只负责在某个固定频率（话题）上发送信息，并不关心谁在收听。
• 订阅者 (Subscriber)： 好比一个收音机。它调到某个频率（话题）上，专心接收该频率传来的所有信息。

在这个项目中，我们将创建：

1. robot_a_talker.py： 扮演机器人A，作为一个发布者，每秒向/greeting话题发送一次问候。
2. robot_b_listener.py： 扮演机器人B，作为一个订阅者，订阅/greeting话题，并打印听到的内容。
3. 进阶任务： 让机器人B也成为一个发布者，在听到问候后，向另一个/reply话题发送回复，从而实现双向“对话”。

为了完成这个项目，你需要准备好以下环境：

• 操作系统： Ubuntu（推荐18.04或20.04，这是ROS最兼容的环境）。可以在物理机、虚拟机或WSL2中安装。
• ROS版本： 推荐ROS Noetic Ninjemys（对应Ubuntu 20.04）或ROS Melodic Morenia（对应Ubuntu 18.04）。本文示例基于ROS Noetic。
• Python版本： ROS Noetic默认使用Python 3。确保你的系统已安装Python 3（通常Ubuntu已预装）。

提示：如果你尚未安装ROS，请前往ROS官方网站（wiki.ros.org）查找对应你Ubuntu版本的安装教程。安装过程通常只需几条命令，但需要稳定的网络连接。

2. 搭建舞台：创建ROS工作空间与功能包

ROS项目通常组织在工作空间（Workspace）中。你可以把它理解为一个专属的项目文件夹，ROS的编译系统catkin会在这里管理你的所有代码和依赖。

2.1 创建并初始化工作空间

打开终端，让我们一步步搭建这个舞台。

# 1. 创建并进入一个名为`ros_chatbot_ws`的工作空间目录
mkdir -p ~/ros_chatbot_ws/src
cd ~/ros_chatbot_ws/src

# 2. 初始化工作空间。这会在src目录下生成CMakeLists.txt的符号链接。
catkin_init_workspace

# 3. 返回工作空间根目录并进行首次编译
cd ~/ros_chatbot_ws
catkin_make

首次运行catkin_make会生成build和devel两个文件夹。devel文件夹下的setup.bash脚本是关键，它包含了当前工作空间的环境变量。

为了让系统在任何新的终端中都能识别这个工作空间，我们需要“激活”它。最方便的方法是将其source命令添加到你的~/.bashrc文件中。

# 将下面这行命令添加到 ~/.bashrc 文件的末尾
echo "source ~/ros_chatbot_ws/devel/setup.bash" >> ~/.bashrc
# 然后让更改立即生效
source ~/.bashrc

现在，每次打开新的终端，你的ROS环境都会自动包含这个自定义工作空间。

2.2 创建功能包存放聊天机器人代码

功能包（Package）是ROS中组织软件的基本单元。我们的聊天机器人代码将放在一个独立的功能包里。

# 确保你在工作空间的src目录下
cd ~/ros_chatbot_ws/src
# 创建一个名为`chatbot_demo`的功能包，它依赖rospy和std_msgs
catkin_create_pkg chatbot_demo rospy std_msgs

这条命令创建了一个名为chatbot_demo的文件夹，里面包含了package.xml和CMakeLists.txt两个配置文件。对于纯Python项目，我们主要关注package.xml，它声明了包的依赖。

接下来，我们需要一个地方存放Python脚本。按照ROS Python代码的常见规范，我们在功能包下创建一个scripts目录。

cd chatbot_demo
mkdir scripts

至此，我们的项目结构如下所示：

ros_chatbot_ws/
└── src/
    └── chatbot_demo/
        ├── CMakeLists.txt
        ├── package.xml
        └── scripts/          # 我们将把Python脚本放在这里

3. 编写核心演员：发布者与订阅者节点

节点（Node）是ROS中可执行文件的实例。我们的robot_a_talker.py和robot_b_listener.py就是两个独立的节点。

3.1 实现发布者节点 (Robot A)

在~/ros_chatbot_ws/src/chatbot_demo/scripts/目录下，创建文件robot_a_talker.py。

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

import rospy
from std_msgs.msg import String
import time

def talker():
    """
    机器人A的发言函数。
    初始化节点，创建发布者，并以固定频率发布问候消息。
    """
    # 初始化ROS节点，节点名为'robot_a_talker'。
    # anonymous=False 意味着节点使用我们指定的名字。如果名字冲突，后启动的节点会失败。
    rospy.init_node('robot_a_talker', anonymous=False)

    # 创建一个Publisher，发布到名为'/greeting'的话题，消息类型为String，队列长度为10。
    # queue_size是一个重要的缓冲参数。如果发布消息的速度快于订阅者处理的速度，
    # 它将保留最多10条最新的消息，旧消息会被丢弃。
    pub = rospy.Publisher('/greeting', String, queue_size=10)

    # 设置发布频率为1Hz（每秒1次）
    rate = rospy.Rate(1)

    # 给机器人A一个简单的自我介绍
    robot_name = "Robot_A"
    rospy.loginfo(f"[{robot_name}] 已上线，开始向世界打招呼...")

    # 循环发布，直到节点被关闭（如按Ctrl+C）
    while not rospy.is_shutdown():
        # 获取当前时间，格式化为易读的字符串
        current_time = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
        # 构造要发布的消息内容
        greeting_msg = f"大家好，我是{robot_name}！现在的系统时间是：{current_time}"

        # 在终端和ROS日志中输出这条消息
        rospy.loginfo(f"[{robot_name}] 发布: {greeting_msg}")

        # 关键一步：将消息发布到'/greeting'话题
        pub.publish(greeting_msg)

        # 按照设定的频率休眠，控制循环速度
        rate.sleep()

if __name__ == '__main__':
    try:
        talker()
    except rospy.ROSInterruptException:
        # 当用户按下Ctrl+C时，会抛出此异常。这里安静地退出。
        rospy.loginfo("机器人A发言结束。")

代码要点解析：

1. #!/usr/bin/env python3： 这行“shebang”告诉系统使用python3来解释这个脚本。如果你直接给脚本赋予可执行权限（chmod +x），就可以像运行命令一样运行它（./robot_a_talker.py）。
2. rospy.init_node()： 这是每个ROS节点的生命起点。它向ROS主节点（Master）注册这个节点。没有这一步，节点无法进行任何通信。
3. rospy.Publisher()： 创建发布者对象。它声明了“我要向哪个话题（/greeting）发送什么类型的消息（String）”。
4. rospy.Rate()： 一个非常实用的工具，帮助你以固定频率运行循环。参数1代表1Hz。
5. rospy.loginfo()： 比print()更强大。它除了打印到屏幕，还会将信息写入节点的日志文件，并发送到rosout，方便用rqt_console等工具统一查看调试信息。
6. pub.publish()： 执行消息发布的实际动作。
7. rospy.is_shutdown()： 检查节点是否应该终止（例如收到了终止信号）。这是一个标准的循环条件。
8. 异常处理： 捕获rospy.ROSInterruptException异常，使得用Ctrl+C终止节点时能优雅退出。

创建完成后，别忘了给脚本添加可执行权限：

cd ~/ros_chatbot_ws/src/chatbot_demo/scripts
chmod +x robot_a_talker.py

3.2 实现订阅者节点 (Robot B)

在同一个scripts目录下，创建文件robot_b_listener.py。

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

import rospy
from std_msgs.msg import String

def greeting_callback(msg):
    """
    订阅'/greeting'话题的回调函数。
    当有新的消息发布到该话题时，此函数会被自动调用。
    :param msg: 接收到的String类型消息对象
    """
    # msg.data 是String消息对象中存储实际字符串数据的字段
    heard_text = msg.data
    # 获取当前节点名（在初始化时指定或由ROS生成）
    node_name = rospy.get_name()
    rospy.loginfo(f"[{node_name}] 我听到了：『{heard_text}』")

def listener():
    """
    机器人B的监听函数。
    初始化节点，订阅话题，并进入等待循环。
    """
    # 初始化节点。anonymous=True让ROS为节点生成一个唯一的名字（在基础名后加随机数），
    # 这样你可以同时运行多个监听器而不会冲突。
    rospy.init_node('robot_b_listener', anonymous=True)

    # 创建一个Subscriber，订阅名为'/greeting'的话题，消息类型为String。
    # 当有新消息时，自动调用greeting_callback函数。
    rospy.Subscriber('/greeting', String, greeting_callback)

    rospy.loginfo(f"[{rospy.get_name()}] 已启动，正在等待来自Robot_A的问候...")

    # rospy.spin() 使Python程序保持运行，直到节点被关闭。
    # 它会监听订阅的话题，并在收到消息时触发回调函数。
    # 这是一个阻塞调用，后面的代码不会被执行。
    rospy.spin()

if __name__ == '__main__':
    listener()

代码要点解析：

1. 回调函数 (Callback Function)： greeting_callback是订阅者模式的核心。你不需要在代码中主动调用它。ROS会在后台监听，一旦有消息发布到/greeting话题，ROS就会自动获取该消息，并将其作为参数msg传递给这个回调函数。
2. rospy.Subscriber()： 创建订阅者对象。它声明了“我要监听哪个话题（/greeting），消息是什么类型（String），收到消息后交给哪个函数处理（greeting_callback）”。
3. rospy.spin()： 这是订阅者节点的“心脏”。它让程序进入一个无限循环，等待并处理到来的消息。如果没有这行代码，你的脚本会瞬间执行完rospy.Subscriber()那一行然后退出，根本来不及接收任何消息。
4. msg.data： std_msgs.msg.String是一个ROS定义的消息类，真正的字符串数据存储在其data属性中。这是新手常犯的错误——直接打印msg对象会得到一串不直观的表示。

同样，赋予可执行权限：

chmod +x robot_b_listener.py

4. 启动与调试：观看机器人的首次对话

现在，激动人心的时刻到了。我们将启动ROS的核心服务，然后让两个机器人节点运行起来。

4.1 启动ROS Master

打开第一个终端。ROS Master是节点之间查找彼此的“名字服务”或“注册中心”。在任何节点启动前，必须先启动它。

# 确保你的工作空间环境已激活（如果已添加到.bashrc则自动生效）
# 启动ROS Master
roscore

保持这个终端运行，你会看到一些日志输出，表明roscore已成功启动。

4.2 启动订阅者Robot B

打开第二个终端。我们先启动监听者，这样当发言者开始说话时，我们能立刻看到效果。

# 切换到工作空间
cd ~/ros_chatbot_ws
# 激活当前工作空间的环境（如果.bashrc已配置，可省略，但显式执行更安全）
source devel/setup.bash
# 启动机器人B监听器
rosrun chatbot_demo robot_b_listener.py

你会看到类似 [INFO] [1712345678.890123]: [/robot_b_listener_12345] 已启动，正在等待来自Robot_A的问候... 的输出。注意节点名后面可能有一串随机数，这是anonymous=True的效果。

4.3 启动发布者Robot A

打开第三个终端。

cd ~/ros_chatbot_ws
source devel/setup.bash
rosrun chatbot_demo robot_a_talker.py

启动后，你会在第三个终端（Robot A）看到每秒输出一条问候信息。同时，在第二个终端（Robot B），你会看到它实时地打印出它“听”到的内容。

恭喜！ 你已经成功实现了一个单向的ROS话题通信。Robot A在说，Robot B在听。

4.4 使用ROS工具进行可视化与调试

ROS的强大之处在于其丰富的工具集。让我们用rqt_graph来图形化地查看当前的节点和话题网络。

打开第四个终端，运行：

rqt_graph

你会看到一个动态更新的图形。节点（椭圆）/robot_a_talker和/robot_b_listener_...通过话题（方框）/greeting连接在一起，箭头从发布者指向话题，再从话题指向订阅者。这直观地展示了我们刚刚构建的通信拓扑。

另一个有用的工具是rostopic。你可以用它来手动查看话题上的消息，甚至手动发布消息进行测试。

# 在新的终端中，查看当前活跃的话题列表
rostopic list
# 你应该能看到 `/greeting` 和 `/rosout` 等话题。

# 手动监听 `/greeting` 话题上的消息（类似于一个临时订阅者）
rostopic echo /greeting

# 手动向 `/greeting` 话题发布一条消息（类似于一个临时发布者）
rostopic pub -1 /greeting std_msgs/String "data: 'Hello from Command Line!'"

5. 进阶挑战：实现双向对话与消息自定义

基础的单向通信已经实现。但真正的对话是双向的。让我们升级系统，让Robot B在听到问候后，能够向Robot A发送一条回复。

5.1 创建自定义消息类型

目前我们使用的是ROS标准消息std_msgs/String。但在复杂系统中，我们经常需要传递结构化的数据。让我们定义一个简单的自定义消息，包含发言者ID和发言内容。

在chatbot_demo功能包下创建msg目录和消息定义文件：

cd ~/ros_chatbot_ws/src/chatbot_demo
mkdir msg
cd msg
# 创建一个名为ChatMessage.msg的文件
gedit ChatMessage.msg

在ChatMessage.msg文件中输入：

string speaker_id
string content
time timestamp

这定义了一个包含三个字段的消息：发言者ID、内容、时间戳。

接下来，我们需要修改package.xml和CMakeLists.txt，告诉ROS构建系统我们要使用自定义消息。

编辑package.xml，确保包含以下两行（如果不存在则添加）：

<build_depend>message_generation</build_depend>
<exec_depend>message_runtime</exec_depend>

编辑CMakeLists.txt，找到相应的部分进行修改：

1. 在find_package(catkin REQUIRED COMPONENTS ...)中添加message_generation。

   find_package(catkin REQUIRED COMPONENTS
     rospy
     std_msgs
     message_generation  # 添加这一行
   )
   

2. 在add_message_files(...)部分取消注释并添加你的.msg文件。

   add_message_files(
     FILES
     ChatMessage.msg
   )
   

3. 在generate_messages(...)部分取消注释。

   generate_messages(
     DEPENDENCIES
     std_msgs
   )
   

4. 在catkin_package(...)的CATKIN_DEPENDS中添加message_runtime。

   catkin_package(
   #  INCLUDE_DIRS include
   #  LIBRARIES chatbot_demo
     CATKIN_DEPENDS rospy std_msgs message_runtime  # 修改这一行
   #  DEPENDS system_lib
   )
   

保存文件，然后回到工作空间根目录重新编译：

cd ~/ros_chatbot_ws
catkin_make

编译成功后，你可以在Python代码中通过from chatbot_demo.msg import ChatMessage来使用这个自定义消息了。

5.2 升级为双向对话节点

我们将创建一个新的节点robot_dialogue.py，它同时是/greeting话题的订阅者，也是/reply话题的发布者。

在scripts目录下创建robot_dialogue.py：

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

import rospy
from chatbot_demo.msg import ChatMessage  # 导入自定义消息
from std_msgs.msg import String  # 保留，用于接收Robot A的旧消息（可选）
import time

class DialogueRobot:
    def __init__(self, robot_id):
        self.robot_id = robot_id
        # 初始化节点，名字基于robot_id
        rospy.init_node(f'{robot_id}_node', anonymous=True)

        # 创建一个发布者，发布到/reply话题，使用自定义的ChatMessage类型
        self.reply_pub = rospy.Publisher('/reply', ChatMessage, queue_size=10)

        # 创建一个订阅者，订阅/greeting话题。
        # 为了兼容性，我们先使用String类型。实际项目中应统一消息类型。
        rospy.Subscriber('/greeting', String, self.greeting_callback)

        rospy.loginfo(f"[{self.robot_id}] 对话机器人已启动。正在监听问候并准备回复...")

    def greeting_callback(self, msg):
        """收到问候后的回调函数"""
        rospy.loginfo(f"[{self.robot_id}] 收到问候：{msg.data}")

        # 模拟一些“思考”时间
        rospy.sleep(0.5)

        # 准备回复消息
        reply_msg = ChatMessage()
        reply_msg.speaker_id = self.robot_id
        reply_msg.content = f"你好！我已收到你的消息：『{msg.data[:30]}...』。这是我的回复。"
        reply_msg.timestamp = rospy.Time.now()  # 使用ROS时间

        # 发布回复
        self.reply_pub.publish(reply_msg)
        rospy.loginfo(f"[{self.robot_id}] 已发送回复。")

    def run(self):
        """保持节点运行"""
        rospy.spin()

if __name__ == '__main__':
    # 可以启动多个对话机器人，赋予不同的ID
    robot = DialogueRobot(robot_id="Robot_C")
    robot.run()

同时，我们需要修改robot_a_talker.py，让它也能监听/reply话题并做出反应，形成一个闭环。为了节省篇幅，这里只概述修改思路：在talker函数中，除了原有的发布者，再添加一个订阅者rospy.Subscriber('/reply', ChatMessage, reply_callback)，并在回调函数中打印收到的回复。这样，一个简单的双向对话循环就建立了。

5.3 调试技巧与常见问题排查

在开发过程中，你肯定会遇到节点无法通信、消息收不到等问题。以下是一些实用的调试命令：

命令	用途	示例
rosnode list	查看所有正在运行的节点。	检查你的节点名是否出现在列表中。
rostopic list	查看所有活跃的话题。	检查/greeting和/reply是否存在。
rostopic echo /topic_name	实时显示某个话题上流动的消息。	rostopic echo /greeting
rostopic info /topic_name	查看某个话题的详细信息，包括消息类型、发布者和订阅者。	rostopic info /greeting
rostopic hz /topic_name	统计某个话题的消息发布频率。	rostopic hz /greeting
rosmsg show message_type	查看某个消息类型的定义。	rosmsg show std_msgs/String
rqt_console	图形化界面，集中查看所有节点的日志信息(rospy.loginfo等)。	非常强大的调试工具。
rqt_graph	图形化显示节点与话题之间的连接关系。	可视化通信拓扑。

常见问题：

1. ImportError: No module named 'chatbot_demo.msg'

   • 原因： 自定义消息没有编译成功，或者编译后没有source devel/setup.bash。
   • 解决： 确保在chatbot_demo/msg/下正确创建了.msg文件，正确修改了package.xml和CMakeLists.txt，并在工作空间根目录执行了catkin_make。每次编译后，在新终端中运行节点前，务必执行source devel/setup.bash。

2. 节点启动后立刻退出

   • 原因： 脚本中没有rospy.spin()或等效的保持运行的代码（如while not rospy.is_shutdown()循环）。对于纯订阅者，必须有rospy.spin()。
   • 解决： 检查订阅者节点的代码，确保在rospy.Subscriber()之后有rospy.spin()。

3. 发布的消息订阅者收不到

   • 原因A： 话题名称或消息类型不匹配。发布到/greeting，订阅/greetings（多了一个s）。
   • 解决： 使用rostopic list和rostopic info仔细核对。
   • 原因B： 订阅者启动晚于发布者，且发布者只发布了一次（queue_size很小或为1，且没有循环）。
   • 解决： 确保发布者在循环中发布，或者增大queue_size。通常先启动订阅者，再启动发布者。

通过这个从零开始的聊天机器人项目，你不仅学会了如何用Python编写ROS的发布者和订阅者，还实践了工作空间管理、自定义消息、以及使用ROS工具链进行调试。这个原型可以轻松扩展，比如将文本消息替换为传感器数据（让一个机器人“告诉”另一个机器人它看到了什么），或者加入更复杂的逻辑（让机器人根据听到的内容决定不同的回复）。ROS话题通信的灵活性与Python的简洁性相结合，为你的机器人创意打开了第一扇大门。