目录

• 一、引言

• 二、酒店服务机器人功能需求分析

• 三、ROS2系统架构设计

• 四、导航模块实现

• 五、语音交互模块

• 六、电梯对接方案

• 七、任务调度系统

• 八、实际部署经验

• 九、总结

一、引言

ROS2（Robot Operating System 2）作为新一代机器人操作系统，相比ROS1在实时性、安全性、通信机制等方面有显著提升，成为酒店服务机器人开发的理想选择。本文将详细介绍基于ROS2的酒店服务机器人系统架构设计，涵盖导航、语音交互、电梯对接、任务调度等核心模块的实现方案。

二、酒店服务机器人功能需求分析

2.1 典型应用场景

酒店服务机器人的主要应用场景包括：

场景	核心功能	特殊需求
大堂迎宾	人脸识别、语音问候、引导至前台/电梯	多人识别、嘈杂环境处理
客房配送	自主导航、物品递送、电梯联动	精确停靠、物品安全
餐饮配送	多点配送、避障、语音通知	路径优化、餐品保温
送物服务	自主呼梯、房间到达通知	电梯高峰期处理

1/21

2.2 功能模块划分

酒店服务机器人功能架构├── 感知模块

│ ├── 激光雷达感知（障碍物检测、定位）

│ ├── 视觉感知（人脸识别、目标跟踪）

│ ├── 超声感知（近端障碍物检测）

│ └── 触觉感知（防撞条、急停按钮）├── 认知模块

│ ├── 语音识别（ASR）

│ ├── 自然语言理解（NLU）

│ ├── 知识问答（FAQ）│ └── 情感识别├── 决策模块│ ├── 任务规划

│ ├── 路径规划│ └── 多机调度├── 执行模块│ ├── 运动控制

│ ├── 电梯控制│ └── 语音合成（TTS）└── 交互模块

├── 触摸屏交互

├── 语音交互

└── APP交互

2.3 系统性能指标

指标	要求
自主导航成功率	≥ 98%
避障成功率	≥ 99%
语音识别准确率	≥ 95%（安静环境）
电梯对接成功率	≥ 99%
连续运行时间	≥ 8小时
平均任务响应时间	≤ 3秒

三、ROS2系统架构设计

2/21

3.1 整体架构

基于ROS2 Humble版本的酒店服务机器人系统架构采用分布式微服务设计理念：

应用层 (Application Layer) │ 迎宾服务 │ │ 配送服务 │ │ 送物服务 │ │ 导览服务 │ ───────── ───────── ───────── ───────── │ │ 业务调度器 │ ──────────				│	││ ││││ │
──────────────────────────────────────────────────────────────── │ ────────────────────────────────────────────────────────────────
──────────		服务层 (Service Layer)		││
│ 导航服务 │ │ 语音服务 │ │ 电梯服务 │ │ 展示服务 │ │ │(Nav2) │ │(ASR/TTS) │ │(IoT) │ │(UI) │ │ ───────── ───────── ───────── ───────── │ ───────────────────────────────────────────────────────────── │ │ │ │ ───────────────────────────────────────────────────────────── 通信层 (Communication Layer - ROS2 DDS) │ │ Topics (异步消息) │ Services (同步调用) │ Actions (长期任务) │
│
│ 雷达驱动 │	│		驱动层 (Driver Layer) 相机驱动 │ │ 底盘驱动 │ │ 传感器驱动│ │		││

3.2 节点设计

# 节点设计示例：配送任务节点 import rclpy

3/21

pyo rclpy .node import Node

from rclpy .action import ActionClient

from nav2_msgs .action import NavigateToPose

from std_msgs .msg import String

from robot_msgs .srv import DeliveryControl

class DeliveryNode (Node):

def __init__(self):

super(). init ('delivery_node')

# 导航Action客户端

self.nav_client = ActionClient(

self ,

NavigateToPose,

'navigate_to_pose'

)

# 电梯服务客户端

self .elevator_client = self .create_client( DeliveryControl,

'/robot/elevator_control'

)

# 状态发布

self .state_pub = self .create_publisher( String,

'/delivery/state',

10

)

# 任务订阅

self .task_sub = self .create_subscription( DeliveryTask,

'/delivery/task',

self.on_task_received,

10

)

def on_task_received (self, task):

self .get_logger().info(f'Received task: {task.task_id}')

# 任务处理流程

self.execute_delivery(task)

3.3 话题通信设计

4/21

核心话题列表：

话题名	消息类型	频率	说明
/scan	LaserScan	10Hz	激光雷达数据
/odom	Odometry	50Hz	里程计数据
/cmd_vel	Twist	-	速度控制命令
/map	OccupancyGrid	1Hz	地图数据
/robot_state	RobotState	1Hz	机器人状态
/speech/text	String	-	TTS文本输入
/face_detected	FaceArray	5Hz	人脸检测结果

# 话题订阅示例

class PerceptionNode (Node):

def __init__(self):

super(). init ('perception_node')

# 激光雷达订阅

self.create_subscription( LaserScan,

'/scan',

self .on_scan,

10

)

# 相机话题订阅

self.create_subscription( Image,

'/camera/rgb/image_raw', self .on_image,

10

)

# 多机位置广播

self.create_subscription( RobotPoseArray,'/robots/positions',

self.on_robots_update,5

5/21

python

)

def on_scan (self, msg):

"""处理激光雷达数据"""

# 障碍物检测

obstacles = self.detect_obstacles(msg)

# 发布障碍物信息

self.obstacle_pub.publish(obstacles)

3.4 服务调用设计

python

# 服务定义示例

from ros2_interface .srv import ElevatorCall

from ros2_interface .srv import RobotCommand

class ElevatorService :

def __init__(self, node):

self .node = node

self.client = node .create_client(

ElevatorCall,

'/elevator/call'

)

async def call_elevator (self, floor, direction):

"""调用电梯"""

request = ElevatorCall.Request()

request .floor = floor

request .direction = direction # 1: up, 0: down

# 异步等待服务响应

future = self.client.call_async(request)

rclpy.spin_until_future_complete(self.node, future)

if future .result().success:

self.node.get_logger().info('电梯调用成功')

return future .result().elevator_id else :

raise RuntimeError(f'电梯调用失败: {future.result().error_msg}')

四、导航模块实现

6/21

4.1 Nav2配置

Nav2是ROS2的官方导航栈，我们采用Nav2 Humble版本：

yaml

# nav2_params.yaml 关键配置

amcl:

ros__parameters: use_sim_time: False

alpha1: 0.2

alpha2: 0.2

alpha3: 0.2

alpha4: 0.2

alpha5: 0.2

base_frame_id: "base_footprint"

beam_skip_distance: 0.5

beacon_strict_checking: false

controller_server:

ros__parameters:

# DWB控制器参数

use_sim_time: False

controller_frequency: 20.0

min_vel_x: 0.0

min_vel_y: 0.0

max_vel_x: 0.5

max_vel_y: 0.5

max_vel_theta: 1.5

min_speed_xy: 0.0

max_speed_xy: 0.5

# 轨迹评分参数

path_distance_bias: 32.0

goal_distance_bias: 24.0

occdist_scale: 0.1

# 路径跟随权重

# 目标接近权重

# 障碍物躲避权重

planner_server:

ros__parameters:

planner_plugin: "nav2_navfn_planner/NavfnPlanner"

use_sim_time: False

tolerance: 0.5

7/21

4.2 Costmap参数调优

yaml

# 全局代价地图配置

global_costmap:

global_costmap:

ros__parameters: update_frequency: 1.0

publish_frequency: 1.0

robot_radius: 0.35

resolution: 0.05

track_unknown_space: true plugins:

- "static_layer"

- "obstacle_layer"

- "inflation_layer"

# 局部代价地图配置

local_costmap:

local_costmap:

ros__parameters: update_frequency: 5.0

publish_frequency: 2.0

width: 3.0

height: 3.0

robot_radius: 0.35

resolution: 0.05

origin_x: -1.5

origin_y: -1.5

# 障碍物层配置

obstacle_layer:

ros__parameters:

enabled: true

observation_sources: scanscan:

sensor_frame: laser_link

topic: /scan

expected_update_rate: 0.1

observation_persistence: 0.5

max_obstacle_height: 2.0

inf_is_valid: false

8/21

4.3 导航任务执行

# 导航执行示例

import rclpy

from rclpy .action import ActionClient

from nav2_msgs .action import NavigateToPose

from geometry_msgs .msg import PoseStamped

class NavigationController :

def __init__(self, node):

self .node = node

self.nav_client = ActionClient(

node,

NavigateToPose,

'navigate_to_pose'

)

async def navigate_to (self, x, y, theta=0.0):

"""导航到指定位置"""

# 等待服务器就绪

self.nav_client .wait_for_server()

# 构建目标点

goal_pose = PoseStamped()

goal_pose .header .stamp = self.node.get_clock().now().to_msg() goal_pose .header.frame_id = 'map'

goal_pose .pose .position .x = x

goal_pose .pose .position .y = y

goal_pose .pose .position .z = 0.0

goal_pose .pose .orientation .z = math.sin(theta/2)

goal_pose .pose .orientation .w = math.cos(theta/2)

# 发送目标

goal_msg = NavigateToPose.Goal()

goal_msg .pose = goal_pose

self .node.get_logger().info(f'导航目标: ({x}, {y})')

# 异步等待结果

send_goal_future = self.nav_client.send_goal_async( goal_msg,

self.feedback_callback

)

rclpy.spin_until_future_complete(self.node, send_goal_future)

9/21

python

goal_handle = send_goal_future.result()

if not goal_handle .accepted:

raise RuntimeError('导航目标被拒绝')

result_future = goal_handle.get_result_async()

rclpy.spin_until_future_complete(self.node, result_future) return result_future .result().result

五、语音交互模块

5.1 ASR+TTS+NLP集成

# 语音交互模块架构

class VoiceInteractionModule :

def __init__(self, node):

self .node = node

# ASR (自动语音识别) 客户端

self.asr_client = ASRClient(

api_url='http://asr-service:8000/recognize')

# NLP (自然语言处理) 客户端

self.nlp_client = NLPClient(

api_url='http://nlp-service:8000/understand')

# TTS (语音合成) 客户端

self.tts_client = TTSClient(

api_url='http://tts-service:8000/synthesize')

# 话题订阅

self.audio_sub = node.create_subscription( AudioData,

'/audio/raw',

self .on_audio,

10

)

10/21

python

self.tts_pub = node.create_publisher(

AudioData,

'/audio/tts_output',

10

)

async def process_voice_input (self, audio_data):

"""处理语音输入"""

# 1. ASR语音转文字

text = await self.asr_client.recognize(audio_data) self .node.get_logger().info(f'识别文本: {text}')

# 2. NLP语义理解

intent = await self.nlp_client.understand(text)

self .node.get_logger().info(f'识别意图: {intent}')

# 3. 生成响应

response = self.generate_response(intent)

# 4. TTS语音合成

tts_audio = await self.tts_client.synthesize(response)

# 5. 播放语音

self.tts_pub.publish(tts_audio)

return response

def generate_response (self, intent):

"""根据意图生成响应"""

responses = {

'greeting' : '您好，请问有什么可以帮您？',

'delivery' : '好的，我马上为您配送。',

'guide' : '请跟我来。',

'cancel' : '好的，已取消任务。',

'unknown' : '抱歉，我没有听清楚，请您再说一次。'

}

return responses.get(intent .action, responses['unknown'])

5.2 唤醒词与语音控制

# 唤醒词检测模块

class WakeWordDetector :

def __init__(self):

11/21

python

self .model = load_wake_word_model('xiaoyu_wakeword.tflite') self.is_awake = False

self.keyword = '小智'

def detect (self, audio_chunk):

"""检测唤醒词"""

features = self.extract_mfcc(audio_chunk)

score = self.model.predict(features)

if score > 0.8 and not self.is_awake:

self.is_awake = True

self.on_wake ()

return True return False

def on_wake (self):

"""唤醒回调"""

# 播放提示音

play_audio('wakeup_tone.wav')

# 点亮屏幕

publish_event('screen_on')

# 开始录音

start_audio_capture()

六、电梯对接方案

6.1 IoT 协议对接

酒店电梯对接采用Modbus RTU/TCP或HTTP API两种主流协议：

# 电梯控制器抽象

class ElevatorController :

def __init__(self, config):

self .protocol = config['protocol']

if self .protocol == 'modbus' :

self.client = ModbusClient(

host=config['host'],

port=config['port']

)

elif self.protocol == 'http':

self.client = HTTPElevatorClient (config['api_url'])

12/21

python

async def call_elevator (self, current_floor, target_floor):

"""呼叫电梯"""

if self .protocol == 'modbus' :

# Modbus协议调用

self.client.write_register(

address=0x00 , # 楼层寄存器

value=target_floor

)

self.client.write_register(

address=0x01 , # 方向寄存器

value=1 if target_floor > current_floor else 0)

else :

# HTTP API调用

await self.client.post ('/api/v1/elevator/call', {

'floor' : target_floor,

'direction' : 'up' if target_floor > current_floor else 'down'})

async def wait_for_door_open (self, timeout=30):

"""等待电梯门打开"""

start_time = time .time()

while time .time() - start_time < timeout:

if self .protocol == 'modbus' : door_status = self.client.read_register(0x10) if door_status == 1 : # 门已打开

return True

else :

status = await self .client.get('/api/v1/elevator/status') if status['door_open']:

return True

await asyncio.sleep(0.5)

raise TimeoutError('电梯门未在规定时间内打开')

6.2 安全互锁机制

# 电梯对接安全机制

class ElevatorSafetyMonitor :

def __init__(self, robot):

self .robot = robot

self .state = 'idle'

async def enter_elevator (self, target_floor):

13/21

python

"""安全进入电梯"""

# 1. 检测电梯状态

if not await self .check_elevator_ready():

raise SafetyError('电梯未就绪')

# 2. 进入电梯

self .state = 'entering'

await self .robot .move_forward(distance=0.8)

# 3. 确认进入完成

if not self .check_entry_complete():

await self .robot .move_backward()

raise SafetyError('进入被打断')

# 4. 按目标楼层

await self.select_floor(target_floor)

# 5. 等待到达

self .state = 'moving'

await self.wait_for_arrival(target_floor)

self .state = 'idle'

def check_entry_complete (self):

"""检查进入是否完成"""

# 使用激光雷达检测后方是否有障碍物

rear_scan = get_laser_scan()

obstacles = [p for p in rear_scan if p.distance < 0.5] return len(obstacles) == 0

七、任务调度系统7.1 多机器人协同调度

# 任务调度器

class TaskScheduler :

def __init__(self, node): self .node = node

self .tasks = []

self .robots = {}

# 任务队列订阅

14/21

node.create_subscription(

TaskMsg,

'/dispatch/task_queue',

self .on_new_task,

10

)

# 机器人状态订阅

node.create_subscription(

RobotStatus,

'/robots/status',

self.on_robot_status,

10

)

def on_new_task (self, task):

"""处理新任务"""

# 任务入队

self .tasks.append(Task(task))

# 触发调度

self .schedule()

def schedule (self):

"""调度任务到合适机器人"""

available_robots = [

r for r in self .robots .values()

if r .status == 'idle'

]

# 按距离排序选择最近机器人

available_robots.sort(

key=lambda r: calculate_distance(r .position, self .tasks[0].location)

)

if available_robots:

robot = available_robots[0]

task = self .tasks.pop(0)

# 分配任务

robot.assign_task(task)

self .node.get_logger().info(

f'任务 {task.id} 分配给机器人 {robot.id}'

)

15/21

python

def optimize_multi_robot (self):

"""多机器人路径优化"""

# 使用K-means聚类或VRP算法优化任务分配

pass

7.2 任务优先级管理

python

# 任务优先级定义

class TaskPriority : EMERGENCY = 100 VIP = 80

NORMAL = 50 LOW = 20

# 紧急任务（如急救）

# VIP客人服务

# 普通配送

# 常规巡检

@dataclass

class Task : task_id: str

task_type: str

priority: int

location: Tuple [float , float]

payload: Any

deadline: Optional[datetime]

created_at: datetime = field(default_factory=datetime.now )

def __lt__(self, other):

# 优先级高的先执行

if self .priority != other .priority:

return self .priority > other .priority

# 同优先级按创建时间

return self.created_at < other .created_at

八、实际部署经验

8.1 常见问题与解决方案

问题1：Nav2路径规划失败（找不到路径）

原因分析：

• 地图中存在未标注的永久障碍物

16/21

• 代价地图inflation层参数设置过小

• 全局路径经过狭窄通道解决方案：

yaml

# 调整planner_server参数

planner_server:

ros__parameters:

tolerance: 1.0 # 增大容差

allow_unknown: true # 允许通过未探索区域

# 增大inflation半径

inflation_layer:

ros__parameters:

cost_scaling_factor: 3.0

inflation_radius: 0.55

问题2：AMCL定位漂移

原因分析：

• 长期运行后粒子多样性丧失

• 地图与实际环境存在差异

• 传感器噪声累积解决方案：

python

# 定期触发重定位

def check_localization_quality (self):

pose_covariance = get_amcl_covariance()

if trace(pose_covariance) > 0.5 :

# 触发重定位

trigger_relocalization()

self.node.get_logger().warn('定位质量差，触发重定位')

问题3：电梯对接超时

17/21

原因分析：

• 电梯控制系统响应慢

• 多机器人同时呼叫电梯

• 网络延迟解决方案：

python

# 实现电梯预约锁

class ElevatorReservationManager :

def __init__(self):

self .reservations = {}

async def reserve_elevator (self, floor, robot_id, timeout=60):

# 检查是否已有预约

if floor in self .reservations:

if time .time() - self .reservations[floor]['time'] < timeout:

# 复用已有预约

return self .reservations[floor]['elevator_id']

# 创建新预约

elevator_id = await self.request_elevator(floor)

self .reservations[floor] = {

'elevator_id' : elevator_id,

'robot_id' : robot_id,

'time' : time .time()

}

return elevator_id

问题4：语音识别在嘈杂环境准确率下降

原因分析：

• 环境噪声干扰

• 混响影响

• 说话人距离远解决方案：

18/21

python

# 麦克风阵列降噪处理

class AudioProcessor :

def __init__(self):

self.mic_array = MicArray(8) # 8麦克风阵列 self.beamformer = Beamformer()

self.noise_reducer = NoiseReducer()

def process (self, raw_audio):

# 波束形成 - 指向声源方向

beamformed = self.beamformer.process(raw_audio)

# 噪声抑制

denoised = self.noise_reducer .reduce(beamformed)

# 自动增益控制

normalized = self.agc.process(denoised) return normalized

问题5：多机调度冲突

原因分析：

• 两机器人同时进入同一区域

• 电梯资源竞争

• 任务分配重复解决方案：

# 分布式资源锁

class ResourceCoordinator :

def __init__(self):

self .locks = {}

async def acquire_lock (self, resource_id, robot_id, timeout=30):

"""获取资源锁"""

start = time .time()

while resource_id in self .locks:

if time .time() - start > timeout:

19/21

python

raise ResourceBusyError(f'{resource_id} 忙')

if self .locks[resource_id] == robot_id:

# 自己的锁，续期

break

await asyncio.sleep(0.5) self.locks[resource_id] = robot_id

def release_lock (self, resource_id, robot_id):

"""释放资源锁"""

if self .locks.get(resource_id) == robot_id: del self .locks[resource_id]

8.2 部署检查清单

检查项	验收标准	检查方法
地图精度	关键点误差<5cm	使用激光测距仪对比
导航成功率	>98%	连续运行100次测试
电梯对接	>99%	连续100次对接测试
语音识别率	>95%	安静环境测试集
连续运行	>8小时	满电到低电量完整测试
安全机制	急停响应<100ms	实测急停距离

九、总结

基于ROS2的酒店服务机器人系统通过模块化架构设计，实现了导航、语音交互、电梯对接、任务调度等核心功能的解耦与协同。从实际部署经验来看，Nav2导航栈的成熟度已足以支撑商用场景，但细节参数调优仍需根据具体环境反复优化。

在实施过程中，建议重点关注：

1. 前期勘察：充分了解部署环境，特别是电梯控制系统和WiFi覆盖

2. 地图质量：建立高精度地图是导航成功的基础

3. 运维监控：部署完善的监控告警系统，及时发现问题

4. 持续迭代：根据用户反馈持续优化系统

20/21