Llamada a herramientas (Function Calling)

Function Calling (llamada a funciones) es el mecanismo central que permite a los LLM utilizar herramientas externas.

Function Calling (llamada a funciones) permite que el modelo decida cuándo llamar a una herramienta, a qué herramienta llamar y qué parámetros pasar.

¿Por qué se necesita Function Calling?

Imagina que el LLM es un asesor inteligente pero indefenso. Sabe mucho, pero no puede:

• Obtener información en tiempo real (como el clima actual, precios de acciones)
• Realizar cálculos (como operaciones matemáticas complejas)
• Operar sistemas externos (como enviar correos, leer y escribir archivos)

Function Calling le da manos al LLM, permitiéndole superar sus limitaciones y realizar tareas reales.

Cómo funciona Function Calling

El flujo básico de Function Calling es el siguiente:

Conceptos clave

• Definición de herramienta: describe la funcionalidad, los parámetros y el valor de retorno de una herramienta.
• Selección de herramienta: el LLM selecciona la herramienta adecuada según la pregunta del usuario.
• Extracción de parámetros: el LLM extrae de la pregunta los parámetros requeridos por la herramienta.
• Procesamiento de resultados: integra los resultados de la ejecución de la herramienta en la respuesta final.

Ejemplo ilustrativo

Supongamos que tenemos una herramienta de consulta del clima, cuando el usuario pregunta¿Cómo está el clima hoy en Beijing?:

1. El LLM identifica que necesita llamar a la herramienta de consulta del clima.
2. Extrae los parámetros de la pregunta:city="北京", date="今天"
3. Llama a la API del clima para obtener datos.
4. Integra los datos meteorológicos en una respuesta amigable y la devuelve al usuario.

Definición de herramientas y redacción de descripciones

Para que el LLM use las herramientas correctamente, primero es necesario definirlas claramente. Una buena definición de herramienta debe ser como un manual claro que permita al LLM entender:

• ¿Qué hace esta herramienta?
• ¿Cuándo usarla?
• ¿Qué parámetros necesita?
• ¿Cuál es el formato de los parámetros?

Estructura de la definición de una herramienta

Una definición completa de herramienta normalmente incluye las siguientes partes:

Escribir descripciones de herramientas de alta calidad

1. La descripción debe ser clara y específica

• Mala descripción: consultar el clima
• Buena descripción: obtener la información meteorológica de una ciudad específica en una fecha determinada, incluyendo temperatura, humedad, velocidad del viento y condiciones climáticas (despejado, lluvia, nublado, etc.).

2. La descripción de los parámetros debe ser detallada

Mala descripción de parámetros：

"city": {"type": "string"}

Buena descripción de parámetros:

"city": {
    "type": "string",
    "description": "完整的城市名称，如'北京市'、'上海市'。不要使用简称或拼音。"
}

3. Usar enumeraciones para limitar las opciones

Para opciones limitadas, usa enumeraciones (enum) para ayudar al LLM a entender:

"unit": {
    "type": "string",
    "description": "温度单位",
    "enum": ["celsius", "fahrenheit"],
    "default": "celsius"
}

4. Proporcionar valores de ejemplo

Proporciona ejemplos en la descripción para ayudar al LLM a entender el formato:

"date": {
    "type": "string",
    "description": "日期，格式应为'YYYY-MM-DD'，例如'2024-06-15'"
}

Ejemplos prácticos de definiciones de herramientas

1. Herramienta calculadora

calculator_tool = {
    "name": "calculate",
    "description": "执行数学计算，支持加减乘除、幂运算等基本运算",
    "parameters": {
        "type": "object",
        "properties": {
            "expression": {
                "type": "string",
                "description": "数学表达式，如'2 + 3 * 4'、'sqrt(16)'、'sin(30)'"
            }
        },
        "required": ["expression"]
    }
}

2. Herramienta de búsqueda

search_tool = {
    "name": "search_web",
    "description": "在互联网上搜索最新信息",
    "parameters": {
        "type": "object",
        "properties": {
            "query": {
                "type": "string",
                "description": "搜索关键词，尽量具体明确"
            },
            "num_results": {
                "type": "integer",
                "description": "返回结果数量，默认为5",
                "default": 5,
                "minimum": 1,
                "maximum": 10
            }
        },
        "required": ["query"]
    }
}

3. Herramienta de operaciones de archivos

file_tool = {
    "name": "read_file",
    "description": "读取指定文件的内容",
    "parameters": {
        "type": "object",
        "properties": {
            "file_path": {
                "type": "string",
                "description": "文件的完整路径，如'/home/user/document.txt'"
            },
            "encoding": {
                "type": "string",
                "description": "文件编码，默认为'utf-8'",
                "default": "utf-8"
            }
        },
        "required": ["file_path"]
    }
}

Mejores prácticas para la definición de herramientas

• Nombre conciso y claro: comienza con un verbo, comoget_、calculate_、search_
• Descripción completa y detallada: explica el propósito, los escenarios de uso y las limitaciones de la herramienta.
• Validación de parámetros suficiente: define el tipo, el rango y los requisitos de formato de los parámetros.
• Proporcionar valores predeterminados: proporciona valores predeterminados razonables para parámetros opcionales.
• Considerar situaciones de error: indica en la descripción los posibles errores y limitaciones.

Validación de la definición de herramientas

Después de definir una herramienta, se deben realizar pruebas de validación:

def validate_tool_definition(tool_def):
    """验证工具定义是否完整"""
    required_fields = ["name", "description", "parameters"]

    for field in required_fields:
        if field not in tool_def:
            return False, f"缺少必要字段: {field}"

    # 检查参数结构
    if "properties" not in tool_def["parameters"]:
        return False, "参数定义缺少properties字段"

    return True, "工具定义完整"

# 测试验证
is_valid, message = validate_tool_definition(weather_tool)
print(f"验证结果: {is_valid}, 消息: {message}")

Extracción y validación de parámetros

Una vez que el LLM extrae los parámetros de la pregunta del usuario, es necesario validarlos y procesarlos para garantizar que la herramienta pueda ejecutarse correctamente.

Proceso de extracción de parámetros

Cuando el LLM decide llamar a una herramienta, analiza la entrada del usuario y extrae los parámetros requeridos por la herramienta:

用户输入: "查询北京明天的天气温度"
工具: get_weather
提取参数: {"city": "北京", "date": "明天"}

Desafíos de la extracción de parámetros

La extracción de parámetros puede enfrentar los siguientes problemas:

• Falta de información: el usuario no proporcionó toda la información necesaria.
• Formato no conforme: el formato proporcionado por el usuario no coincide con el requerido por la herramienta.
• Ambigüedad: la misma información puede tener múltiples interpretaciones.
• Dependencia del contexto: los parámetros deben entenderse en conjunto con el historial de la conversación.

Métodos de validación de parámetros

1. Validación de tipo

Asegura que el tipo de parámetro cumpla con los requisitos:

def validate_parameters(params, tool_def):
    """验证参数类型"""
    errors = []

    for param_name, param_def in tool_def["parameters"]["properties"].items():
        if param_name in params:
            param_value = params[param_name]
            expected_type = param_def.get("type")

            # 类型检查
            if expected_type == "string" and not isinstance(param_value, str):
                errors.append(f"参数'{param_name}'应为字符串类型")
            elif expected_type == "integer" and not isinstance(param_value, int):
                errors.append(f"参数'{param_name}'应为整数类型")
            elif expected_type == "number" and not isinstance(param_value, (int, float)):
                errors.append(f"参数'{param_name}'应为数字类型")
            elif expected_type == "boolean" and not isinstance(param_value, bool):
                errors.append(f"参数'{param_name}'应为布尔类型")

    return errors

2. Validación de rango

Comprueba si el valor del parámetro se encuentra dentro del rango permitido:

def validate_range(params, tool_def):
    """验证参数范围"""
    errors = []

    for param_name, param_def in tool_def["parameters"]["properties"].items():
        if param_name in params:
            param_value = params[param_name]

            # 检查最小值
            if "minimum" in param_def and param_value < param_def["minimum"]:
                errors.append(f"参数'{param_name}'不能小于{param_def['minimum']}")

            # 检查最大值
            if "maximum" in param_def and param_value > param_def["maximum"]:
                errors.append(f"参数'{param_name}'不能大于{param_def['maximum']}")

            # 检查枚举值
            if "enum" in param_def and param_value not in param_def["enum"]:
                errors.append(f"参数'{param_name}'必须是{param_def['enum']}中的一个")

    return errors

3. Validación de parámetros obligatorios

Asegura que todos los parámetros obligatorios estén presentes:

def validate_required(params, tool_def):
    """验证必填参数"""
    errors = []
    required_params = tool_def["parameters"].get("required", [])

    for param_name in required_params:
        if param_name not in params:
            errors.append(f"缺少必填参数: {param_name}")

    return errors

Sistema completo de validación de parámetros

Flujo completo de extracción y validación de parámetros

Estrategias de corrección de parámetros

Cuando la validación de parámetros falla, se pueden adoptar las siguientes estrategias:

• Preguntar al usuario：Preguntar directamente al usuario por los parámetros faltantes o incorrectos
• Usar valores predeterminados：Para parámetros opcionales, usar valores predeterminados predefinidos
• Inferencia inteligente：Inferir valores de parámetros razonables según el contexto
• Conversión de formato：Convertir el formato proporcionado por el usuario al formato requerido por la herramienta

def fix_parameters(params, errors, tool_def):
    """尝试修正参数错误"""
    fixed_params = params.copy()

    for error in errors:
        if "缺少必填参数" in error:
            param_name = error.split(": ")[1]
            # 尝试从上下文推断或使用默认值
            if param_name == "date":
                fixed_params[param_name] = "今天"  # 使用当天作为默认值

        elif "应为字符串类型" in error:
            param_name = error.split("'")[1]
            # 尝试转换为字符串
            fixed_params[param_name] = str(params[param_name])

    return fixed_params

Manejo de errores y mecanismos de reintento

En aplicaciones reales, las llamadas a herramientas pueden fallar. Un manejo de errores y mecanismos de reintentos sólidos son clave para construir agentes de IA confiables.

<Errores comunes en la llamada a herramientas

• Error de red：Tiempo de espera agotado en la llamada a la API, fallo de conexión
• Error de parámetros：Fallo de validación de parámetros, formato incorrecto
• Error de permisos：Clave de API inválida, permisos insuficientes
• Error de recursos：Servicio no disponible, se alcanzó el límite de llamadas
• Error lógico：Error lógico interno de la herramienta

Estrategias de manejo de errores

1. Manejo de errores por niveles

Adoptar diferentes estrategias de manejo según el tipo de error:

class ErrorHandler:
    """错误处理器"""

    def handle(self, error, context):
        """处理错误"""
        error_type = self.classify_error(error)

        if error_type == "network":
            return self.handle_network_error(error, context)
        elif error_type == "parameter":
            return self.handle_parameter_error(error, context)
        elif error_type == "authentication":
            return self.handle_auth_error(error, context)
        elif error_type == "resource":
            return self.handle_resource_error(error, context)
        else:
            return self.handle_unknown_error(error, context)

    def classify_error(self, error):
        """分类错误类型"""
        error_str = str(error).lower()

        if any(word in error_str for word in ["timeout", "connection", "network"]):
            return "network"
        elif any(word in error_str for word in ["parameter", "invalid", "missing"]):
            return "parameter"
        elif any(word in error_str for word in ["auth", "key", "permission"]):
            return "authentication"
        elif any(word in error_str for word in ["limit", "quota", "unavailable"]):
            return "resource"
        else:
            return "unknown"

    def handle_network_error(self, error, context):
        """处理网络错误"""
        return {
            "success": False,
            "error": "网络连接失败，请检查网络后重试",
            "retryable": True,
            "retry_after": 5  # 5秒后重试
        }

    def handle_parameter_error(self, error, context):
        """处理参数错误"""
        return {
            "success": False,
            "error": f"参数错误: {str(error)}",
            "retryable": False,
            "suggestion": "请检查输入参数是否正确"
        }

    # 其他错误处理方法类似...

2. Mensajes de error amigables para el usuario

Convertir errores técnicos en mensajes que el usuario pueda entender:

def user_friendly_error(error):
    """生成用户友好的错误消息"""
    error_map = {
        "timeout": "请求超时，请稍后重试",
        "connection_error": "网络连接失败，请检查网络",
        "invalid_api_key": "API 密钥无效，请检查配置",
        "rate_limit_exceeded": "调用频率过高，请稍后再试",
        "invalid_parameters": "输入参数不正确，请检查后重试"
    }

    error_str = str(error).lower()
    for key, message in error_map.items():
        if key in error_str:
            return message

    return "系统繁忙，请稍后重试"

Mecanismo de reintento

Para errores que se pueden reintentar, se debe implementar una estrategia inteligente de reintentos:

1. Reintento con retroceso exponencial

import time
import random

class ExponentialBackoffRetry:
    """指数退避重试"""

    def __init__(self, max_retries=3, base_delay=1, max_delay=30):
        self.max_retries = max_retries
        self.base_delay = base_delay
        self.max_delay = max_delay

    def retry(self, func, *args, **kwargs):
        """执行带重试的函数调用"""
        last_error = None

        for attempt in range(self.max_retries):
            try:
                return func(*args, **kwargs)

            except Exception as e:
                last_error = e

                # 检查是否可重试
                if not self.is_retryable(e):
                    raise

                # 最后一次尝试，不再重试
                if attempt == self.max_retries - 1:
                    break

                # 计算等待时间（指数退避 + 随机抖动）
                delay = min(
                    self.base_delay * (2 ** attempt) + random.uniform(0, 1),
                    self.max_delay
                )

                print(f"第{attempt + 1}次尝试失败，{delay:.1f}秒后重试...")
                time.sleep(delay)

        # 所有重试都失败
        raise last_error

    def is_retryable(self, error):
        """判断错误是否可重试"""
        error_str = str(error).lower()
        retryable_errors = ["timeout", "connection", "busy", "temporarily"]

        for retryable_error in retryable_errors:
            if retryable_error in error_str:
                return True

        return False

# 使用示例
retry = ExponentialBackoffRetry(max_retries=3)

def call_weather_api(city, date):
    """调用天气API（模拟可能失败）"""
    # 模拟API调用
    if random.random() < 0.3:  # 30%概率失败
        raise ConnectionError("API连接超时")
    return {"temperature": 25, "condition": "晴"}

try:
    result = retry.retry(call_weather_api, "北京", "今天")
    print(f"成功获取天气: {result}")
except Exception as e:
    print(f"获取天气失败: {e}")

2. Patrón de disyuntor

Evitar que fallos consecutivos provoquen una avalancha del sistema:

class CircuitBreaker:
    """熔断器"""

    def __init__(self, failure_threshold=5, reset_timeout=60):
        self.failure_threshold = failure_threshold
        self.reset_timeout = reset_timeout
        self.failure_count = 0
        self.last_failure_time = None
        self.state = "closed"  # closed, open, half-open

    def execute(self, func, *args, **kwargs):
        """通过熔断器执行函数"""
        if self.state == "open":
            # 检查是否应该尝试恢复
            if self.should_try_reset():
                self.state = "half-open"
            else:
                raise Exception("熔断器开启，服务暂时不可用")

        try:
            result = func(*args, **kwargs)

            # 成功调用，重置状态
            if self.state == "half-open":
                self.state = "closed"
                self.failure_count = 0

            return result

        except Exception as e:
            self.record_failure()
            raise e

    def record_failure(self):
        """记录失败"""
        self.failure_count += 1
        self.last_failure_time = time.time()

        if self.failure_count >= self.failure_threshold:
            self.state = "open"

    def should_try_reset(self):
        """检查是否应该尝试重置"""
        if self.last_failure_time is None:
            return True

        elapsed = time.time() - self.last_failure_time
        return elapsed > self.reset_timeout

# 使用示例
breaker = CircuitBreaker(failure_threshold=3, reset_timeout=30)

def unreliable_service():
    """模拟不可靠的服务"""
    if random.random() < 0.5:
        raise Exception("服务异常")
    return "服务正常"

for i in range(10):
    try:
        result = breaker.execute(unreliable_service)
        print(f"调用{i+1}: {result}")
    except Exception as e:
        print(f"调用{i+1}: {e}")
        time.sleep(1)

Marco completo de llamada a herramientas

Combinando todos los componentes anteriores, podemos construir un marco completo de invocación de herramientas: