机器人情绪系统设计
范围:Reachy Mini 机器人的情绪动作系统设计,涵盖情绪/舞蹈/呼吸状态切换、音效同步、产品鲜活感设计
综合自:reachy-mini-conversation-app
优先级:P0
概述
Reachy Mini 的情绪系统是一个精心设计的分层动画架构,它将机器人的"鲜活感"(aliveness)分解为多个独立的运动层,并通过智能融合创造出自然、持续的交互体验。这个系统的核心思想是:机器人永远 不应该完全静止。
系统采用"主要动作 + 次要偏移"的双层融合架构,在 100Hz 控制循环中实时合成最终姿态。情绪动作、舞蹈、头部定位和呼吸状态作为主要动作顺序执行,而语音摆动和人脸追踪作为次要偏移叠加在主要动作之上。
核心架构
1. 分层运动系统
主要动作(Primary Moves)
主要动作是互斥的、顺序执行的运动:
队列顺序:情绪 → 舞蹈 → Goto 定位 → 呼吸
代码实现(moves.py):
class MovementManager:
"""协调顺序动作、附加偏移和机器人输出(100Hz)。
职责:
- 拥有实时循环,采样当前主要动作,融合次要偏移,调用 set_target
- 在 idle_inactivity_delay 后启动 BreathingMove
- 暴露线程安全 API
"""
def __init__(self, current_robot: ReachyMini, camera_worker=None):
self.move_queue: deque[Move] = deque() # 主要动作队列
self.state = MovementState()
self.idle_inactivity_delay = 0.3 # 秒
self.target_frequency = 100.0 # Hz
设计理由:
- 互斥执行:避免动作冲突,确保流畅过渡
- 队列管理:支持动作预排队,实现无缝衔接
- 单控制点:所有运动通过
set_target统一输出
次要偏移(Secondary Offsets)
次要偏移是叠加在主要动作上的实时偏移:
# 次要偏移组合
secondary_offsets = [
self.state.speech_offsets[0] + self.state.face_tracking_offsets[0], # x
self.state.speech_offsets[1] + self.state.face_tracking_offsets[1], # y
self.state.speech_offsets[2] + self.state.face_tracking_offsets[2], # z
self.state.speech_offsets[3] + self.state.face_tracking_offsets[3], # roll
self.state.speech_offsets[4] + self.state.face_tracking_offsets[4], # pitch
self.state.speech_offsets[5] + self.state.face_tracking_offsets[5], # yaw
]
设计理由:
- 加性融合:多个偏移可以同时作用
- 世界坐标系:使用
compose_world_offset进行姿态合成 - 线程安全:通过锁保护偏移更新
2. 呼吸状态配置
呼吸是机器人在空闲时自动进入的"待机动画",创造持续鲜活感。
呼吸参数设计
class BreathingMove(Move):
def __init__(self, interpolation_start_pose, interpolation_start_antennas,
interpolation_duration=1.0):
# 中性位置
self.neutral_head_pose = create_head_pose(0, 0, 0, 0, 0, 0, degrees=True)
self.neutral_antennas = np.array([0.0, 0.0])
# 呼吸参数
self.breathing_z_amplitude = 0.005 # 5mm 轻微上下浮动
self.breathing_frequency = 0.1 # Hz(每分钟 6 次呼吸)
self.antenna_sway_amplitude = np.deg2rad(15) # 天线摆动 15 度
self.antenna_frequency = 0.5 # Hz(天线更快摆动)
呼 吸动画分两阶段:
- 插值阶段:从当前姿态平滑过渡到中性位置
- 呼吸循环:持续的 Z 轴浮动 + 天线交替摆动
def evaluate(self, t: float):
if t < self.interpolation_duration:
# 阶段 1:插值到中性位置
interpolation_t = t / self.interpolation_duration
head_pose = linear_pose_interpolation(
self.interpolation_start_pose,
self.neutral_head_pose,
interpolation_t
)
else:
# 阶段 2:呼吸循环
breathing_time = t - self.interpolation_duration
# Z 轴轻微浮动
z_offset = self.breathing_z_amplitude * np.sin(
2 * np.pi * self.breathing_frequency * breathing_time
)
# 天线交替摆动(增加鲜活感)
antenna_sway = self.antenna_sway_amplitude * np.sin(
2 * np.pi * self.antenna_frequency * breathing_time
)
antennas = np.array([antenna_sway, -antenna_sway])
设计考量:
- 5mm 浮动:足够被感知但不分散注意力
- 6 次/分钟:接近人类呼吸频率,创造亲和感
- 天线交替:打破完全对称,增加有机感
呼吸触发条件
def _manage_breathing(self, current_time: float):
"""管理空闲时的自动呼吸"""
if (self.state.current_move is None
and not self.move_queue
and not self._is_listening
and not self._breathing_active):
idle_for = current_time - self.state.last_activity_time
if idle_for >= self.idle_inactivity_delay: # 0.3 秒
# 获取当前姿态作为插值起点
current_head_pose = self.current_robot.get_current_head_pose()
_, current_antennas = self.current_robot.get_current_joint_positions()
breathing_move = BreathingMove(
interpolation_start_pose=current_head_pose,
interpolation_start_antennas=current_antennas,
interpolation_duration=1.0
)
self.move_queue.append(breathing_move)
中断机制:任何新动作入队时,呼吸立即被中断:
if isinstance(self.state.current_move, BreathingMove) and self.move_queue:
self.state.current_move = None
self._breathing_active = False
logger.debug("由于新动作活动停止呼吸")
3. 情绪动作系统
情绪库加载
情绪动作从 Hugging Face 数据集动态加载:
# play_emotion.py
try:
from reachy_mini.motion.recorded_move import RecordedMoves
# 自动从环境变量读取 HF_TOKEN
RECORDED_MOVES = RecordedMoves("pollen-robotics/reachy-mini-emotions-library")
EMOTION_AVAILABLE = True
except ImportError as e:
logger.warning(f"情绪库不可用: {e}")
EMOTION_AVAILABLE = False
情绪工具定义
class PlayEmotion(Tool):
name = "play_emotion"
description = "播放预录制的情绪动作"
parameters_schema = {
"type": "object",
"properties": {
"emotion": {
"type": "string",
"description": """要播放的情绪名称。
可用情绪列表:
{get_available_emotions_and_descriptions()}
""",
},
},
"required": ["emotion"],
}
async def __call__(self, deps: ToolDependencies, **kwargs):
emotion_name = kwargs.get("emotion")
# 验证情绪存在
emotion_names = RECORDED_MOVES.list_moves()
if emotion_name not in emotion_names:
return {"error": f"未知情绪 '{emotion_name}'"}
# 加入动作队列
emotion_move = EmotionQueueMove(emotion_name, RECORDED_MOVES)
movement_manager.queue_move(emotion_move)
return {"status": "queued", "emotion": emotion_name}
情绪动作包装器
class EmotionQueueMove(Move):
"""将情绪动作包装为队列兼容的 Move 对象"""
def __init__(self, emotion_name: str, recorded_moves: RecordedMoves):
self.emotion_move = recorded_moves.get(emotion_name)
self.emotion_name = emotion_name
@property
def duration(self) -> float:
return float(self.emotion_move.duration)
def evaluate(self, t: float):
head_pose, antennas, body_yaw = self.emotion_move.evaluate(t)
# 转换为标准格式
if isinstance(antennas, tuple):
antennas = np.array([antennas[0], antennas[1]])
return (head_pose, antennas, body_yaw)
4. 舞蹈动作系统
可用舞蹈动作
舞蹈动作提供更丰富的表现力:
AVAILABLE_MOVES = {
"simple_nod": "简单的连续上下点头",
"head_tilt_roll": "连续的侧向头部滚动(耳朵贴肩膀)",
"side_to_side_sway": "平滑的左右摇摆",
"dizzy_spin": "结合滚转和俯仰的眩晕圆形运动",
"stumble_and_recover": "模拟踉跄和恢复,多轴运动,氛围感好",
"interwoven_spirals": "三轴不同频率的复杂螺旋运动",
"sharp_side_tilt": "使用三角波形的锐利快速侧倾",
"side_peekaboo": "多阶段躲猫猫表演,两侧躲藏和窥视",
"yeah_nod": "强调性的两部分 yeah 点头",
"uh_huh_tilt": "滚转和俯仰组合的同意手势",
"neck_recoil": "快速瞬态颈部后缩",
"chin_lead": "下巴引导的前向运动",
"groovy_sway_and_roll": "左右摇摆配合相应滚转的律动效果",
"chicken_peck": "锐利的前向啄食动作",
"side_glance_flick": "快速侧视-停留-返回",
"polyrhythm_combo": "3拍摇摆+2拍点头创造复节奏感",
"grid_snap": "使用方波的机器人网格运动",
"pendulum_swing": "简单平滑的钟摆式滚转摆动",
"jackson_square": "5点路径描绘矩形,到达检查点时锐利抽动",
}
舞蹈工具定义
class Dance(Tool):
name = "dance"
description = "播放命名或随机舞蹈动作。非阻塞。"
parameters_schema = {
"type": "object",
"properties": {
"move": {
"type": "string",
"description": "动作名称;使用 'random' 或省略则随机",
},
"repeat": {
"type": "integer",
"description": "重复次数(默认 1)",
},
},
}
async def __call__(self, deps: ToolDependencies, **kwargs):
move_name = kwargs.get("move")
repeat = int(kwargs.get("repeat", 1))
if not move_name or move_name == "random":
move_name = random.choice(list(AVAILABLE_MOVES.keys()))
# 支持重复排队
for _ in range(repeat):
dance_move = DanceQueueMove(move_name)
movement_manager.queue_move(dance_move)
return {"status": "queued", "move": move_name, "repeat": repeat}
5. 语音同步摆动系统
音频驱动的头部运动
这是让机器人"说话时自然动起来"的关键系统:
# speech_tapper.py - 音频分析参数
SR = 16_000 # 采样率
FRAME_MS = 20 # 帧长度(毫秒)
HOP_MS = 50 # 跳跃长度(毫秒)
# 摆动参数
SWAY_MASTER = 1.5 # 主增益
VAD_DB_ON = -35.0 # 语音活动检测开启阈值
VAD_DB_OFF = -45.0 # 语音活动检测关闭阈值
# 各轴摆动频率和幅度
SWAY_F_PITCH = 2.2 # 俯仰频率 Hz
SWAY_A_PITCH_DEG = 4.5 # 俯仰幅度(度)
SWAY_F_YAW = 0.6 # 偏航频率 Hz
SWAY_A_YAW_DEG = 7.5 # 偏航幅度(度)
SWAY_F_ROLL = 1.3 # 滚转频率 Hz
SWAY_A_ROLL_DEG = 2.25 # 滚转幅度(度)
# 位移摆动
SWAY_F_X = 0.35 # X 轴频率
SWAY_A_X_MM = 4.5 # X 轴幅度(毫米)
SWAY_F_Y = 0.45 # Y 轴频率
SWAY_A_Y_MM = 3.75 # Y 轴幅度
SWAY_F_Z = 0.25 # Z 轴频率
SWAY_A_Z_MM = 2.25 # Z 轴幅度
实时音频处理流程
class SwayRollRT:
"""输入音频块 → 每跳摆动输出"""
def feed(self, pcm: NDArray, sr: int) -> List[Dict]:
# 1. 转换为 float32 单声道
x = _to_float32_mono(pcm)
# 2. 如需要则重采样
if sr_in != SR:
x = _resample_linear(x, sr_in, SR)
# 3. 按 HOP 大小处理
while self.carry.size >= HOP:
hop = self.carry[:HOP]
# 4. 计算 RMS 响度(dBFS)
db = _rms_dbfs(frame)
# 5. VAD 带滞后的语音检测
if db >= VAD_DB_ON:
self.vad_on = True
elif db <= VAD_DB_OFF:
self.vad_on = False
# 6. 计算响度增益
loud = _loudness_gain(db) * SWAY_MASTER
# 7. 生成各轴摆动
pitch = math.radians(SWAY_A_PITCH_DEG) * loud * env * \
math.sin(2 * math.pi * SWAY_F_PITCH * self.t + phase)
yaw = math.radians(SWAY_A_YAW_DEG) * loud * env * \
math.sin(2 * math.pi * SWAY_F_YAW * self.t + phase)
# ... 其他轴
out.append({
"pitch_rad": pitch,
"yaw_rad": yaw,
"roll_rad": roll,
"x_mm": x_mm,
"y_mm": y_mm,
"z_mm": z_mm,
})
头部摆动器
class HeadWobbler:
"""将音频增量转换为头部运动偏移"""
def __init__(self, set_speech_offsets: Callable):
self._apply_offsets = set_speech_offsets
self.sway = SwayRollRT()
self.audio_queue = Queue()
# 延迟对齐参数
MOVEMENT_LATENCY_S = 0.2 # 音频和运动之间的延迟
def feed(self, delta_b64: str):
"""线程安全:将音频推入队列"""
buf = np.frombuffer(base64.b64decode(delta_b64), dtype=np.int16)
self.audio_queue.put((generation, SAMPLE_RATE, buf))
def working_loop(self):
"""音频 → 头部运动偏移"""
while not self._stop_event.is_set():
chunk = self.audio_queue.get_nowait()
# 处理音频生成摆动数据
results = self.sway.feed(pcm, sr)
for r in results:
# 时间对齐
target = base_ts + MOVEMENT_LATENCY_S + hops_done * hop_dt
if target > now:
time.sleep(target - now)
# 应用偏移
offsets = (
r["x_mm"] / 1000.0, # mm → m
r["y_mm"] / 1000.0,
r["z_mm"] / 1000.0,
r["roll_rad"],
r["pitch_rad"],
r["yaw_rad"],
)
self._apply_offsets(offsets)
设计考量:
- 200ms 延迟:补偿音频处理和机械延迟
- 随机相位:每次会话使用不同的初始相位,避免机械感
- 响度驱动:幅度随音量动态调整
6. 动作时长设计
Move 基类的时长协议
所有动作必须实现 duration 属性,控制循环依赖它来判断动作是否完成:
class Move(abc.ABC):
"""动作基类,所有动作必须实现 duration 属性"""
@property
@abc.abstractmethod
def duration(self) -> float:
"""