🤖 第10课：多模态融合

📌 多模态融合概述

人类交互天然是多模态的——说话时配合手势、表情、注视方向。服务机器人需要融合多种感知通道才能准确理解用户意图：

┌─────────────────────────────────────┐
│      决策层融合 (意图级)            │
│  语音意图 + 手势意图 → 最终意图     │
├─────────────────────────────────────┤
│      特征层融合 (语义级)            │
│  语音情感 + 表情情感 → 情感状态     │
├─────────────────────────────────────┤
│      数据层融合 (信号级)            │
│  激光雷达 + 相机 + IMU → 位置估计   │
└─────────────────────────────────────┘

📌 多模态输入融合

class MultiModalInput:
    """多模态输入融合处理器"""
    def __init__(self):
        self.modalities = {
            "voice": {"available": True, "latency": 0.3, "reliability": 0.85},
            "gesture": {"available": True, "latency": 0.1, "reliability": 0.7},
            "gaze": {"available": True, "latency": 0.05, "reliability": 0.6},
            "touch": {"available": True, "latency": 0.02, "reliability": 0.95},
        }
        self.fusion_rules = {
            "confirm": {"voice_any": True, "gesture_nod": True},
            "cancel": {"voice_cancel": True, "gesture_wave": True},
            "point": {"gaze_target": True, "gesture_point": True},
        }

    def process(self, inputs):
        """处理多模态输入"""
        results = {}
        for modality, data in inputs.items():
            if modality in self.modalities and self.modalities[modality]["available"]:
                results[modality] = {
                    "data": data,
                    "confidence": self.modalities[modality]["reliability"],
                    "latency": self.modalities[modality]["latency"],
                }
        
        fused = self._fuse(results)
        return {"modalities": results, "fused": fused}

    def _fuse(self, results):
        """融合决策"""
        intent = None; target = None; confidence = 0
        
        if "voice" in results and "intent" in results["voice"]["data"]:
            intent = results["voice"]["data"]["intent"]
            confidence = results["voice"]["confidence"]
        
        if "gesture" in results:
            gesture = results["gesture"]["data"].get("type")
            if gesture == "point" and intent == "navigate":
                target = results["gesture"]["data"].get("direction")
                confidence = min(1.0, confidence + 0.15)
            elif gesture == "nod" and intent:
                confidence = min(1.0, confidence + 0.1)
            elif gesture == "wave" and intent:
                intent = "cancel"
                confidence = results["gesture"]["confidence"]
        
        if "gaze" in results:
            gaze_target = results["gaze"]["data"].get("target")
            if gaze_target and not target:
                target = gaze_target
        
        return {"intent": intent, "target": target, "confidence": round(confidence, 2)}

mm = MultiModalInput()
print("多模态输入融合")
print("=" * 55)

scenarios = [
    {"name": "语音+手势指向", "inputs": {
        "voice": {"intent": "navigate", "text": "去那边"},
        "gesture": {"type": "point", "direction": "会议室A"},
        "gaze": {"target": "会议室A"}}},
    {"name": "语音+点头确认", "inputs": {
        "voice": {"intent": "confirm", "text": "是的"},
        "gesture": {"type": "nod"}}},
    {"name": "挥手取消", "inputs": {
        "voice": {"intent": "navigate", "text": "去..."},
        "gesture": {"type": "wave"}}},
]

for s in scenarios:
    result = mm.process(s["inputs"])
    print(f"\n📋 {s['name']}:")
    print(f"   融合: 意图={result['fused']['intent']}, 目标={result['fused']['target']}, 置信度={result['fused']['confidence']}")

print("\n✅ 多模态融合验证通过")

📌 传感器数据融合

卡尔曼滤波是多传感器融合的经典方法，在导航定位中广泛应用：

import math

class SensorFusion:
    """传感器数据融合 - 卡尔曼滤波简化版"""
    def __init__(self):
        self.state = {"x": 0, "y": 0, "theta": 0}
        self.covariance = [[1,0,0],[0,1,0],[0,0,0.5]]
        self.process_noise = 0.1
        self.sensors = {
            "lidar": {"noise": 0.05, "rate": 10},
            "camera": {"noise": 0.15, "rate": 30},
            "imu": {"noise": 0.02, "rate": 100},
            "odometry": {"noise": 0.1, "rate": 50},
        }

    def predict(self, vx, vy, vtheta, dt):
        """预测步骤"""
        self.state["x"] += vx * dt
        self.state["y"] += vy * dt
        self.state["theta"] += vtheta * dt
        for i in range(3):
            self.covariance[i][i] += self.process_noise

    def update(self, sensor_name, measurement):
        """更新步骤（简化卡尔曼增益）"""
        if sensor_name not in self.sensors:
            return
        noise = self.sensors[sensor_name]["noise"]
        for key in ["x", "y", "theta"]:
            if key in measurement:
                idx = ["x","y","theta"].index(key)
                kalman_gain = self.covariance[idx][idx] / (self.covariance[idx][idx] + noise)
                self.state[key] += kalman_gain * (measurement[key] - self.state[key])
                self.covariance[idx][idx] *= (1 - kalman_gain)

    def get_state(self):
        return dict(self.state)

fusion = SensorFusion()
print("传感器融合(卡尔曼滤波)")
print("=" * 55)

# 模拟机器人运动与感知
true_pos = {"x": 0, "y": 0, "theta": 0}
import random
random.seed(42)

for step in range(10):
    vx, vy, vtheta = 0.5, 0.3, 0.05
    true_pos["x"] += vx * 0.1; true_pos["y"] += vy * 0.1; true_pos["theta"] += vtheta * 0.1
    
    fusion.predict(vx, vy, vtheta, 0.1)
    
    lidar_meas = {k: true_pos[k] + random.gauss(0, 0.05) for k in ["x","y"]}
    camera_meas = {"theta": true_pos["theta"] + random.gauss(0, 0.15)}
    imu_meas = {"theta": true_pos["theta"] + random.gauss(0, 0.02)}
    
    fusion.update("lidar", lidar_meas)
    fusion.update("camera", camera_meas)
    fusion.update("imu", imu_meas)
    
    if step % 3 == 0:
        est = fusion.get_state()
        err_x = abs(est["x"]-true_pos["x"])
        err_y = abs(est["y"]-true_pos["y"])
        print(f"步骤{step}: 真实({true_pos['x']:.3f},{true_pos['y']:.3f}) "
              f"估计({est['x']:.3f},{est['y']:.3f}) 误差({err_x:.4f},{err_y:.4f})")

print("\n✅ 传感器融合验证通过")

📌 跨模态注意力

不同模态间的相关性和一致性决定融合权重：

class CrossModalAttention:
    """跨模态注意力机制模拟"""
    def __init__(self):
        self.attention_weights = {
            ("voice", "face"): 0.8,   # 语音-面部强关联
            ("voice", "gesture"): 0.6, # 语音-手势中关联
            ("gaze", "gesture"): 0.7,  # 注视-手势强关联
            ("voice", "gaze"): 0.5,    # 语音-注视弱关联
        }
        self.history = []

    def compute_attention(self, modal_data):
        """计算跨模态注意力"""
        modalities = list(modal_data.keys())
        attention_map = {}
        
        for i, m1 in enumerate(modalities):
            for j, m2 in enumerate(modalities):
                if i < j:
                    key = (m1, m2)
                    rev_key = (m2, m1)
                    weight = self.attention_weights.get(key, self.attention_weights.get(rev_key, 0.3))
                    
                    d1 = modal_data[m1]; d2 = modal_data[m2]
                    consistency = self._check_consistency(m1, d1, m2, d2)
                    attention_map[(m1, m2)] = weight * consistency
        
        self.history.append(attention_map)
        return attention_map

    def _check_consistency(self, m1, d1, m2, d2):
        """检查两个模态数据的一致性"""
        if m1 == "voice" and m2 == "face":
            voice_emotion = d1.get("emotion", "neutral")
            face_emotion = d2.get("emotion", "neutral")
            return 1.0 if voice_emotion == face_emotion else 0.3
        if "gaze" in (m1, m2) and "gesture" in (m1, m2):
            return 0.8
        return 0.5

    def get_focus(self, attention_map):
        """确定关注焦点"""
        max_pair = max(attention_map, key=attention_map.get)
        return max_pair

attn = CrossModalAttention()
print("跨模态注意力机制")
print("=" * 55)

scenarios = [
    {"name": "一致信号", "data": {
        "voice": {"emotion": "happy", "text": "谢谢"},
        "face": {"emotion": "happy"},
        "gaze": {"target": "robot"},
        "gesture": {"type": "wave"}}},
    {"name": "冲突信号", "data": {
        "voice": {"emotion": "neutral", "text": "没事"},
        "face": {"emotion": "sad"},
        "gaze": {"target": "floor"},
        "gesture": {"type": "none"}}},
]

for s in scenarios:
    result = attn.compute_attention(s["data"])
    focus = attn.get_focus(result)
    print(f"\n📋 {s['name']}:")
    for pair, weight in result.items():
        print(f"   {pair[0]}↔{pair[1]}: {weight:.2f}")
    print(f"   关注焦点: {focus}")

print("\n✅ 跨模态注意力验证通过")

📌 融合策略选择

📌 练习

📌 成就