摘要

当机器人具备了感知（视觉+传感器融合）、导航（SLAM+Nav2）、操作（视觉抓取）三大基础能力后，如何让它真正“理解”人类意图并自主完成复杂任务？答案是端侧多模态大模型。本文以高通跃龙IQ-9100（100 TOPS NPU，可运行Llama 2 7B @ 22 tok/s）为平台，实战部署视觉语言模型（VLM）实现场景理解与视觉问答，为具身智能体构建“看懂环境”的能力。

📌 本系列共两篇，本文为第一篇，聚焦背景、系统架构和VLM部署；第二篇将介绍LLM任务规划、语音交互及完整系统集成。

1. 为什么具身智能需要端侧大模型

1.1 从规则驱动到理解驱动

传统机器人（规则驱动）：

用户：“把桌子上的红色杯子放到厨房”
系统：无法理解 → 需预编程每种指令
流程：IF command == "move_cup" THEN execute_predefined_sequence()
问题：无法处理开放指令，无法理解新场景

具身智能体（大模型驱动）：

用户：“把桌子上的红色杯子放到厨房”
系统：LLM分解任务：
  1. 视觉搜索 → 在当前场景中找到“红色杯子”
  2. 确认位置 → 在“桌子上”（坐标验证）
  3. 导航到桌子 → Nav2 规划路径
  4. 抓取杯子 → 视觉引导抓取
  5. 导航到厨房 → Nav2 规划路径
  6. 放置杯子 → 选择合适位置放下
  7. 确认完成 → “好的，红色杯子已经放到厨房了”

1.2 端侧 vs 云端大模型

1.3 IQ-9100平台介绍

高通跃龙IQ-9100是高通打造的高性能工业级平台，

可以完美应用到具身智能机器人场景。

• 在《基于高通跃龙IQ-9100打造具身智能机器人多传感器融合感知系统》文中，我们基于高通跃龙IQ-9100平台完成了具身智能机器人的硬件选型，并利用其强大的100 TOPS NPU实现了多摄像头AI感知系统。
• 在《基于高通跃龙IQ-9100打造具身智能机器人视觉SLAM与自主导航系统》》文中，我们基于高通跃龙IQ-9100平台搭建了一套完整的视觉SLAM建图 + 自主导航 + 动态避障系统。
• 在《基于高通跃龙IQ-9100打造具身智能机器人视觉引导抓取系统》》中，我们基于高通跃龙IQ-9100工业机器人平台，打造具身智能机器人场景下一套完整的“眼-手”协作系统。
• 本文（系列），我们基于高通跃龙IQ-9100工业机器人平台，利用端侧多模态大模型赋能具身智能交互系统。

1.3 系统架构

┌─────────────────────────────────────────────────────────────────────┐
│               IQ-9100 具身智能交互系统                                │
│                                                                     │
│  ┌─────────────────── 输入层 ───────────────────────────────────┐    │
│  │                                                             │    │
│  │  ┌─────────┐  ┌──────────┐  ┌──────────┐  ┌────────────┐    │    │
│  │  │ 语音输入 │  │ 文本输入  │  │ 摄像头   │  │ 触摸屏      │    │    │
│  │  │ 麦克风   │  │ 终端/APP │  │ 实时画面  │  │ 手势识别    │    │    │
│  │  └────┬────┘  └────┬─────┘  └────┬─────┘  └─────┬──────┘     │    │
│  └───────┼────────────┼─────────────┼──────────────┼────────────┘    │
│          │            │             │              │                 │
│  ┌───────▼────────────▼─────────────▼──────────────▼────────────┐    │
│  │              多模态理解层 (NPU TP0 + TP1)                     │    │
│  │                                                              │    │
│  │  ┌─────────────┐  ┌──────────────┐  ┌─────────────────────┐  │    │
│  │  │ ASR 语音识别 │  │ VLM 视觉     │  │ LLM 语言理解         │  │    │
│  │  │ Whisper     │  │ 语言模型     │  │ Llama 2 7B           │  │    │
│  │  │ (NPU)       │  │ 场景描述     │  │ 意图识别+任务分解     │   │    │
│  │  └─────────────┘  └──────────────┘  └─────────────────────┘   │    │
│  └──────────────────────────┬────────────────────────────────────┘    │
│                             │                                         │
│  ┌──────────────────────────▼───────────────────────────────────┐     │
│  │                任务规划层 (CPU + LLM)                         │     │
│  │                                                              │     │
│  │  ┌─────────────────┐  ┌──────────────┐  ┌────────────┐       │     │
│  │  │ 任务链生成       │  │ 行为树动态    │  │ 执行监控    │      │     │
│  │  │ LLM → JSON Plan │  │ 构建/修改     │  │ 异常处理    │      │    │
│  │  └─────────────────┘  └──────────────┘  └────────────┘       │    │
│  └──────────────────────────┬───────────────────────────────────┘    │
│                             │                                       │
│  ┌──────────────────────────▼──────────────────────────────────┐    │
│  │              技能执行层 (已有能力)                            │    │
│  │                                                             │    │
│  │  ┌──────┐ ┌──────┐ ┌──────┐ ┌──────┐ ┌──────┐ ┌────────┐    │    │
│  │  │ 导航 │ │ 抓取  │ │ 放置 │ │ 搜索 │ │ 跟随  │ │ 语音播 │    │    │
│  │  │      │ │      │ │      │ │      │ │      │ │ 报     │    │    │
│  │  └──────┘ └──────┘ └──────┘ └──────┘ └──────┘ └────────┘    │    │
│  └─────────────────────────────────────────────────────────────┘    │
└─────────────────────────────────────────────────────────────────────┘

系统主要包含：

• 感知层：摄像头、传感器、VLM场景理解
• 任务规划层：LLM意图识别 + 任务分解 + 行为树动态构建/修改 + 异常处理
• 技能执行层：导航、抓取、放置、搜索、跟随、语音播报等原子能力

2. 视觉语言模型（VLM）部署

2.1 模型选型

在IQ-9100的36GB内存和100 TOPS NPU上，可运行的VLM选择：

推荐方案：VLM（场景理解） + LLM（任务规划）分离部署，按需切换使用NPU。

2.2 VLM场景理解节点（核心代码）

以下为vlm_scene_node.py的关键实现，展示如何在ROS2中集成VLM并使用NPU加速。

# ...（省略导入部分）

class VLMSceneNode(Node):
    def __init__(self):
        super().__init__('vlm_scene')
        
        self.declare_parameter('model_dir', '/opt/models/llava_7b')
        self.declare_parameter('max_tokens', 256)
        self.declare_parameter('image_size', 336)
        
        self.bridge = CvBridge()
        self._latest_frame = None
        self._lock = threading.Lock()
        
        model_dir = self.get_parameter('model_dir').value
        self._load_vlm(model_dir)
        
        # 订阅摄像头原始图像
        self.create_subscription(Image, '/camera/color/image_raw', self._image_callback, 5)
        # 订阅VLM查询请求
        self.create_subscription(String, '/vlm/query', self._query_callback, 10)
        
        self.response_pub = self.create_publisher(String, '/vlm/response', 10)
        self.scene_pub = self.create_publisher(String, '/scene_description', 10)
        
        self.create_timer(5.0, self._periodic_scene_update)
        self.get_logger().info('VLM Scene node initialized')
    
    def _load_vlm(self, model_dir):
        """加载 VLM 模型"""
        try:
            from qnn_sdk import QNNContext
            self.vision_encoder = QNNContext(
                model_path=f"{model_dir}/vision_encoder_int8.ctx",
                backend="http"
            )
            self.llm_decoder = QNNContext(
                model_path=f"{model_dir}/llm_decoder_w4a16.ctx",
                backend="http"
            )
            self.use_npu = True
            self.get_logger().info('VLM loaded on NPU')
        except (ImportError, FileNotFoundError):
            self.use_npu = False
            self.get_logger().warning('VLM NPU not available, using simulation mode')
            # 回退到transformers或模拟模式
            try:
                from transformers import AutoTokenizer
                self.tokenizer = AutoTokenizer.from_pretrained(model_dir, local_files_only=True)
            except Exception:
                self.tokenizer = None
    
    def _image_callback(self, msg):
        frame = self.bridge.imgmsg_to_cv2(msg, 'bgr8')
        with self._lock:
            self._latest_frame = frame
    
    def _query_callback(self, msg):
        query = msg.data
        with self._lock:
            frame = self._latest_frame.copy() if self._latest_frame is not None else None
        if frame is None:
            self._publish_response("无法获取摄像头画面")
            return
        
        t0 = time.perf_counter()
        response = self._vlm_inference(frame, query)
        latency = (time.perf_counter() - t0) * 1000
        self.get_logger().info(f'VLM response ({latency:.0f}ms): "{response[:100]}"')
        self._publish_response(response)
    
    def _vlm_inference(self, frame: np.ndarray, query: str) -> str:
        """VLM推理: 图像 + 问题 → 回答"""
        if self.use_npu and self.tokenizer:
            return self._npu_inference(frame, query)
        else:
            return self._simulation_inference(frame, query)
    
    def _npu_inference(self, frame: np.ndarray, query: str) -> str:
        """NPU加速的VLM推理"""
        img_size = self.get_parameter('image_size').value
        # 图像预处理（resize、归一化、标准化）
        image = cv2.resize(frame, (img_size, img_size))
        image = image.astype(np.float32) / 255.0
        mean = np.array([0.48145466, 0.4578275, 0.40821073])
        std = np.array([0.26862954, 0.26130258, 0.27577711])
        image = (image - mean) / std
        image = image.transpose(2, 0, 1)[np.newaxis, ...]
        
        # 视觉编码器推理
        image_features = self.vision_encoder.execute({"pixel_values": image.astype(np.float32)})
        
        # 构造文本prompt
        prompt = f"<image>\nUSER: {query}\nASSISTANT:"
        input_ids = self.tokenizer.encode(prompt, return_tensors="np")
        
        max_tokens = self.get_parameter('max_tokens').value
        generated = []
        for _ in range(max_tokens):
            logits = self.lm_decoder.execute({
                "input_ids": input_ids,
                "image_features": image_features["last_hidden_state"]
            })
            next_token = int(np.argmax(logits[0, -1, :]))
            if next_token == self.tokenizer.eos_token_id:
                break
            generated.append(next_token)
            input_ids = np.concatenate([input_ids, np.array([[next_token]])], axis=1)
        
        return self.tokenizer.decode(generated, skip_special_tokens=True)
    
    def _simulation_inference(self, frame: np.ndarray, query: str) -> str:
        """模拟推理（开发/调试用）—— 基于颜色检测的简单逻辑"""
        h, w = frame.shape[:2]
        hsv = cv2.cvtColor(frame, cv2.COLOR_BGR2HSV)
        colors_detected = []
        color_ranges = {
            "红色": ((0, 100, 100), (10, 255, 255)),
            "蓝色": ((100, 100, 100), (130, 255, 255)),
            "绿色": ((40, 100, 100), (80, 255, 255)),
            "黄色": ((20, 100, 100), (40, 255, 255)),
        }
        for name, (lower, upper) in color_ranges.items():
            mask = cv2.inRange(hsv, np.array(lower), np.array(upper))
            if np.sum(mask > 0) > 500:
                colors_detected.append(name)
        
        if "在哪" in query or "找" in query or "搜索" in query:
            return json.dumps({
                "type": "search_result",
                "found": len(colors_detected) > 0,
                "objects": colors_detected,
                "description": f"场景中检测到: {', '.join(colors_detected) if colors_detected else '无物体'}"
            }, ensure_ascii=False)
        elif "描述" in query or "看到" in query or "场景" in query:
            brightness = np.mean(frame)
            desc = f"当前画面分辨率{w}x{h}, 亮度{'较亮' if brightness > 128 else '较暗'}, 检测到颜色: {', '.join(colors_detected) if colors_detected else '无'}"
            return desc
        else:
            return f"收到查询: {query}。场景分析: 检测到{len(colors_detected)}种颜色物体。"
    
    def _periodic_scene_update(self):
        """定期更新场景描述（后台）"""
        with self._lock:
            frame = self._latest_frame.copy() if self._latest_frame is not None else None
        if frame is None:
            return
        desc = self._vlm_inference(frame, "请简要描述当前场景中有什么物体")
        msg = String()
        msg.data = desc
        self.scene_pub.publish(msg)
    
    def _publish_response(self, text: str):
        msg = String()
        msg.data = text
        self.response_pub.publish(msg)

def main(args=None):
    rclpy.init(args=args)
    node = VLMSceneNode()
    rclpy.spin(node)
    node.destroy_node()
    rclpy.shutdown()

说明：以上代码展示了VLM节点完整实现，包括NPU推理、图像预处理、模拟回退模式以及定期场景更新。实际部署时需根据QNN SDK具体接口调整。

小结

本文（第一篇）介绍了具身智能为何需要端侧大模型，对比了端云部署差异，并详细给出了IQ-9100上VLM的部署代码。下一篇我们将继续深入LLM任务链规划、语音交互集成以及完整的系统演示，敬请期待。