HumansXドキュメントセンター 公式サイトへ ↗

LLM による意味理解:ハードウェア統合ガイド

概要

LLM モジュールは完全なハードウェア統合フレームワークを提供し、Function Calling インターフェースを通じてさまざまな物理デバイスの制御をサポートします。システムはモジュール設計を採用し、シリアルポートや GPIO などの通信方式をサポートするとともに、安全なハードウェア制御の抽象化レイヤーを提供します。

対応ハードウェアデバイス

ハードウェア機器一覧

1. 6 自由度ロボットアーム

  • 型式:汎用 6DOF ロボットアーム
  • 通信:USB シリアル通信
  • 機能:把持、配置、精密な位置決め
  • 制御精度:± 1mm

2. 2 軸パン・チルト

  • 型式:汎用サーボ式パン・チルト
  • 通信:シリアルポートによる PWM 制御
  • 機能:水平/垂直回転
  • 角度範囲:0°~270°

3. オーディオ再生機器

  • インターフェース:システムのオーディオ出力
  • 形式:MP3、WAV、OGG など
  • 機能:再生制御、音量調整

4. カメラ機器

  • インターフェース:USB/CSI カメラ
  • 機能:画像取得、視覚認識
  • 解像度:複数の解像度に対応

ロボットアーム制御システム

システムアーキテクチャ

+-------------------+
|   LLM指令处理    |
+-------------------+
         |
+-------------------+
|   机械臂函数接口 |
+-------------------+
         |
+-------------------+
|   RobotArm类      |
+-------------------+
         |
    +----+----+
    |         |
    |串口通信 | 运动规划算法
    |协议     | 
    +----+----+
         |
+-------------------+
|   物理机械臂     |
+-------------------+

コアクラスの定義

class RobotArm:
    """6自由度机械臂控制类"""

    def __init__(self, port: str = "/dev/ttyS1", baudrate: int = 115200):
        """
        初始化机械臂控制器

        Args:
            port: 串口设备路径
            baudrate: 波特率
        """
        self.port = port
        self.baudrate = baudrate
        self.serial_conn = None
        self.current_position = [0, 0, 0]  # X, Y, Z
        self.current_angles = [0, 0, 0, 0, 0, 0]  # 6个关节角度
        self.is_grabbing = False

    def connect(self) -> bool:
        """连接机械臂"""
        try:
            self.serial_conn = serial.Serial(self.port, self.baudrate, timeout=2.0)
            time.sleep(2)  # 等待初始化
            return self.serial_conn.is_open
        except Exception as e:
            logger.error(f"机械臂连接失败: {e}")
            return False

    def move_to_position(self, x: float, y: float, z: float, speed: float = 50.0) -> bool:
        """移动到指定位置"""
        # 安全检查
        if not self.is_position_safe(x, y, z):
            logger.warning(f"位置超出安全范围: ({x}, {y}, {z})")
            return False

        # 逆运动学计算
        joint_angles = self.inverse_kinematics(x, y, z)
        if joint_angles is None:
            logger.error("逆运动学求解失败")
            return False

        # 发送运动指令
        success = self.move_joints(joint_angles, speed)
        if success:
            self.current_position = [x, y, z]
            self.current_angles = joint_angles

        return success

    def grab_object(self, x: float, y: float, z: float) -> bool:
        """在指定位置抓取物体"""
        try:
            # 移动到目标上方
            if not self.move_to_position(x, y, z + 50):
                return False

            # 下降到目标位置
            if not self.move_to_position(x, y, z):
                return False

            # 闭合夹爪
            if not self.close_gripper():
                return False

            # 提升物体
            if not self.move_to_position(x, y, z + 50):
                return False

            self.is_grabbing = True
            logger.info(f"成功抓取物体于位置: ({x}, {y}, {z})")
            return True

        except Exception as e:
            logger.error(f"抓取物体失败: {e}")
            return False

運動学アルゴリズム

順運動学

def forward_kinematics(self, joint_angles: List[float]) -> Tuple[float, float, float]:
    """
    正运动学计算:根据关节角度计算末端位置

    Args:
        joint_angles: 6个关节角度 (度)

    Returns:
        末端执行器的 (x, y, z) 坐标
    """
    # 机械臂DH参数 (Denavit-Hartenberg parameters)
    dh_params = [
        [0,   90,  85,  0],    # 关节1
        [105,  0,   0, -90],   # 关节2  
        [95,   0,   0,   0],   # 关节3
        [0,   90,  82,   0],   # 关节4
        [0,  -90,   0,   0],   # 关节5
        [0,    0,  45,   0]    # 关节6
    ]

    # 转换为弧度
    angles_rad = [math.radians(angle) for angle in joint_angles]

    # 计算变换矩阵
    T = np.eye(4)
    for i, (a, alpha, d, theta_offset) in enumerate(dh_params):
        theta = angles_rad[i] + math.radians(theta_offset)

        # DH变换矩阵
        Ti = np.array([
            [math.cos(theta), -math.sin(theta)*math.cos(math.radians(alpha)), 
             math.sin(theta)*math.sin(math.radians(alpha)), a*math.cos(theta)],
            [math.sin(theta), math.cos(theta)*math.cos(math.radians(alpha)), 
             -math.cos(theta)*math.sin(math.radians(alpha)), a*math.sin(theta)],
            [0, math.sin(math.radians(alpha)), 
             math.cos(math.radians(alpha)), d],
            [0, 0, 0, 1]
        ])

        T = T @ Ti

    return T[0, 3], T[1, 3], T[2, 3]

逆運動学

def inverse_kinematics(self, x: float, y: float, z: float) -> Optional[List[float]]:
    """
    逆运动学求解:根据目标位置计算关节角度
    """
    try:
        from scipy.optimize import fsolve

        def equations(angles):
            fx, fy, fz = self.forward_kinematics(angles)
            return [fx - x, fy - y, fz - z]

        initial_guess = self.current_angles.copy()
        solution = fsolve(equations, initial_guess)

        # 验证解的有效性
        error_x, error_y, error_z = equations(solution)
        error_magnitude = math.sqrt(error_x**2 + error_y**2 + error_z**2)

        if error_magnitude < 1.0 and self.are_angles_valid(solution):
            return solution.tolist()

        return None

    except Exception as e:
        logger.error(f"逆运动学计算错误: {e}")
        return None

安全制御システム

class SafetyController:
    """机械臂安全控制器"""

    def __init__(self, arm: RobotArm):
        self.arm = arm
        self.workspace_limits = {
            'x': [-200, 200],  # mm
            'y': [-200, 200],  # mm
            'z': [0, 300]      # mm
        }
        self.joint_limits = {
            'joint1': [-180, 180],  # 度
            'joint2': [-90, 90],
            'joint3': [-90, 90], 
            'joint4': [-180, 180],
            'joint5': [-90, 90],
            'joint6': [-180, 180]
        }
        self.max_speed = 100  # mm/s
        self.emergency_stop = False

    def is_position_safe(self, x: float, y: float, z: float) -> bool:
        """检查位置是否在安全范围内"""
        return (self.workspace_limits['x'][0] <= x <= self.workspace_limits['x'][1] and
                self.workspace_limits['y'][0] <= y <= self.workspace_limits['y'][1] and
                self.workspace_limits['z'][0] <= z <= self.workspace_limits['z'][1])

    def emergency_stop_all(self):
        """紧急停止所有运动"""
        self.emergency_stop = True
        try:
            if self.arm.serial_conn and self.arm.serial_conn.is_open:
                self.arm.serial_conn.write(b"EMERGENCY_STOP\n")
            logger.critical("机械臂紧急停止已激活")
        except Exception as e:
            logger.error(f"紧急停止指令发送失败: {e}")

パン・チルト制御システム

パン・チルトコントローラ

class GimbalController:
    """双轴云台控制器"""

    def __init__(self, port: str, baudrate: int, pan_pin: int, tilt_pin: int):
        """
        初始化云台控制器

        Args:
            port: 串口设备路径
            baudrate: 波特率
            pan_pin: 水平舵机控制引脚
            tilt_pin: 垂直舵机控制引脚
        """
        self.port = port
        self.baudrate = baudrate
        self.pan_pin = pan_pin
        self.tilt_pin = tilt_pin
        self.serial_conn = None
        self.current_pan = 135.0  # 中心位置
        self.current_tilt = 135.0  # 中心位置
        self.angle_range = [0, 270]  # 角度范围

    def connect(self) -> bool:
        """连接云台控制器"""
        try:
            self.serial_conn = serial.Serial(self.port, self.baudrate, timeout=1.0)
            time.sleep(1)
            return self.serial_conn.is_open
        except Exception as e:
            logger.error(f"云台连接失败: {e}")
            return False

    def set_position(self, pan: float, tilt: float) -> bool:
        """
        设置云台位置

        Args:
            pan: 水平角度 (0 到 270 度)
            tilt: 垂直角度 (0 到 270 度)

        Returns:
            设置是否成功
        """
        # 角度限制
        pan = max(self.angle_range[0], min(self.angle_range[1], pan))
        tilt = max(self.angle_range[0], min(self.angle_range[1], tilt))

        try:
            # 发送控制指令
            command = f"G1 X{pan} Y{tilt}\n"
            self.serial_conn.write(command.encode())

            # 等待响应
            response = self.serial_conn.readline().decode().strip()

            if response == "OK":
                self.current_pan = pan
                self.current_tilt = tilt
                logger.info(f"云台位置设置: Pan={pan}°, Tilt={tilt}°")
                return True
            else:
                logger.error(f"云台控制失败: {response}")
                return False

        except Exception as e:
            logger.error(f"云台控制错误: {e}")
            return False

    def move_relative(self, pan_delta: float, tilt_delta: float) -> bool:
        """相对移动"""
        new_pan = self.current_pan + pan_delta
        new_tilt = self.current_tilt + tilt_delta
        return self.set_position(new_pan, new_tilt)

    def reset_position(self) -> bool:
        """重置到中心位置"""
        return self.set_position(135.0, 135.0)

パン・チルトのキャリブレーションシステム

class GimbalCalibration:
    """云台标定系统"""

    def __init__(self, gimbal: GimbalController):
        self.gimbal = gimbal
        self.calibration_data = {}

    def run_calibration(self) -> bool:
        """运行标定程序"""
        logger.info("开始云台标定...")

        # 标定点列表
        calibration_points = [
            (0, 0), (0, 135), (0, 270),
            (135, 0), (135, 135), (135, 270),
            (270, 0), (270, 135), (270, 270)
        ]

        for pan, tilt in calibration_points:
            logger.info(f"标定点: ({pan}, {tilt})")

            if not self.gimbal.set_position(pan, tilt):
                logger.error(f"无法移动到标定点 ({pan}, {tilt})")
                return False

            # 等待稳定
            time.sleep(2)

            # 记录实际位置
            actual_pan, actual_tilt = self.measure_actual_position()
            self.calibration_data[(pan, tilt)] = (actual_pan, actual_tilt)

        self.calculate_calibration_parameters()
        logger.info("云台标定完成")
        return True

    def measure_actual_position(self) -> Tuple[float, float]:
        """测量实际位置"""
        return self.gimbal.current_pan, self.gimbal.current_tilt

    def calculate_calibration_parameters(self):
        """计算标定参数"""
        pan_errors = []
        tilt_errors = []

        for (cmd_pan, cmd_tilt), (actual_pan, actual_tilt) in self.calibration_data.items():
            pan_errors.append(actual_pan - cmd_pan)
            tilt_errors.append(actual_tilt - cmd_tilt)

        self.pan_offset = sum(pan_errors) / len(pan_errors)
        self.tilt_offset = sum(tilt_errors) / len(tilt_errors)

        logger.info(f"标定偏移: Pan={self.pan_offset:.2f}°, Tilt={self.tilt_offset:.2f}°")

オーディオシステムの統合

オーディオプレーヤー

class AudioPlayer:
    """音频播放器"""

    def __init__(self, music_directory: str = "./music"):
        """
        初始化音频播放器

        Args:
            music_directory: 音乐文件目录
        """
        self.music_directory = Path(music_directory)
        self.current_music = None
        self.is_playing = False
        self.volume = 0.7
        self.supported_formats = ['.mp3', '.wav', '.ogg', '.m4a']

        # 初始化pygame音频系统
        pygame.mixer.pre_init(frequency=22050, size=-16, channels=2, buffer=512)
        pygame.mixer.init()

    def list_music(self) -> List[str]:
        """获取音乐文件列表"""
        if not self.music_directory.exists():
            self.music_directory.mkdir(parents=True)
            return []

        music_files = []
        for ext in self.supported_formats:
            music_files.extend(self.music_directory.glob(f"*{ext}"))

        return [f.name for f in music_files]

    def play_music(self, music_name: str) -> bool:
        """播放指定音乐"""
        music_file = self.find_music_file(music_name)
        if not music_file:
            logger.warning(f"未找到音乐文件: {music_name}")
            return False

        try:
            self.stop_music()
            pygame.mixer.music.load(str(music_file))
            pygame.mixer.music.set_volume(self.volume)
            pygame.mixer.music.play()

            self.current_music = music_name
            self.is_playing = True

            logger.info(f"开始播放音乐: {music_name}")
            return True

        except Exception as e:
            logger.error(f"音乐播放失败: {e}")
            return False

    def stop_music(self):
        """停止音乐播放"""
        try:
            pygame.mixer.music.stop()
            self.is_playing = False
            self.current_music = None
            logger.info("音乐播放已停止")
        except Exception as e:
            logger.error(f"停止音乐失败: {e}")

    def set_volume(self, volume: float):
        """设置音量"""
        self.volume = max(0.0, min(1.0, volume))
        pygame.mixer.music.set_volume(self.volume)
        logger.info(f"音量设置为: {self.volume * 100:.0f}%")

    def get_status(self) -> dict:
        """获取播放状态"""
        return {
            "is_playing": self.is_playing,
            "current_music": self.current_music,
            "volume": self.volume,
            "music_count": len(self.list_music())
        }

ビジョンシステムの統合

カメラ制御

class CameraController:
    """摄像头控制器"""

    def __init__(self, device_id: int = 0, resolution: tuple = (640, 480)):
        """
        初始化摄像头

        Args:
            device_id: 摄像头设备ID
            resolution: 分辨率 (宽, 高)
        """
        self.device_id = device_id
        self.resolution = resolution
        self.cap = None
        self.is_opened = False

    def open_camera(self) -> bool:
        """打开摄像头"""
        try:
            self.cap = cv2.VideoCapture(self.device_id)
            self.cap.set(cv2.CAP_PROP_FRAME_WIDTH, self.resolution[0])
            self.cap.set(cv2.CAP_PROP_FRAME_HEIGHT, self.resolution[1])

            self.is_opened = self.cap.isOpened()
            if self.is_opened:
                logger.info(f"摄像头已打开: 设备{self.device_id}, 分辨率{self.resolution}")
            else:
                logger.error(f"无法打开摄像头设备: {self.device_id}")

            return self.is_opened

        except Exception as e:
            logger.error(f"摄像头打开失败: {e}")
            return False

    def capture_image(self, save_path: str = None) -> Optional[np.ndarray]:
        """捕获图像"""
        if not self.is_opened:
            logger.warning("摄像头未打开")
            return None

        try:
            ret, frame = self.cap.read()
            if ret:
                if save_path:
                    cv2.imwrite(save_path, frame)
                    logger.info(f"图像已保存到: {save_path}")
                return frame
            else:
                logger.error("图像捕获失败")
                return None

        except Exception as e:
            logger.error(f"图像捕获错误: {e}")
            return None

    def close_camera(self):
        """关闭摄像头"""
        if self.cap:
            self.cap.release()
            self.is_opened = False
            logger.info("摄像头已关闭")

視覚認識の統合

class VisionSystem:
    """视觉识别系统"""

    def __init__(self, camera: CameraController, vlm_client):
        self.camera = camera
        self.vlm_client = vlm_client
        self.temp_image_dir = Path("./temp_images")
        self.temp_image_dir.mkdir(exist_ok=True)

    def analyze_current_view(self, query: str = "请描述看到的内容") -> str:
        """分析当前摄像头画面"""
        timestamp = int(time.time())
        image_path = self.temp_image_dir / f"capture_{timestamp}.jpg"

        image = self.camera.capture_image(str(image_path))
        if image is None:
            return "无法获取摄像头图像"

        try:
            result = self.vlm_client.analyze_image(str(image_path), query)
            image_path.unlink(missing_ok=True)  # 清理临时文件
            return result

        except Exception as e:
            logger.error(f"视觉分析失败: {e}")
            return f"视觉分析错误: {str(e)}"

    def detect_objects_in_view(self) -> List[dict]:
        """检测当前视野中的物体"""
        timestamp = int(time.time())
        image_path = self.temp_image_dir / f"detect_{timestamp}.jpg"

        image = self.camera.capture_image(str(image_path))
        if image is None:
            return []

        try:
            objects = self.vlm_client.detect_objects(str(image_path))
            image_path.unlink(missing_ok=True)  # 清理临时文件
            return objects

        except Exception as e:
            logger.error(f"物体检测失败: {e}")
            return []

⚙️ ハードウェア設定管理

デバイス設定ファイル

# 硬件设备配置
hardware:
  # 机械臂配置
  robot_arm:
    enabled: true
    port: "/dev/ttyS1"
    baudrate: 115200
    timeout: 5.0

    # 工作空间限制 (mm)
    workspace:
      x_range: [-200, 200]
      y_range: [-200, 200]  
      z_range: [0, 300]

    # 关节角度限制 (度)
    joint_limits:
      joint1: [-180, 180]
      joint2: [-90, 90]
      joint3: [-90, 90]
      joint4: [-180, 180]
      joint5: [-90, 90]
      joint6: [-180, 180]

    # 运动参数
    max_speed: 100  # mm/s
    acceleration: 50  # mm/s²

  # 云台配置  
  gimbal:
    enabled: true
    port: "/dev/ttyS1"
    baudrate: 115200
    timeout: 3.0

    # 舵机引脚
    pan_pin: 3
    tilt_pin: 4

    # 角度限制
    angle_range: [0, 270]
    init_pan_angle: 135.0
    init_tilt_angle: 135.0

  # 音频配置
  audio:
    enabled: true
    music_directory: "./music"
    default_volume: 0.7
    supported_formats: [".mp3", ".wav", ".ogg", ".m4a"]

  # 摄像头配置
  camera:
    enabled: true
    device_id: 0
    resolution: [640, 480]
    fps: 30

# 安全配置
safety:
  emergency_stop_enabled: true
  collision_detection: true
  workspace_monitoring: true

  # 监控参数
  position_check_interval: 0.1  # 秒
  force_threshold: 50  # N
  temperature_threshold: 70  # °C

デバイスマネージャー

class HardwareManager:
    """硬件设备管理器"""

    def __init__(self, config: dict):
        self.config = config
        self.devices = {}
        self.safety_monitor = SafetyMonitor()

    def initialize_devices(self) -> bool:
        """初始化所有硬件设备"""
        success = True

        # 初始化机械臂
        if self.config['hardware']['robot_arm']['enabled']:
            success &= self._init_robot_arm()

        # 初始化云台
        if self.config['hardware']['gimbal']['enabled']:
            success &= self._init_gimbal()

        # 初始化音频系统
        if self.config['hardware']['audio']['enabled']:
            success &= self._init_audio()

        # 初始化摄像头
        if self.config['hardware']['camera']['enabled']:
            success &= self._init_camera()

        # 启动安全监控
        if success:
            self.safety_monitor.start()

        return success

    def get_device(self, device_name: str):
        """获取设备实例"""
        return self.devices.get(device_name)

    def shutdown_all(self):
        """关闭所有设备"""
        for device_name, device in self.devices.items():
            try:
                if hasattr(device, 'disconnect'):
                    device.disconnect()
                elif hasattr(device, 'close'):
                    device.close()
                logger.info(f"设备已关闭: {device_name}")
            except Exception as e:
                logger.error(f"设备关闭失败 {device_name}: {e}")

        self.safety_monitor.stop()

トラブルシューティングガイド

よくあるハードウェアの問題

  1. シリアルデバイスにアクセスできない
# 检查设备文件
ls -l /dev/ttyS*

# 添加用户到dialout组
sudo usermod -a -G dialout $USER

# 重新登录或重启
  1. ロボットアームの通信障害
  • シリアル接続を確認します
  • ボーレートが正しく設定されているか(115200)を確認します
  • 機器の電源を確認します
  • 通信プロトコルが一致しているかを確認します
  1. パン・チルト制御の異常
  • サーボの電源電圧を確認します
  • ピンが正しく接続されているかを確認します
  • 角度範囲の設定を確認します
  • 制御ボードの状態を確認します
  1. オーディオ再生の問題
# 检查音频系统
aplay -l

# 安装pygame
pip install pygame==2.5.2

# 检查音频文件格式
file ./music/*.mp3

デバッグツール

class HardwareDebugger:
    """硬件调试工具"""

    def __init__(self, hardware_manager: HardwareManager):
        self.hardware_manager = hardware_manager

    def run_diagnostic(self) -> dict:
        """运行硬件诊断"""
        results = {}

        # 机械臂诊断
        if 'robot_arm' in self.hardware_manager.devices:
            results['robot_arm'] = self._test_robot_arm()

        # 云台诊断
        if 'gimbal' in self.hardware_manager.devices:
            results['gimbal'] = self._test_gimbal()

        return results

    def _test_robot_arm(self) -> dict:
        """测试机械臂功能"""
        arm = self.hardware_manager.get_device('robot_arm')
        tests = {}

        # 连接测试
        tests['connection'] = arm.serial_conn.is_open if arm.serial_conn else False

        # 运动测试
        try:
            original_pos = arm.current_position.copy()
            tests['movement'] = arm.move_to_position(
                original_pos[0] + 10, 
                original_pos[1], 
                original_pos[2]
            )
            arm.move_to_position(*original_pos)  # 回到原位置
        except Exception:
            tests['movement'] = False

        return tests

このハードウェア統合システムは、LLM モジュールに完全な物理デバイス制御機能を提供します。簡単なオーディオ再生から複雑なロボットアーム制御まで、自然言語の指示によって正確に制御できます。