ドキュメントLLM による意味理解：Function Calling ツールシステム
概要
LLM モジュールの Function Calling システムは強力なツール統合機能を提供し、大規模言語モデルがさまざまな外部ツールを呼び出して複雑なタスクを実行できるようにします。システムはプラグイン設計を採用しており、ツール関数の動的な登録と管理をサポートします。
️ システムアーキテクチャ
コアコンポーネント
+-------------------+
| LLM客户端 |
+-------------------+
|
工具调用识别
|
+-------------------+
| 函数注册器 |
| FunctionRegistry |
+-------------------+
|
+-----+-----+
| |
| 工具分发 | Schema生成
| 器 |
+-----------+
|
+-----+-----+-----+-----+
| | | | |
|计算器|音乐播放|机械臂|云台|
|工具 |器工具 |控制 |控制|
+-----+-----+-----+-----+
ツール登録の仕組み
すべてのツールはデコレーターパターンによって関数レジストリに登録されます：
@FunctionRegistry.register_function('tool_name', 'Tool description')
def my_tool_func(param1: str, param2: int) -> str:
"""工具函数实现"""
return f"Result: {param1} - {param2}"
コアツールの詳細
1. 電卓ツール calculator.py
基本電卓（calculator）
四則演算、べき乗、剰余演算などの基本的な数学演算をサポートします。
機能の特徴：
- 安全な式の解析により、悪意のあるコードの実行を防止
- 基本演算子をサポート：
+,-,*,/,**,% - 括弧と演算の優先順位をサポート
- 数学定数：
pi,e
使用例：
# 用户输入：帮我计算 (123 + 456) * 2
# 系统调用：calculator(expression="(123 + 456) * 2")
# 返回结果：1158
実装の詳細：
@FunctionRegistry.register_function('calculator', '基本数学计算工具')
def calculator_func(expression: str) -> str:
"""执行基本数学计算"""
try:
# 安全评估数学表达式
allowed_names = {
"abs": abs, "round": round, "min": min, "max": max,
"sum": sum, "pow": pow, "pi": math.pi, "e": math.e
}
# 编译并执行表达式
code = compile(expression, "<string>", "eval")
result = eval(code, {"__builtins__": {}}, allowed_names)
return f"计算结果: {expression} = {result}"
except Exception as e:
return f"计算错误: {str(e)}"
高度な電卓（advanced_calculator）
三角関数、対数、平方根などの高度な数学関数をサポートします。
サポートする演算：
- 三角関数：
sin,cos,tan,asin,acos,atan - 対数関数：
log,log10,log2 - 指数とべき根：
exp,sqrt,cbrt - その他の関数：
abs,ceil,floor
使用例：
# 用户输入：计算sin(30度)的值
# 系统调用：advanced_calculator(operation="sin", value=30, unit="degree")
# 返回结果：0.5
2. 音楽プレーヤーツール music_player.py
音楽プレーヤーツールセットは、pygame ライブラリに基づく完全なオーディオ再生制御機能を提供します。
音楽リスト管理（music_list）
@FunctionRegistry.register_function('music_list', '获取可播放的音乐列表')
def music_list_func() -> str:
"""获取音乐目录中所有可播放的音乐文件"""
player = get_music_player()
music_files = player.list_music()
if not music_files:
return "当前没有可播放的音乐文件"
music_names = [file.rsplit('.', 1)[0] for file in music_files]
return f"可播放音乐：{', '.join(music_names)}"
再生制御ツール
音楽の再生（music_play）：
@FunctionRegistry.register_function('music_play', '播放指定的音乐')
def music_play_func(music_name: str) -> str:
"""播放指定音乐文件"""
player = get_music_player()
success = player.play_music(music_name)
if success:
return f"开始播放音乐: {music_name}"
else:
return f"播放失败: 未找到音乐 {music_name}"
音楽の停止（music_stop）：
@FunctionRegistry.register_function('music_stop', '停止音乐播放')
def music_stop_func() -> str:
"""停止当前音乐播放"""
player = get_music_player()
player.stop_music()
return "音乐播放已停止"
ランダム再生（music_play_random）：
@FunctionRegistry.register_function('music_play_random', '随机播放音乐')
def music_play_random_func() -> str:
"""随机选择并播放一首音乐"""
player = get_music_player()
music_name = player.play_random()
if music_name:
return f"随机播放: {music_name}"
else:
return "没有可播放的音乐文件"
音量制御ツール
音量の設定（music_volume_set）：
@FunctionRegistry.register_function('music_volume_set', '设置音量')
def music_volume_set_func(volume: int) -> str:
"""设置音乐播放音量"""
if not 0 <= volume <= 100:
return "音量值必须在0-100之间"
player = get_music_player()
player.set_volume(volume / 100.0)
return f"音量已设置为: {volume}%"
スマート音量調整（music_volume_set_smart）：
@FunctionRegistry.register_function('music_volume_set_smart', '智能设置音量')
def music_volume_set_smart_func(description: str) -> str:
"""根据描述智能设置音量"""
volume_map = {
"很小声": 10, "小声": 25, "轻声": 35,
"正常": 50, "标准": 50, "适中": 50,
"大声": 75, "很大声": 90, "最大声": 100
}
for key, value in volume_map.items():
if key in description:
player = get_music_player()
player.set_volume(value / 100.0)
return f"音量已设置为{key}({value}%)"
return "无法理解音量描述，请使用如'小声'、'正常'、'大声'等描述"
3. ロボットアーム制御ツール arm_functions.py
ロボットアーム制御ツールは、6 自由度ロボットアームの精密な制御機能を提供します。
基本的な動作制御
物体の把持（arm_grab_object）：
@FunctionRegistry.register_function('arm_grab_object', '抓取指定位置的物体')
def arm_grab_object_func(x: float, y: float, z: float) -> str:
"""在指定坐标抓取物体"""
try:
arm = get_arm_controller()
# 安全检查坐标范围
if not arm.is_position_safe(x, y, z):
return f"坐标超出安全范围: ({x}, {y}, {z})"
# 执行抓取动作
success = arm.grab_object(x, y, z)
if success:
return f"成功在位置({x}, {y}, {z})抓取物体"
else:
return "抓取失败，请检查目标位置是否有物体"
except Exception as e:
return f"机械臂控制错误: {str(e)}"
物体の配置（arm_place_object）：
@FunctionRegistry.register_function('arm_place_object', '将物体放置到指定位置')
def arm_place_object_func(x: float, y: float, z: float) -> str:
"""将抓取的物体放置到指定位置"""
try:
arm = get_arm_controller()
success = arm.place_object(x, y, z)
if success:
return f"成功将物体放置到位置({x}, {y}, {z})"
else:
return "放置失败，请确认机械臂正在抓取物体"
except Exception as e:
return f"机械臂控制错误: {str(e)}"
高度な制御機能
指定位置への移動（arm_move_to_position）：
@FunctionRegistry.register_function('arm_move_to_position', '移动机械臂到指定位置')
def arm_move_to_position_func(x: float, y: float, z: float, speed: float = 50.0) -> str:
"""移动机械臂末端到指定位置"""
try:
arm = get_arm_controller()
success = arm.move_to_position(x, y, z, speed)
if success:
return f"机械臂已移动到位置({x}, {y}, {z})"
else:
return "移动失败，请检查目标位置是否可达"
except Exception as e:
return f"机械臂控制错误: {str(e)}"
関節角度の制御（arm_set_joint_angles）：
@FunctionRegistry.register_function('arm_set_joint_angles', '设置机械臂关节角度')
def arm_set_joint_angles_func(angles: List[float]) -> str:
"""设置所有关节的角度"""
if len(angles) != 6:
return "必须提供6个关节的角度值"
try:
arm = get_arm_controller()
success = arm.set_joint_angles(angles)
if success:
return f"关节角度已设置为: {angles}"
else:
return "角度设置失败，请检查角度范围"
except Exception as e:
return f"机械臂控制错误: {str(e)}"
4. パン・チルト制御ツール gimbal_functions.py
パン・チルト制御ツールは、カメラ用パン・チルトの精密な制御機能を提供します。
基本制御（gimbal_control）
@FunctionRegistry.register_function('gimbal_control', '控制云台转动')
def gimbal_control_func(pan: float, tilt: float) -> str:
"""控制云台水平和垂直转动"""
try:
# 角度范围检查
if not (0 <= pan <= 270) or not (0 <= tilt <= 270):
return "云台角度必须在0到270度之间"
gimbal = get_gimbal_controller()
success = gimbal.set_position(pan, tilt)
if success:
return f"云台已转动到位置: 水平{pan}°, 垂直{tilt}°"
else:
return "云台控制失败，请检查硬件连接"
except Exception as e:
return f"云台控制错误: {str(e)}"
相対移動（gimbal_move_relative）
@FunctionRegistry.register_function('gimbal_move_relative', '云台相对移动')
def gimbal_move_relative_func(pan_delta: float, tilt_delta: float) -> str:
"""基于当前位置进行相对移动"""
try:
gimbal = get_gimbal_controller()
current_pan, current_tilt = gimbal.get_current_position()
new_pan = current_pan + pan_delta
new_tilt = current_tilt + tilt_delta
# 范围限制
new_pan = max(0, min(270, new_pan))
new_tilt = max(0, min(270, new_tilt))
success = gimbal.set_position(new_pan, new_tilt)
if success:
return f"云台相对移动完成: ({pan_delta:+.1f}°, {tilt_delta:+.1f}°)"
else:
return "云台移动失败"
except Exception as e:
return f"云台控制错误: {str(e)}"
5. 視覚認識ツール vlm_functions.py
視覚認識ツールは、視覚言語モデル（VLM）に基づく画像解析機能を提供します。
画像解析（vlm_analyze_image）
@FunctionRegistry.register_function('vlm_analyze_image', '分析图像内容')
def vlm_analyze_image_func(image_path: str, query: str = "请描述这张图片的内容") -> str:
"""使用VLM分析图像内容"""
try:
# 检查图片文件是否存在
if not Path(image_path).exists():
return f"图像文件不存在: {image_path}"
# 调用VLM进行图像分析
vlm_client = get_vlm_client()
result = vlm_client.analyze_image(image_path, query)
return f"图像分析结果:\n{result}"
except Exception as e:
return f"图像分析错误: {str(e)}"
物体検出（vlm_detect_objects）
@FunctionRegistry.register_function('vlm_detect_objects', '检测图像中的物体')
def vlm_detect_objects_func(image_path: str) -> str:
"""检测图像中的物体和位置"""
try:
vlm_client = get_vlm_client()
objects = vlm_client.detect_objects(image_path)
if not objects:
return "图像中未检测到任何物体"
result = "检测到的物体:\n"
for obj in objects:
result += f"- {obj['name']} (置信度: {obj['confidence']:.2f})\n"
return result
except Exception as e:
return f"物体检测错误: {str(e)}"
ツール Schema システム
Schema の自動生成
システムは関数のシグネチャと docstring に基づいて、OpenAI Function Schema を自動生成できます：
def generate_function_schema(func_name: str, func: callable) -> dict:
"""生成函数的OpenAI Function Schema"""
signature = inspect.signature(func)
doc = func.__doc__ or ""
schema = {
"name": func_name,
"description": doc.split('\n')[0] if doc else "",
"parameters": {
"type": "object",
"properties": {},
"required": []
}
}
# 解析参数
for param_name, param in signature.parameters.items():
param_info = {
"type": _get_type_string(param.annotation),
"description": _extract_param_description(doc, param_name)
}
schema["parameters"]["properties"][param_name] = param_info
if param.default is param.empty:
schema["parameters"]["required"].append(param_name)
return schema
Schema の例
以下は電卓ツールに対して自動生成された Schema です：
{
"name": "calculator",
"description": "基本数学计算工具",
"parameters": {
"type": "object",
"properties": {
"expression": {
"type": "string",
"description": "要计算的数学表达式"
}
},
"required": ["expression"]
}
}
ツール呼び出しのプロセス
呼び出しの検出
LLM クライアントがユーザー入力を解析し、呼び出すべきツールを特定します：
async def process_message(self, message: str, session_id: str):
# 构建消息和工具Schema
messages = self.conversation_manager.get_messages(session_id)
tools = get_enabled_schemas()
# 调用LLM API
response = await self.llm_client.chat_with_tools(
messages=messages,
tools=tools,
stream=True
)
# 处理工具调用
for chunk in response:
if chunk.get("tool_calls"):
await self._handle_tool_calls(chunk["tool_calls"])
ツールの実行
関数レジストリがツール呼び出しの振り分けと実行を担当します：
async def execute_tool_call(self, tool_call: dict) -> dict:
"""执行工具调用"""
func_name = tool_call["function"]["name"]
arguments = json.loads(tool_call["function"]["arguments"])
try:
# 获取注册的函数
func = FunctionRegistry.get_function(func_name)
if not func:
return {"error": f"未知工具: {func_name}"}
# 执行函数（支持同步和异步）
if FunctionRegistry.is_async(func_name):
result = await func(**arguments)
else:
result = func(**arguments)
return {"result": result}
except Exception as e:
logger.error(f"工具调用失败: {func_name}, 错误: {e}")
return {"error": str(e)}
結果の統合
ツールの実行結果は対話コンテキストに統合されます：
def integrate_tool_result(self, messages: list, tool_result: dict) -> list:
"""将工具结果集成到对话中"""
if "error" in tool_result:
messages.append({
"role": "function",
"name": tool_result["name"],
"content": f"工具调用失败: {tool_result['error']}"
})
else:
messages.append({
"role": "function",
"name": tool_result["name"],
"content": str(tool_result["result"])
})
return messages
️ ツール開発ガイド
新しいツールの作成
ステップ 1：ツール関数を定義する
@FunctionRegistry.register_function('my_tool', '我的自定义工具')
def my_tool_func(param1: str, param2: int = 10) -> str:
"""
执行自定义操作
Args:
param1: 字符串参数，必需
param2: 整数参数，可选，默认值为10
Returns:
操作结果的描述
"""
try:
# 工具逻辑实现
result = do_something(param1, param2)
return f"操作完成: {result}"
except Exception as e:
return f"操作失败: {str(e)}"
ステップ 2：設定のサポートを追加する
config/llm_config.yaml にツールの設定を追加します：
function_calling:
tools:
my_tool:
enabled: true
description: "我的自定义工具"
timeout: 10.0
# 工具特定配置
custom_param: "value"
ステップ 3：非同期処理に対応する
非同期処理が必要なツールの場合：
@FunctionRegistry.register_function('async_tool', '异步工具')
async def async_tool_func(param: str) -> str:
"""异步工具示例"""
# 异步操作，如网络请求
async with httpx.AsyncClient() as client:
response = await client.get(f"https://api.example.com/{param}")
return f"异步结果: {response.text}"
ツール開発のベストプラクティス
エラー処理
- 例外を処理するため、必ず try-catch ブロックを含めます
- ユーザーにわかりやすいエラーメッセージを返します
- デバッグ用に詳細なエラーログを記録します
パラメータの検証
def validate_parameters(func):
"""参数验证装饰器"""
def wrapper(*args, **kwargs):
# 验证参数有效性
if not all(isinstance(arg, str) for arg in args):
return "参数类型错误"
return func(*args, **kwargs)
return wrapper
@FunctionRegistry.register_function('safe_tool', '安全工具')
@validate_parameters
def safe_tool_func(text: str) -> str:
return f"处理结果: {text}"
タイムアウト制御
import asyncio
from functools import wraps
def timeout_control(seconds: int):
"""超时控制装饰器"""
def decorator(func):
@wraps(func)
async def wrapper(*args, **kwargs):
try:
return await asyncio.wait_for(
func(*args, **kwargs),
timeout=seconds
)
except asyncio.TimeoutError:
return f"操作超时({seconds}秒)"
return wrapper
return decorator
ハードウェアツールの開発
シリアルデバイス制御のテンプレート
import serial
import time
@FunctionRegistry.register_function('serial_device', '串口设备控制')
def serial_device_func(command: str, port: str = "/dev/ttyUSB0") -> str:
"""控制串口设备"""
try:
with serial.Serial(port, 9600, timeout=1) as ser:
# 发送命令
ser.write(command.encode())
time.sleep(0.1)
# 读取响应
response = ser.readline().decode().strip()
return f"设备响应: {response}"
except serial.SerialException as e:
return f"串口通信错误: {str(e)}"
except Exception as e:
return f"设备控制错误: {str(e)}"
⚡ パフォーマンス最適化
ツールのキャッシュ
計算負荷の高いツールにはキャッシュを使用できます：
from functools import lru_cache
@FunctionRegistry.register_function('expensive_calc', '计算密集型工具')
@lru_cache(maxsize=128)
def expensive_calc_func(input_data: str) -> str:
"""使用缓存的计算密集型工具"""
# 复杂计算逻辑
result = complex_calculation(input_data)
return str(result)
同時実行の制御
ツールの同時呼び出し数を制限します：
import asyncio
# 创建信号量限制并发
tool_semaphore = asyncio.Semaphore(5)
@FunctionRegistry.register_function('concurrent_tool', '并发控制工具')
async def concurrent_tool_func(param: str) -> str:
"""使用并发控制的工具"""
async with tool_semaphore:
# 工具逻辑
await asyncio.sleep(1) # 模拟工作
return f"处理完成: {param}"
監視とデバッグ
ツール呼び出しの監視
import time
from functools import wraps
def monitor_tool_calls(func):
"""工具调用监控装饰器"""
@wraps(func)
def wrapper(*args, **kwargs):
start_time = time.time()
try:
result = func(*args, **kwargs)
duration = time.time() - start_time
logger.info(f"工具调用成功: {func.__name__}, 耗时: {duration:.2f}s")
return result
except Exception as e:
duration = time.time() - start_time
logger.error(f"工具调用失败: {func.__name__}, 耗时: {duration:.2f}s, 错误: {e}")
raise
return wrapper
ツールの使用統計
from collections import defaultdict
class ToolUsageStats:
def __init__(self):
self.call_count = defaultdict(int)
self.success_count = defaultdict(int)
self.error_count = defaultdict(int)
self.total_time = defaultdict(float)
def record_call(self, tool_name: str, success: bool, duration: float):
self.call_count[tool_name] += 1
if success:
self.success_count[tool_name] += 1
else:
self.error_count[tool_name] += 1
self.total_time[tool_name] += duration
def get_stats(self) -> dict:
return {
tool: {
"calls": self.call_count[tool],
"success_rate": self.success_count[tool] / self.call_count[tool],
"avg_time": self.total_time[tool] / self.call_count[tool]
}
for tool in self.call_count
}
この Function Calling ツールシステムは、LLM モジュールに強力かつ柔軟なツール統合機能を提供し、簡単な計算から複雑なハードウェア制御まで、さまざまな応用シーンをサポートします。標準化された登録メカニズムと豊富なツールエコシステムにより、さまざまな用途のニーズに合わせて新しい機能を容易に拡張できます。