Úvod: Volání nástroje jako most do skutečného světa
Il volání nástroje je to mechanismus, který umožňuje agentům AI překročit hranici generování textu e jednat v reálném světě. Bez volání nástroje může agent stačí vytvořit slova: odpovědi, vysvětlení, kód, který zůstane na obrazovce. S voláním nástroje to jde vyhledávat informace na webu, dotazovat se v databázích, volat externí API, vytvářet soubory, odesílat e-maily, spravovat nasazení a automatizovat složité procesy.
Volání nástroje transformuje jazykový model ze systému pasivního generování na a aktivní orchestrátor kdo autonomně rozhoduje o tom, které nástroje použít, se kterými parametry pro jejich vyvolání a jak sestavit výsledky do koherentní odpovědi. Tato schopnost je taková což odlišuje chatbota od agenta: první odpovídá, druhý akty.
V tomto článku prozkoumáme volání nástrojů do hloubky: od formální definice nástroje se schématem JSON, po ověření vstupu, analýzu výstupu až po integraci s REST API, GraphQL a databáze. Vybudujeme rámec opakovaně použitelných nástrojů a analyzujeme pokročilé vzory, jako je dynamické zjišťování nástrojů a dlouhodobá správa nástrojů.
Co se dozvíte v tomto článku
- Jak definovat nástroje se schématem JSON: název, popis, parametry a výstup
- Ověření a dezinfekce vstupů, aby se zabránilo vstřikování a chybám
- Analýza strukturovaného výstupu s obnovou chyb
- Integrace s REST API a automatické generování ze specifikace OpenAPI
- Databázové nástroje s bezpečnými, parametrizovanými dotazy
- Jak vytvořit znovu použitelný vlastní nástrojový rámec
- Dynamické zjišťování a registrace nástrojů za běhu
- Správa dlouhotrvajících nástrojů se streamováním a časovými limity
Specifikace nástroje: Schéma JSON
Každý nástroj, který může agent použít, musí být formálně popsán tak, aby model jazykově rozumět co to dělá, jaké parametry přijímá e jaký druh výstupu produkuje. De facto standard pro tento popis je Schéma JSON, deklarativní formát, který umožňuje specifikovat strukturu, datové typy a omezení.
Dobrá specifikace nástroje je základem kvality agenta: pokud jde o popis je vágní, model nebude vědět, kdy nástroj použít; pokud jsou parametry nejednoznačné, vygenerujte volání s nesprávnými hodnotami; pokud výstup není zdokumentován, nebude schopen interpretovat výsledky správně.
Anatomie nástroje Definice
Definice nástroje se skládá ze čtyř klíčových prvků:
- Jméno: podle vzoru
verb_noun(např.search_database,create_file,analyze_code). Název musí být popisný a jedinečný v kontextu agenta - Popis: nejkritičtější část definice. Model k rozhodování používá popis Když vyvolat nástroj. Musí vysvětlovat, co nástroj dělá, kdy jej použít a kdy NEPOUŽÍVAT
- Parametry: definované pomocí schématu JSON, určují typ, omezení, výchozí hodnoty a popis pro každý parametr akceptovaný nástrojem
- Výstupy: Očekávaný formát odpovědi, který modelu pomáhá interpretovat výsledky
{
"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"]
}
}
Nejlepší postupy pro popisy nástrojů
- Buďte konkrétní: "Prohledat databázi projektu" je lepší než "Vyhledat data"
- Uveďte, kdy jej NEPOUŽÍVAT: Pomáhá modelu vyhnout se zbytečným vyvoláním
- Uveďte příklady: Konkrétní příklad dotazu nebo parametrů vydá za tisíc slov
- Zdokumentujte limity: Pokud má nástroj omezení rychlosti, časové limity nebo omezení, uveďte to v popisu
- Použijte vzor sloveso_podstatné jméno:
search_database,create_issue,delete_file,analyze_code
Ověření vstupu a dezinfekce
Když jazykový model generuje parametry pro volání nástroje, není zaručeno, že hodnoty jsou správné, bezpečné nebo v očekávaném formátu. Ověření vstupu je první linií obrany: první Chcete-li provést jakoukoli operaci, každý parametr musí být porovnán se schématem a dezinfikováno, aby se zabránilo injekčním útokům.
Úrovně ověření
Robustní validační systém funguje na třech úrovních:
- Kontrola typu: Ověřuje, že typ každého parametru odpovídá schématu. Řetězec, kde se očekává číslo, pole, kde se očekává objekt, null, kde se očekává hodnota bez možnosti null, to vše jsou chyby, které mají být zachyceny okamžitě
- Kontrola hranic: kontroluje, zda jsou číselné hodnoty v povolených mezích, že řetězce respektují minimální a maximální délky, že pole nepřekračují maximální počet prvků
- Ověření omezení: kontroluje složitější omezení, jako jsou formáty (e-mail, URL, data ISO), hodnoty výčtu, vzory regulárních výrazů, závislosti mezi parametry
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
Ochrana vstřikování SQL
Když nástroj přijímá parametry, které budou použity v dotazech SQL, ochrana vkládání je to naprosto kritické. Jazykový model může generovat parametry obsahující kód Škodlivý SQL, a to jak proto, že byl ovlivněn nepřátelským obsahem ve výzvě, tak proto uživatel se pokouší o úmyslný útok.
Základní pravidlo je jednoduché a nepřipouští výjimky: VŽDY používejte dotaz parametrizované, nikdy nezřetězujte řetězce pro vytváření SQL dotazů. Tento princip platí bez ohledu na kontext, důvěru v model nebo tlak na výkon.
# 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()
Varování: Vstřikování v parametrech nástroje
Nikdy nepodceňujte riziko vstřikování přes parametry nástroje. Útočník mohl vložit do výzvy uživatele příkazy, které ovlivňují parametry generované modelem. Může to být například výzva jako „Prohledat databázi; DROP TABLE uživatele; --“. přeměněny modelem na nebezpečné parametry, pokud neexistují adekvátní validace. Zásadní je ochrana do hloubky: ověřování, sanitace a parametrizované dotazy na každé úrovni.
Analýza výstupu a zotavení po chybě
Když nástroj vrátí svůj výsledek, agent jej musí být schopen interpretovat správně. Špatně tvarovaný výstup, neočekávaná chyba nebo časový limit mohou přerušit tok agentů, pokud není správně řízen. Klíčem je robustní výstupní analýza pro odolnost celého systému.
Strukturovaný výstup
Každý nástroj by měl vrátit výstup ve strukturovaném a předvídatelném formátu. Standard očekává objekt JSON se standardizovanými poli:
@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 obnovy chyb
Když nástroj selže, agent má několik možností, jak jej obnovit bez přerušení interakce s uživatelem:
- Zkuste to znovu s exponenciálním stažením: pro přechodné chyby (timeout, rychlostní limit, síťové chyby) opakujte s prodlužujícími se intervaly (1s, 2s, 4s, 8s)
- Alternativní parametry: pokud nástroj selže s určitými parametry, agent to může zkusit s jinými parametry (jednodušší dotaz, širší časové období)
- Alternativní nástroje: Pokud nástroj není k dispozici, agent může použít alternativní nástroj, který poskytuje podobné informace
- Půvabná degradace: Pokud nefunguje žádný nástroj, agent informuje uživatele a odpoví informacemi dostupnými v jeho kontextu
- Ukládání výsledků do mezipaměti: Úspěšné výsledky jsou ukládány do mezipaměti, aby se zabránilo opakovaným voláním a jako záložní zdroj v případě následných chyb
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)
Integrace REST API
REST API jsou nejběžnějším integračním bodem pro agenty AI. Většina webových služeb odhaluje RESTful API a integrace těchto API jako nástroje agentů umožňuje pro přístup k rozsáhlému ekosystému funkcí: z GitHubu pro správu kódu, do Jira pro sledování projektu, do Slacku pro komunikaci, do jakékoli služby SaaS s veřejným API.
Automatické generování z OpenAPI Spec
Mnoho REST API jedno poskytuje Specifikace OpenAPI (dříve Swagger) popisující formálně všechny koncové body, parametry, typy a odpovědi. Tato specifikace může být parsata automaticky generuje definice nástrojů bez psaní ručního kódu.
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", "")
}
}
Správa autentizace
Integrace externích rozhraní API vyžaduje správu různých mechanismů ověřování. Rámec agentních nástrojů musí podporovat nejběžnější vzory transparentně, bez odhalení pověření v kontextu jazykového modelu.
Autentizační vzor API
| Metoda | Implementace | Bezpečnost | Use Case |
|---|---|---|---|
| Klíč API | Záhlaví X-API-Key nebo dotaz param |
Průměrný | Jednoduchá API, interní služby |
| Token na doručitele | Záhlaví Authorization: Bearer <token> |
Vysoký | Standardní RESTful API, JWT |
| OAuth 2.0 | Autorizační tok s obnovovacím tokenem | Velmi vysoká | API třetí strany, delegovaný přístup |
| mTLS | Dvoustranné klientské a serverové certifikáty | Maximum | Enterprise API, interní mikroslužby |
Omezení rychlosti a logika opakování
Každé API má limity rychlosti, které je třeba respektovat. API volajícího agenta bez správy rychlosti a omezování hrozí zablokování a zhoršení uživatelské zkušenosti. Management musí být proaktivní: vykreslit zbývající limity a zpomalit předtím dosáhnout limitu, nejen reagovat poté.
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"])
Integrace GraphQL
GraphQL nabízí flexibilní alternativu k REST API pro integraci agenti AI. Na rozdíl od REST, kde každý koncový bod vrací pevnou strukturu, GraphQL umožňuje klientovi přesně specifikovat, která pole chce v odpovědi obsahovat. To je výhodné zejména pro agenty, protože to snižuje množství dat přenesené a tokeny potřebné ke zpracování odpovědi.
Dotaz a mutace jako nástroje
V GraphQL jsou operace rozděleny na dotaz (čtení) e mutace (psaní). Každý dotaz nebo mutaci lze vystavit jako samostatný nástroj agenta, s parametry GraphQL namapovanými na parametry nástroje.
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"]}
}
)
Introspekční schéma pro generování nástrojů
Jednou z nejvýkonnějších funkcí GraphQL je schéma introspekce: schopnost dotazovat se na samotné schéma a objevovat všechny typy, dotazy a mutace k dispozici. To vám umožní automaticky generovat definice nástrojů z GraphQL API bez externí dokumentace, analyzování schématu za běhu.
Databázové nástroje
Databázové nástroje umožňují agentovi dotazovat se a upravovat relační databáze a NoSQL. Patří mezi nejvýkonnější a zároveň nejrizikovější nástroje: nesprávný dotaz může odhalit citlivá data, poškodit záznamy nebo způsobit problémy s výkonem.
Bezpečnostní architektura pro databázové nástroje
Přístup do databáze musí být zprostředkován víceúrovňovým bezpečnostním systémem:
- Oprávnění pouze pro čtení vs: Ve výchozím nastavení by databázové nástroje měly mít přístup pouze pro čtení. Operace zápisu musí vyžadovat výslovné povolení a potvrzení od uživatele
- Seznam povolených dotazů: Omezte typy dotazů, které lze spustit. Pouze SELECT, žádné DROP, ALTER nebo TRUNCATE
- Limit řádku: Vždy uplatněte LIMIT na dotazy, abyste se vyhnuli vracení milionů řádků a nasycení kontextu
- Řízení přístupu ke stolu: Definujte seznam povolených tabulek, s výjimkou citlivých tabulek, jako jsou tabulky s hesly nebo finančními údaji
- Protokolování auditu: Zaznamenává každý provedený dotaz s časovým razítkem, uživatelem a výsledkem pro účely odpovědnosti a ladění
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)}")
Custom Tool Framework
Ve výrobním systému s desítkami nebo stovkami nástrojů potřebujete rámec, který standardizuje vytváření, registrace a spouštění nástrojů. Dobrý rámec snižuje standardy, zaručuje konzistenci a snadné testování.
Vzor pro opakovaně použitelný rámec
Rám je založen na dekoratérovi @tool který transformuje funkci Pythonu
běžné v nástroji, který může agent zaregistrovat. Dekoratér se o to automaticky postará
generovat definici nástroje z docstringu a tipů typu, ověřovat vstupy, spravovat
chyby a formátujte výstup.
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
Objevování nástrojů a dynamická registrace
V pokročilých agentech nejsou dostupné nástroje pevně dané: mohou se měnit za běhu na kontext, uživatelská oprávnění nebo dostupnost externích služeb. The objev nástroje je to mechanismus, který agentovi umožňuje objevovat nové věci nástroj a dynamicky je registrujte bez restartování systému.
Odvozování schopností
Když má agent přístup k mnoha nástrojům (desítky nebo stovky), není praktické je všechny zahrnout ve výzvě: zabrali by příliš mnoho žetonů a zmátli model. Tam schopnost závěr dynamicky vybírá pouze nástroje relevantní pro aktuální dotaz, na základě sémantické podobnosti mezi popisem nástroje a požadavkem uživatele.
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
]
Výhody nástroje Dynamic Discovery Tool
- Škálovatelnost: podporuje stovky nástrojů bez nasycení kontextového okna
- Přesnost: Model vidí pouze relevantní nástroje, což snižuje zmatky a chyby
- Rozšiřitelnost: Nové nástroje lze přidávat za běhu bez změny kódu agenta
- Bezpečnost: Nástroje lze filtrovat na základě uživatelských oprávnění
- Snížení nákladů: Méně nástrojů ve výzvě znamená méně tokenů spotřebovaných na volání
Streamovací a dlouhotrvající nástroje
Not all tools complete their execution in milliseconds. Některé operace vyžadují významný čas: analýza velkých kódových základen, nasazení aplikací, generování zpráv komplexní provedení testovací sady. Pro tyto případy musí rámec podporovat asynchronní provádění s hlášením o pokroku.
Vzor pro nástroj s dlouhou životností
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
Zpětná volání postupu
U časově náročných nástrojů jsou zpětná volání postupu zásadní pro udržení uživatelů informován. Agent může ukázat průběh v reálném čase, což umožňuje uživatel se může rozhodnout, zda má operaci počkat nebo zrušit.
- Procento dokončení: pro operace s kvantifikovatelným postupem (analýza souborů, dávkové zpracování)
- Stavové zprávy: pro operace s diskrétními fázemi ("Připojování k serveru...", "Analýza výsledků...", "Generování zprávy...")
- Dílčí výsledky: pro operace, které vytvářejí přírůstkový výstup (streamování výsledků vyhledávání, progresivní analýza)
Vzor Složení nástroje
Síla agentů nespočívá v jednotlivých nástrojích, ale v jejich schopnosti skládat je ve složitých pracovních postupech. Zralý agent může řetězit více nástrojů pro dokončení úkolů, které by žádný nástroj nezvládl.
Příklad složení: Analýza a oprava chyby
Když uživatel nahlásí chybu, agent může automaticky zorganizovat sekvenci volání nástrojů:
search_database: Prohledejte nástroj pro sledování chyb, pokud již byla chyba nahlášenasearch_codebase: Vyhledá soubory kódu relevantní pro postiženou komponentuanalyze_code: Analyzujte nalezený kód a identifikujte příčinu chybygenerate_fix: Vygeneruje navrhovanou opravu na základě analýzyrun_tests: spustí testy k ověření, že oprava nezavádí regresecreate_pull_request: Vytvořte PR s opravou a podrobným popisem
Každý nástroj používá výstup předchozího nástroje jako vstup a vytváří automatizované potrubí což změní hlášení o chybě na ověřený požadavek na stažení.
Závěry
Volání nástroje je mechanismus, který transformuje agenty AI z pasivních generátorů textu aktivním orchestrátorům schopným působit v reálném světě. Kvalita vyvolání nástroje závisí na třech pilířích: přesné definice s podrobným schématem JSON, přísné ověřování vstupů pro prevenci chyb a útoků, e robustní zpracování chyb s vhodným opakováním a nouzovým řešením.
Viděli jsme, jak bezpečně integrovat REST API, GraphQL a databáze, jak stavět rámec opakovaně použitelných nástrojů s dynamickým zjišťováním a jak řídit operace dlouhotrvající s asynchronním prováděním a hlášením průběhu. Tyto vzory tvoří základ, na kterém lze vybudovat spolehlivé a škálovatelné výrobní prostředky.
V příštím článku se budeme věnovat testování agentů AI: jak testovat toky volání nástrojů, jak simulovat odezvy modelu, jak měřit kvalitu rozhodnutí agenta a jak implementovat regresní testování, aby to bylo zajištěno změny nezavedou neočekávané chování.