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

関数呼び出しによる音楽再生の実装

  1. LLM（大規模言語モデル）に基づき、音声対話による音楽再生を実現します。

技術ポイント

  1. Function Calling：大規模言語モデルのツール呼び出しの仕組みを学びます。
  2. 非同期プログラミング：Python の asyncio と FastAPI の非同期フレームワークを体験します。
  3. Web 開発：FastAPI フレームワークの高度な使い方と WebSocket によるリアルタイム通信
  4. システム統合：複雑なシステムのモジュール設計と統合を習得します。

応用シーン

スマートホームの音声操作、パーソナライズされた音楽コンパニオン、車載音声アシスタント、子ども向けの教育・エンターテインメントのお供など、さまざまなシーンに活用できます。

機能要件

  • シナリオ説明：運転者や同乗者が手を使わずに、音声で車内の音楽を安全に操作できます。
  • 典型的な対話：
  • 運転者：「ジェイ・チョウの曲を流して。」
  • LLM：「ジェイ・チョウの『Mojito』を再生しています。道路状況が複雑なので、運転に集中してください。」

ハードウェアの準備

ディスプレイ * 1、Bluetooth キーボード・マウス * 1、マイク * 1、メインコンソール * 1

回路接続

ハードウェアモジュール

ヒント：ハードウェアモジュールの使用説明

ハードウェアシステムの全体組み立て

組み込み AI 開発ボード RDK_X5

Uni mega2560 拡張ボード【V1.1】使用説明

電子モジュールの紹介

物理配線

具体的な配線・取り付け方法については、ハードウェアシステムの全体組み立てを参照してください。

ソフトウェア設計

音楽再生のワークフロー

音声による音楽再生の事例として、「xx の曲を再生する」場合のワークフローを示します。

  1. ユーザー入力：「『稲香』を再生して」
  2. LLM による解析：
  • 関数呼び出しが必要であることを識別
  • パラメータを抽出：曲名 =「稲香」

3. Function Calling で音楽ツールを呼び出し

4. 実行：

音楽プレーヤーツールが「稲香」の再生を開始します

5. 結果の返却：

「『稲香』の再生を開始しました。」

ソースコードのダウンロード

（HumansX サポートより提供）

プログラムコード

"""
音乐播放器功能函数
为LLM提供音乐播放的Function Calling接口
"""

import logging
from typing import Dict, Any
from .music_player import get_music_player
from .function_registry import FunctionRegistry

logger = logging.getLogger(__name__)

@FunctionRegistry.register_function('music_list', '获取可播放的音乐列表')
def music_list_func() -> str:
    """获取音乐列表"""
    try:
        player = get_music_player()
        music_list = player.list_music()

        if not music_list:
            return "很抱歉，当前没有可播放的音乐。请将音乐文件放入music目录中，并确保已安装pygame库：pip install pygame==2.5.2"

        # 生成友好的回复
        music_names = []
        for music_file in music_list:
            # 去掉文件扩展名
            name = music_file.rsplit('.', 1)[0]
            music_names.append(name)

        return f"当前可播放的音乐有：{', '.join(music_names)}，共{len(music_names)}首歌曲。"

    except Exception as e:
        if "pygame库未安装" in str(e):
            return "音乐播放器需要pygame库支持。请先安装：pip install pygame==2.5.2"
        logger.error(f"获取音乐列表失败: {e}")
        return "获取音乐列表时出现了问题，请稍后重试。"

@FunctionRegistry.register_function('music_play', '播放指定的歌曲')
def music_play_func(music_name: str) -> str:
    """播放指定歌曲"""
    try:
        player = get_music_player()
        result = player.play_music(music_name)
        return result

    except Exception as e:
        logger.error(f"播放音乐失败: {e}")
        return f"播放音乐时出现了问题：{str(e)}"

@FunctionRegistry.register_function('music_play_random', '随机播放一首歌曲')
def music_play_random_func() -> str:
    """随机播放歌曲"""
    try:
        player = get_music_player()
        result = player.play_random_music()
        return result

    except Exception as e:
        if "pygame库未安装" in str(e):
            return "音乐播放器需要pygame库支持。请先安装：pip install pygame==2.5.2"
        logger.error(f"随机播放失败: {e}")
        return f"随机播放时出现了问题：{str(e)}"

@FunctionRegistry.register_function('music_stop', '停止当前播放的音乐')
def music_stop_func() -> str:
    """停止播放音乐"""
    try:
        player = get_music_player()
        result = player.stop_music()
        return result

    except Exception as e:
        logger.error(f"停止音乐失败: {e}")
        return f"停止音乐时出现了问题：{str(e)}"

@FunctionRegistry.register_function('music_volume_set', '设置音乐播放音量')
def music_volume_set_func(volume: float) -> str:
    """设置音量"""
    try:
        # 验证音量参数
        if not isinstance(volume, (int, float)):
            return "音量参数必须是数字"

        if volume < 0 or volume > 1:
            return "音量范围应在0.0到1.0之间"

        player = get_music_player()
        result = player.set_volume(volume)
        return result

    except Exception as e:
        if "pygame库未安装" in str(e):
            return "音乐播放器需要pygame库支持。请先安装：pip install pygame==2.5.2"
        logger.error(f"设置音量失败: {e}")
        return f"设置音量时出现了问题：{str(e)}"

@FunctionRegistry.register_function('music_volume_up', '增加音乐播放音量')
def music_volume_up_func(step: float = 0.1) -> str:
    """增加音量"""
    try:
        # 验证步长参数
        if not isinstance(step, (int, float)):
            return "音量步长必须是数字"

        if step <= 0 or step > 1:
            return "音量步长应在0到1之间"

        player = get_music_player()
        result = player.increase_volume(step)
        return result

    except Exception as e:
        if "pygame库未安装" in str(e):
            return "音乐播放器需要pygame库支持。请先安装：pip install pygame==2.5.2"
        logger.error(f"增加音量失败: {e}")
        return f"增加音量时出现了问题：{str(e)}"

@FunctionRegistry.register_function('music_volume_down', '减少音乐播放音量')
def music_volume_down_func(step: float = 0.1) -> str:
    """减少音量"""
    try:
        # 验证步长参数
        if not isinstance(step, (int, float)):
            return "音量步长必须是数字"

        if step <= 0 or step > 1:
            return "音量步长应在0到1之间"

        player = get_music_player()
        result = player.decrease_volume(step)
        return result

    except Exception as e:
        if "pygame库未安装" in str(e):
            return "音乐播放器需要pygame库支持。请先安装：pip install pygame==2.5.2"
        logger.error(f"减少音量失败: {e}")
        return f"减少音量时出现了问题：{str(e)}"

@FunctionRegistry.register_function('music_status', '获取当前音乐播放状态')
def music_status_func() -> str:
    """获取播放状态"""
    try:
        player = get_music_player()
        status = player.get_current_status()

        if status["is_playing"] and status["current_music"]:
            music_name = status["current_music"].rsplit('.', 1)[0]
            volume_percent = int(status["volume"] * 100)
            return f"🎵 正在播放《{music_name}》，音量：{volume_percent}%"
        else:
            volume_percent = int(status["volume"] * 100)
            return f"⏸ 当前没有播放音乐，音量设置为{volume_percent}%，共有{status['music_count']}首歌曲可供播放。"

    except Exception as e:
        if "pygame库未安装" in str(e):
            return "音乐播放器需要pygame库支持。请先安装：pip install pygame==2.5.2"
        logger.error(f"获取播放状态失败: {e}")
        return "获取播放状态时出现了问题。"

def parse_volume_value(value) -> tuple:
    """解析音量值，支持多种格式"""
    if isinstance(value, (int, float)):
        v = float(value)
        if v < 0:
            return 0.0, 'mute'
        elif v > 1:
            return 1.0, 'max'
        else:
            return v, 'normal'

    if isinstance(value, str):
        v = value.strip().lower()

        # 支持中文描述
        if v in ['静音', 'mute']:
            return 0.0, 'mute'
        elif v in ['最大', '最大音量', '全开']:
            return 1.0, 'max'
        elif v in ['最小', '最小音量']:
            return 0.0, 'mute'
        elif v in ['一半', '中等', '50%']:
            return 0.5, 'normal'

        # 支持百分比
        if v.endswith('%'):
            try:
                percent = float(v[:-1])
                if percent <= 0:
                    return 0.0, 'mute'
                elif percent >= 100:
                    return 1.0, 'max'
                else:
                    return percent / 100.0, 'normal'
            except ValueError:
                pass

        # 尝试直接转换为数字
        try:
            f = float(v)
            if f < 0:
                return 0.0, 'mute'
            elif f > 1:
                return 1.0, 'max'
            else:
                return f, 'normal'
        except ValueError:
            pass

    # 默认值
    return 0.5, 'normal'

@FunctionRegistry.register_function('music_volume_set_smart', '智能设置音乐音量（支持数字、百分比、中文描述）')
def music_volume_set_smart_func(value) -> str:
    """智能设置音量，支持多种输入格式"""
    try:
        volume, vtype = parse_volume_value(value)
        player = get_music_player()
        result = player.set_volume(volume)

        # 根据输入类型优化回复
        if vtype == 'mute':
            return "音乐已为您静音。"
        elif vtype == 'max':
            return "音乐音量已调到最大。"
        else:
            volume_percent = int(volume * 100)
            return f"音乐音量已设置为{volume_percent}%。"

    except Exception as e:
        logger.error(f"智能设置音量失败: {e}")
        return f"设置音量时出现了问题：{str(e)}"

"""
音乐播放器工具实现
只支持实际音频播放的完整版本
"""

import os
import logging
import random
import threading
import time
from typing import List, Optional, Dict, Any
from pathlib import Path

# 尝试导入pygame，如果不可用则抛出错误
try:
    import pygame
    import pygame.mixer
    _pygame_available = True
    logger = logging.getLogger(__name__)
    logger.info("pygame音频播放库可用")
except ImportError:
    pygame = None
    pygame.mixer = None
    _pygame_available = False
    logger = logging.getLogger(__name__)
    logger.error("pygame不可用，音乐播放器无法工作。请运行: pip install pygame==2.5.2")

class SimpleMusicPlayer:
    """完整的音乐播放器类，只支持实际音频播放"""

    def __init__(self, music_dir: str = "music"):
        if not _pygame_available:
            raise RuntimeError("pygame库未安装，音乐播放器无法工作。请安装: pip install pygame==2.5.2")

        self.music_dir = Path(music_dir)
        self.current_music = None
        self.is_playing = False
        self.volume = 0.7
        self._music_list = []
        self._playback_thread = None
        self._stop_event = threading.Event()

        # 创建音乐目录
        self.music_dir.mkdir(exist_ok=True)

        # 初始化pygame mixer
        self._init_audio()

        # 初始化音乐列表
        self._load_music_list()
        logger.info(f"音乐播放器初始化完成，目录: {self.music_dir}")

    def _init_audio(self):
        """初始化音频系统"""
        try:
            pygame.mixer.init(frequency=22050, size=-16, channels=2, buffer=512)
            pygame.mixer.music.set_volume(self.volume)
            logger.info("pygame音频系统初始化成功")
        except Exception as e:
            logger.error(f"pygame音频系统初始化失败: {e}")
            raise RuntimeError(f"音频系统初始化失败: {e}")

    def _load_music_list(self) -> List[str]:
        """加载音乐文件列表"""
        try:
            if not self.music_dir.exists():
                logger.warning(f"音乐目录不存在: {self.music_dir}")
                self._music_list = []
                return []

            # 查找音频文件
            audio_extensions = ['.mp3', '.wav', '.flac', '.m4a', '.ogg']
            files = []

            for file in self.music_dir.iterdir():
                if file.is_file() and file.suffix.lower() in audio_extensions:
                    files.append(file.name)

            self._music_list = files
            logger.info(f"加载了 {len(files)} 个音乐文件")
            return files

        except Exception as e:
            logger.error(f"加载音乐列表失败: {e}")
            self._music_list = []
            return []

    def list_music(self) -> List[str]:
        """获取音乐列表"""
        return self._music_list

    def _monitor_playback(self):
        """监控播放状态"""
        while not self._stop_event.wait(1.0):
            try:
                if not pygame.mixer.music.get_busy() and self.is_playing:
                    # 音乐自然播放结束
                    logger.info("音乐播放自然结束")
                    self.is_playing = False
                    self.current_music = None
                    break
            except Exception as e:
                logger.error(f"监控播放状态时出错: {e}")
                break

        logger.debug("音乐播放监控线程结束")

    def play_music(self, music_name: str) -> str:
        """播放指定音乐"""
        try:
            # 查找音乐文件
            music_file = self._find_music_file(music_name)
            if not music_file:
                return f"很抱歉，未找到歌曲《{music_name}》"

            # 停止当前播放
            if self.is_playing:
                self.stop_music()

            music_path = self.music_dir / music_file

            # 实际播放音频
            try:
                pygame.mixer.music.load(str(music_path))
                pygame.mixer.music.set_volume(self.volume)
                pygame.mixer.music.play()

                self.current_music = music_file
                self.is_playing = True

                # 启动监控线程
                self._stop_event.clear()
                self._playback_thread = threading.Thread(target=self._monitor_playback, daemon=True)
                self._playback_thread.start()

                music_title = Path(music_file).stem
                logger.info(f"开始播放音乐: {music_path}")
                return f"🎵 正在为您播放《{music_title}》，音量: {int(self.volume * 100)}%"

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

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

    def play_random_music(self) -> str:
        """随机播放音乐"""
        if not self._music_list:
            return "很抱歉，当前没有可播放的音乐"

        random_music = random.choice(self._music_list)
        return self.play_music(random_music)

    def stop_music(self) -> str:
        """停止播放音乐"""
        if not self.is_playing:
            return "当前没有正在播放的音乐"

        try:
            pygame.mixer.music.stop()
            logger.info("停止音频播放")
        except Exception as e:
            logger.error(f"停止播放失败: {e}")

        # 停止监控线程
        self._stop_event.set()
        if self._playback_thread and self._playback_thread.is_alive():
            self._playback_thread.join(timeout=2.0)

        self.is_playing = False
        music_title = Path(self.current_music).stem if self.current_music else "未知歌曲"
        self.current_music = None

        logger.info(f"停止播放音乐: {music_title}")
        return f"🛑 已停止播放《{music_title}》"

    def set_volume(self, volume: float) -> str:
        """设置音量"""
        try:
            # 验证音量范围
            volume = max(0.0, min(1.0, volume))
            self.volume = volume

            # 实时调整音量
            try:
                pygame.mixer.music.set_volume(volume)
                logger.info(f"调整音量到: {volume}")
            except Exception as e:
                logger.error(f"音量调整失败: {e}")

            percentage = int(volume * 100)
            logger.info(f"音量设置为: {percentage}%")
            return f"🔊 音乐音量已设置为 {percentage}%"

        except Exception as e:
            logger.error(f"设置音量失败: {e}")
            return f"设置音量失败: {str(e)}"

    def increase_volume(self, step: float = 0.1) -> str:
        """增加音量"""
        new_volume = min(1.0, self.volume + step)
        return self.set_volume(new_volume)

    def decrease_volume(self, step: float = 0.1) -> str:
        """减少音量"""
        new_volume = max(0.0, self.volume - step)
        return self.set_volume(new_volume)

    def get_current_status(self) -> Dict[str, Any]:
        """获取当前播放状态"""
        return {
            "is_playing": self.is_playing,
            "current_music": self.current_music,
            "volume": self.volume,
            "music_count": len(self._music_list)
        }

    def _find_music_file(self, music_name: str) -> Optional[str]:
        """查找音乐文件"""
        # 精确匹配
        if music_name in self._music_list:
            return music_name

        # 不带扩展名匹配
        name_without_ext = Path(music_name).stem
        for music_file_item in self._music_list:
            if Path(music_file_item).stem == name_without_ext:
                return music_file_item

        # 模糊匹配
        for music_file_item in self._music_list:
            if name_without_ext.lower() in Path(music_file_item).stem.lower():
                return music_file_item

        return None

# 全局音乐播放器实例
_music_player_instance: Optional[SimpleMusicPlayer] = None

def get_music_player() -> SimpleMusicPlayer:
    """获取全局音乐播放器实例"""
    global _music_player_instance
    if _music_player_instance is None:
        _music_player_instance = SimpleMusicPlayer()
    return _music_player_instance

def initialize_music_player(music_dir: str = "music"):
    """初始化音乐播放器"""
    global _music_player_instance
    _music_player_instance = SimpleMusicPlayer(music_dir)
    return _music_player_instance

操作手順：

操作動画

リンク

具体的な操作（テキスト版）

回路接続が完了したら：

  1. ターミナルからリモートアクセスします。ターミナルで次のコマンドを入力し、Enter キーを押します。

ssh sunrise@192.168.8.111

  1. サービスを起動します。llm フォルダに移動し、メインプログラム python main.py を起動します。

Enter キーを押して起動を待ちます。次の画面が表示されたら、http:// 0.0.0.0:8002 を開けることを示しています。

  1. ブラウザを開き、URL http:// 0.0.0.0:8002 を入力してアクセスします。起動すると LLM インテリジェント対話アシスタントのメイン画面が表示されます。

  1. 右側のメニューの「音楽コントロール」をクリックすると、音楽リストの表示、ランダム再生、音量調整などの操作ができます。

実験結果

音楽リストを表示し、音楽をランダム再生できます。また、音量の調整（上げる・下げる）や再生の停止もできます。

リンク

サポート窓口

資料のダウンロード・技術サポートについては、HumansX株式会社までお問い合わせください。

← 前へMCP 機能：天気・地図の検索次へ →関数呼び出しによるロボットアームの動作制御の実装