Wprowadzenie: Wywoływanie narzędzi jako pomost do prawdziwego świata
Il wywołanie narzędzia jest to mechanizm, który umożliwia agentom AI przekraczanie granicy generowania tekstu, np działać w realnym świecie. Bez wywoływania narzędzi agent może to zrobić po prostu twórz słowa: odpowiedzi, wyjaśnienia, kod, który pozostaje na ekranie. Dzięki wywoływaniu narzędzi jest to możliwe szukać informacji w sieci, przeszukiwać bazy danych, wywoływać zewnętrzne API, tworzyć pliki, wysyłać e-maile, zarządzaj wdrożeniami i automatyzuj złożone procesy.
Wywołanie narzędzia przekształca model języka z pasywnego systemu generowania na model języka aktywny orkiestrator kto samodzielnie decyduje, jakich narzędzi i za pomocą jakich użyć parametry, aby je wywołać i jak ułożyć wyniki w spójną odpowiedź. Ta zdolność to jest to co odróżnia chatbota od agenta: ten pierwszy odpowiada, drugi dzieje.
W tym artykule szczegółowo przyjrzymy się wywoływaniu narzędzi: od formalnej definicji narzędzia ze schematem JSON, do sprawdzania poprawności danych wejściowych, analizowania wyników, aż do integracji z REST API, GraphQL i bazy danych. Zbudujemy framework narzędzi wielokrotnego użytku i przeanalizujemy zaawansowane wzorce, takie jak dynamiczne odkrywanie narzędzi i długotrwałe zarządzanie narzędziami.
Czego dowiesz się w tym artykule
- Jak zdefiniować narzędzia za pomocą schematu JSON: nazwa, opis, parametry i dane wyjściowe
- Walidacja i odkażanie danych wejściowych, aby zapobiec wstrzykiwaniu i błędom
- Ustrukturyzowane analizowanie wyników z odzyskiwaniem błędów
- Integracja z REST API i automatyczne generowanie ze specyfikacji OpenAPI
- Narzędzia bazodanowe z bezpiecznymi, sparametryzowanymi zapytaniami
- Jak zbudować niestandardową strukturę narzędzi wielokrotnego użytku
- Dynamiczne wykrywanie i rejestracja narzędzi w czasie wykonywania
- Zarządzanie długotrwałymi narzędziami ze strumieniowaniem i limitami czasu
Specyfikacja narzędzia: Schemat JSON
Każde narzędzie, z którego może skorzystać agent, musi być formalnie opisane tak, aby powstał model językowo zrozumieć co to robi, jakie parametry akceptuje e jaki rodzaj produkcji generuje. De facto standardem dla tego opisu jest Schemat JSON, format deklaratywny, który pozwala określić strukturę, typy danych i ograniczenia.
Dobra specyfikacja narzędzia ma fundamentalne znaczenie dla jakości środka: jeśli chodzi o opis jest niejasne, model nie będzie wiedział, kiedy użyć narzędzia; jeśli parametry są niejednoznaczne, wygeneruj wywołania z nieprawidłowymi wartościami; jeśli wynik nie zostanie udokumentowany, nie będzie w stanie zinterpretować wyniki poprawnie.
Anatomia definicji narzędzia
Definicja narzędzia składa się z czterech kluczowych elementów:
- Nazwa: podąża za wzorem
verb_noun(np.search_database,create_file,analyze_code). Nazwa musi być opisowa i unikalna w kontekście agenta - Opis: najbardziej krytyczna część definicji. Model podejmuje decyzję na podstawie opisu Gdy wywołać narzędzie. Musi wyjaśniać, do czego służy dane narzędzie, kiedy go używać, a kiedy NIE
- Parametry: zdefiniowane za pomocą JSON Schema, określają typ, ograniczenia, wartości domyślne i opis każdego parametru akceptowanego przez narzędzie
- Wyjścia: Oczekiwany format odpowiedzi, który pomaga modelowi zinterpretować wyniki
{
"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"]
}
}
Najlepsze praktyki dotyczące opisów narzędzi
- Bądź konkretny: „Przeszukaj bazę danych projektu” jest lepsze niż „Wyszukaj dane”
- Podaj, kiedy NIE należy go używać: Pomaga modelowi uniknąć niepotrzebnych wywołań
- Podaj przykłady: Konkretny przykład zapytania lub parametrów jest wart tysiąca słów
- Dokumentuj limity: Jeśli narzędzie ma ograniczenia szybkości, przekroczenia limitu czasu lub ograniczenia, powiedz to w opisie
- Użyj wzorca czasownik_noun:
search_database,create_issue,delete_file,analyze_code
Walidacja danych wejściowych i oczyszczanie
Gdy model języka generuje parametry wywołania narzędzia, nie ma gwarancji, że wartości te zostaną wygenerowane są prawidłowe, bezpieczne lub mają oczekiwany format. Walidacja danych wejściowych to pierwsza linia obrony: pierwsza Aby wykonać jakąkolwiek operację, każdy parametr musi zostać sprawdzony pod kątem jego schematu i odkażane, aby zapobiec atakom polegającym na wstrzyknięciu.
Poziomy walidacji
Solidny system walidacji działa na trzech poziomach:
- Wpisz Sprawdzanie: sprawdza, czy typ każdego parametru jest zgodny ze schematem. Ciąg znaków, w którym oczekiwana jest liczba, tablica, w której oczekiwany jest obiekt, wartość null, w której oczekiwana jest wartość nie dopuszczająca wartości null – wszystkie błędy mają zostać wykryte natychmiast
- Sprawdzanie granic: sprawdza, czy wartości liczbowe mieszczą się w dozwolonych zakresach, czy ciągi znaków mają minimalną i maksymalną długość, czy tablice nie przekraczają maksymalnej liczby elementów
- Walidacja ograniczeń: sprawdza bardziej złożone ograniczenia, takie jak formaty (e-mail, adres URL, daty ISO), wartości wyliczeniowe, wzorce wyrażeń regularnych, zależności między parametrami
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
Ochrona przed wtryskiem SQL
Gdy narzędzie akceptuje parametry, które będą wykorzystywane w zapytaniach SQL, następuje zabezpieczenie przed wstrzyknięciem jest to absolutnie krytyczne. Model języka może generować parametry zawierające kod Złośliwy SQL, zarówno dlatego, że miała na niego wpływ kontradyktoryjna treść monitu, jak i dlatego, że użytkownik podejmuje próbę celowego ataku.
Podstawowa zasada jest prosta i nie dopuszcza wyjątków: ZAWSZE używaj zapytania sparametryzowane, nigdy nie łącz ciągów znaków w celu tworzenia zapytań SQL. Ta zasada sprawdza się niezależnie od kontekstu, zaufania do modelu lub presji na wydajność.
# 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()
Ostrzeżenie: Wstrzyknięcie parametrów narzędzia
Nigdy nie lekceważ ryzyka wtrysku na podstawie parametrów narzędzia. Napastnik mógłby wstaw instrukcje do podpowiedzi użytkownika, które wpływają na parametry generowane przez model. Na przykład monit typu „Przeszukaj bazę danych; Użytkownicy DROP TABLE; --” może brzmieć przekształcane przez model w niebezpieczne parametry, jeśli nie ma odpowiednich walidacji. Niezbędna jest dogłębna obrona: weryfikacja, oczyszczanie i sparametryzowane zapytania na każdym poziomie.
Analiza wyników i odzyskiwanie błędów
Kiedy narzędzie zwraca swój wynik, agent musi być w stanie go zinterpretować poprawnie. Zniekształcone dane wyjściowe, nieoczekiwany błąd lub przekroczenie limitu czasu mogą spowodować przerwanie przepływ agentów, jeśli nie jest odpowiednio zarządzany. Kluczem jest solidne analizowanie wyników dla odporności całego systemu.
Ustrukturyzowane dane wyjściowe
Każde narzędzie powinno zwracać dane wyjściowe w ustrukturyzowanym i przewidywalnym formacie. Standard oczekuje obiektu JSON ze standardowymi polami:
@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}
)
Strategie odzyskiwania błędów
W przypadku awarii narzędzia agent ma kilka możliwości przywrócenia działania bez przerywania interakcja z użytkownikiem:
- Spróbuj ponownie z wykładniczym wycofaniem: w przypadku błędów przejściowych (przekroczenie limitu czasu, limit szybkości, błędy sieciowe) ponów próbę w rosnących odstępach czasu (1 s, 2 s, 4 s, 8 s)
- Parametry alternatywne: jeśli narzędzie zawiedzie z określonymi parametrami, agent może spróbować z innymi parametrami (prostsze zapytanie, szerszy zakres dat)
- Alternatywne narzędzia: Jeśli narzędzie nie jest dostępne, agent może skorzystać z alternatywnego narzędzia, które dostarcza podobnych informacji
- Urocza degradacja: Jeśli żadne narzędzie nie działa, agent informuje użytkownika i odpowiada, podając informacje dostępne w jego kontekście
- Buforowanie wyników: Pomyślne wyniki są zapisywane w pamięci podręcznej, aby uniknąć powtarzających się wywołań i jako rozwiązanie awaryjne w przypadku kolejnych błędów
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)
Integracja API REST
Interfejsy API REST są najpopularniejszym punktem integracji agentów AI. Większość usług internetowych udostępnia interfejsy API RESTful i integruje je w miarę możliwości narzędzi agenta aby uzyskać dostęp do szerokiego ekosystemu funkcji: od GitHuba do zarządzania kodem, do Jira w celu śledzenia projektu, do Slacka w celu komunikacji, do dowolnej usługi SaaS z publicznym API.
Automatyczne generowanie ze specyfikacji OpenAPI
Wiele interfejsów API REST udostępnia taki interfejs Specyfikacja OpenAPI (dawniej Swagger) opisując formalnie wszystkie punkty końcowe, parametry, typy i odpowiedzi. Ta specyfikacja może być parsata automatycznie, aby wygenerować definicje narzędzi bez ręcznego pisania kodu.
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", "")
}
}
Zarządzanie uwierzytelnianiem
Integracja zewnętrznych interfejsów API wymaga zarządzania różnymi mechanizmami uwierzytelniania. Ramy narzędzi agenta musi obsługiwać najpopularniejsze wzorce w sposób przejrzysty, bez ujawniania referencje w kontekście modelu języka.
Wzór uwierzytelniania API
| Metoda | Realizacja | Bezpieczeństwo | Przypadek użycia |
|---|---|---|---|
| Klucz API | Chodnikowiec X-API-Key lub parametr zapytania |
Przeciętny | Proste API, usługi wewnętrzne |
| Token okaziciela | Chodnikowiec Authorization: Bearer <token> |
Wysoki | Standardowe API RESTful, JWT |
| OAuth 2.0 | Przepływ autoryzacji z tokenem odświeżania | Bardzo wysoki | API innej firmy, dostęp delegowany |
| mTLS | Dwustronne certyfikaty klienta i serwera | Maksymalny | Enterprise API, wewnętrzne mikroserwisy |
Logika ograniczania szybkości i ponawiania prób
Każdy interfejs API ma ograniczenia szybkości, których należy przestrzegać. Agent wywołujący API bez zarządzania ograniczaniem szybkości istnieje ryzyko zablokowania, co pogarsza komfort użytkownika. Zarządzanie musi być proaktywne: wyznaczyć pozostałe limity i wcześniej zwolnić osiągnąć limit, a nie tylko reagować później.
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"])
Integracja z GraphQL
WykresQL oferuje elastyczną alternatywę dla interfejsów API REST do integracji z Agenci AI. W przeciwieństwie do REST, gdzie każdy punkt końcowy zwraca stałą strukturę, GraphQL pozwala klientowi dokładnie określić, które pola chce w odpowiedzi. Jest to szczególnie korzystne dla agentów, ponieważ zmniejsza ilość danych przesłane oraz tokeny potrzebne do przetworzenia odpowiedzi.
Zapytania i mutacje jako narzędzia
W GraphQL operacje są podzielone na zapytanie (czytanie) e mutacja (pismo). Każde zapytanie lub mutacja może zostać ujawnione jako osobne narzędzie agenta, z parametrami GraphQL odwzorowanymi na parametry narzędzia.
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"]}
}
)
Schemat introspekcji dla generowania narzędzi
Jedną z najpotężniejszych funkcji GraphQL jest schemat introspekcji: możliwość wysyłania zapytań do samego schematu w celu wykrycia wszystkich typów, zapytań i mutacji dostępne. Pozwala to na automatyczne generowanie definicji narzędzi z API GraphQL bez dokumentacji zewnętrznej, analizując schemat w czasie wykonywania.
Narzędzia baz danych
Narzędzia baz danych umożliwiają agentowi wysyłanie zapytań do relacyjnych baz danych i modyfikowanie ich i NoSQL. Należą do najpotężniejszych i jednocześnie najbardziej ryzykownych narzędzi: nieprawidłowe zapytanie może narażać wrażliwe dane, uszkadzać zapisy lub powodować problemy z wydajnością.
Architektura zabezpieczeń dla narzędzi bazodanowych
Dostęp do bazy danych musi być zapewniony poprzez wielopoziomowy system bezpieczeństwa:
- Uprawnienia tylko do odczytu a uprawnienia do odczytu i zapisu: Domyślnie narzędzia bazy danych powinny mieć dostęp tylko do odczytu. Operacje zapisu muszą wymagać wyraźnej zgody i potwierdzenia od użytkownika
- Zapytanie o listę dozwolonych: Ogranicz typy zapytań, które można uruchomić. Tylko WYBIERZ, bez DROP, ALTER i TRUNCATE
- Limit wierszy: Zawsze nakładaj LIMIT na zapytania, aby uniknąć zwracania milionów wierszy i nasycania kontekstu
- Kontrola dostępu do stołu: zdefiniuj białą listę dostępnych tabel, z wyłączeniem wrażliwych tabel, takich jak te z hasłami lub danymi finansowymi
- Rejestrowanie audytu: Rejestruje każde wykonane zapytanie ze znacznikiem czasu, użytkownikiem i wynikiem na potrzeby odpowiedzialności i debugowania
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)}")
Niestandardowe ramy narzędzi
W systemie produkcyjnym z dziesiątkami lub setkami narzędzi potrzebny jest framework, który standaryzuje tworzenia, rejestrowania i uruchamiania narzędzi. Dobre ramy ograniczają schematy, gwarantuje spójność i łatwość testowania.
Wzór struktury wielokrotnego użytku
Framework opiera się na dekoratorze @tool który przekształca funkcję Pythona
zwykłe w narzędziu rejestrowanym przez agenta. Dekorator zajmuje się tym automatycznie
wygeneruj definicję narzędzia na podstawie dokumentacji i wpisz wskazówki, sprawdź poprawność danych wejściowych, zarządzaj
błędy i sformatuj dane wyjściowe.
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
Wykrywanie narzędzi i rejestracja dynamiczna
W zaawansowanych agentach dostępne narzędzia nie są stałe: można je zmieniać w czasie wykonywania do kontekstu, uprawnień użytkownika lub dostępności usług zewnętrznych. The odkrycie narzędzia jest to mechanizm, który pozwala agentowi odkrywać nowe rzeczy narzędzie i zarejestruj je dynamicznie, bez konieczności ponownego uruchamiania systemu.
Wnioskowanie o możliwościach
Gdy agent ma dostęp do wielu narzędzi (dziesiątek lub setek), niepraktyczne jest uwzględnienie ich wszystkich w monicie: zajęłyby zbyt wiele żetonów i zmyliłyby model. Tam zdolność wnioskowanie dynamicznie wybiera tylko te narzędzia, które odpowiadają aktualnemu zapytaniu, w oparciu o podobieństwo semantyczne pomiędzy opisem narzędzia a żądaniem użytkownika.
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
]
Zalety narzędzia Dynamic Discovery Tool
- Skalowalność: obsługuje setki narzędzi bez obciążania okna kontekstowego
- Precyzja: Model widzi tylko odpowiednie narzędzia, co ogranicza zamieszanie i błędy
- Rozciągliwość: Nowe narzędzia można dodawać w czasie wykonywania bez zmiany kodu agenta
- Bezpieczeństwo: Narzędzia można filtrować na podstawie uprawnień użytkownika
- Redukcja kosztów: Mniej narzędzi w podpowiedzi oznacza mniej tokenów zużywanych na połączenie
Narzędzia do przesyłania strumieniowego i długotrwałe
Nie wszystkie narzędzia kończą swoje działanie w ciągu milisekund. Niektóre operacje wymagają znaczny czas: analiza dużych baz kodów, wdrażanie aplikacji, generowanie raportów złożone wykonanie zestawu testów. W takich przypadkach ramy muszą obsługiwać wykonanie asynchroniczne z raportowaniem postępów.
Wzór długotrwałego działania narzędzia
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
Wywołania zwrotne postępu
W przypadku czasochłonnych narzędzi wywołania zwrotne postępu są niezbędne do utrzymania użytkownika poinformowany. Agent może pokazywać postęp w czasie rzeczywistym, co pozwala użytkownik może zdecydować, czy poczekać, czy anulować operację.
- Procent ukończenia: dla operacji z wymiernym postępem (analiza plików, przetwarzanie wsadowe)
- Komunikaty o stanie: dla operacji w fazie dyskretnej („Łączenie z serwerem…”, „Analiza wyników…”, „Generowanie raportu…”)
- Częściowe wyniki: dla operacji generujących przyrostowy wynik (strumieniowe przesyłanie wyników wyszukiwania, analiza progresywna)
Wzór składu narzędzia
Siła agentów nie leży w indywidualnych narzędziach, ale w ich zdolnościach skomponować je w złożonych przepływach pracy. Dojrzały agent może łączyć się w łańcuchy wiele narzędzi do wykonywania zadań, z którymi nie poradzi sobie żadne pojedyncze narzędzie.
Przykład kompozycji: analiza i naprawa błędu
Gdy użytkownik zgłosi błąd, agent może automatycznie zaaranżować sekwencję wywołań narzędzi:
search_database: Przeszukaj narzędzie do śledzenia błędów, jeśli błąd został już zgłoszonysearch_codebase: Znajduje pliki kodu powiązane z komponentem, którego dotyczy problemanalyze_code: Przeanalizuj znaleziony kod, aby zidentyfikować przyczynę błędugenerate_fix: Generuje sugerowaną poprawkę na podstawie analizyrun_tests: uruchamia testy, aby sprawdzić, czy poprawka nie powoduje regresjicreate_pull_request: Utwórz PR z poprawką i szczegółowym opisem
Każde narzędzie wykorzystuje dane wyjściowe poprzedniego narzędzia jako dane wejściowe, tworząc zautomatyzowany potok co zamienia raport o błędzie w zweryfikowane żądanie ściągnięcia.
Wnioski
Wywoływanie narzędzi to mechanizm, który przekształca agentów AI z pasywnych generatorów tekstu aktywnym orkiestratorom zdolnym do działania w realnym świecie. Jakość wywołania narzędzia zależy na trzech filarach: precyzyjne definicje ze szczegółowym schematem JSON, rygorystyczna weryfikacja wejść, aby zapobiec błędom i atakom, np solidna obsługa błędów z odpowiednią ponowną próbą i powrotem.
Widzieliśmy jak bezpiecznie integrować API REST, GraphQL i bazy danych, jak budować strukturę narzędzi wielokrotnego użytku z dynamicznym odkrywaniem oraz sposób zarządzania operacjami długotrwałe z asynchronicznym wykonaniem i raportowaniem postępu. Wzory te stanowią podstawa, na której można budować niezawodne i skalowalne środki produkcyjne.
W następnym artykule zajmiemy się testowanie agentów AI: jak przetestować przepływy wywoływania narzędzi, jak symulować odpowiedzi modelu, jak mierzyć jakość decyzji agenta i jak wdrożyć testy regresyjne, aby to zapewnić zmiany nie powodują nieoczekiwanego zachowania.