Function Calling ツール呼び出しシステム

️ システム概要

Function Calling は音声対話アシスタントの中核となる拡張メカニズムで、大規模言語モデルが事前定義されたツール関数を呼び出して特定のタスクを実行できるようにします。本システムはレジストリパターンとアダプターパターンを採用し、ローカルツールと MCP リモートツールを統一的に呼び出せます。

️ アーキテクチャ設計

コアコンポーネント

┌─────────────────────────────────────────────────────────────┐
│                Function Calling 系统                        │
├─────────────────────────────────────────────────────────────┤
│  ┌─────────────────┐  ┌─────────────────┐  ┌─────────────────┐ │
│  │ Function Schema │  │ Function Tools  │  │ Function Impl   │ │
│  │   (Schema定义)  │  │   (注册调用)    │  │   (具体实现)    │ │
│  └─────────────────┘  └─────────────────┘  └─────────────────┘ │
│                                │                               │
│  ┌─────────────────────────────┼───────────────────────────┐   │
│  │           MCP Integration   │                           │   │
│  │  ┌─────────────────┐       │        ┌─────────────────┐ │   │
│  │  │ MCP Adapter     │       │        │ MCP Client Mgr  │ │   │
│  │  └─────────────────┘       │        └─────────────────┘ │   │
│  └─────────────────────────────┼───────────────────────────┘   │
└─────────────────────────────────┼───────────────────────────────┘
                                  │
                ┌─────────────────┼─────────────────┐
                │                 ▼                 │
    ┌─────────────────┐                  ┌─────────────────┐
    │   Local Tools   │                  │   MCP Servers   │
    │  (本地工具集)    │                  │  (远程工具集)    │
    └─────────────────┘                  └─────────────────┘

対応ツールの分類

1. 基本計算ツール

電卓（`calculator`）

機能：四則演算 対応する演算：加算（add）、減算（sub）、乗算（mul）、除算（div）

# 使用示例
await FunctionRegistry.call("calculator", op="add", a=10, b=5)
# 返回: "计算结果：15.00"

Schema 定義：

{
    "type": "function",
    "function": {
        "name": "calculator",
        "description": "加减乘除计算器",
        "parameters": {
            "type": "object",
            "properties": {
                "op": {"type": "string", "enum": ["add", "sub", "mul", "div"]},
                "a": {"type": "number", "description": "第一个数字"},
                "b": {"type": "number", "description": "第二个数字"}
            },
            "required": ["op", "a", "b"]
        }
    }
}

2. オーディオ制御ツール

音楽プレーヤー関連

対応する音楽フォーマット：MP3、WAV、FLAC、OGG、AAC、M4A

ツール名	機能説明	パラメーター
`music_player_list`	再生可能な楽曲の一覧を取得	なし
`music_player_play`	指定した楽曲を再生	`music_name: str`
`music_player_play_random`	楽曲をランダム再生	`sample_rate: int = 48000`
`music_player_stop`	楽曲の再生を停止	なし
`music_player_set_volume`	音量を設定	`value: float`
`kan`	音量を上げる	`step: float = 0.1`
`kan`	音量を下げる	`step: float = 0.1`

音量制御の特長：

数値指定に対応：0.8、50
パーセント指定に対応："50%"、"80%"
言葉による指定に対応："Maximum"（最大）、"Minimum"（最小）、"Half"（半分）、"Mute"（ミュート）

# 播放音乐示例
await FunctionRegistry.call("music_player_play", music_name="彩虹")

# 音量控制示例  
await FunctionRegistry.call("music_player_set_volume", value="80%")
await FunctionRegistry.call("music_player_increase_volume", step=0.1)

TTS の音量制御

ツール名	機能説明
`tts_player_set_volume`	音声アシスタントの音量を設定
`kan`	音声アシスタントの音量を上げる
`kan`	音声アシスタントの音量を下げる

3. MCP 拡張ツール

ロボットアーム制御（MCP サーバー：vision-arm-control）

MCP サーバーに移行したツール：

物体の検出と認識
ロボットアームの動作制御
ビジュアルナビゲーション機能
座標系の変換

カメラ制御

ツール名	機能説明
`yolo_move_servo_relative`	パン・チルトの相対移動
`yolo_reset_servo`	雲台を中央にリセット
`yolo_toggle_tracking`	顔追跡の切り替え
`yolo_get_face_count`	検出された顔の数を取得

コアとなる実装の仕組み

1. ツールレジストリ（`FunctionRegistry`）

class FunctionRegistry:
    _registry: Dict[str, Dict[str, Any]] = {}

    @classmethod
    def register_function(cls, name: str, description: str = ""):
        def decorator(func: Callable):
            cls._registry[name] = {
                'func': func,
                'description': description,
                'is_async': inspect.iscoroutinefunction(func)
            }
            return func
        return decorator

    @classmethod
    async def call(cls, name: str, *args, **kwargs):
        entry = cls._registry.get(name)
        if not entry:
            raise ValueError(f'未注册的函数: {name}')
        func = entry['func']
        if entry['is_async']:
            return await func(*args, **kwargs)
        else:
            loop = asyncio.get_event_loop()
            return await loop.run_in_executor(None, lambda: func(*args, **kwargs))

特長：

デコレーターによる登録：@FunctionRegistry.register_function() で自動登録
非同期への対応：同期/非同期関数を自動検出して適応
統一された呼び出し：FunctionRegistry.call() が統一された呼び出し窓口を提供

2. ツール実装層

音楽プレーヤーの実装（`MusicPlayer`）

class MusicPlayer:
    def __init__(self, audio_device_manager=None):
        self.audio_device_manager = audio_device_manager
        self.current_music = None
        self._monitor_thread = None
        self.player = None  # AudioNewPlayer实例

        # 从配置文件加载设置
        config_loader = load_config()
        self.config = config_loader.get_music_player_config()

        # 初始化pygame mixer
        self._init_pygame_mixer()

    def play_music(self, music_name, sample_rate=48000):
        # 1. 请求音频设备访问权限
        access_granted = self.audio_device_manager.request_device_access_sync(
            AudioDeviceType.MUSIC_PLAYER,
            {"music_name": music_name},
            force_stop_others=True
        )

        # 2. 加载并播放音乐
        pygame.mixer.music.load(music_path)
        pygame.mixer.music.set_volume(self._volume)
        pygame.mixer.music.play()

        # 3. 启动监控线程
        self._start_monitor_thread()

デバイス競合の管理：

オーディオデバイスのアクセス権を要求する仕組み
他のプレーヤーを強制停止するオプション
再生終了後にデバイスのアクセス権を自動解放

音量解析システム

def parse_volume_value(value):
    """智能音量值解析"""
    if isinstance(value, (int, float)):
        v = float(value)
        if v == 0: return 0.0, 'mute'
        if 0 < v < 0.08: return 0.05, 'min'
        return max(0.0, min(1.0, v)), 'normal'

    if isinstance(value, str):
        v = value.strip()
        if v in ['静音', 'mute']: return 0.0, 'mute'
        if v in ['最大', '最大音量']: return 1.0, 'max'
        if v in ['一半', '50%']: return 0.5, 'normal'

        # 百分比解析
        m = re.match(r'([0-9]+)%', v)
        if m:
            percent = int(m.group(1))
            return max(0.0, min(1.0, percent / 100.0)), 'normal'

3. MCP 統合アダプター

MCP Function Adapter のアーキテクチャ

class MCPFunctionAdapter:
    def __init__(self):
        self.integrator: Optional[MCPFunctionCallIntegrator] = None
        self._initialized = False

    async def get_enhanced_tools_schema(self) -> List[Dict[str, Any]]:
        """获取增强的工具schema，包含本地工具和MCP工具"""
        # 1. 获取本地工具schema
        local_schemas = get_all_schemas()

        # 2. 获取MCP工具schema  
        mcp_schemas = []
        if self.integrator:
            mcp_tools = self.integrator.get_mcp_tools()
            for tool in mcp_tools:
                if tool.active:
                    schema = {
                        "type": "function",
                        "function": {
                            "name": tool.name,
                            "description": tool.description,
                            "parameters": tool.parameters
                        }
                    }
                    mcp_schemas.append(schema)

        # 3. 合并schema
        return local_schemas + mcp_schemas

    async def call_tool_enhanced(self, tool_name: str, **kwargs) -> Any:
        """增强的工具调用，支持本地工具和MCP工具"""
        if tool_name.startswith("mcp:"):
            # MCP工具调用
            return await self.integrator.call_tool(tool_name, kwargs)
        else:
            # 本地工具调用
            return await FunctionRegistry.call(tool_name, **kwargs)

MCP ツールの命名規則：

ローカルツール：tool_name
MCP ツール：mcp:server_name:tool_name

ツール呼び出しの流れ

1. LLM によるツール呼び出しの流れ

1. 用户输入 → Turn Manager
2. 构建消息历史 + 工具Schema → LLM API
3. LLM生成Function Call → JSON格式解析
4. 工具名称路由判断 → 本地 vs MCP
5. 执行对应工具函数 → 获取结果
6. 格式化工具结果 → 回传LLM
7. LLM生成最终回复 → 用户

2. 複数ラウンドのツール呼び出しへの対応

# 在turn_manager.py中实现
MAX_TOOL_CALL_ROUNDS = 5
tool_call_round_count = 0

while tool_call_round_count < MAX_TOOL_CALL_ROUNDS:
    # 调用LLM
    response = await llm_client.stream_chat(messages_history)

    # 检查是否有工具调用
    if response.get("tool_calls"):
        for tool_call in response["tool_calls"]:
            # 执行工具
            result = await execute_tool(tool_call)

            # 将结果添加到消息历史
            messages_history.append({
                "role": "tool",
                "name": tool_call["function"]["name"],
                "content": str(result)
            })

        tool_call_round_count += 1
        continue  # 继续下一轮
    else:
        break  # 没有工具调用，结束循环

3. WebSocket メッセージプロトコル

ツール呼び出しの開始：

{
    "type": "TOOL_CALL_START",
    "payload": {
        "tool": "music_player_play",
        "params": {
            "music_name": "彩虹",
            "sample_rate": 48000
        },
        "round": 1
    }
}

ツール呼び出しの結果：

{
    "type": "TOOL_CALL_RESULT", 
    "payload": {
        "tool": "music_player_play",
        "result": "已为您播放《彩虹》，祝您聆听愉快！",
        "status": "success",
        "round": 1
    }
}

インテリジェントなプロンプト設計

ツール使用のガイダンス

システムプロンプトには、ツールの使用に関する詳細なガイドラインが含まれています。

## 🛠️ 可用工具功能

你拥有以下工具，请在适合的时候主动使用：

### 🧮 计算器
- 用于数学运算（加减乘除）
- 示例：用户问"12乘以8等于多少"时，调用calculator工具

### 🎵 音乐播放控制  
- music_player_list: 查看可播放的音乐
- music_player_play: 播放指定音乐（支持模糊匹配）
- music_player_play_random: 随机播放
- music_player_stop: 停止播放
- music_player_set_volume: 设置音量（支持百分比、描述词）

### 🔊 语音助手音量
- tts_player_set_volume: 调节我的说话音量
- tts_player_increase_volume/decrease_volume: 增减音量

## 📋 工具使用原则
1. **工具优先**：能用工具解决的问题，优先使用工具
2. **参数验证**：确保传递正确的参数类型和值
3. **错误处理**：工具调用失败时，向用户解释并提供替代方案
4. **用户体验**：多轮工具调用时，及时向用户反馈进度

エラー処理ポリシー

# 在LLM提示词中定义错误处理规则
"""
🚨 工具调用错误处理：

1. **参数错误**：
   - 检查参数类型和范围
   - 提供正确的使用示例
   - 引导用户重新输入

2. **工具不可用**：
   - 解释工具当前状态
   - 提供替代解决方案
   - 记录问题供后续处理

3. **执行失败**：
   - 分析失败原因
   - 尝试重试或降级处理
   - 向用户诚实反馈结果
"""

拡張開発ガイド

1. 新しいローカルツールの追加

ステップ 1：function_tools_impl.py にツールクラスを実装します

class NewTool:
    def __init__(self):
        pass

    def execute_action(self, param1: str, param2: int) -> str:
        """执行具体操作"""
        # 实现逻辑
        return "操作结果"

ステップ 2：function_call_tools.py にツールを登録します

@FunctionRegistry.register_function('new_tool_action', '新工具功能描述')
def new_tool_action_func(param1: str, param2: int = 10):
    return NewTool().execute_action(param1, param2)

ステップ 3：function_schema.py に Schema を定義します

def get_new_tool_schema():
    return {
        "type": "function",
        "function": {
            "name": "new_tool_action",
            "description": "新工具功能描述",
            "parameters": {
                "type": "object",
                "properties": {
                    "param1": {"type": "string", "description": "参数1描述"},
                    "param2": {"type": "integer", "description": "参数2描述", "default": 10}
                },
                "required": ["param1"]
            }
        }
    }

ステップ 4：get_all_schemas() 関数を更新します

def get_all_schemas():
    schemas = []
    schemas.append(get_calculator_schema())
    schemas.extend(get_music_player_schemas())
    schemas.extend(get_tts_player_schemas())
    schemas.append(get_new_tool_schema())  # 添加新工具
    return schemas

2. MCP ツールサーバーの開発

MCP サーバーの作成：

from mcp.server import Server
from mcp.server.models import InitializationOptions
import mcp.server.stdio

app = Server("custom-tools")

@app.list_tools()
async def handle_list_tools() -> list[types.Tool]:
    return [
        types.Tool(
            name="custom_action",
            description="自定义工具功能",
            inputSchema={
                "type": "object",
                "properties": {
                    "param": {"type": "string"}
                },
                "required": ["param"]
            }
        )
    ]

@app.call_tool()
async def handle_call_tool(name: str, arguments: dict) -> list[types.TextContent]:
    if name == "custom_action":
        result = perform_custom_action(arguments["param"])
        return [types.TextContent(type="text", text=result)]

MCP サーバー接続の設定：

# 在config.yaml中添加MCP配置
mcp:
  servers:
    custom-tools:
      command: "python"
      args: ["path/to/custom_tools_server.py"]
      env:
        CUSTOM_API_KEY: "your-api-key"

セキュリティ上の考慮事項

1. パラメーターの検証

厳密な型チェックと範囲の検証
パストラバーサル攻撃の防止
入力長の制限

2. 権限の制御

機密性の高い操作には追加の検証が必要
ツール呼び出し頻度の制限
リソース使用量の監視

3. エラーメッセージのフィルタリング

内部実装の詳細を公開しない
機密情報のマスキング処理
統一されたエラーレスポンス形式

パフォーマンスの最適化

1. 非同期実行

すべてのツールが非同期呼び出しに対応
I/O 負荷の高い処理には asyncio を使用
CPU 負荷の高い処理にはスレッドプールを使用

2. キャッシュ戦略

ツール Schema のキャッシュ
頻繁な呼び出し結果のキャッシュ
MCP 接続の再利用

3. リソース管理

オーディオデバイスの適時解放
同時ツール呼び出し数の制限
メモリ使用量の監視

Function Calling システムは、統一されたインターフェースと柔軟なアーキテクチャにより、音声対話アシスタントに強力な機能拡張能力を提供し、簡単な計算から複雑なロボット制御まで、さまざまなタスクに対応します。