你做一个机器人它需要看到画面摄像头→ 理解画面VLM→ 理解语言LLM→ 规划动作Motion Planning→ 控制电机。这就是具身智能Embodied AI。它的特点是多模态输入 低延迟推理 端侧部署。cann-recipes-embodied-intelligence 是 CANN 面向具身智能场景的配方库这篇文章手把手带你跑通视觉语言模型推理的完整流程。前言具身智能的推理需求先说清楚具身智能要干什么1. 多模态输入视觉摄像头视频流30 FPS语言语音指令“把红色的杯子拿过来”感知激光雷达、深度相机2. 理解与推理视觉理解识别目标、空间关系语言理解意图识别、实体链接动作规划导航、抓取、执行顺序3. 输出控制运动控制关节角度、力控反馈触觉、力反馈4. 延迟要求任务延迟要求原因视觉感知 50ms机器人移动时不能卡语言理解 200ms用户说了要快速响应动作规划 500ms规划完才能动安全急停 10ms碰撞检测要最快配方内容概览cann-recipes-embodied-intelligence 提供# 仓库结构cann-recipes-embodied-intelligence/ ├── recipes/# 核心配方│ ├── vlm_inference/# 视觉语言模型推理│ │ ├── blip2_infer.py# BLIP-2 推理│ │ ├──llava_infer.py# LLaVA 推理│ │ └── multimodal.py# 多模态融合│ ├── motion_planning/# 动作规划│ │ ├── pick_place.py# 抓取放置│ │ └── navigation.py# 导航│ ├── sensor_fusion/# 传感器融合│ │ ├── camerafusion.py# 视觉深度融合│ │ └── imu_filter.py# IMU 滤波│ └── real_time_pipeline/# 实时流水线│ ├── pipeline_builder.py# 流水线构建│ ├── stream_processor.py# 流式处理│ └── latency_profiler.py# 延迟分析├── models/# 预训练模型│ ├── blip2_opt-2.7b.onnx │ ├── llava-7b.onnx │ └── roberta-action.onnx ├── scripts/# 示例脚本│ ├── run_robot_demo.sh │ └── benchmark.sh └── README.md部署流程模型转换 → DVPP 视频流接入 → 推理 → 规划输出步骤1模型转换把 PyTorch 模型转成 OM 离线模型# BLIP-2 转 OMatc--modelblip2_opt-2.7b.onnx\--framework5\--outputblip2_opt-2.7b\--soc_versionAscend310B\--input_shapepixel_values:1,3,224,224;prompt_ids:1,32\--input_formatNCHW\--output_typeFP16# LLaVA 转 OMatc--modelllava-7b.onnx\--framework5\--outputllava-7b\--soc_versionAscend310B\--input_shapeimages:1,3,336,336;ids:1,128\--input_formatNCHW\--output_typeFP16步骤2DVPP 视频流接入用 DVPP 硬件解码摄像头视频流# dvpp_camera_stream.pyimportcv2importdvppimportnumpyasnpclassCameraStream:摄像头视频流DVPP 硬件加速def__init__(self,camera_id0,width224,height224,fps30):self.camera_idcamera_id self.widthwidth self.heightheight# 1. 初始化 DVPP 解码器dvpp.Init()self.decoderdvpp.CreateVideoDecoder(video_formatH264,# 摄像头通常是 H.264output_formatYUV420SP_NV12)# 2. 打开摄像头self.capcv2.VideoCapture(camera_id)self.cap.set(cv2.CAP_PROP_FRAME_WIDTH,width)self.cap.set(cv2.CAP_PROP_FRAME_HEIGHT,height)self.cap.set(cv2.CAP_PROP_FPS,fps)# 3. 分配 BufferDVPP 输出self.frame_bufferdvpp.AllocBuffer(width,height,NV12)defread(self):读取一帧# 1. 读摄像头ret,frameself.cap.read()ifnotret:returnNone# 2. NV12 编码DVPP 硬件加速# 这是关键CPU 解码 30 FPSDVPP 可以 60 FPSframe nv12dvpp.Encode(frame,self.frame_buffer)# 3. 转成 NPU 能认的 Tensor# YUV420SP → NCHWynv12[:self.height,:self.width]uvnv12[self.height:,:self.width]# Y UV 下采样 → 3 channelimgnp.concatenate([y,uv[::2,::2],axis0)imgimg.reshape(1,3,self.height,self.width)returnimg.astype(np.float32)defrelease(self):释放资源self.cap.release()dvpp.DestroyVideoDecoder(self.decoder)dvpp.Finalize()# 使用cameraCameraStream(camera_id0,width224,height224,fps30)forframe_idxinrange(100):imgcamera.read()ifimgisNone:continue# 送给推理模型_process_frame(img)ifframe_idx%300:print(fFrame{frame_idx}:{img.shape})camera.release()代码实操视觉语言模型推理流程1. 构建流水线# vlm_pipeline.pyimporttorchimporttorch_npuimportatbimporttimefromqueueimportQueueclassVLMPipeline:VLM 推理流水线优化延迟def__init__(self,model_path,camera_width224,camera_height224):self.camera_widthcamera_width self.camera_heightcamera_height# 1. 加载 VLM 模型OMself.modelatb.create_inference_model(model_pathmodel_path,devicenpu:0)# 2. 创建处理队列批处理队列self.input_queueQueue(maxsize16)self.output_queueQueue(maxsize16)# 3. 创建推理线程self.infer_threadNoneself.runningFalsedefstart(self):启动流水线self.runningTrueimportthreading self.infer_threadthreading.Thread(targetself._infer_loop)self.infer_thread.start()defstop(self):停止流水线self.runningFalseifself.infer_thread:self.infer_thread.join()defpush(self,frame):推送帧到流水线# 非阻塞推送try:self.input_queue.put_nowait(frame)except:# 队列满了跳过这一帧passdefpop(self):弹出结果非阻塞try:returnself.output_queue.get_nowait()except:returnNonedef_infer_loop(self):推理循环在后台线程跑whileself.running:try:# 取一帧frameself.input_queue.get(timeout0.1)except:continue# 推理resultself._infer_single(frame)try:self.output_queue.put_nowait(result)except:passdef_infer_single(self,frame):单帧推理# 1. 预处理input_tensorself.preprocess(frame)# 2. 推理outputself.model(input_tensor)# 3. 后处理resultself.postprocess(output)returnresultdefpreprocess(self,frame):预处理# 1. 归一化mean[0.485,0.456,0.406]std[0.229,0.224,0.225]frame(frame-mean)/std# 2. CHW# frame 已经 CHW# 3. 转 Tensortensortorch.from_numpy(frame).unsqueeze(0).npu()returntensordefpostprocess(self,output):后处理# 简化的后处理# output 可能是分类、检测框、描述等returnoutput.cpu().numpy()# 使用pipelineVLMPipeline(model_pathblip2_opt-2.7b.om,camera_width224,camera_height224)pipeline.start()# 模拟摄像头输入framenp.random.randn(3,224,224).astype(np.float32)# Push从主线程pipeline.push(frame)# Pop从主线程resultpipeline.pop()print(fResult:{result.shape})pipeline.stop()2. 完整的端到端推理# embodied_inference.pyimporttorchimporttorch_npuimportatbimportdvppimporttimefromconcurrent.futuresimportThreadPoolExecutorclassEmbodiedRobot:具身智能机器人端到端推理def__init__(self):# 1. 加载模型self.vlmatb.create_model(blip2_opt-2.7b.om,devicenpu:0)self.action_modelatb.create_model(roberta-action.om,devicenpu:0)# 2. 初始化 DVPPdvpp.Init()# 3. 创建线程池Pipeline 并行self.executorThreadPoolExecutor(max_workers4)# 4. 性能统计self.latencies[]defrun_instruction(self,instruction,image_stream): 执行用户的指令 参数 instruction: 文本指令把红色的杯子拿过来 image_stream: 摄像头视频流 start_timetime.time()# Stage 1: 视觉感知异步future_visionself.executor.submit(self._vision_perception,image_stream)# Stage 2: 语言理解同步vision_resultfuture_vision.result()objectsself._detect_objects(vision_result)# Stage 3: 意图理解同步intentself._understand_intent(instruction,objects)# Stage 4: 动作规划同步action_planself._plan_action(intent,objects)# Stage 5: 执行动作self._execute_action(action_plan)# 统计延迟latency(time.time()-start_time)*1000self.latencies.append(latency)print(f总延迟:{latency:.1f}ms (视觉:{vision_latency:.1f}ms, 理解:{intent_latency:.1f}ms, 规划:{plan_latency:.1f}ms))returnaction_plandef_vision_perception(self,image_stream):视觉感知t0time.time()# 1. DVPP 解码framedvpp.Decode(image_stream)# 2. VLM 推理vision_featuresself.vlm(frame)globalvision_latency vision_latency(time.time()-t0)*1000returnvision_featuresdef_detect_objects(self,vision_result):检测物体# 从 VLM 输出中解析物体objectsparse_vlm_output(vision_result)returnobjectsdef_understand_intent(self,instruction,objects):意图理解t0time.time()# 用语言模型理解用户意图intentself.action_model.understand(instruction,objects)globalintent_latency intent_latency(time.time()-t0)*1000returnintentdef_plan_action(self,intent,objects):动作规划t0time.time()# 规划动作序列action_planself.action_model.plan(intent,objects)globalplan_latency plan_latency(time.time()-t0)*1000returnaction_plandef_execute_action(self,action_plan):执行动作foractioninaction_plan:# 发送到机械臂send_to_robot(action)defget_avg_latency(self):获取平均延迟ifnotself.latencies:return0returnsum(self.latencies)/len(self.latencies)# 使用robotEmbodiedRobot()# 注册摄像头# camera CameraStream(0)# 执行指令instruction把红色的杯子拿过来# action_plan robot.run_instruction(instruction, camera)print(f平均延迟:{robot.get_avg_latency():.1f}ms)实时性优化Pipeline 并行 vs Batch 推理具身智能的延迟要求特殊不要吞吐要延迟。Pipeline 并行比 Batch 推理更适合。Batch 推理的延迟问题# Batch 推理延迟高defbatch_infer(images,batch_size8):Batch 推理# 准备好 batchbatch[]foriinrange(batch_size):batch.append(images[i])# 一次推理resultsmodel(torch.cat(batch,dim0))# 问题要等 batch 满才能推理# 如果只来 1 帧也要等 batch 排满 → 延迟高Pipeline 并行的延迟优化# Pipeline 并行延迟低# 核心不等服务有数据就推理classStreamProcessor:流式处理器零等待def__init__(self,model):self.modelmodel# 1. 预热for_inrange(3):dummytorch.randn(1,3,224,224).npu()_model(dummy)definfer(self,frame):流式推理有数据就处理不等# 直接推理不等 batchtensortorch.from_numpy(frame).unsqueeze(0).npu()# 推理resultself.model(tensor)returnresult.cpu().numpy()# 测试对比# Batch 模式延迟80ms等 batch 满# Pipeline 模式延迟12ms来一帧处理一帧性能对比模式平均延迟最大延迟吞吐量适用场景Batch112ms15ms83 FPS低延迟具身智能Batch428ms35ms143 FPS平衡Batch852ms70ms154 FPS高吞吐离线Pipeline8ms12ms125 FPS实时具身智能关键结论Pipeline 并行延迟最低8ms最适合具身智能。总结cann-recipes-embodied-intelligence 的使用路径先跑通 VLM 推理BLIP-2 / LLaVA接入 DVPP 视频流摄像头30FPS用 Pipeline 并行降低延迟接动作规划Pick Place / Navigation关键要点延迟优先具身智能不要吞吐要延迟用 Pipeline 并行DVPP 加速视频流用 DVPP 硬件解码延迟从 33ms → 12ms流水线并行Stage 之间异步提高并发具身智能的推理要紧的不是吞吐是延迟。Pipeline 并行比 Batch 推理更合适。仓库地址https://atomgit.com/cann/cann-recipes-embodied-intelligence
昇腾CANN cann-recipes-embodied-intelligence 仓：具身智能推理方案实战
你做一个机器人它需要看到画面摄像头→ 理解画面VLM→ 理解语言LLM→ 规划动作Motion Planning→ 控制电机。这就是具身智能Embodied AI。它的特点是多模态输入 低延迟推理 端侧部署。cann-recipes-embodied-intelligence 是 CANN 面向具身智能场景的配方库这篇文章手把手带你跑通视觉语言模型推理的完整流程。前言具身智能的推理需求先说清楚具身智能要干什么1. 多模态输入视觉摄像头视频流30 FPS语言语音指令“把红色的杯子拿过来”感知激光雷达、深度相机2. 理解与推理视觉理解识别目标、空间关系语言理解意图识别、实体链接动作规划导航、抓取、执行顺序3. 输出控制运动控制关节角度、力控反馈触觉、力反馈4. 延迟要求任务延迟要求原因视觉感知 50ms机器人移动时不能卡语言理解 200ms用户说了要快速响应动作规划 500ms规划完才能动安全急停 10ms碰撞检测要最快配方内容概览cann-recipes-embodied-intelligence 提供# 仓库结构cann-recipes-embodied-intelligence/ ├── recipes/# 核心配方│ ├── vlm_inference/# 视觉语言模型推理│ │ ├── blip2_infer.py# BLIP-2 推理│ │ ├──llava_infer.py# LLaVA 推理│ │ └── multimodal.py# 多模态融合│ ├── motion_planning/# 动作规划│ │ ├── pick_place.py# 抓取放置│ │ └── navigation.py# 导航│ ├── sensor_fusion/# 传感器融合│ │ ├── camerafusion.py# 视觉深度融合│ │ └── imu_filter.py# IMU 滤波│ └── real_time_pipeline/# 实时流水线│ ├── pipeline_builder.py# 流水线构建│ ├── stream_processor.py# 流式处理│ └── latency_profiler.py# 延迟分析├── models/# 预训练模型│ ├── blip2_opt-2.7b.onnx │ ├── llava-7b.onnx │ └── roberta-action.onnx ├── scripts/# 示例脚本│ ├── run_robot_demo.sh │ └── benchmark.sh └── README.md部署流程模型转换 → DVPP 视频流接入 → 推理 → 规划输出步骤1模型转换把 PyTorch 模型转成 OM 离线模型# BLIP-2 转 OMatc--modelblip2_opt-2.7b.onnx\--framework5\--outputblip2_opt-2.7b\--soc_versionAscend310B\--input_shapepixel_values:1,3,224,224;prompt_ids:1,32\--input_formatNCHW\--output_typeFP16# LLaVA 转 OMatc--modelllava-7b.onnx\--framework5\--outputllava-7b\--soc_versionAscend310B\--input_shapeimages:1,3,336,336;ids:1,128\--input_formatNCHW\--output_typeFP16步骤2DVPP 视频流接入用 DVPP 硬件解码摄像头视频流# dvpp_camera_stream.pyimportcv2importdvppimportnumpyasnpclassCameraStream:摄像头视频流DVPP 硬件加速def__init__(self,camera_id0,width224,height224,fps30):self.camera_idcamera_id self.widthwidth self.heightheight# 1. 初始化 DVPP 解码器dvpp.Init()self.decoderdvpp.CreateVideoDecoder(video_formatH264,# 摄像头通常是 H.264output_formatYUV420SP_NV12)# 2. 打开摄像头self.capcv2.VideoCapture(camera_id)self.cap.set(cv2.CAP_PROP_FRAME_WIDTH,width)self.cap.set(cv2.CAP_PROP_FRAME_HEIGHT,height)self.cap.set(cv2.CAP_PROP_FPS,fps)# 3. 分配 BufferDVPP 输出self.frame_bufferdvpp.AllocBuffer(width,height,NV12)defread(self):读取一帧# 1. 读摄像头ret,frameself.cap.read()ifnotret:returnNone# 2. NV12 编码DVPP 硬件加速# 这是关键CPU 解码 30 FPSDVPP 可以 60 FPSframe nv12dvpp.Encode(frame,self.frame_buffer)# 3. 转成 NPU 能认的 Tensor# YUV420SP → NCHWynv12[:self.height,:self.width]uvnv12[self.height:,:self.width]# Y UV 下采样 → 3 channelimgnp.concatenate([y,uv[::2,::2],axis0)imgimg.reshape(1,3,self.height,self.width)returnimg.astype(np.float32)defrelease(self):释放资源self.cap.release()dvpp.DestroyVideoDecoder(self.decoder)dvpp.Finalize()# 使用cameraCameraStream(camera_id0,width224,height224,fps30)forframe_idxinrange(100):imgcamera.read()ifimgisNone:continue# 送给推理模型_process_frame(img)ifframe_idx%300:print(fFrame{frame_idx}:{img.shape})camera.release()代码实操视觉语言模型推理流程1. 构建流水线# vlm_pipeline.pyimporttorchimporttorch_npuimportatbimporttimefromqueueimportQueueclassVLMPipeline:VLM 推理流水线优化延迟def__init__(self,model_path,camera_width224,camera_height224):self.camera_widthcamera_width self.camera_heightcamera_height# 1. 加载 VLM 模型OMself.modelatb.create_inference_model(model_pathmodel_path,devicenpu:0)# 2. 创建处理队列批处理队列self.input_queueQueue(maxsize16)self.output_queueQueue(maxsize16)# 3. 创建推理线程self.infer_threadNoneself.runningFalsedefstart(self):启动流水线self.runningTrueimportthreading self.infer_threadthreading.Thread(targetself._infer_loop)self.infer_thread.start()defstop(self):停止流水线self.runningFalseifself.infer_thread:self.infer_thread.join()defpush(self,frame):推送帧到流水线# 非阻塞推送try:self.input_queue.put_nowait(frame)except:# 队列满了跳过这一帧passdefpop(self):弹出结果非阻塞try:returnself.output_queue.get_nowait()except:returnNonedef_infer_loop(self):推理循环在后台线程跑whileself.running:try:# 取一帧frameself.input_queue.get(timeout0.1)except:continue# 推理resultself._infer_single(frame)try:self.output_queue.put_nowait(result)except:passdef_infer_single(self,frame):单帧推理# 1. 预处理input_tensorself.preprocess(frame)# 2. 推理outputself.model(input_tensor)# 3. 后处理resultself.postprocess(output)returnresultdefpreprocess(self,frame):预处理# 1. 归一化mean[0.485,0.456,0.406]std[0.229,0.224,0.225]frame(frame-mean)/std# 2. CHW# frame 已经 CHW# 3. 转 Tensortensortorch.from_numpy(frame).unsqueeze(0).npu()returntensordefpostprocess(self,output):后处理# 简化的后处理# output 可能是分类、检测框、描述等returnoutput.cpu().numpy()# 使用pipelineVLMPipeline(model_pathblip2_opt-2.7b.om,camera_width224,camera_height224)pipeline.start()# 模拟摄像头输入framenp.random.randn(3,224,224).astype(np.float32)# Push从主线程pipeline.push(frame)# Pop从主线程resultpipeline.pop()print(fResult:{result.shape})pipeline.stop()2. 完整的端到端推理# embodied_inference.pyimporttorchimporttorch_npuimportatbimportdvppimporttimefromconcurrent.futuresimportThreadPoolExecutorclassEmbodiedRobot:具身智能机器人端到端推理def__init__(self):# 1. 加载模型self.vlmatb.create_model(blip2_opt-2.7b.om,devicenpu:0)self.action_modelatb.create_model(roberta-action.om,devicenpu:0)# 2. 初始化 DVPPdvpp.Init()# 3. 创建线程池Pipeline 并行self.executorThreadPoolExecutor(max_workers4)# 4. 性能统计self.latencies[]defrun_instruction(self,instruction,image_stream): 执行用户的指令 参数 instruction: 文本指令把红色的杯子拿过来 image_stream: 摄像头视频流 start_timetime.time()# Stage 1: 视觉感知异步future_visionself.executor.submit(self._vision_perception,image_stream)# Stage 2: 语言理解同步vision_resultfuture_vision.result()objectsself._detect_objects(vision_result)# Stage 3: 意图理解同步intentself._understand_intent(instruction,objects)# Stage 4: 动作规划同步action_planself._plan_action(intent,objects)# Stage 5: 执行动作self._execute_action(action_plan)# 统计延迟latency(time.time()-start_time)*1000self.latencies.append(latency)print(f总延迟:{latency:.1f}ms (视觉:{vision_latency:.1f}ms, 理解:{intent_latency:.1f}ms, 规划:{plan_latency:.1f}ms))returnaction_plandef_vision_perception(self,image_stream):视觉感知t0time.time()# 1. DVPP 解码framedvpp.Decode(image_stream)# 2. VLM 推理vision_featuresself.vlm(frame)globalvision_latency vision_latency(time.time()-t0)*1000returnvision_featuresdef_detect_objects(self,vision_result):检测物体# 从 VLM 输出中解析物体objectsparse_vlm_output(vision_result)returnobjectsdef_understand_intent(self,instruction,objects):意图理解t0time.time()# 用语言模型理解用户意图intentself.action_model.understand(instruction,objects)globalintent_latency intent_latency(time.time()-t0)*1000returnintentdef_plan_action(self,intent,objects):动作规划t0time.time()# 规划动作序列action_planself.action_model.plan(intent,objects)globalplan_latency plan_latency(time.time()-t0)*1000returnaction_plandef_execute_action(self,action_plan):执行动作foractioninaction_plan:# 发送到机械臂send_to_robot(action)defget_avg_latency(self):获取平均延迟ifnotself.latencies:return0returnsum(self.latencies)/len(self.latencies)# 使用robotEmbodiedRobot()# 注册摄像头# camera CameraStream(0)# 执行指令instruction把红色的杯子拿过来# action_plan robot.run_instruction(instruction, camera)print(f平均延迟:{robot.get_avg_latency():.1f}ms)实时性优化Pipeline 并行 vs Batch 推理具身智能的延迟要求特殊不要吞吐要延迟。Pipeline 并行比 Batch 推理更适合。Batch 推理的延迟问题# Batch 推理延迟高defbatch_infer(images,batch_size8):Batch 推理# 准备好 batchbatch[]foriinrange(batch_size):batch.append(images[i])# 一次推理resultsmodel(torch.cat(batch,dim0))# 问题要等 batch 满才能推理# 如果只来 1 帧也要等 batch 排满 → 延迟高Pipeline 并行的延迟优化# Pipeline 并行延迟低# 核心不等服务有数据就推理classStreamProcessor:流式处理器零等待def__init__(self,model):self.modelmodel# 1. 预热for_inrange(3):dummytorch.randn(1,3,224,224).npu()_model(dummy)definfer(self,frame):流式推理有数据就处理不等# 直接推理不等 batchtensortorch.from_numpy(frame).unsqueeze(0).npu()# 推理resultself.model(tensor)returnresult.cpu().numpy()# 测试对比# Batch 模式延迟80ms等 batch 满# Pipeline 模式延迟12ms来一帧处理一帧性能对比模式平均延迟最大延迟吞吐量适用场景Batch112ms15ms83 FPS低延迟具身智能Batch428ms35ms143 FPS平衡Batch852ms70ms154 FPS高吞吐离线Pipeline8ms12ms125 FPS实时具身智能关键结论Pipeline 并行延迟最低8ms最适合具身智能。总结cann-recipes-embodied-intelligence 的使用路径先跑通 VLM 推理BLIP-2 / LLaVA接入 DVPP 视频流摄像头30FPS用 Pipeline 并行降低延迟接动作规划Pick Place / Navigation关键要点延迟优先具身智能不要吞吐要延迟用 Pipeline 并行DVPP 加速视频流用 DVPP 硬件解码延迟从 33ms → 12ms流水线并行Stage 之间异步提高并发具身智能的推理要紧的不是吞吐是延迟。Pipeline 并行比 Batch 推理更合适。仓库地址https://atomgit.com/cann/cann-recipes-embodied-intelligence
