はじめに: 現実世界への架け橋としてのツール呼び出し
Il ツール呼び出し AIエージェントが国境を越えることを可能にする仕組みです テキスト生成 e 現実世界で行動する。ツールを呼び出さなくても、エージェントは次のことができます。 答え、説明、画面上に残るコードなど、単語を生成するだけです。ツール呼び出しを使用すると、 Web 上の情報の検索、データベースのクエリ、外部 API の呼び出し、ファイルの作成、電子メールの送信、 導入を管理し、複雑なプロセスを自動化します。
ツール呼び出しにより、言語モデルが受動的生成システムから アクティブなオーケストレーター どのツールをどのツールで使用するかを誰が自律的に決定するか それらを呼び出すためのパラメーターと、結果を一貫した応答に組み立てる方法。この能力こそが、 チャットボットとエージェントを区別するものです。前者は応答し、後者は応答します。 行為.
この記事では、ツールの正式な定義から、ツール呼び出しについて詳しく説明します。 JSON スキーマを使用して、入力検証、出力解析、統合まで REST API、GraphQL、データベース。再利用可能なツールのフレームワークを構築し、分析します 動的なツール検出や長期にわたるツール管理などの高度なパターン。
この記事で学べること
- JSON スキーマを使用してツールを定義する方法: 名前、説明、パラメーター、出力
- インジェクションとエラーを防ぐための入力検証とサニタイズ
- エラー回復機能を備えた構造化された出力解析
- REST APIとの統合とOpenAPI仕様からの自動生成
- 安全なパラメータ化されたクエリを備えたデータベース ツール
- 再利用可能なカスタム ツール フレームワークを構築する方法
- 実行時の動的なツールの検出と登録
- ストリーミングとタイムアウトによる長時間実行ツールの管理
ツール仕様: JSON スキーマ
エージェントが使用できる各ツールは、モデルが次のとおりであるように正式に記述される必要があります。 言語的に理解する それは何をするのか, どのパラメータを受け入れるか e どのような出力が生成されるか。この説明の事実上の標準は、 JSONスキーマ、構造を指定できる宣言形式、 データ型と制約。
適切なツール仕様はエージェントの品質の基礎です。 曖昧な場合、モデルはいつツールを使用すればよいかわかりません。パラメータがあいまいな場合は、生成 不正な値で呼び出します。出力が文書化されていない場合、解釈することができません。 結果が正しく表示されます。
ツール定義の構造
ツール定義は、次の 4 つの主要な要素で構成されます。
- 名前: パターンに従います
verb_noun（例えば。search_database,create_file,analyze_code）。名前は、エージェントのコンテキスト内で説明的かつ一意である必要があります - 説明: 定義の最も重要な部分。モデルは説明を使用して決定します いつ ツールを呼び出します。ツールが何をするのか、いつ使用するのか、いつ使用しないのかを説明する必要があります
- パラメータ: JSON スキーマで定義され、ツールで受け入れられる各パラメーターのタイプ、制約、デフォルト値、および説明を指定します。
- 出力: 応答の予期される形式。モデルが結果を解釈するのに役立ちます。
{
"name": "search_database",
"description": "Search the project database for records matching a query. Use this tool when the user asks about stored data, project records, or needs to find specific entries. Do NOT use this tool for general knowledge questions - those should be answered directly. Supports filtering by date range, status, and category.",
"parameters": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "The search query. Supports full-text search with boolean operators (AND, OR, NOT). Example: 'authentication AND bug NOT resolved'",
"minLength": 1,
"maxLength": 500
},
"table": {
"type": "string",
"description": "The database table to search in",
"enum": ["issues", "pull_requests", "commits", "users", "projects"]
},
"filters": {
"type": "object",
"description": "Optional filters to narrow results",
"properties": {
"status": {
"type": "string",
"enum": ["open", "closed", "in_progress", "resolved"],
"description": "Filter by record status"
},
"date_from": {
"type": "string",
"format": "date",
"description": "Start date for date range filter (ISO 8601)"
},
"date_to": {
"type": "string",
"format": "date",
"description": "End date for date range filter (ISO 8601)"
},
"assignee": {
"type": "string",
"description": "Filter by assigned user"
}
}
},
"limit": {
"type": "integer",
"description": "Maximum number of results to return",
"default": 10,
"minimum": 1,
"maximum": 100
},
"sort_by": {
"type": "string",
"description": "Field to sort results by",
"enum": ["relevance", "date_created", "date_updated", "priority"],
"default": "relevance"
}
},
"required": ["query", "table"]
}
}
ツールの説明のベスト プラクティス
- 具体的にする: 「プロジェクト データベースの検索」は「データの検索」よりも優れています。
- 使用しない場合を含める: モデルが不必要な呼び出しを回避できるようにします。
- 例を提供する: クエリまたはパラメータの具体的な例は、百聞は一見に如かずです
- 限界を文書化する: ツールにレート制限、タイムアウト、制限がある場合は、説明にその旨を記載してください
- 動詞_名詞パターンを使用する:
search_database,create_issue,delete_file,analyze_code
入力の検証とサニタイズ
言語モデルがツール呼び出しのパラメーターを生成するとき、その値が 正しく、安全であるか、期待される形式であるか。入力検証は防御の最前線です。 何らかの操作を実行するには、各パラメータをそのスキーマと照合してチェックする必要があります 注射攻撃を防ぐために消毒されています。
検証レベル
堅牢な検証システムは、次の 3 つのレベルで動作します。
- 型チェック: 各パラメータの型がスキーマと一致することを検証します。数値が期待される文字列、オブジェクトが期待される配列、null 非許容値が期待される null はすべて、すぐに捕捉されるエラーです。
- 境界チェック: 数値が許容範囲内であること、文字列が最小長と最大長を遵守していること、配列が最大要素数を超えていないことをチェックします。
- 制約の検証: 形式 (電子メール、URL、ISO 日付)、列挙値、正規表現パターン、パラメーター間の依存関係など、より複雑な制約をチェックします。
from dataclasses import dataclass
from typing import Any
import re
@dataclass
class ValidationError:
field: str
message: str
received_value: Any
class ToolInputValidator:
"""Validatore generico per input di tool call."""
def validate(self, schema: dict, params: dict) -> list[ValidationError]:
errors = []
properties = schema.get("properties", {})
required = schema.get("required", [])
# Verifica campi required
for field in required:
if field not in params or params[field] is None:
errors.append(ValidationError(
field=field,
message=f"Required field '{field}' is missing",
received_value=None
))
# Valida ogni parametro fornito
for field, value in params.items():
if field not in properties:
errors.append(ValidationError(
field=field,
message=f"Unknown field '{field}'",
received_value=value
))
continue
field_schema = properties[field]
errors.extend(self._validate_field(field, value, field_schema))
return errors
def _validate_field(self, field: str, value: Any, schema: dict) -> list[ValidationError]:
errors = []
expected_type = schema.get("type")
# Type checking
type_map = {"string": str, "integer": int, "number": (int, float),
"boolean": bool, "array": list, "object": dict}
if expected_type and not isinstance(value, type_map.get(expected_type, object)):
errors.append(ValidationError(field, f"Expected {expected_type}", value))
return errors # Skip further validation
# String constraints
if expected_type == "string":
if "minLength" in schema and len(value) < schema["minLength"]:
errors.append(ValidationError(field, f"Min length: {schema['minLength']}", value))
if "maxLength" in schema and len(value) > schema["maxLength"]:
errors.append(ValidationError(field, f"Max length: {schema['maxLength']}", value))
if "enum" in schema and value not in schema["enum"]:
errors.append(ValidationError(field, f"Must be one of: {schema['enum']}", value))
if "pattern" in schema and not re.match(schema["pattern"], value):
errors.append(ValidationError(field, f"Must match: {schema['pattern']}", value))
# Numeric constraints
if expected_type in ("integer", "number"):
if "minimum" in schema and value < schema["minimum"]:
errors.append(ValidationError(field, f"Minimum: {schema['minimum']}", value))
if "maximum" in schema and value > schema["maximum"]:
errors.append(ValidationError(field, f"Maximum: {schema['maximum']}", value))
return errors
SQLインジェクション保護
ツールが SQL クエリで使用されるパラメーターを受け入れる場合、インジェクション保護 それは絶対に重要です。言語モデルはコードを含むパラメータを生成する場合があります 悪意のある SQL。プロンプト内の敵対的なコンテンツの影響を受けたことと、 ユーザーが意図的な攻撃を試みています。
基本的なルールは単純で、例外は認められません。 常にクエリを使用する パラメータ化された、文字列を連結して SQL クエリを作成しないでください。この原則 これは、コンテキスト、モデルの信頼性、パフォーマンスのプレッシャーに関係なく当てはまります。
# MAI fare questo - vulnerabile a SQL injection
def search_unsafe(query: str, table: str):
sql = f"SELECT * FROM {table} WHERE content LIKE '%{query}%'"
return db.execute(sql) # PERICOLOSO!
# SEMPRE fare questo - query parameterizzata
def search_safe(query: str, table: str):
# Whitelist delle tabelle ammesse
allowed_tables = {"issues", "pull_requests", "commits", "users"}
if table not in allowed_tables:
raise ValueError(f"Table '{table}' not allowed")
sql = "SELECT * FROM " + table + " WHERE content LIKE ?"
return db.execute(sql, (f"%{query}%",)) # Parametro separato
# Ancora meglio: usa un ORM o query builder
def search_orm(query: str, table: str):
model = get_model(table) # Mappa tabella -> modello ORM
return model.objects.filter(content__icontains=query).all()
警告: ツールパラメータへの挿入
ツールパラメータによる注入のリスクを過小評価しないでください。攻撃者は次の可能性があります モデルによって生成されたパラメータに影響を与えるステートメントをユーザー プロンプトに挿入します。 たとえば、「データベースの検索; DROP TABLE ユーザー; --」のようなプロンプトは次のようになります。 適切な検証がない場合、モデルによって危険なパラメータに変換されます。 多層防御が不可欠です: 検証、サニタイズ、パラメータ化されたクエリ あらゆるレベルで。
出力解析とエラー回復
ツールが結果を返すとき、エージェントはそれを解釈できなければなりません 正しく。不正な出力、予期しないエラー、またはタイムアウトにより中断される可能性があります エージェントのフローが適切に管理されていない場合。堅牢な出力解析が鍵 システム全体の回復力のために。
構造化された出力
すべてのツールは、構造化された予測可能な形式で出力を返す必要があります。標準 標準化されたフィールドを持つ JSON オブジェクトが必要です。
@dataclass
class ToolResult:
"""Formato standard per il risultato di un tool."""
success: bool
data: Any = None
error: str | None = None
metadata: dict | None = None
def to_message(self) -> str:
"""Converte il risultato in un messaggio leggibile per il modello."""
if self.success:
if isinstance(self.data, list):
return f"Found {len(self.data)} results:\n" + \
"\n".join(str(item) for item in self.data)
return str(self.data)
else:
return f"Error: {self.error}"
# Esempio di output strutturato
result = ToolResult(
success=True,
data=[
{"id": 1, "title": "Auth bug", "status": "open"},
{"id": 2, "title": "Login issue", "status": "resolved"}
],
metadata={"total_count": 42, "page": 1, "execution_time_ms": 150}
)
エラー回復戦略
ツールに障害が発生した場合、エージェントには中断せずに回復するためのいくつかのオプションがあります。 ユーザーとの対話:
- 指数バックオフを使用して再試行します: 一時的なエラー (タイムアウト、レート制限、ネットワーク エラー) の場合は、間隔を増やして (1 秒、2 秒、4 秒、8 秒) 再試行します。
- 代替パラメータ: ツールが特定のパラメーターで失敗した場合、エージェントは別のパラメーター (より単純なクエリ、より広い日付範囲) で試行できます。
- 代替ツール: ツールが利用できない場合、エージェントは同様の情報を提供する代替ツールを使用できます。
- グレースフルデグラデーション: どのツールも機能しない場合、エージェントはユーザーに通知し、そのコンテキストで入手可能な情報で応答します。
- 結果のキャッシュ: 成功した結果は、繰り返しの呼び出しを回避し、後続のエラーが発生した場合のフォールバックとしてキャッシュされます。
import time
class ToolExecutor:
"""Esecutore di tool con retry e error recovery."""
def __init__(self, max_retries: int = 3, base_delay: float = 1.0):
self.max_retries = max_retries
self.base_delay = base_delay
self.cache: dict = {}
def execute(self, tool_name: str, params: dict) -> ToolResult:
# 1. Controlla la cache
cache_key = f"{tool_name}:{hash(str(sorted(params.items())))}"
if cache_key in self.cache:
cached = self.cache[cache_key]
if time.time() - cached["timestamp"] < 300: # 5 min TTL
return cached["result"]
# 2. Valida gli input
errors = self.validator.validate(tool_name, params)
if errors:
return ToolResult(
success=False,
error=f"Validation failed: {'; '.join(e.message for e in errors)}"
)
# 3. Esegui con retry
last_error = None
for attempt in range(self.max_retries):
try:
result = self._call_tool(tool_name, params)
# 4. Cache il risultato positivo
self.cache[cache_key] = {
"result": result, "timestamp": time.time()
}
return result
except RateLimitError:
delay = self.base_delay * (2 ** attempt)
time.sleep(delay)
last_error = "Rate limit exceeded"
except TimeoutError:
last_error = "Tool execution timed out"
except Exception as e:
last_error = str(e)
break # Non riprovare per errori non transienti
return ToolResult(success=False, error=last_error)
REST APIの統合
REST API は、AI エージェントの最も一般的な統合ポイントです。ほとんど の Web サービスは RESTful API を公開しており、これらの API をエージェント ツールとして統合することで、 コード管理のための GitHub から、機能の広大なエコシステムにアクセスします。 プロジェクト追跡には Jira、コミュニケーションには Slack、任意の SaaS サービスに接続 パブリック API を使用します。
OpenAPI仕様からの自動生成
多くの REST API は、 OpenAPI仕様 (旧 Swagger) の説明 正式には、すべてのエンドポイント、パラメータ、タイプ、および応答。この仕様は、 parsata を使用すると、手動でコードを記述せずにツール定義を自動的に生成できます。
import yaml
class OpenAPIToolGenerator:
"""Genera tool definitions da una OpenAPI specification."""
def generate_tools(self, spec_path: str) -> list[dict]:
with open(spec_path) as f:
spec = yaml.safe_load(f)
tools = []
for path, methods in spec.get("paths", {}).items():
for method, details in methods.items():
if method in ("get", "post", "put", "patch", "delete"):
tool = self._endpoint_to_tool(path, method, details, spec)
tools.append(tool)
return tools
def _endpoint_to_tool(self, path, method, details, spec) -> dict:
# Genera nome: GET /users/{id} -> get_user
operation_id = details.get("operationId", f"{method}_{path}")
name = operation_id.replace("-", "_").replace("/", "_")
# Genera parametri da path params, query params e request body
properties = {}
required = []
# Path parameters
for param in details.get("parameters", []):
prop = self._param_to_property(param)
properties[param["name"]] = prop
if param.get("required", False):
required.append(param["name"])
# Request body
if "requestBody" in details:
body_schema = self._resolve_ref(
details["requestBody"]["content"]["application/json"]["schema"],
spec
)
properties["body"] = body_schema
if details["requestBody"].get("required", False):
required.append("body")
return {
"name": name,
"description": details.get("summary", "") + ". " +
details.get("description", ""),
"parameters": {
"type": "object",
"properties": properties,
"required": required
},
"_metadata": {
"http_method": method.upper(),
"path": path,
"base_url": spec.get("servers", [{}])[0].get("url", "")
}
}
認証管理
外部 API を統合するには、さまざまな認証メカニズムを管理する必要があります。枠組み のエージェント ツールは、最も一般的なパターンを透過的にサポートする必要があります。 言語モデルのコンテキストでの資格情報。
API認証パターン
| 方法 | 実装 | 安全性 | 使用事例 |
|---|---|---|---|
| APIキー | ヘッダ X-API-Key またはクエリパラメータ |
平均 | シンプルなAPI、内部サービス |
| 無記名トークン | ヘッダ Authorization: Bearer <token> |
高い | 標準 RESTful API、JWT |
| OAuth 2.0 | リフレッシュトークンを使用した認可フロー | 非常に高い | サードパーティ API、委任されたアクセス |
| mTLS | 双方向のクライアント証明書とサーバー証明書 | 最大 | エンタープライズ API、内部マイクロサービス |
レート制限と再試行ロジック
各 API には、遵守する必要があるレート制限があります。 APIを呼び出すエージェント レート制限を管理しないとブロックされる危険があり、ユーザー エクスペリエンスが低下します。 管理者は積極的に行動する必要があります。残りの制限を引き出し、その前に速度を落とします。 限界に達してから反応するだけではありません。
import time
from collections import defaultdict
class RateLimiter:
"""Rate limiter con token bucket per API esterne."""
def __init__(self):
self.limits: dict[str, dict] = {}
def configure(self, api_name: str, requests_per_minute: int):
self.limits[api_name] = {
"rpm": requests_per_minute,
"tokens": requests_per_minute,
"last_refill": time.time()
}
def acquire(self, api_name: str) -> bool:
"""Tenta di acquisire un token. Restituisce False se rate limited."""
if api_name not in self.limits:
return True
limit = self.limits[api_name]
now = time.time()
elapsed = now - limit["last_refill"]
# Refill tokens proporzionalmente al tempo trascorso
refill = elapsed * (limit["rpm"] / 60.0)
limit["tokens"] = min(limit["rpm"], limit["tokens"] + refill)
limit["last_refill"] = now
if limit["tokens"] >= 1:
limit["tokens"] -= 1
return True
return False
def wait_time(self, api_name: str) -> float:
"""Tempo di attesa stimato prima del prossimo token disponibile."""
if api_name not in self.limits:
return 0
limit = self.limits[api_name]
if limit["tokens"] >= 1:
return 0
return (1 - limit["tokens"]) * (60.0 / limit["rpm"])
GraphQLの統合
グラフQL と統合するための REST API に代わる柔軟な代替手段を提供します。 AIエージェント。各エンドポイントが固定構造を返す REST とは異なり、 GraphQL を使用すると、クライアントは応答に必要なフィールドを正確に指定できます。 これはデータ量を削減できるため、エージェントにとって特に有益です。 転送され、応答を処理するために必要なトークン。
ツールとしてのクエリとミューテーション
GraphQL では、操作は次のように分割されます。 クエリ （読み）え 突然変異 （書き込み）。各クエリまたはミューテーションは、個別のエージェント ツールとして公開できます。 GraphQL パラメータがツール パラメータにマップされています。
class GraphQLToolAdapter:
"""Adatta operazioni GraphQL a tool per l'agente."""
def __init__(self, endpoint: str, headers: dict = None):
self.endpoint = endpoint
self.headers = headers or {}
def create_query_tool(self, name: str, query: str, variables_schema: dict) -> dict:
"""Crea un tool da una query GraphQL."""
return {
"name": name,
"description": f"Execute GraphQL query: {name}",
"parameters": {
"type": "object",
"properties": variables_schema
},
"_executor": lambda params: self._execute(query, params)
}
def _execute(self, query: str, variables: dict) -> ToolResult:
import requests
response = requests.post(
self.endpoint,
json={"query": query, "variables": variables},
headers=self.headers
)
data = response.json()
if "errors" in data:
return ToolResult(
success=False,
error=str(data["errors"])
)
return ToolResult(success=True, data=data.get("data"))
# Esempio di utilizzo
adapter = GraphQLToolAdapter("https://api.example.com/graphql")
tool = adapter.create_query_tool(
name="get_project_issues",
query="""
query GetIssues($projectId: ID!, $status: String) {
project(id: $projectId) {
name
issues(status: $status) {
id
title
assignee { name }
priority
}
}
}
""",
variables_schema={
"projectId": {"type": "string", "description": "Project ID"},
"status": {"type": "string", "enum": ["OPEN", "CLOSED", "IN_PROGRESS"]}
}
)
ツール生成のためのイントロスペクション スキーマ
GraphQL の最も強力な機能の 1 つは、 イントロスペクションスキーム: スキーマ自体をクエリして、すべての型、クエリ、および変異を検出する機能 利用可能です。これにより、GraphQL API からツール定義を自動的に生成できます。 外部ドキュメントなしで、実行時にスキーマを分析します。
データベースツール
データベース ツールを使用すると、エージェントはリレーショナル データベースのクエリと変更を行うことができます。 そしてNoSQL。これらは最も強力なツールであると同時に、最もリスクの高いツールの 1 つです。間違ったクエリは、 機密データが公開されたり、レコードが破損したり、パフォーマンスの問題が発生したりする可能性があります。
データベースツールのセキュリティアーキテクチャ
データベースへのアクセスは、マルチレベルのセキュリティ システムによって仲介される必要があります。
- 読み取り専用アクセス許可と読み取り/書き込みアクセス許可: デフォルトでは、データベース ツールには読み取り専用アクセスが必要です。書き込み操作にはユーザーからの明示的な許可と確認が必要です
- クエリ許可リスト: 実行できるクエリの種類を制限します。 SELECT のみ。DROP、ALTER、または TRUNCATE はなし
- 行制限: 何百万もの行が返されてコンテキストが飽和するのを避けるために、クエリには常に LIMIT を課します。
- テーブルのアクセス制御: パスワードや財務データが含まれるテーブルなどの機密テーブルを除く、アクセス可能なテーブルのホワイトリストを定義します。
- 監査ログ: 説明責任とデバッグのために、実行されたすべてのクエリをタイムスタンプ、ユーザー、結果とともに記録します。
class SecureDatabaseTool:
"""Tool di database con sicurezza a più livelli."""
ALLOWED_TABLES = {"issues", "projects", "sprints", "tasks", "comments"}
MAX_ROWS = 100
BLOCKED_KEYWORDS = {"DROP", "ALTER", "TRUNCATE", "DELETE", "INSERT", "UPDATE"}
def __init__(self, db_connection, read_only: bool = True):
self.db = db_connection
self.read_only = read_only
self.audit_log = []
def execute_query(self, query: str, params: tuple = ()) -> ToolResult:
# 1. Sanitization: controlla per keyword pericolose
query_upper = query.upper().strip()
if self.read_only:
if not query_upper.startswith("SELECT"):
return ToolResult(False, error="Only SELECT queries allowed in read-only mode")
for keyword in self.BLOCKED_KEYWORDS:
if keyword in query_upper:
return ToolResult(False, error=f"Blocked keyword: {keyword}")
# 2. Verifica le tabelle accedute
tables_in_query = self._extract_tables(query)
unauthorized = tables_in_query - self.ALLOWED_TABLES
if unauthorized:
return ToolResult(False, error=f"Access denied to tables: {unauthorized}")
# 3. Aggiungi LIMIT se mancante
if "LIMIT" not in query_upper:
query = query.rstrip(";") + f" LIMIT {self.MAX_ROWS}"
# 4. Esegui con parametri (MAI concatenazione di stringhe)
try:
cursor = self.db.execute(query, params)
columns = [desc[0] for desc in cursor.description]
rows = [dict(zip(columns, row)) for row in cursor.fetchall()]
# 5. Audit log
self.audit_log.append({
"query": query, "params": params,
"rows_returned": len(rows),
"timestamp": time.time()
})
return ToolResult(
success=True,
data=rows,
metadata={"columns": columns, "row_count": len(rows)}
)
except Exception as e:
return ToolResult(False, error=f"Query error: {str(e)}")
カスタムツールフレームワーク
数十、数百のツールを備えた運用システムでは、標準化するフレームワークが必要です ツールの作成、登録、実行。優れたフレームワークは定型文を削減し、 一貫性とテストの容易さを保証します。
再利用可能なフレームワークのパターン
フレームワークはデコレータに基づいています @tool Python 関数を変換します
エージェントが登録できるツールでは通常のものです。デコレータが自動的に処理します
docstring とタイプヒントからツール定義を生成し、入力を検証し、
エラーが発生し、出力をフォーマットします。
import inspect
from functools import wraps
from typing import get_type_hints
class ToolRegistry:
"""Registry centralizzato per tutti i tool dell'agente."""
def __init__(self):
self._tools: dict[str, dict] = {}
def tool(self, name: str = None, description: str = None):
"""Decoratore per registrare una funzione come tool."""
def decorator(func):
tool_name = name or func.__name__
tool_desc = description or func.__doc__ or "No description"
# Genera lo schema dei parametri dai type hints
hints = get_type_hints(func)
sig = inspect.signature(func)
properties = {}
required = []
for param_name, param in sig.parameters.items():
if param_name == "self":
continue
param_type = hints.get(param_name, str)
prop = self._type_to_schema(param_type)
# Usa il docstring per la descrizione del parametro
prop["description"] = f"Parameter: {param_name}"
properties[param_name] = prop
if param.default is inspect.Parameter.empty:
required.append(param_name)
# Registra il tool
self._tools[tool_name] = {
"definition": {
"name": tool_name,
"description": tool_desc,
"parameters": {
"type": "object",
"properties": properties,
"required": required
}
},
"executor": func
}
@wraps(func)
def wrapper(*args, **kwargs):
try:
result = func(*args, **kwargs)
return ToolResult(success=True, data=result)
except Exception as e:
return ToolResult(success=False, error=str(e))
return wrapper
return decorator
def get_definitions(self) -> list[dict]:
"""Restituisce le definizioni di tutti i tool registrati."""
return [t["definition"] for t in self._tools.values()]
def execute(self, tool_name: str, params: dict) -> ToolResult:
"""Esegue un tool per nome con i parametri forniti."""
if tool_name not in self._tools:
return ToolResult(False, error=f"Unknown tool: {tool_name}")
executor = self._tools[tool_name]["executor"]
return executor(**params)
# Utilizzo del framework
registry = ToolRegistry()
@registry.tool(
name="analyze_code",
description="Analyze source code for potential issues, complexity, and style."
)
def analyze_code(code: str, language: str, checks: list[str] = None) -> dict:
"""Analizza il codice sorgente cercando problemi e suggerimenti."""
results = {
"issues": [],
"complexity_score": 0,
"suggestions": []
}
# ... logica di analisi ...
return results
@registry.tool(
name="search_documentation",
description="Search the project documentation for relevant articles and guides."
)
def search_documentation(query: str, section: str = None, limit: int = 5) -> list:
"""Cerca nella documentazione del progetto."""
# ... logica di ricerca ...
return results
ツールの検出と動的登録
高度なエージェントでは、利用可能なツールは固定されておらず、実行時にベースで変更される可能性があります。 コンテキスト、ユーザー権限、外部サービスの可用性などに影響します。の ツールの発見 エージェントが新しいことを発見できるようにするメカニズムです ツールを使用して、システムを再起動せずに動的に登録します。
能力の推論
エージェントが多数のツール (数十、数百) にアクセスできる場合、それらすべてを含めることは現実的ではありません。 プロンプトで: トークンが多すぎるため、モデルが混乱します。そこには 能力 推論 現在のクエリに関連するツールのみを動的に選択します。 ツールの説明とユーザー要求の間の意味上の類似性に基づいて行われます。
class DynamicToolSelector:
"""Seleziona i tool più rilevanti per la query corrente."""
def __init__(self, registry: ToolRegistry, embedding_model):
self.registry = registry
self.embedding_model = embedding_model
self._tool_embeddings: dict[str, list[float]] = {}
self._index_tools()
def _index_tools(self):
"""Indicizza le descrizioni dei tool per la ricerca semantica."""
for tool in self.registry.get_definitions():
text = f"{tool['name']}: {tool['description']}"
self._tool_embeddings[tool["name"]] = \
self.embedding_model.embed(text)
def select_tools(self, query: str, max_tools: int = 10) -> list[dict]:
"""Seleziona i tool più rilevanti per la query."""
query_embedding = self.embedding_model.embed(query)
# Calcola la similarità con ogni tool
scores = []
for tool_name, tool_embedding in self._tool_embeddings.items():
similarity = cosine_similarity(query_embedding, tool_embedding)
scores.append((tool_name, similarity))
# Ordina per similarità decrescente
scores.sort(key=lambda x: x[1], reverse=True)
# Restituisci i top-k tool
selected_names = [name for name, _ in scores[:max_tools]]
return [
t for t in self.registry.get_definitions()
if t["name"] in selected_names
]
動的検出ツールの利点
- スケーラビリティ: コンテキスト ウィンドウを飽和させることなく、数百のツールをサポートします
- 精度: モデルは関連するツールのみを参照するため、混乱やエラーが軽減されます。
- 拡張性: エージェント コードを変更せずに、実行時に新しいツールを追加できます。
- 安全性: ユーザー権限に基づいてツールをフィルタリングできます
- コスト削減: プロンプト内のツールが少ないということは、呼び出しごとに消費されるトークンが少ないことを意味します
ストリーミングおよび長時間実行ツール
すべてのツールがミリ秒以内に実行を完了するわけではありません。一部の操作では、 重要な時間: 大規模なコードベースの分析、アプリケーションのデプロイ、レポートの生成 複雑なテストスイートの実行。このような場合、フレームワークはサポートする必要があります。 非同期実行 進捗報告付き。
長期間実行されるツールのパターン
import asyncio
from enum import Enum
class ToolStatus(Enum):
PENDING = "pending"
RUNNING = "running"
COMPLETED = "completed"
FAILED = "failed"
CANCELLED = "cancelled"
class AsyncToolExecutor:
"""Esecutore asincrono per tool long-running."""
def __init__(self, timeout_seconds: int = 300):
self.timeout = timeout_seconds
self.tasks: dict[str, dict] = {}
async def execute_async(self, tool_name: str, params: dict,
on_progress=None) -> str:
"""Avvia l'esecuzione asincrona e restituisce un task ID."""
task_id = f"task_{tool_name}_{int(time.time())}"
self.tasks[task_id] = {
"status": ToolStatus.PENDING,
"progress": 0,
"result": None,
"started_at": time.time()
}
# Avvia il task in background
asyncio.create_task(
self._run_with_timeout(task_id, tool_name, params, on_progress)
)
return task_id
async def _run_with_timeout(self, task_id, tool_name, params, on_progress):
self.tasks[task_id]["status"] = ToolStatus.RUNNING
try:
result = await asyncio.wait_for(
self._execute(tool_name, params, task_id, on_progress),
timeout=self.timeout
)
self.tasks[task_id]["status"] = ToolStatus.COMPLETED
self.tasks[task_id]["result"] = result
except asyncio.TimeoutError:
self.tasks[task_id]["status"] = ToolStatus.FAILED
self.tasks[task_id]["result"] = ToolResult(
False, error=f"Timeout after {self.timeout}s"
)
except Exception as e:
self.tasks[task_id]["status"] = ToolStatus.FAILED
self.tasks[task_id]["result"] = ToolResult(False, error=str(e))
def get_status(self, task_id: str) -> dict:
"""Ottieni lo stato corrente di un task."""
if task_id not in self.tasks:
return {"error": "Task not found"}
task = self.tasks[task_id]
return {
"status": task["status"].value,
"progress": task["progress"],
"elapsed": time.time() - task["started_at"],
"result": task["result"] if task["status"] == ToolStatus.COMPLETED else None
}
async def cancel(self, task_id: str) -> bool:
"""Annulla un task in esecuzione."""
if task_id in self.tasks:
self.tasks[task_id]["status"] = ToolStatus.CANCELLED
return True
return False
進行状況コールバック
時間のかかるツールの場合、ユーザー維持のために進行状況コールバックが不可欠です 知らされた。エージェントはリアルタイムで進行状況を表示できるため、 ユーザーは操作を待つかキャンセルするかを決定します。
- 完了率: 進捗状況が定量化できる操作 (ファイル分析、バッチ処理) 用
- ステータスメッセージ: 離散フェーズ操作用 (「サーバーに接続しています...」、「結果を分析しています...」、「レポートを生成しています...」)
- 部分的な結果: 増分出力を生成する操作用 (ストリーミング検索結果、プログレッシブ分析)
ツール構成パターン
エージェントの力は個々のツールにあるのではなく、次のことを行う能力にあります。 それらを構成する 複雑なワークフローで。成熟したエージェントは連鎖することができます 複数のツールを使用して、単一のツールでは処理できないタスクを完了します。
構成例：バグの解析と修正
ユーザーがバグを報告すると、エージェントは自動的にシーケンスを調整できます。 ツール呼び出しの数:
search_database: バグがすでに報告されている場合は、バグ トラッカーを検索します。search_codebase: 影響を受けるコンポーネントに関連するコード ファイルを検索します。analyze_code: 見つかったコードを分析してバグの原因を特定します。generate_fix: 分析に基づいて推奨パッチを生成します。run_tests: テストを実行して、修正によってリグレッションが発生しないことを確認します。create_pull_request: 修正と詳細な説明を含む PR を作成します
各ツールは前のツールの出力を入力として使用し、自動化されたパイプラインを作成します。 これにより、バグレポートが検証済みのプルリクエストに変わります。
結論
ツール呼び出しは、AI エージェントを受動的なテキスト ジェネレーターから変換するメカニズムです 現実世界で行動できるアクティブなオーケストレーターに。ツール呼び出しの品質は依存します 3つの柱によって： 正確な定義 詳細な JSON スキーマを使用して、 厳格な検証 エラーや攻撃を防ぐための入力、e 堅牢なエラー処理 適切な再試行とフォールバックを使用します。
REST API、GraphQL、データベースを安全に統合する方法、構築方法について見てきました。 動的検出を備えた再利用可能なツールのフレームワークと運用の管理方法 非同期実行と進捗レポートによる長時間実行。これらのパターンは、 信頼性が高くスケーラブルな運用エージェントを構築するための基礎となります。
次の記事では、 AIエージェントのテスト: テスト方法 ツール呼び出しフロー、モデル応答をシミュレートする方法、品質を測定する方法 エージェントの決定と、それを確実にするための回帰テストの実装方法について説明します。 この変更によって予期しない動作が発生することはありません。