Herramientas de Agente de IA e Integración Externa

Este capítulo presenta cómo los Agentes interactúan con el mundo exterior.

La capacidad de invocar herramientas es clave para la conexión del Agente con el entorno externo.

Al integrar diversas herramientas, el Agente puede ejecutar código, acceder a API, manipular sistemas de archivos, etc.

Computer Use

Computer Use permite que el Agente opere interfaces de computadora como un humano.

Incluye interfaces gráficas como navegadores, aplicaciones de escritorio, etc.

Este es un paso importante hacia la inteligencia artificial general (AGI).

Capacidades principales

Análisis de capturas de pantalla: comprender el contenido en pantalla e identificar elementos interactivos.

Reconocimiento de elementos GUI: identificar elementos de interfaz como botones, campos de entrada, menús, etc.

Control de ratón/teclado: simular operaciones humanas, realizar acciones como clics, escritura, etc.

Automatización de navegador: navegación web, relleno de formularios, búsqueda, etc.

Principio de funcionamiento

El flujo de trabajo de Computer Use se puede resumir en: observar, comprender, decidir y ejecutar.

Primer paso, captura de pantalla. Capturar el contenido de la pantalla o ventana actual.

Segundo paso, análisis visual. Usar modelos de visión para analizar la captura e identificar elementos y estados de la interfaz.

Tercer paso, decisión. Basado en el objetivo de la tarea, determinar la siguiente acción.

Cuarto paso, ejecución. Realizar acciones como clics de ratón, entrada de teclado, etc.

Ejecutar en bucle hasta que la tarea esté completa.

Implementación de código

Implementación básica de Computer Use Agent

class ComputerUseAgent:
    """
Implementación de Computer Use Agent
Capaz de operar interfaces de computadora como un humano
    """
   
    def __init__(self, vision_model, action_executor, planner):
        # Modelo de visión: analizar captura de pantalla
        self.vision_model = vision_model
        # Ejecutor de acciones: realizar operaciones de ratón y teclado
        self.action_executor = action_executor
        # Planificador: decidir la siguiente acción
        self.planner = planner
        # Máximo de pasos de ejecución
        self.max_steps = 100
   
    def observe(self, screenshot):
        """
Analizar la captura de pantalla, identificar elementos interactivos
:param screenshot: imagen de captura de pantalla
:return: lista de elementos de interfaz
        """
        # Usar el modelo de visión para analizar la captura
        analysis = self.vision_model.analyze(screenshot)
       
        # Devolver los elementos de interfaz identificados
        # Cada elemento incluye: tipo, posición, contenido, interactividad
        return analysis.elements
   
    def act(self, action):
        """
Ejecutar acción
:param action: descripción de la acción, como {"type": "click", "x": 100, "y": 200}
        """
        return self.action_executor.execute(action)
   
    def run(self, task):
        """
Ejecutar el bucle de tarea
:param task: descripción de la tarea
:return: resultado de la tarea
        """
        # Inicializar el estado de la tarea
        self.planner.set_task(task)
       
        for step in range(self.max_steps):
            # Primer paso: capturar la pantalla actual
            screenshot = self.get_screen()
           
            # Segundo paso: observar - identificar elementos de interfaz
            elements = self.observe(screenshot)
           
            # Tercer paso: decidir - determinar la siguiente acción
            action = self.planner.decide_action(elements)
           
            # Verificar si la tarea está completa
            if action.is_final:
                return action.result
           
            # Cuarto paso: ejecutar la acción
            self.act(action)
           
            # Opcional: esperar la actualización de la interfaz
            self.wait_for_update()
       
        return "Se alcanzó el límite máximo de pasos"


class VisionModel:
    """Modelo de visión: analizar captura de pantalla"""
   
    def analyze(self, screenshot):
        """
Analizar captura de pantalla
Devolver lista de elementos de interfaz
        """
        # Analizar utilizando modelo multimodal
        prompt = """
Analiza esta captura de pantalla e identifica todos los elementos de interfaz interactivos.
Incluye: botones, campos de entrada, enlaces, menús, etc.

Para cada elemento, proporciona:
1. Tipo (button, input, link, etc.)
2. Posición (coordenadas del cuadro delimitador)
3. Contenido (texto del botón, marcador de posición del campo de entrada, etc.)
4. Interactividad (visible o no, habilitado o no)
"""
        result = self.vision_model.analyze_image(screenshot, prompt)
        return ScreenAnalysisResult(elements=result.elements)


class ActionExecutor:
    """Ejecutor de acciones: realizar operaciones de ratón y teclado"""
   
    def execute(self, action):
        """
Ejecutar acción
:param action: objeto de acción
        """
        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:
    """Planificador: decidir la siguiente acción"""
   
    def __init__(self, llm):
        self.llm = llm
        self.task = None
        self.history = []
   
    def set_task(self, task):
        """Establecer la tarea actual"""
        self.task = task
        self.history = []
   
    def decide_action(self, elements):
        """
Decidir la siguiente acción basándose en el estado actual de la interfaz
        """
        prompt = f"""
Tarea actual: {self.task}

Acciones históricas ejecutadas:
{self.history}

Elementos interactivos en la interfaz actual:
{elements}

Por favor, decide la siguiente acción.
Si la tarea está completa, devuelve is_final=True.
De lo contrario, devuelve la acción a ejecutar (type, coordenadas, parámetros, etc.).
"""
        response = self.llm.generate(prompt)
        return Action.parse(response)

Casos de uso

Computer Use es especialmente adecuado para los siguientes escenarios:

Automatización web: como rellenar formularios automáticamente, extraer datos y realizar operaciones web.

Operación de aplicaciones de escritorio: como abrir archivos, editar documentos y manejar software.

Automatización de pruebas: como ejecutar automáticamente pruebas de interfaz de usuario y pruebas de regresión.

Nota: Computer Use todavía está en desarrollo y presenta problemas como velocidad de ejecución lenta y errores ocasionales. Para escenarios con una API clara, llamar directamente a la API suele ser más eficiente que Computer Use.

Explicación detallada del protocolo MCP

MCP (Model Context Protocol) es un protocolo estándar abierto.

Permite que los modelos de IA se conecten de forma segura con herramientas y fuentes de datos externas.

El objetivo de diseño de MCP es convertirse en el 'puerto USB' del campo de la IA.

Concepto central de diseño

Estandarización: formato de protocolo unificado; diferentes Agentes y herramientas pueden interoperar.

Seguridad: control de permisos claro; el Agente solo puede acceder a recursos autorizados.

Escalabilidad: fácil de añadir nuevas herramientas y fuentes de datos.

Componentes de la arquitectura

MCP Host: entorno host que ejecuta aplicaciones de IA, como Claude Desktop, asistentes de codificación de IA, etc.

MCP Client: cliente que mantiene una conexión 1:1 con el MCP Server.

MCP Server: programa del lado del servidor que proporciona herramientas y recursos.

Flujo de comunicación

Primer paso, establecimiento de la conexión. El Cliente establece conexión con el Servidor e intercambia información de capacidades.

Segundo paso, descubrimiento de herramientas. El Cliente consulta qué herramientas y recursos ofrece el Servidor.

Tercer paso, invocación de herramientas. El Cliente envía una solicitud de invocación, el Servidor la ejecuta y devuelve el resultado.

Cuarto paso, acceso a recursos. El Cliente puede leer y escribir los recursos proporcionados por el Servidor.

Ejemplo de implementación de MCP Server

Implementación de MCP Server

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

# Crear una instancia de MCP Server
app = Server("filesystem")

# Definir herramienta de sistema de archivos
@app.list_tools()
async def list_tools():
    """Listar todas las herramientas disponibles"""
    return [
        Tool(
            name="read_file",
            description="Leer contenido del archivo",
            inputSchema={
                "type": "object",
                "properties": {
                    "path": {
                        "type": "string",
                        "description": "Ruta del archivo"
                    },
                    "encoding": {
                        "type": "string",
                        "description": "Codificación del archivo, por defecto utf-8",
                        "default": "utf-8"
                    }
                },
                "required": ["path"]
            }
        ),
        Tool(
            name="write_file",
            description="Escribir contenido del archivo",
            inputSchema={
                "type": "object",
                "properties": {
                    "path": {
                        "type": "string",
                        "description": "Ruta del archivo"
                    },
                    "content": {
                        "type": "string",
                        "description": "Contenido del archivo"
                    }
                },
                "required": ["path", "content"]
            }
        ),
        Tool(
            name="list_directory",
            description="Listar contenido del directorio",
            inputSchema={
                "type": "object",
                "properties": {
                    "path": {
                        "type": "string",
                        "description": "Ruta del directorio"
                    }
                }
            }
        )
    ]

@app.call_tool()
async def call_tool(name, arguments):
    """Ejecutar invocación de herramienta"""
    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"Unknown tool: {name}")

async def read_file(path, encoding):
    """Leer archivo"""
    with open(path, "r", encoding=encoding) as f:
        content = f.read()
    return [{"type": "text", "text": content}]

async def write_file(path, content):
    """Escribir archivo"""
    with open(path, "w", encoding="utf-8") as f:
        f.write(content)
    return [{"type": "text", "text": "File written successfully"}]

async def list_directory(path):
    """Listar directorio"""
    import os
    entries = os.listdir(path)
    return [{"type": "text", "text": "\n".join(entries)}]

# Ejecutar el servidor
if __name__ == "__main__":
    import mcp.server.stdio
    asyncio.run(mcp.server.stdio.serve(app))

Modos de integración de API

Existen tres formas comunes de que un Agente integre API externas.

REST API es el estilo de API web más utilizado.

GraphQL ofrece capacidades de consulta de datos más flexibles.

gRPC es adecuado para llamadas a procedimientos remotos en escenarios de alto rendimiento.

Integración de API REST

REST (Representational State Transfer) es un estilo de diseño de API web.

Utiliza métodos HTTP (GET, POST, PUT, DELETE) para realizar operaciones.

Implementación del cliente REST API

import requests
from typing import Dict, Any, Optional

class RESTAPIClient:
    """
Cliente REST API
Encapsula solicitudes HTTP y proporciona una interfaz de llamada sencilla
    """
   
    def __init__(self, base_url: str, headers: Optional[Dict] = None):
        """
Inicializar el cliente de API
:param base_url: URL base de la API
:param headers: Cabeceras de solicitud predeterminadas
        """
        self.base_url = base_url.rstrip("/")
        self.session = requests.Session()
       
        # Establecer cabeceras de solicitud predeterminadas
        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]:
        """
Enviar solicitud de API
:param method: Método HTTP (GET, POST, PUT, DELETE)
:param endpoint: Endpoint de la API
:param params: Parámetros de consulta de URL
:param data: Datos del cuerpo de la solicitud
:return: Datos de la respuesta
        """
        url = f"{self.base_url}/{endpoint.lstrip('/')}"
       
        response = self.session.request(
            method=method.upper(),
            url=url,
            params=params,
            json=data,
            **kwargs
        )
       
        # Verificar el código de estado HTTP
        response.raise_for_status()
       
        # Analizar la respuesta
        if response.content:
            return response.json()
        return {}
   
    def get(self, endpoint: str, params: Optional[Dict] = None):
        """Solicitud GET"""
        return self.call("GET", endpoint, params=params)
   
    def post(self, endpoint: str, data: Dict):
        """Solicitud POST"""
        return self.call("POST", endpoint, data=data)
   
    def put(self, endpoint: str, data: Dict):
        """Solicitud PUT"""
        return self.call("PUT", endpoint, data=data)
   
    def delete(self, endpoint: str, params: Optional[Dict] = None):
        """Solicitud DELETE"""
        return self.call("DELETE", endpoint, params=params)


class APIAgent:
    """
    API Agent
Llamar a APIs externas mediante lenguaje natural
    """
   
    def __init__(self, api_client: RESTAPIClient):
        self.api_client = api_client
        self.tool_descriptions = self.define_tools()
   
    def define_tools(self):
        """
Definir las herramientas disponibles para el Agente
Devuelve la descripción de la herramienta para que el LLM entienda cómo llamarla
        """
        return [
            {
                "name": "get_weather",
                "description": "Obtener información meteorológica de una ciudad específica",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "city": {
                            "type": "string",
                            "description": "Nombre de la ciudad, como Beijing, Shanghai"
                        }
                    },
                    "required": ["city"]
                }
            },
            {
                "name": "get_forecast",
                "description": "Obtener el pronóstico del tiempo de una ciudad específica",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "city": {
                            "type": "string",
                            "description": "Nombre de la ciudad"
                        },
                        "days": {
                            "type": "integer",
                            "description": "Días de pronóstico, por defecto 3",
                            "default": 3
                        }
                    },
                    "required": ["city"]
                }
            }
        ]
   
    def execute_tool(self, tool_name: str, arguments: Dict):
        """
Ejecutar la llamada a la herramienta
:param tool_name: Nombre de la herramienta
:param arguments: Argumentos de la herramienta
:return: Resultado de la ejecución
        """
        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"Unknown tool: {tool_name}")

Mejores prácticas de integración de API

Manejo de errores: Manejar con elegancia errores de red, tiempos de espera, servicios no disponibles, etc.

Mecanismo de reintento: Para errores temporales, implementar reintentos con retroceso exponencial

Control de límite de tasa: Respetar los límites de tasa de la API para evitar ser bloqueado

Estrategia de caché: Aplicar caché a los datos solicitados con frecuencia para reducir las llamadas a la API

Consideraciones de seguridad: No codificar información sensible como claves de API; usar variables de entorno para gestionarlas

Resumen del capítulo

Este capítulo presenta las tecnologías clave para que el Agente interactúe con el mundo exterior

Computer UsePermite que el Agente manipule interfaces GUI, adecuado para escenarios que requieren interactuar con aplicaciones gráficas

Protocolo MCPProporciona una forma estandarizada de integración de herramientas; es el "puerto USB" de la era de los Agentes

Integración de APIEs la forma más común de integración con sistemas externos, incluyendo REST, GraphQL, gRPC, etc.

Elegir la forma de integración adecuada depende del escenario específico

Para la operación de interfaces gráficas, Computer Use es una solución universal

Para la integración estandarizada de herramientas, MCP es la dirección de desarrollo futura

Para la integración de APIs web, REST API sigue siendo la opción más popular

Otras extensiones