소개: 현실 세계로의 연결 역할을 하는 도구 호출
Il 도구 호출 AI 에이전트가 국경을 넘을 수 있게 해주는 메커니즘이다. 텍스트 생성 e 현실 세계에서 행동하다. 도구 호출 없이 상담원은 다음을 수행할 수 있습니다. 화면에 남아 있는 답변, 설명, 코드 등 단어만 생성하면 됩니다. 도구 호출을 사용하면 다음과 같은 작업을 수행할 수 있습니다. 웹에서 정보 검색, 데이터베이스 쿼리, 외부 API 호출, 파일 생성, 이메일 보내기, 배포를 관리하고 복잡한 프로세스를 자동화합니다.
도구 호출은 언어 모델을 수동적 생성 시스템에서 활성 오케스트레이터 어떤 도구를 사용할지 자율적으로 결정하는 사람 이를 호출하는 매개변수와 결과를 일관된 응답으로 구성하는 방법을 설명합니다. 이 능력은 챗봇과 에이전트를 구별하는 점은 전자는 응답하고 후자는 에이전트입니다. 행위.
이 기사에서는 도구 호출을 심층적으로 살펴보겠습니다. 도구의 형식적 정의부터 JSON 스키마를 사용하여 입력 검증, 출력 구문 분석, 통합까지 REST API, GraphQL 및 데이터베이스. 재사용 가능한 도구의 프레임워크를 구축하고 분석합니다. 동적 도구 검색 및 장기 실행 도구 관리와 같은 고급 패턴.
이 기사에서 배울 내용
- JSON 스키마를 사용하여 도구를 정의하는 방법: 이름, 설명, 매개변수 및 출력
- 입력 검증 및 삭제를 통해 주입 및 오류 방지
- 오류 복구를 통한 구조화된 출력 구문 분석
- REST API와의 통합 및 OpenAPI 사양의 자동 생성
- 안전하고 매개변수화된 쿼리가 포함된 데이터베이스 도구
- 재사용 가능한 사용자 정의 도구 프레임워크를 구축하는 방법
- 런타임 시 동적 도구 검색 및 등록
- 스트리밍 및 시간 제한이 포함된 장기 실행 도구 관리
도구 사양: JSON 스키마
에이전트가 사용할 수 있는 각 도구는 모델이 다음과 같이 공식적으로 설명되어야 합니다. 언어적으로 이해하다 그것이 무엇을 하는가, 어떤 매개변수를 허용하는지 e 어떤 종류의 출력이 생성되는지. 이 설명에 대한 사실상의 표준은 다음과 같습니다. JSON 스키마, 구조를 지정할 수 있는 선언적 형식, 데이터 유형 및 제약 조건.
좋은 도구 사양은 에이전트 품질의 기본입니다. 모호하므로 모델은 언제 도구를 사용해야 할지 알 수 없습니다. 매개변수가 모호한 경우 생성 잘못된 값으로 호출; 출력이 문서화되지 않으면 해석할 수 없습니다. 결과는 제대로.
도구 정의 분석
도구 정의는 다음 네 가지 핵심 요소로 구성됩니다.
- 이름: 패턴을 따른다
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"]
}
}
도구 설명에 대한 모범 사례
- 구체적으로 말하라: "프로젝트 데이터베이스 검색"이 "데이터 검색"보다 낫습니다.
- 사용하지 않을 때 포함: 모델이 불필요한 호출을 방지하도록 도와줍니다.
- 예시 제공: 쿼리나 매개변수의 구체적인 예는 천 단어의 가치가 있습니다.
- 한계를 문서화하라: 도구에 속도 제한, 시간 초과 또는 제한 사항이 있는 경우 설명에 이를 명시하세요.
- verb_noun 패턴을 사용하세요:
search_database,create_issue,delete_file,analyze_code
입력 검증 및 정리
언어 모델이 도구 호출에 대한 매개변수를 생성할 때 값이 보장되지 않습니다. 정확하고 안전하며 예상되는 형식입니다. 입력 검증은 첫 번째 방어선입니다. 작업을 수행하려면 각 매개변수를 해당 스키마와 비교하여 확인해야 합니다. 주입 공격을 방지하기 위해 소독되었습니다.
검증 수준
강력한 검증 시스템은 세 가지 수준에서 작동합니다.
- 유형 확인: 각 매개변수의 유형이 스키마와 일치하는지 확인합니다. 숫자가 예상되는 문자열, 객체가 예상되는 배열, nullable이 아닌 값이 예상되는 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()
경고: 도구 매개변수에 주입
도구 매개변수를 통한 주입 위험을 과소평가하지 마십시오. 공격자는 다음을 수행할 수 있습니다. 모델에 의해 생성된 매개변수에 영향을 미치는 명령문을 사용자 프롬프트에 삽입합니다. 예를 들어, "Search Database; DROP TABLE users; --"와 같은 프롬프트는 다음과 같습니다. 적절한 검증이 없으면 모델에 의해 위험한 매개변수로 변환됩니다. 심층적인 방어가 필수적입니다: 검증, 삭제 및 매개변수화된 쿼리 모든 수준에서.
출력 구문 분석 및 오류 복구
도구가 결과를 반환하면 에이전트는 이를 해석할 수 있어야 합니다. 올바르게. 잘못된 출력, 예상치 못한 오류 또는 시간 초과로 인해 중단될 수 있음 제대로 관리되지 않으면 에이전트 흐름. 강력한 출력 구문 분석이 핵심입니다. 전체 시스템의 탄력성을 위해.
구조화된 출력
모든 도구는 구조화되고 예측 가능한 형식으로 출력을 반환해야 합니다. 표준 표준화된 필드가 있는 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 에이전트의 가장 일반적인 통합 지점입니다. 대부분 웹 서비스는 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 통합
GraphQL 통합을 위해 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의 가장 강력한 기능 중 하나는 자기 성찰 계획: 모든 유형, 쿼리 및 변형을 발견하기 위해 스키마 자체를 쿼리하는 기능 가능합니다. 이를 통해 GraphQL API에서 도구 정의를 자동으로 생성할 수 있습니다. 외부 문서 없이 런타임에 스키마를 분석합니다.
데이터베이스 도구
데이터베이스 도구를 사용하면 에이전트가 관계형 데이터베이스를 쿼리하고 수정할 수 있습니다. 그리고 NoSQL. 이는 가장 강력하면서도 가장 위험한 도구 중 하나입니다. 잘못된 쿼리는 민감한 데이터를 노출하거나, 기록을 손상시키거나, 성능 문제를 일으킬 수 있습니다.
데이터베이스 도구를 위한 보안 아키텍처
데이터베이스에 대한 액세스는 다단계 보안 시스템에 의해 조정되어야 합니다.
- 읽기 전용 및 읽기-쓰기 권한: 기본적으로 데이터베이스 도구에는 읽기 전용 액세스 권한이 있어야 합니다. 쓰기 작업에는 사용자의 명시적인 권한과 확인이 필요합니다.
- 쿼리 허용 목록: 실행할 수 있는 쿼리 유형을 제한합니다. 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 함수를 변환하는 것입니다.
에이전트가 등록할 수 있는 도구에서 일반적입니다. 데코레이터가 자동으로 처리합니다.
독스트링에서 도구 정의를 생성하고 힌트를 입력하고, 입력을 검증하고,
오류가 발생하고 출력 형식을 지정합니다.
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 에이전트를 변환하는 메커니즘입니다. 현실 세계에서 활동할 수 있는 활동적인 오케스트레이터. 도구 호출의 품질은 다음과 같습니다. 세 개의 기둥으로: 정확한 정의 자세한 JSON 스키마를 사용하여 엄격한 검증 오류와 공격을 방지하기 위한 입력 e 강력한 오류 처리 적절한 재시도 및 폴백을 사용합니다.
REST API, GraphQL 및 데이터베이스를 안전하게 통합하는 방법, 구축하는 방법을 살펴보았습니다. 동적 검색 기능을 갖춘 재사용 가능한 도구 프레임워크 및 운영 관리 방법 비동기 실행 및 진행률 보고를 통한 장기 실행. 이러한 패턴은 안정적이고 확장 가능한 프로덕션 에이전트를 구축하기 위한 기반입니다.
다음 기사에서는 이에 대해 다루겠습니다. AI 에이전트 테스트: 테스트 방법 도구 호출 흐름, 모델 응답 시뮬레이션 방법, 품질 측정 방법 에이전트의 결정과 회귀 테스트를 구현하여 다음을 보장하는 방법 변경 사항으로 인해 예기치 않은 동작이 발생하지 않습니다.