具身智能视觉特征库匹配：让机器人"认识"它看到的世界

黑漂技术佬的第10篇实战笔记 —— 当向量检索遇上机械臂，机器人终于知道该抓什么了

一、具身智能中的视觉检索需求

具身智能（Embodied AI）是当下AI领域最火的方向之一。简单说，就是给AI一个"身体"（机器人），让它不仅能"看"能"想"，还能"动"。传统AI是"脑子在罐子里"——只能处理数据，具身智能是"脑子装在身体里"——能感知环境、做决策、执行动作。

在这个体系里，视觉特征库匹配扮演什么角色？是机器人的"记忆"。

举个具体场景：一个补货机器人站在货架前，它需要：

1. 看到环境中有哪些物体（YOLO检测）
2. 认识这些物体是什么（向量检索匹配特征库）
3. 决定怎么操作（根据匹配结果选择抓取策略）
4. 执行动作（机械臂运动控制）

第2步就是本文的重点——视觉特征库匹配。没有这一步，机器人看到了物体但不知道是什么，后面的一切都是空中楼阁。

1.1 机械臂抓取：识别目标 → 匹配特征库 → 确定抓取策略

不同物体需要不同的抓取方式。可乐瓶要捏瓶身，鸡蛋要轻托底部，纸巾盒要夹两侧。机器人怎么知道面前这个东西该用什么方式抓？靠特征库匹配——识别出物体类别后，从预设的策略库里取出对应的抓取方案。

1.2 场景理解：检测环境物体 → 特征匹配 → 理解场景

机器人不仅要认识单个物体，还要理解整个场景。比如看到"货架+缺货格子+散落商品"，推断出"需要补货"。这种场景理解也建立在物体识别的基础上。

二、视觉特征库匹配完整流程

摄像头输入 → YOLO检测 → 框出物体位置
                           ↓
                    裁剪目标区域
                           ↓
                 ResNet/MobileNet提取特征
                           ↓
              FAISS/Milvus向量检索匹配特征库
                           ↓
                    返回匹配结果+置信度
                           ↓
              根据匹配结果选择动作策略

2.1 YOLO检测：框出物体位置

from ultralytics import YOLO
import cv2
import numpy as np

class SceneDetector:
    """场景物体检测器"""
    
    def __init__(self, model_path='yolov8n.pt', conf_threshold=0.5):
        self.model = YOLO(model_path)
        self.conf_threshold = conf_threshold
    
    def detect(self, image):
        """
        检测场景中的所有物体
        :param image: 图片路径或numpy数组(BGR)
        :return: 检测结果列表
        """
        results = self.model(image, conf=self.conf_threshold)
        
        detections = []
        for result in results:
            boxes = result.boxes
            for box in boxes:
                x1, y1, x2, y2 = box.xyxy[0].cpu().numpy().astype(int)
                confidence = float(box.conf[0].cpu().numpy())
                class_id = int(box.cls[0].cpu().numpy())
                class_name = result.names[class_id]
                
                detections.append({
                    'bbox': (x1, y1, x2, y2),
                    'confidence': confidence,
                    'class_name': class_name,
                    'class_id': class_id
                })
        
        return detections
    
    def visualize(self, image, detections):
        """可视化检测结果"""
        img = image.copy()
        for det in detections:
            x1, y1, x2, y2 = det['bbox']
            label = f"{det['class_name']} {det['confidence']:.2f}"
            
            cv2.rectangle(img, (x1, y1), (x2, y2), (0, 255, 0), 2)
            cv2.putText(img, label, (x1, y1 - 10),
                       cv2.FONT_HERSHEY_SIMPLEX, 0.6, (0, 255, 0), 2)
        
        return img

2.2 特征提取：裁剪目标 → 提取特征

import torch
import torchvision.models as models
import torchvision.transforms as transforms
from PIL import Image

class FeatureExtractor:
    """物体特征提取器"""
    
    def __init__(self, model_name='resnet18'):
        if model_name == 'resnet18':
            base = models.resnet18(pretrained=True)
            self.dim = 512
        elif model_name == 'mobilenet_v3_small':
            base = models.mobilenet_v3_small(pretrained=True)
            self.dim = 576
        else:
            raise ValueError(f"不支持: {model_name}")
        
        self.model = torch.nn.Sequential(*list(base.children())[:-1])
        self.model.eval()
        
        self.transform = transforms.Compose([
            transforms.Resize((224, 224)),
            transforms.ToTensor(),
            transforms.Normalize(mean=[0.485, 0.456, 0.406],
                                 std=[0.229, 0.224, 0.225])
        ])
    
    def extract_from_crop(self, crop_image):
        """从裁剪的图像区域提取特征"""
        if isinstance(crop_image, np.ndarray):
            # BGR → RGB → PIL
            crop_rgb = cv2.cvtColor(crop_image, cv2.COLOR_BGR2RGB)
            pil_img = Image.fromarray(crop_rgb)
        else:
            pil_img = crop_image
        
        tensor = self.transform(pil_img).unsqueeze(0)
        
        with torch.no_grad():
            feature = self.model(tensor)
        
        feature = feature.squeeze().numpy().astype('float32')
        norm = np.linalg.norm(feature)
        if norm > 0:
            feature = feature / norm
        
        return feature
    
    def extract_batch(self, crop_images):
        """批量提取特征（多个目标同时处理）"""
        tensors = []
        for crop in crop_images:
            if isinstance(crop, np.ndarray):
                crop_rgb = cv2.cvtColor(crop, cv2.COLOR_BGR2RGB)
                pil_img = Image.fromarray(crop_rgb)
            else:
                pil_img = crop
            tensors.append(self.transform(pil_img))
        
        batch = torch.stack(tensors)
        
        with torch.no_grad():
            features = self.model(batch)
        
        features = features.squeeze(-1).squeeze(-1).numpy().astype('float32')
        # L2归一化
        norms = np.linalg.norm(features, axis=1, keepdims=True)
        features = features / np.maximum(norms, 1e-8)
        
        return features

2.3 向量检索：匹配特征库

import faiss
import json

class FeatureLibrary:
    """视觉特征库管理"""
    
    def __init__(self, index_path, metadata_path):
        self.index = faiss.read_index(index_path)
        with open(metadata_path, 'r', encoding='utf-8') as f:
            self.metadata = json.load(f)
        print(f"特征库加载完成: {self.index.ntotal} 条记录")
    
    def search(self, query_feature, top_k=5):
        """检索单个特征"""
        query = np.array([query_feature]).astype('float32')
        scores, indices = self.index.search(query, top_k)
        
        results = []
        for i in range(top_k):
            idx = indices[0][i]
            if idx >= 0:
                match = self.metadata[idx].copy()
                match['score'] = float(scores[0][i])
                results.append(match)
        
        return results
    
    def search_batch(self, query_features, top_k=5):
        """批量检索多个特征"""
        queries = np.array(query_features).astype('float32')
        scores, indices = self.index.search(queries, top_k)
        
        all_results = []
        for q_idx in range(len(query_features)):
            results = []
            for i in range(top_k):
                idx = indices[q_idx][i]
                if idx >= 0:
                    match = self.metadata[idx].copy()
                    match['score'] = float(scores[q_idx][i])
                    results.append(match)
            all_results.append(results)
        
        return all_results

2.4 结果决策：根据匹配结果选择动作

class ActionDecisionEngine:
    """动作决策引擎：根据识别结果选择操作策略"""
    
    # 物体类别 → 抓取策略映射
    GRASP_STRATEGIES = {
        'bottle':       {'method': 'lateral_grip', 'force': 0.5, 'approach': 'side'},
        'can':          {'method': 'lateral_grip', 'force': 0.7, 'approach': 'side'},
        'box':          {'method': 'parallel_grip', 'force': 0.8, 'approach': 'top'},
        'fruit':        {'method': 'soft_cup',     'force': 0.3, 'approach': 'top'},
        'egg':          {'method': 'soft_cup',     'force': 0.2, 'approach': 'top'},
        'snack_bag':    {'method': 'pinch_grip',   'force': 0.4, 'approach': 'top'},
    }
    
    def decide(self, match_result):
        """
        根据匹配结果生成动作策略
        :param match_result: 特征库检索返回的最佳匹配
        :return: 动作策略字典
        """
        if not match_result:
            return {'action': 'skip', 'reason': '未匹配到已知物体'}
        
        best_match = match_result[0]
        object_category = best_match.get('category', 'unknown')
        confidence = best_match.get('score', 0)
        
        # 置信度太低，不执行操作
        if confidence < 0.6:
            return {
                'action': 'skip',
                'reason': f'置信度过低({confidence:.2f})',
                'best_guess': best_match.get('name', 'unknown')
            }
        
        # 获取抓取策略
        strategy = self.GRASP_STRATEGIES.get(
            object_category, 
            {'method': 'default_grip', 'force': 0.5, 'approach': 'top'}
        )
        
        return {
            'action': 'grasp',
            'object_name': best_match.get('name'),
            'object_category': object_category,
            'confidence': confidence,
            'grasp_strategy': strategy,
            'bbox': best_match.get('bbox')
        }

三、完整Pipeline整合

class EmbodiedVisionPipeline:
    """具身智能视觉Pipeline：检测→提特征→检索→决策"""
    
    def __init__(self, index_path, metadata_path, 
                 yolo_model='yolov8n.pt', feature_model='resnet18'):
        self.detector = SceneDetector(yolo_model)
        self.extractor = FeatureExtractor(feature_model)
        self.library = FeatureLibrary(index_path, metadata_path)
        self.decision_engine = ActionDecisionEngine()
    
    def process(self, image):
        """
        处理一帧图像，完成从检测到决策的全流程
        :param image: 输入图片(BGR numpy数组或路径)
        :return: 处理结果列表
        """
        # 第1步：YOLO检测
        detections = self.detector.detect(image)
        print(f"[1] 检测到 {len(detections)} 个目标")
        
        if not detections:
            return []
        
        # 读取图片用于裁剪
        if isinstance(image, str):
            frame = cv2.imread(image)
        else:
            frame = image
        
        # 第2步：裁剪所有目标区域
        crops = []
        for det in detections:
            x1, y1, x2, y2 = det['bbox']
            crop = frame[y1:y2, x1:x2]
            crops.append(crop)
        
        # 第3步：批量提取特征
        features = self.extractor.extract_batch(crops)
        print(f"[2] 特征提取完成: {features.shape}")
        
        # 第4步：批量向量检索
        search_results = self.library.search_batch(features, top_k=3)
        print(f"[3] 向量检索完成")
        
        # 第5步：动作决策
        decisions = []
        for i, (det, results) in enumerate(zip(detections, search_results)):
            # 把检测位置信息加入匹配结果
            for r in results:
                r['bbox'] = det['bbox']
            
            decision = self.decision_engine.decide(results)
            decisions.append({
                'detection': det,
                'search_results': results,
                'decision': decision
            })
            
            obj_name = decision.get('object_name', 'unknown')
            conf = decision.get('confidence', 0)
            action = decision.get('action', 'skip')
            print(f"  目标{i+1}: {det['class_name']} → {obj_name} "
                  f"(置信度:{conf:.2f}) → 动作:{action}")
        
        return decisions

# 使用示例
pipeline = EmbodiedVisionPipeline(
    index_path="object_library.faiss",
    metadata_path="object_metadata.json"
)

# 处理一帧货架图片
results = pipeline.process("shelf_photo.jpg")

# 输出示例:
# [1] 检测到 5 个目标
# [2] 特征提取完成: (5, 512)
# [3] 向量检索完成
#   目标1: bottle → 可口可乐330ml (置信度:0.95) → 动作:grasp
#   目标2: bottle → 雪碧330ml (置信度:0.91) → 动作:grasp
#   目标3: box → 纸巾盒 (置信度:0.88) → 动作:grasp

四、特征库管理

特征库不是一成不变的，需要支持动态增删改：

class FeatureLibraryManager:
    """特征库管理器：支持增删改查"""
    
    def __init__(self, index_path, metadata_path, dimension=512):
        self.index_path = index_path
        self.metadata_path = metadata_path
        self.dimension = dimension
        
        # 加载已有库
        self.index = faiss.read_index(index_path)
        with open(metadata_path, 'r', encoding='utf-8') as f:
            self.metadata = json.load(f)
        
        self.extractor = FeatureExtractor('resnet18')
    
    def add_object(self, object_id, name, category, images, **extra_info):
        """
        新增物体到特征库
        :param object_id: 物体唯一ID
        :param name: 物体名称
        :param category: 物体类别
        :param images: 物体图片路径列表（多角度）
        :param extra_info: 额外元数据（如抓取策略）
        """
        # 提取多角度特征
        features = []
        for img_path in images:
            feat = self.extractor.extract_from_crop(img_path)
            features.append(feat)
        
        # 多角度特征取平均，生成原型向量
        centroid = np.mean(features, axis=0).astype('float32')
        centroid = centroid / np.linalg.norm(centroid)
        
        # 添加到索引
        self.index.add(np.array([centroid]))
        
        # 添加元数据
        meta = {
            'object_id': object_id,
            'name': name,
            'category': category,
            'num_images': len(images),
            **extra_info
        }
        self.metadata.append(meta)
        
        # 保存
        self._save()
        print(f"物体 '{name}' 已添加到特征库")
    
    def update_object(self, object_id, new_images=None, **updated_info):
        """
        更新物体特征（替换特征向量）
        FAISS不支持直接删除单条记录，需要重建索引
        """
        # 找到要更新的记录
        target_idx = None
        for i, meta in enumerate(self.metadata):
            if meta['object_id'] == object_id:
                target_idx = i
                break
        
        if target_idx is None:
            print(f"未找到物体: {object_id}")
            return
        
        # 更新元数据
        self.metadata[target_idx].update(updated_info)
        
        # 如果提供了新图片，重新提取特征
        if new_images:
            features = []
            for img_path in new_images:
                feat = self.extractor.extract_from_crop(img_path)
                features.append(feat)
            
            new_centroid = np.mean(features, axis=0).astype('float32')
            new_centroid = new_centroid / np.linalg.norm(new_centroid)
            
            # 重建索引（小规模库可以接受）
            self._rebuild_index_with_update(target_idx, new_centroid)
        
        self._save()
        print(f"物体 '{object_id}' 已更新")
    
    def remove_object(self, object_id):
        """删除物体（标记删除，定期重建索引）"""
        target_idx = None
        for i, meta in enumerate(self.metadata):
            if meta['object_id'] == object_id:
                target_idx = i
                break
        
        if target_idx is None:
            print(f"未找到物体: {object_id}")
            return
        
        # 从元数据中移除
        removed = self.metadata.pop(target_idx)
        
        # 重建索引（去掉对应向量）
        all_vectors = []
        for i in range(self.index.ntotal):
            if i == target_idx:
                continue
            vec = self.index.reconstruct(i)
            all_vectors.append(vec)
        
        # 重建
        quantizer = faiss.IndexFlatIP(self.dimension)
        new_index = faiss.IndexIVFFlat(quantizer, self.dimension, 
                                        min(100, len(all_vectors)//10),
                                        faiss.METRIC_INNER_PRODUCT)
        if len(all_vectors) > 0:
            vectors_array = np.array(all_vectors).astype('float32')
            new_index.train(vectors_array)
            new_index.add(vectors_array)
        
        self.index = new_index
        self._save()
        print(f"物体 '{removed['name']}' 已从特征库删除")
    
    def _save(self):
        faiss.write_index(self.index, self.index_path)
        with open(self.metadata_path, 'w', encoding='utf-8') as f:
            json.dump(self.metadata, f, ensure_ascii=False, indent=2)
    
    def _rebuild_index_with_update(self, target_idx, new_vector):
        """重建索引（更新单条向量）"""
        all_vectors = []
        for i in range(self.index.ntotal):
            if i == target_idx:
                all_vectors.append(new_vector)
            else:
                all_vectors.append(self.index.reconstruct(i))
        
        vectors_array = np.array(all_vectors).astype('float32')
        quantizer = faiss.IndexFlatIP(self.dimension)
        new_index = faiss.IndexIVFFlat(quantizer, self.dimension,
                                        min(100, len(all_vectors)//10),
                                        faiss.METRIC_INNER_PRODUCT)
        new_index.train(vectors_array)
        new_index.add(vectors_array)
        self.index = new_index

五、多模态融合：视觉特征 + 文本描述（CLIP）

纯视觉特征有个问题：长得像的东西分不清。比如红富士苹果和花牛苹果，视觉特征几乎一样。怎么办？引入文本描述，做多模态融合。

CLIP模型能同时理解图像和文本，把两者映射到同一个向量空间。这样就可以用文字描述来辅助识别。

from transformers import CLIPModel, CLIPProcessor

class MultiModalMatcher:
    """多模态匹配器：视觉+文本融合"""
    
    def __init__(self, model_name='openai/clip-vit-base-patch32'):
        self.model = CLIPModel.from_pretrained(model_name)
        self.processor = CLIPProcessor.from_pretrained(model_name)
        self.model.eval()
        self.dimension = 512  # CLIP输出维度
    
    def extract_image_feature(self, image):
        """提取图像特征"""
        if isinstance(image, np.ndarray):
            image = Image.fromarray(cv2.cvtColor(image, cv2.COLOR_BGR2RGB))
        
        inputs = self.processor(images=image, return_tensors="pt")
        
        with torch.no_grad():
            features = self.model.get_image_features(**inputs)
        
        features = features.squeeze().numpy().astype('float32')
        features = features / np.linalg.norm(features)
        return features
    
    def extract_text_feature(self, text):
        """提取文本特征"""
        inputs = self.processor(text=[text], return_tensors="pt", 
                                padding=True, truncation=True)
        
        with torch.no_grad():
            features = self.model.get_text_features(**inputs)
        
        features = features.squeeze().numpy().astype('float32')
        features = features / np.linalg.norm(features)
        return features
    
    def build_multimodal_index(self, items):
        """
        构建多模态特征库
        :param items: [{"image": path, "description": "红色成熟番茄", ...}, ...]
        """
        image_features = []
        text_features = []
        metadata = []
        
        for item in items:
            # 图像特征
            img_feat = self.extract_image_feature(item['image'])
            image_features.append(img_feat)
            
            # 文本特征
            text_feat = self.extract_text_feature(item['description'])
            text_features.append(text_feat)
            
            metadata.append(item)
        
        # 融合策略：加权平均（图像权重0.7，文本权重0.3）
        image_features = np.array(image_features)
        text_features = np.array(text_features)
        fused_features = 0.7 * image_features + 0.3 * text_features
        
        # 重新归一化
        norms = np.linalg.norm(fused_features, axis=1, keepdims=True)
        fused_features = fused_features / np.maximum(norms, 1e-8)
        fused_features = fused_features.astype('float32')
        
        # 构建FAISS索引
        index = faiss.IndexFlatIP(self.dimension)
        index.add(fused_features)
        
        return index, metadata
    
    def search_with_description(self, query_image, query_text, index, 
                                 metadata, top_k=5, image_weight=0.7):
        """
        用图像+文本描述联合检索
        """
        img_feat = self.extract_image_feature(query_image)
        text_feat = self.extract_text_feature(query_text)
        
        # 融合查询特征
        query_fused = image_weight * img_feat + (1 - image_weight) * text_feat
        query_fused = query_fused / np.linalg.norm(query_fused)
        
        # 检索
        query = np.array([query_fused]).astype('float32')
        scores, indices = index.search(query, top_k)
        
        results = []
        for i in range(top_k):
            idx = indices[0][i]
            if idx >= 0:
                match = metadata[idx].copy()
                match['score'] = float(scores[0][i])
                results.append(match)
        
        return results

六、实际落地案例

6.1 无人售货柜补货机器人

场景：仓库里有一堆商品，售货柜里缺了几个格子，补货机器人需要从仓库中找到对应商品并放到货架上。

class RestockingRobot:
    """无人售货柜补货机器人"""
    
    def __init__(self, feature_library):
        self.pipeline = EmbodiedVisionPipeline(
            index_path="warehouse_objects.faiss",
            metadata_path="warehouse_objects.json"
        )
        self.library = feature_library
    
    def scan_warehouse(self, camera_image):
        """
        扫描仓库，建立可用商品清单
        """
        results = self.pipeline.process(camera_image)
        
        available_items = []
        for r in results:
            if r['decision']['action'] == 'grasp':
                available_items.append({
                    'name': r['decision']['object_name'],
                    'category': r['decision']['object_category'],
                    'bbox': r['decision']['bbox'],
                    'confidence': r['decision']['confidence']
                })
        
        return available_items
    
    def find_target_item(self, camera_image, target_product_id):
        """
        在画面中找到目标商品的位置
        """
        results = self.pipeline.process(camera_image)
        
        for r in results:
            best = r['search_results'][0] if r['search_results'] else None
            if best and best.get('object_id') == target_product_id:
                return {
                    'found': True,
                    'bbox': r['detection']['bbox'],
                    'confidence': r['decision']['confidence'],
                    'grasp_strategy': r['decision'].get('grasp_strategy')
                }
        
        return {'found': False}
    
    def execute_restock(self, target_product_id, max_attempts=3):
        """执行补货流程"""
        for attempt in range(max_attempts):
            print(f"\n=== 补货尝试 {attempt + 1} ===")
            
            # 1. 拍照
            frame = self.capture_frame()
            
            # 2. 找目标商品
            target = self.find_target_item(frame, target_product_id)
            
            if not target['found']:
                print("未找到目标商品，调整位置...")
                self.robot_move("adjust_position")
                continue
            
            # 3. 执行抓取
            print(f"找到目标商品，置信度: {target['confidence']:.2f}")
            strategy = target['grasp_strategy']
            
            grasp_success = self.execute_grasp(
                target['bbox'], 
                strategy['method'],
                strategy['force'],
                strategy['approach']
            )
            
            if grasp_success:
                # 4. 放到目标位置
                self.place_to_shelf(target_product_id)
                print("补货成功!")
                return True
            else:
                print("抓取失败，重试...")
        
        return False
    
    def capture_frame(self):
        """从相机捕获一帧"""
        # 实际实现调用相机SDK
        pass
    
    def execute_grasp(self, bbox, method, force, approach):
        """执行机械臂抓取"""
        # 实际实现调用机械臂SDK
        print(f"  抓取方式: {method}, 力度: {force}, 接近方向: {approach}")
        return True  # 模拟成功
    
    def place_to_shelf(self, product_id):
        """放到货架"""
        print(f"  将商品 {product_id} 放到货架")
    
    def robot_move(self, action):
        """机器人移动"""
        pass

6.2 智慧农业采摘机器人

场景：果园里，机器人需要识别成熟的果实，判断成熟度，然后采摘。特征库里存的是不同成熟度果实的特征模板。

class HarvestingRobot:
    """智慧农业采摘机器人"""
    
    def __init__(self):
        # 特征库包含不同成熟度的果实特征
        # metadata中包含ripeness字段: "unripe" / "semi_ripe" / "ripe" / "overripe"
        self.pipeline = EmbodiedVisionPipeline(
            index_path="fruit_library.faiss",
            metadata_path="fruit_metadata.json"
        )
    
    def scan_orchard(self, camera_image):
        """扫描果树，识别成熟果实"""
        results = self.pipeline.process(camera_image)
        
        harvestable = []
        for r in results:
            if not r['search_results']:
                continue
            
            best = r['search_results'][0]
            ripeness = best.get('ripeness', 'unknown')
            confidence = best.get('score', 0)
            
            # 只采摘成熟果实
            if ripeness == 'ripe' and confidence > 0.7:
                harvestable.append({
                    'fruit_type': best.get('name'),
                    'ripeness': ripeness,
                    'bbox': r['detection']['bbox'],
                    'confidence': confidence
                })
                print(f"  发现成熟果实: {best['name']} "
                      f"(成熟度:{ripeness}, 置信度:{confidence:.2f})")
            elif ripeness in ['semi_ripe', 'unripe']:
                print(f"  果实未成熟: {best['name']} ({ripeness}), 跳过")
        
        return harvestable
    
    def harvest(self, camera_image):
        """执行采摘流程"""
        # 1. 扫描识别
        targets = self.scan_orchard(camera_image)
        
        if not targets:
            print("未发现可采摘的成熟果实")
            return 0
        
        print(f"\n发现 {len(targets)} 个可采摘果实，开始采摘...")
        
        harvested = 0
        for target in targets:
            # 2. 定位果实三维位置（结合深度相机）
            fruit_3d_pos = self.locate_3d(target['bbox'])
            
            # 3. 机械臂移动到果实位置
            self.move_arm_to(fruit_3d_pos)
            
            # 4. 执行采摘动作（剪切/夹取/吸取，取决于果实类型）
            if self.pick_fruit(target['fruit_type']):
                harvested += 1
                print(f"  采摘成功: {target['fruit_type']}")
            else:
                print(f"  采摘失败: {target['fruit_type']}")
        
        print(f"\n采摘完成: {harvested}/{len(targets)}")
        return harvested
    
    def locate_3d(self, bbox):
        """结合深度相机获取三维坐标"""
        # 实际实现使用RGB-D相机
        x1, y1, x2, y2 = bbox
        cx = (x1 + x2) / 2
        cy = (y1 + y2) / 2
        return (cx, cy, 0.5)  # 模拟深度0.5米
    
    def move_arm_to(self, position):
        """机械臂移动"""
        print(f"  机械臂移动到: {position}")
    
    def pick_fruit(self, fruit_type):
        """执行采摘"""
        print(f"  采摘 {fruit_type}")
        return True

七、项目架构总结

一个完整的具身智能视觉检索系统，架构如下：

┌─────────────────────────────────────────────────────────┐
│                    感知层 (Perception)                    │
│  ┌──────────┐  ┌──────────────┐  ┌───────────────────┐  │
│  │ RGB相机   │  │ RGB-D深度相机  │  │ 其他传感器(触觉等)  │  │
│  └────┬─────┘  └──────┬───────┘  └────────┬──────────┘  │
│       └───────────────┼────────────────────┘            │
│                       ↓                                  │
│  ┌────────────────────────────────────────────────────┐ │
│  │           YOLO物体检测 → 目标区域裁剪               │ │
│  └────────────────────────┬───────────────────────────┘ │
└───────────────────────────┼─────────────────────────────┘
                            ↓
┌─────────────────────────────────────────────────────────┐
│                   认知层 (Cognition)                      │
│  ┌──────────────┐  ┌──────────────────┐                  │
│  │ 特征提取模型   │  │ 多模态融合(CLIP)  │                  │
│  │ ResNet/Mobile │  │ 视觉+文本融合     │                  │
│  └──────┬───────┘  └────────┬─────────┘                  │
│         └──────────┬────────┘                            │
│                    ↓                                     │
│  ┌────────────────────────────────────────────────────┐ │
│  │     向量检索 (FAISS/Milvus) → 匹配特征库           │ │
│  │     返回: 物体名称 + 类别 + 置信度                  │ │
│  └────────────────────────┬───────────────────────────┘ │
│                           ↓                              │
│  ┌────────────────────────────────────────────────────┐ │
│  │     动作决策引擎 → 选择抓取策略/操作方案            │ │
│  └────────────────────────┬───────────────────────────┘ │
└───────────────────────────┼─────────────────────────────┘
                            ↓
┌─────────────────────────────────────────────────────────┐
│                    执行层 (Action)                        │
│  ┌──────────┐  ┌──────────┐  ┌──────────────────────┐   │
│  │ 机械臂控制 │  │ 移动底盘  │  │ 末端执行器(夹爪/吸盘)  │   │
│  └──────────┘  └──────────┘  └──────────────────────┘   │
└─────────────────────────────────────────────────────────┘

整个闭环就是六个字：感知→认知→执行。感知层负责"看见"，认知层负责"认识"，执行层负责"动手"。向量检索在认知层扮演核心角色——它是连接"看见"和"认识"的桥梁，把像素转化成语义。

具身智能的视觉检索和普通视觉检索最大的区别在于：检索结果直接驱动物理动作。普通检索错了最多返回错误结果，具身智能检索错了可能导致机械臂抓坏东西。所以这里对置信度阈值的要求更严格，对特征库的维护更精细，对实时性的要求也更高——一个闭环控制系统，检索延迟太高会导致机械臂动作卡顿甚至失控。