Что такое Tool System в Harness (defineTool, registry, JSON schema validation, rate limiting)?

Краткий тезис

Tool System в фреймворке Harness — это центральный механизм (слой 4, harness-one/tools), который позволяет агенту взаимодействовать с внешними API, функциями и базами данных через единый интерфейс. Он включает defineTool — декоратор для регистрации инструмента с описанием параметров на JSON Schema, Registry — каталог всех доступных инструментов, JSON Schema validation — автоматическую проверку аргументов перед вызовом, и rate limiting — контроль частоты вызовов для предотвращения перегрузки. Эта система делает агента безопасным, предсказуемым и расширяемым.

1. Термин: Tool System (инструментальная система)

Tool System — это компонент архитектуры Agentic RAG, который управляет жизненным циклом инструментов — от объявления до выполнения. Инструмент — это любая функция, которую агент может вызвать: поиск по векторной БД, вычисление, запрос к SQL, вызов внешнего API. Без Tool System агент не знал бы, какие инструменты существуют, как их вызывать и какие ограничения на вызовы наложены.

В Harness Tool System располагается в пакете harness-one/tools и состоит из четырёх ключевых частей:

Каждый инструмент после регистрации становится доступным агенту через унифицированный интерфейс call_tool(name, args).

2. defineTool — декоратор для регистрации

defineTool — это декоратор Python, который берёт обычную функцию и оборачивает её в объект Tool, добавляя метаданные (имя, описание, JSON Schema параметров). Пример:

from harness_one.tools import defineTool, ToolRegistry

@defineTool(
    name="web_search",
    description="Выполняет поиск в Google и возвращает список ссылок",
    parameters={
        "type": "object",
        "properties": {
            "query": {
                "type": "string",
                "description": "Поисковый запрос"
            },
            "num_results": {
                "type": "integer",
                "description": "Количество результатов (1-10)",
                "minimum": 1,
                "maximum": 10,
                "default": 5
            }
        },
        "required": ["query"]
    }
)
def web_search(query: str, num_results: int = 5) -> list[dict]:
    # реальная реализация вызова API
    return [{"title": "...", "url": "..."} for _ in range(num_results)]

Что происходит «под капотом»

• Декоратор создаёт объект Tool с полями name, description, parameters_schema, fn.
• Вызывает ToolRegistry.register(tool) (см. следующий раздел).
• Функция после обёртки сохраняет своё поведение, но агент будет вызывать её только через ToolRegistry.call().

Расшифровка терминов

• JSON Schema — стандарт описания структуры JSON-данных. Позволяет задать тип, обязательность, ограничения (min, max, enum). Используется для валидации аргументов.
• Декоратор — синтаксический сахар Python, позволяющий модифицировать функцию или класс. Здесь @defineTool(...) заменяет исходную функцию на объект Tool, но сохраняет её как атрибут .fn.

3. Registry — каталог инструментов

Registry (каталог) — это глобальный объект-синглтон, реализованный как словарь {name: Tool}. После регистрации инструмент можно получить по имени:

from harness_one.tools import ToolRegistry

# Доступные инструменты
print(ToolRegistry.list_tools())  # ["web_search", "calculate", "sql_query"]

# Вызов через registry
result = ToolRegistry.call("web_search", {"query": "погода Москва", "num_results": 3})

Роль Registry в Agentic RAG

• Интроспекция — LLM-агент получает список всех инструментов с их описаниями (сериализованными в JSON Schema) при формировании промпта.
• Динамическое добавление / удаление — можно регистрировать инструменты во время выполнения (например, подключать внешние плагины).
• Централизованное логирование — вызовы проходят через один метод, что упрощает мониторинг (сколько раз вызван инструмент X?).

Пример сериализованного описания (то, что видит LLM):

{
  "name": "web_search",
  "description": "Выполняет поиск в Google",
  "parameters": {
    "type": "object",
    "properties": {
      "query": {"type": "string"},
      "num_results": {"type": "integer", "minimum": 1, "maximum": 10}
    },
    "required": ["query"]
  }
}

Агент формирует вызов в формате JSON, содержащем name и arguments. Registry парсит этот JSON, валидирует и передаёт в реализацию.

4. JSON Schema validation — проверка аргументов

JSON Schema validation — это механизм, который до выполнения инструмента проверяет, соответствуют ли переданные аргументы описанной схеме. Он встроен в ToolRegistry.call() и выполняется автоматически.

Зачем нужна валидация

• Безопасность — агент (LLM) может «галлюцинировать» параметры: передать строку вместо числа, отрицательное число для num_results, отсутствие обязательного поля. Валидация отлавливает это до вызова реального API.
• Предсказуемость — инструмент получает только корректные данные, не нужно писать внутреннюю проверку.
• Обратная связь — если валидация не прошла, агенту возвращается понятная ошибка: «параметр query не может быть пустым». Агент может перепланировать действие.

Реализация в Harness (упрощённо):

import jsonschema
from jsonschema import validate, ValidationError

class ToolRegistry:
    _tools: dict[str, Tool] = {}

    @classmethod
    def call(cls, name: str, args: dict) -> Any:
        tool = cls._tools[name]
        try:
            validate(instance=args, schema=tool.parameters_schema)
            return tool.fn(**args)
        except ValidationError as e:
            raise ToolArgumentError(f"Ошибка валидации для {name}: {e.message}")

Расшифровка термина

• JSON Schema validation — процесс проверки JSON-объекта на соответствие JSON Schema. Использует библиотеки вроде jsonschema (Python) или ajv (JavaScript).

5. Rate limiting — ограничение частоты вызовов

Rate limiting (ограничение частоты) — механизм, не позволяющий агенту вызывать один и тот же инструмент слишком часто за короткий промежуток времени. Это защищает как самого агента (от бесконечных циклов), так и внешние API (от превышения лимитов).

Как реализовано в Harness

Чаще всего используется токен-ба́кет (token bucket algorithm):

• Каждому инструменту назначается лимит: rate = 10 (вызовов в секунду), burst = 20 (максимальный краткосрочный всплеск).
• Токены добавляются в bucket со скоростью rate (например, 10 в секунду).
• На каждый вызов тратится 1 токен. Если токенов нет — вызов блокируется и возвращается ошибка RateLimitExceeded.

Пример настройки в defineTool

@defineTool(
    name="web_search",
    rate_limit={"rate": 5, "burst": 10}  # 5 запросов/с с пиком до 10
)
def web_search(query: str) -> list:
    ...

Как rate limiting интегрируется с агентом:

• Агент получает информацию о лимитах (или не получает — это внутренняя деталь).
• Если лимит превышен, ToolRegistry.call() возвращает специальную ошибку.
• Оркестратор (слой 3) может принять решение: подождать и повторить, или выбрать другой инструмент.

Расшифровка термина

• Token bucket — алгоритм, где в «ведро» (bucket) с определённой скоростью капают токены. Каждый вызов забирает токен. Если ведро пусто — вызов запрещён. Пиковый размер ведра (burst) позволяет обрабатывать кратковременные всплески.

6. Интеграция с агентом

Tool System — это не изолированный модуль. Он тесно взаимодействует с другими слоями Harness:

• Слой 1 (LLM) — агент получает JSON список инструментов в системном промпте.
• Слой 2 (Planner) — планировщик выбирает, какой инструмент вызвать, и формирует аргументы (JSON).
• Слой 3 (Orchestrator) — оркестратор вызывает ToolRegistry.call() и обрабатывает результат.
• Слой 4 (Tools) — сама Tool System.

Типичная последовательность:

LLM: "Нужно узнать погоду. Использую инструмент web_search с query='погода'"
Orchestrator: вызов ToolRegistry.call("web_search", {"query":"погода", "num_results":3})
  -> валидация (JSON Schema) : OK
  -> rate limit check : allow (токен взят)
  -> выполнение web_search(query="погода", num_results=3) -> результат
  -> возврат результата LLM

7. Обработка ошибок и идемпотентность

Tool System также может обеспечивать:

• Идемпотентность — для инструментов, выполняющих побочные эффекты (например, запись в БД), можно гарантировать, что повторный вызов с теми же аргументами не навредит (используется поле idempotency_key).
• Время ожидания (timeout) — если инструмент выполняется дольше заданного порога, вызов прерывается и агенту сообщается об ошибке.
• Retry-логика — при временных сбоях (ошибка API) можно автоматически повторить вызов (с учётом rate limit).

Пример расширенного инструмента с таймаутом:

@defineTool(
    name="sql_query",
    parameters={...},
    timeout=10,          # секунд
    idempotent=True       # гарантировано идемпотентно (только SELECT)
)
def sql_query(sql: str) -> list:
    ...

8. Пример полного цикла работы Tool System на Python

Смоделируем упрощённую версию Harness:

# harness_one/tools.py (упрощённая имитация)

import functools
import time
import threading
from typing import Callable, Any

class Tool:
    def __init__(self, name, description, parameters_schema, fn, rate_limit):
        self.name = name
        self.description = description
        self.parameters_schema = parameters_schema
        self.fn = fn
        self.rate_limit = rate_limit
        self._bucket = rate_limit['burst']
        self._last_refill = time.time()
        self._lock = threading.Lock()

    def _acquire_token(self):
        with self._lock:
            now = time.time()
            elapsed = now - self._last_refill
            self._bucket = min(
                self.rate_limit['burst'],
                self._bucket + elapsed * self.rate_limit['rate']
            )
            self._last_refill = now
            if self._bucket >= 1:
                self._bucket -= 1
                return True
            return False

    def call(self, args: dict) -> Any:
        if not self._acquire_token():
            raise RateLimitExceeded(f"Tool {self.name} limit exceeded")
        # валидация (упрощённо)
        self._validate(args)
        return self.fn(**args)

    def _validate(self, args):
        import jsonschema
        jsonschema.validate(instance=args, schema=self.parameters_schema)


class ToolRegistry:
    _tools = {}

    @classmethod
    def register(cls, tool: Tool):
        cls._tools[tool.name] = tool

    @classmethod
    def call(cls, name: str, args: dict) -> Any:
        tool = cls._tools.get(name)
        if not tool:
            raise ToolNotFound(f"Tool {name} not found")
        return tool.call(args)

    @classmethod
    def list_tools(cls):
        return {name: {
            "description": t.description,
            "parameters": t.parameters_schema
        } for name, t in cls._tools.items()}


class RateLimitExceeded(Exception):
    pass

class ToolNotFound(Exception):
    pass


# Декоратор
def defineTool(name, description, parameters, rate_limit=None):
    def decorator(fn):
        tool = Tool(
            name=name,
            description=description,
            parameters_schema=parameters,
            fn=fn,
            rate_limit=rate_limit or {"rate": 100, "burst": 200}
        )
        ToolRegistry.register(tool)
        @functools.wraps(fn)
        def wrapper(*args, **kwargs):
            return fn(*args, **kwargs)
        return wrapper
    return decorator

Использование

@defineTool(name="add", description="Складывает два числа", parameters={
    "type": "object",
    "properties": {"a": {"type": "number"}, "b": {"type": "number"}},
    "required": ["a", "b"]
}, rate_limit={"rate": 10, "burst": 20})
def add(a, b):
    return a + b

# Агент вызывает
result = ToolRegistry.call("add", {"a": 5, "b": 3})  # 8

9. Пет-проект для закрепления

Задача Написать микроверсию Tool System на Python (без фреймворка Harness), используемую агентом на LangGraph или AutoGen.

Инструменты Python, json, jsonschema (или валидация вручную), threading, time.

Шаги:

1. Определить класс Tool с полями name, description, parameters_schema, fn, rate_limit.
2. Реализовать декоратор define_tool (аналогично описанному).
3. Реализовать ToolRegistry (синглтон со словарём).
4. Реализовать валидацию аргументов с помощью jsonschema.validate().
5. Реализовать токен-ба́кет для rate limiting, привязанный к каждому инструменту.
6. Создать 2-3 инструмента: current_time, random_number, fetch_url.
7. Написать скрипт, который эмулирует агента: цикл, читающий JSON-запросы из stdin и вызывающий инструменты.
8. Проверить поведение при превышении rate limit (должна возвращаться ошибка) и при невалидных аргументах.

Ожидаемый результат Работающий агент, который корректно обрабатывает вызовы, валидирует данные и не выходит за лимиты. Это можно использовать как основу для более сложного проекта (например, интеграция с LLM через промпты).

10. Связь с другими вопросами

Навигация

• Предыдущий: 750
• Следующий: 752
• Индекс: 00. Индекс разборов