Инструменты AI-агента и внешняя интеграция

В этой главе рассказывается, как агенты взаимодействуют с внешним миром.

Возможность вызова инструментов — ключ к подключению агентов к внешней среде.

Интегрируя различные инструменты, агенты могут выполнять код, получать доступ к API, работать с файловыми системами и многое другое.

Использование компьютера

Использование компьютера позволяет агентам управлять компьютерными интерфейсами, как люди.

Это включает графические интерфейсы, такие как браузеры и настольные приложения.

Это важный шаг к достижению общего искусственного интеллекта (AGI).

Основные возможности

Анализ скриншотов: Понимать содержимое на экране и определять интерактивные элементы.

Распознавание элементов графического интерфейса: распознавание элементов интерфейса, таких как кнопки, поля ввода и меню.

Управление мышью/клавиатурой: имитация действий человека, выполнение таких действий, как щелчки и ввод текста.

Автоматизация браузера: навигация по веб-сайтам, заполнение форм, поиск и т. д.

Принцип работы

Рабочий процесс использования компьютера можно кратко описать так: наблюдать, понимать, принимать решения и выполнять.

Первый шаг — скриншот. Захват содержимого текущего экрана или окна.

Второй шаг — визуальный анализ. Используйте модель компьютерного зрения для анализа скриншота и определения элементов интерфейса и состояний.

Третий шаг — принятие решений. Определите следующее действие на основе цели задачи.

Четвёртый шаг — выполнение. Выполняйте такие действия, как щелчки мышью и ввод с клавиатуры.

Повторяйте цикл, пока задача не будет выполнена.

Реализация кода

Базовая реализация агента использования компьютера

class ComputerUseAgent:
    """
Реализация агента использования компьютера
Способен работать с компьютерными интерфейсами, как человек
    """
   
    def __init__(self, vision_model, action_executor, planner):
        # Модель зрения: анализ скриншота
        self.vision_model = vision_model
        # Исполнитель действий: выполнение операций мышью и клавиатурой
        self.action_executor = action_executor
        # Планировщик: определить следующее действие
        self.planner = planner
        # Максимальное количество шагов выполнения
        self.max_steps = 100
   
    def observe(self, screenshot):
        """
Разобрать скриншот, определить интерактивные элементы
:param screenshot: изображение скриншота
:return: список элементов интерфейса
        """
        # Использовать модель компьютерного зрения для анализа скриншота
        analysis = self.vision_model.analyze(screenshot)
       
        # Возвращает распознанные элементы интерфейса
        # Каждый элемент включает: type, location, content, interactivity
        return analysis.elements
   
    def act(self, action):
        """
Выполнить действие
:param action: Описание действия, например, {"type": "click", "x": 100, "y": 200}
        """
        return self.action_executor.execute(action)
   
    def run(self, task):
        """
Запустить цикл задачи
:param task: Описание задачи
:return: Результат задачи
        """
        Инициализировать состояние задачи
        self.planner.set_task(task)
       
        for step in range(self.max_steps):
            # Шаг 1: Захватить текущий экран
            screenshot = self.get_screen()
           
            # Шаг 2: Наблюдать - Определить элементы интерфейса
            elements = self.observe(screenshot)
           
            # Шаг 3: Решить - Определить следующее действие
            action = self.planner.decide_action(elements)
           
            # Проверить, завершена ли задача
            if action.is_final:
                return action.result
           
            # Шаг 4: Выполнить действие
            self.act(action)
           
            # Необязательно: дождаться обновления интерфейса
            self.wait_for_update()
       
        return "Достигнут максимальный предел шагов"


class VisionModel:
    """Модель зрения: анализ скриншота"""
   
    def analyze(self, screenshot):
        """
Анализ скриншота
Возвращает список элементов интерфейса
        """
        # Использовать мультимодальную модель для анализа
        prompt = """
Проанализируйте этот скриншот и определите все интерактивные элементы интерфейса.
Включая: кнопки, поля ввода, ссылки, меню и т. д.

Для каждого элемента, пожалуйста, укажите:
1. Тип (кнопка, поле ввода, ссылка и т. д.)
2. Расположение (координаты ограничивающего прямоугольника)
3. Содержимое (текст кнопки, подсказка поля ввода и т. д.)
4. Интерактивность (видимость, состояние доступности)
"""
        result = self.vision_model.analyze_image(screenshot, prompt)
        return ScreenAnalysisResult(elements=result.elements)


class ActionExecutor:
    """Исполнитель действий: выполняет операции мышью и клавиатурой"""
   
    def execute(self, action):
        """
Выполнить действие
:param action: объект Action
        """
        if action.type == "click":
            self.mouse.click(action.x, action.y)
        elif action.type == "type":
            self.keyboard.type_text(action.text)
        elif action.type == "scroll":
            self.mouse.scroll(action.direction, action.amount)
        elif action.type == "press":
            self.keyboard.press_key(action.key)
       
        return ActionResult(success=True)


class Planner:
    """Планировщик: определить следующее действие"""
   
    def __init__(self, llm):
        self.llm = llm
        self.task = None
        self.history = []
   
    def set_task(self, task):
        """Установить текущую задачу"""
        self.task = task
        self.history = []
   
    def decide_action(self, elements):
        """
Определить следующее действие на основе текущего состояния интерфейса
        """
        prompt = f"""
Текущая задача: {self.task}

История выполненных действий:
{self.history}

Интерактивные элементы на текущем интерфейсе:
{elements}

Пожалуйста, определите следующее действие.
Если задача выполнена, верните is_final=True.
В противном случае верните действие, которое нужно выполнить (тип, координаты, параметры и т. д.).
"""
        response = self.llm.generate(prompt)
        return Action.parse(response)

Сценарии применения

Computer Use особенно подходит для следующих сценариев:

Веб-автоматизация: например, автоматическое заполнение форм, сбор данных и выполнение веб-операций.

Операции с настольными приложениями: например, открытие файлов, редактирование документов и работа с программным обеспечением.

Автоматизация тестирования: например, автоматическое выполнение UI-тестов и регрессионных тестов.

Примечание: Computer Use всё ещё находится в разработке и имеет такие проблемы, как медленное выполнение и периодические ошибки. Для сценариев с понятным API прямой вызов API обычно эффективнее, чем Computer Use.

Подробное объяснение протокола MCP

MCP (Model Context Protocol) — это открытый стандартный протокол.

Он позволяет ИИ-моделям безопасно подключаться к внешним инструментам и источникам данных.

Цель дизайна MCP — стать «USB-интерфейсом» в области ИИ.

Основные принципы проектирования

Стандартизация: Унифицированный формат протокола, позволяющий разным агентам и инструментам взаимодействовать.

Безопасность: Чёткий контроль разрешений, агенты могут получать доступ только к авторизованным ресурсам.

Расширяемость: Легко добавлять новые инструменты и источники данных.

Компоненты архитектуры

MCP Host: Хост-среда для запуска AI-приложений, таких как Claude Desktop, AI-ассистенты для программирования и т.д.

MCP Client: Клиент, который поддерживает соединение 1:1 с MCP Server.

MCP Server: Серверная программа, предоставляющая инструменты и ресурсы.

Поток взаимодействия

Первый шаг — установка соединения. Клиент устанавливает соединение с Сервером и обменивается информацией о возможностях.

Второй шаг — обнаружение инструментов. Клиент запрашивает у Сервера доступные инструменты и ресурсы.

Третий шаг — вызов инструмента. Клиент отправляет запрос на вызов инструмента, а Сервер выполняет его и возвращает результат.

Четвёртый шаг — доступ к ресурсам. Клиент может читать и записывать ресурсы, предоставляемые Сервером.

Пример реализации MCP Server

Реализация MCP Server

from mcp.server import Server
from mcp.types import Tool, Resource
import asyncio

# Создать экземпляр MCP Server
app = Server("filesystem")

# Определить инструменты файловой системы
@app.list_tools()
async def list_tools():
    """Перечислить все доступные инструменты"""
    return [
        Инструмент(
            name="read_file",
            description="Прочитать содержимое файла",
            inputSchema={
                "type": "object",
                "properties": {
                    "path": {
                        "type": "string",
                        "description": "Путь к файлу"
                    },
                    "encoding": {
                        "type": "string",
                        "description": "Кодировка файла, по умолчанию utf-8",
                        "default": "utf-8"
                    }
                },
                "required": ["path"]
            }
        ),
        Инструмент(
            name="write_file",
            description="Записать содержимое файла",
            inputSchema={
                "type": "object",
                "properties": {
                    "path": {
                        "type": "string",
                        "description": "Путь к файлу"
                    },
                    "content": {
                        "type": "string",
                        "description": "Содержимое файла"
                    }
                },
                "required": ["path", "content"]
            }
        ),
        Инструмент(
            name="list_directory",
            description="Список содержимого каталога",
            inputSchema={
                "type": "object",
                "properties": {
                    "path": {
                        "type": "string",
                        "description": "Путь к каталогу"
                    }
                }
            }
        )
    ]

@app.call_tool()
async def call_tool(name, arguments):
    """Выполнить вызов инструмента"""
    if name == "read_file":
        return await read_file(arguments["path"], arguments.get("encoding", "utf-8"))
    elif name == "write_file":
        return await write_file(arguments["path"], arguments["content"])
    elif name == "list_directory":
        return await list_directory(arguments["path"])
    else:
        raise ValueError(f"Неизвестный инструмент: {name}")

async def read_file(path, encoding):
    """Прочитать файл"""
    with open(path, "r", encoding=encoding) as f:
        content = f.read()
    return [{"type": "text", "text": content}]

async def write_file(path, content):
    """Записать файл"""
    with open(path, "w", encoding="utf-8") as f:
        f.write(content)
    return [{"type": "text", "text": "Файл успешно записан"}]

async def list_directory(path):
    """Вывести содержимое каталога"""
    import os
    entries = os.listdir(path)
    return [{"type": "text", "text": "\n".join(entries)}]

# Запуск сервера
if __name__ == "__main__":
    import mcp.server.stdio
    asyncio.run(mcp.server.stdio.serve(app))

Паттерны интеграции API

Существует три распространённых способа интеграции внешних API агентом.

REST API — наиболее часто используемый стиль Web API.

GraphQL предоставляет более гибкие возможности запросов к данным.

gRPC подходит для удалённых вызовов процедур в высокопроизводительных сценариях.

Интеграция REST API

REST (Representational State Transfer) — это стиль проектирования веб-API.

Он использует HTTP-методы (GET, POST, PUT, DELETE) для операций.

Реализация клиента REST API

import requests
from typing import Dict, Any, Optional

class RESTAPIClient:
    """
Клиент REST API
Инкапсулирует HTTP-запросы и предоставляет лаконичный интерфейс вызова
    """
   
    def __init__(self, base_url: str, headers: Optional[Dict] = None):
        """
Инициализировать клиент API
:param base_url: базовый URL API
:param headers: заголовки запроса по умолчанию
        """
        self.base_url = base_url.rstrip("/")
        self.session = requests.Session()
       
        # Установить заголовки запроса по умолчанию
        default_headers = {
            "Content-Type": "application/json",
            "Accept": "application/json"
        }
        if headers:
            default_headers.update(headers)
        self.session.headers.update(default_headers)
   
    def call(
        self,
        method: str,
        endpoint: str,
        params: Optional[Dict] = None,
        data: Optional[Dict] = None,
        **kwargs
    ) -> Dict[Any, Any]:
        """
Отправить API-запрос
:param method: HTTP-метод (GET, POST, PUT, DELETE)
:param endpoint: конечная точка API
:param params: параметры запроса URL
:param data: данные тела запроса
:return: данные ответа
        """
        url = f"{self.base_url}/{endpoint.lstrip('/')}"
       
        response = self.session.request(
            method=method.upper(),
            url=url,
            params=params,
            json=data,
            **kwargs
        )
       
        # Проверить код состояния HTTP
        response.raise_for_status()
       
        # Разобрать ответ
        if response.content:
            return response.json()
        return {}
   
    def get(self, endpoint: str, params: Optional[Dict] = None):
        """GET-запрос"""
        return self.call("GET", endpoint, params=params)
   
    def post(self, endpoint: str, data: Dict):
        """POST-запрос"""
        return self.call("POST", endpoint, data=data)
   
    def put(self, endpoint: str, data: Dict):
        """PUT-запрос"""
        return self.call("PUT", endpoint, data=data)
   
    def delete(self, endpoint: str, params: Optional[Dict] = None):
        """DELETE-запрос"""
        return self.call("DELETE", endpoint, params=params)


class APIAgent:
    """
    API-агент
Вызов внешних API с помощью естественного языка
    """
   
    def __init__(self, api_client: RESTAPIClient):
        self.api_client = api_client
        self.tool_descriptions = self.define_tools()
   
    def define_tools(self):
        """
Определить инструменты, доступные агенту
Вернуть описания инструментов, чтобы LLM понимала, как их вызывать
        """
        return [
            {
                "name": "get_weather",
                "description": "Получить информацию о погоде для указанного города",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "city": {
                            "type": "string",
                            "description": "Название города, например, Пекин, Шанхай"
                        }
                    },
                    "required": ["city"]
                }
            },
            {
                "name": "get_forecast",
                "description": "Получить прогноз погоды для указанного города",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "city": {
                            "type": "string",
                            "description": "Название города"
                        },
                        "days": {
                            "type": "integer",
                            "description": "Количество дней прогноза, по умолчанию 3",
                            "default": 3
                        }
                    },
                    "required": ["city"]
                }
            }
        ]
   
    def execute_tool(self, tool_name: str, arguments: Dict):
        """
Выполнить вызов инструмента
:param tool_name: имя инструмента
:param arguments: аргументы инструмента
:return: результат выполнения
        """
        if tool_name == "get_weather":
            return self.api_client.get(f"/weather/{arguments['city']}")
        elif tool_name == "get_forecast":
            params = {"days": arguments.get("days", 3)}
            return self.api_client.get(f"/forecast/{arguments['city']}", params=params)
        else:
            raise ValueError(f"Неизвестный инструмент: {tool_name}")

Лучшие практики интеграции API

Обработка ошибок: корректно обрабатывайте сетевые ошибки, тайм-ауты, недоступность сервиса и т. д.

Механизм повторных попыток: для временных ошибок реализуйте повторные попытки с экспоненциальной задержкой.

Ограничение скорости: соблюдайте ограничения скорости API, чтобы избежать блокировки.

Стратегия кэширования: кэшируйте часто запрашиваемые данные, чтобы сократить количество вызовов API.

Вопросы безопасности: не встраивайте в код конфиденциальную информацию, такую как ключи API; используйте переменные окружения для управления ими.

Краткое содержание главы

В этой главе представлены основные технологии, позволяющие агентам взаимодействовать с внешним миром.

Использование компьютераПозволяет агентам работать с интерфейсами GUI, подходит для сценариев, требующих взаимодействия с графическими приложениями.

Протокол MCPПредоставляет стандартизированный способ интеграции инструментов, служа «USB-интерфейсом» эпохи агентов.

Интеграция APIЯвляется самым распространённым способом интеграции внешних систем, включая REST, GraphQL, gRPC и т. д.

Выбор подходящего способа интеграции зависит от конкретного сценария.

Для операций с графическим интерфейсом Computer Use — универсальное решение.

Для стандартизированной интеграции инструментов MCP — направление будущего развития.

Для интеграции через Web API REST API остаётся наиболее распространённым выбором.

Другие расширения