ツール設計原則

優れたAgentツールは以下の原則に従うべきです:

  • 単一責任:各ツールは一つのことを行い、それをしっかり行う
  • 明確なインターフェース:入出力が明確で、理解しやすく使いやすい
  • 堅牢なエラーハンドリング:さまざまな異常な状況を優雅に処理する
  • 可観測性:デバッグと監視のためのログとメトリクスを提供する
  • セキュリティ:機密操作には権限制御と監査がある

ツールインターフェース設計

from pydantic import BaseModel, Field
from typing import Optional, List, Any
from enum import Enum

class ToolCategory(str, Enum):
    SEARCH = "search"
    DATA = "data"
    ACTION = "action"
    UTILITY = "utility"

class ToolParameter(BaseModel):
    name: str
    type: str
    description: str
    required: bool = True
    enum: Optional[List[str]] = None

class ToolDefinition(BaseModel):
    name: str = Field(description="ツール名")
    description: str = Field(description="Agentがいつ使用するかを理解するための詳細な説明")
    category: ToolCategory
    parameters: List[ToolParameter]
    risk_level: str = Field(default="low", pattern="^(low|medium|high|critical)$")

    def to_openai_function(self) -> dict:
        """OpenAI Function Calling形式に変換"""
        properties = {}
        required = []

        for param in self.parameters:
            prop = {"type": param.type, "description": param.description}
            if param.enum:
                prop["enum"] = param.enum
            properties[param.name] = prop
            if param.required:
                required.append(param.name)

        return {
            "name": self.name,
            "description": self.description,
            "parameters": {
                "type": "object",
                "properties": properties,
                "required": required
            }
        }

完全なツールの実装

import logging
import time
from functools import wraps

logger = logging.getLogger(__name__)

def tool_logger(func):
    """ツールデコレータ:呼び出しログと所要時間を自動記録"""
    @wraps(func)
    def wrapper(*args, **kwargs):
        start = time.time()
        try:
            result = func(*args, **kwargs)
            elapsed = time.time() - start
            logger.info(f"Tool {func.__name__} succeeded in {elapsed:.2f}s")
            return result
        except Exception as e:
            elapsed = time.time() - start
            logger.error(f"Tool {func.__name__} failed in {elapsed:.2f}s: {e}")
            raise
    return wrapper

class DatabaseQueryTool:
    """データベースクエリツール"""

    definition = ToolDefinition(
        name="query_database",
        description="データベースをクエリし、SQLクエリをサポートします。ユーザーデータ、注文情報、統計レポートなどの取得に使用します",
        category=ToolCategory.DATA,
        parameters=[
            ToolParameter(
                name="query_type",
                type="string",
                description="クエリタイプ",
                enum=["user", "order", "stats"]
            ),
            ToolParameter(
                name="filters",
                type="object",
                description="クエリフィルタ条件",
                required=False
            )
        ],
        risk_level="medium"
    )

    def __init__(self, db_connection):
        self.db = db_connection
        self.allowed_tables = ["users", "orders", "products"]

    @tool_logger
    def execute(self, query_type: str, filters: dict = None):
        # セキュリティ検証
        if query_type not in ["user", "order", "stats"]:
            return {"error": f"サポートされていないクエリタイプ: {query_type}"}

        # 安全なクエリを構築
        query = self._build_safe_query(query_type, filters or {})

        # クエリを実行
        result = self.db.execute(query)

        return {
            "success": True,
            "data": result,
            "count": len(result)
        }

    def _build_safe_query(self, query_type: str, filters: dict):
        # パラメータ化クエリを使用してSQLインジェクションを防ぐ
        # 事前定義されたテーブルのみをクエリ可能にする
        pass

ツールの登録と発見

class ToolRegistry:
    """ツール登録センター"""

    def __init__(self):
        self._tools: dict[str, Any] = {}

    def register(self, tool):
        """ツールを登録"""
        name = tool.definition.name
        if name in self._tools:
            raise ValueError(f"ツール {name} は既に登録されています")
        self._tools[name] = tool
        logger.info(f"ツールを登録: {name}")

    def get_all_definitions(self):
        """すべてのツール定義を取得(Function Calling用)"""
        return [
            tool.definition.to_openai_function()
            for tool in self._tools.values()
        ]

    def execute(self, name: str, params: dict):
        """ツールを実行"""
        if name not in self._tools:
            return {"error": f"ツールが存在しません: {name}"}

        tool = self._tools[name]

        # 高リスク操作は確認が必要
        if tool.definition.risk_level in ["high", "critical"]:
            if not self._confirm_execution(name, params):
                return {"error": "操作はユーザーによってキャンセルされました"}

        return tool.execute(**params)

ツールテスト

import pytest

class TestDatabaseQueryTool:
    def setup_method(self):
        self.tool = DatabaseQueryTool(MockDB())

    def test_valid_query(self):
        result = self.tool.execute("user", {"name": "張三"})
        assert result["success"] is True
        assert "data" in result

    def test_invalid_query_type(self):
        result = self.tool.execute("delete_all")
        assert "error" in result

    def test_sql_injection_prevention(self):
        result = self.tool.execute("user", {"name": "'; DROP TABLE users; --"})
        assert result["success"] is True  # 悪意のあるSQLを実行すべきではない

ツールパフォーマンス最適化

  • 結果キャッシュ:同一パラメータのクエリ結果をキャッシュする
  • タイムアウト制御:各ツールに実行タイムアウトを設定する
  • 並列実行:独立したツール呼び出しは並列に実行できる
  • 結果のトリミング:LLMに返す結果は簡潔にし、必要な情報のみを含める

まとめ

ツールはAgentの能力の延長です。優れたツール設計はAgentに力を与え、悪いツール設計はAgentの動きを妨げます。高品質なAgentシステムを構築するには、ツールの設計とテストに時間を投資することが重要です。