Tool-Aufruf

Function Calling ist der Kernmechanismus, der es LLMs ermöglicht, externe Tools zu verwenden.

Function Calling ermöglicht es dem Modell zu entscheiden, wann es Tools aufruft, welches Tool es aufruft und welche Parameter übergeben werden.

Warum wird Function Calling benötigt?

Stellen Sie sich vor, ein LLM ist ein kluger, aber unbewaffneter Berater. Es weiß viel, kann aber nicht:

• Echtzeitinformationen abrufen (wie aktuelle Wetterdaten, Aktienkurse)
• Berechnungen durchführen (wie komplexe mathematische Operationen).
• Externe Systeme bedienen (wie E-Mails senden, Dateien lesen/schreiben)

Function Calling verleiht dem LLM Hände, sodass es seine eigenen Grenzen durchbrechen und reale Aufgaben ausführen kann.

Das Funktionsprinzip von Function Calling.

Der grundlegende Ablauf von Function Calling ist wie folgt:

Kernkonzept

• Tool-DefinitionBeschreiben Sie die Funktion, die Parameter und den Rückgabewert eines Tools.
• Tool-Auswahl: LLM wählt das geeignete Tool basierend auf der Frage des Benutzers aus.
• Parameterextraktion: : Das LLM extrahiert die vom Tool benötigten Parameter aus der Frage.
• ErgebnisverarbeitungDie Ergebnisse der Tool-Ausführung in die endgültige Antwort integrieren

Beispielerklärung

Angenommen, wir haben ein Wetterabfrage-Tool. Wenn ein Benutzer fragtWie ist das Wetter heute in Beijing?Zeit:

1. Das LLM erkennt, dass es das Wetterabfrage-Tool aufrufen muss
2. Parameter aus der Frage extrahieren:city="北京", date="今天"
3. Die Wetter-API aufrufen, um Daten abzurufen
4. Integriert die Wetterdaten in eine freundliche Antwort und gibt sie an den Benutzer zurück

Tool-Definition und Beschreibung verfassen

Damit das LLM das Tool korrekt verwenden kann, muss das Tool zuerst klar definiert werden. Eine gute Tool-Definition sollte wie eine klare Bedienungsanleitung sein, die das LLM verstehen lässt:

• Was macht dieses Tool?
• Wann sollte es verwendet werden?
• Welche Parameter werden benötigt?
• Wie ist das Format der Parameter?

Die Struktur von Tool-Definitionen.

Eine vollständige Tool-Definition enthält normalerweise die folgenden Teile:

Verfassen Sie hochwertige Tool-Beschreibungen.

1. Machen Sie die Beschreibung klar und spezifisch

• Schlechte Beschreibung.: Wetter abfragen
• Gute Beschreibung: Ruft die Wetterinformationen für eine angegebene Stadt an einem bestimmten Datum ab, einschließlich Temperatur, Luftfeuchtigkeit, Windgeschwindigkeit und Wetterbedingungen (sonnig, regnerisch, bewölkt usw.)

2. Machen Sie die Parameterbeschreibungen detailliert

Schlechte Parameterbeschreibung：

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

Gute Parameterbeschreibung:

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

3. Verwenden Sie Enums, um Optionen einzuschränken

Verwenden Sie bei begrenzten Optionen Enums, um dem LLM das Verständnis zu erleichtern:

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

4. Geben Sie Beispielwerte an

Geben Sie Beispiele in der Beschreibung an, um dem LLM zu helfen, das Format zu verstehen:

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

Beispiel für eine Utility-Definition

1. Taschenrechner-Tool

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

2. Such-Tool

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. Dateioperationen-Tool

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"]
    }
}

Best Practices für Tool-Definitionen.

• Prägnanter und klarer Name：Verwenden Sie Verben am Anfang, wie z. B.get_、calculate_、search_
• Vollständige und detaillierte Beschreibung: Beschreiben Sie den Zweck des Tools, anwendbare Szenarien und Einschränkungen
• Ausreichende Parametervalidierung: Definieren Sie Parametertypen, Bereiche und Formatanforderungen
• Geben Sie einen Standardwert an.: Geben Sie sinnvolle Standardwerte für optionale Parameter an
• Berücksichtigen Sie Fehlerfälle: Beschreiben Sie mögliche Fehler und Einschränkungen in der Beschreibung

Validierung der Tool-Definition

Nachdem ein Tool definiert wurde, sollten Validierungstests durchgeführt werden:

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}")

Parameterextraktion und -validierung.

Nachdem das LLM Parameter aus der Frage des Benutzers extrahiert hat, müssen diese Parameter validiert und verarbeitet werden, um sicherzustellen, dass das Tool korrekt ausgeführt werden kann.

Prozess der Parameterextraktion

Wenn das LLM entscheidet, ein Tool aufzurufen, analysiert es die Benutzereingabe und extrahiert die vom Tool benötigten Parameter:

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

Herausforderungen bei der Parameterextraktion

Bei der Parameterextraktion können die folgenden Probleme auftreten:

• Fehlende Informationen: Der Benutzer hat nicht alle erforderlichen Informationen angegeben
• Formatabweichung: Das vom Benutzer angegebene Format entspricht nicht den Anforderungen des Tools
• Auflösung von Mehrdeutigkeiten: Dieselbe Information kann mehrere Interpretationen haben
• Kontextabhängigkeit: Parameter müssen im Kontext des Gesprächsverlaufs verstanden werden

Parameter-Validierungsmethode

1. Typvalidierung

Stellen Sie sicher, dass die Parametertypen den Anforderungen entsprechen:

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. Bereichsvalidierung

Prüfen Sie, ob die Parameterwerte innerhalb des erlaubten Bereichs liegen:

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. Validierung erforderlicher Parameter

Stellen Sie sicher, dass alle erforderlichen Parameter angegeben wurden:

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

Vollständiges Parametervalidierungssystem

Der vollständige Workflow für Parameterextraktion und -validierung.

Strategie zur Parameterkorrektur

Wenn die Parameterprüfung fehlschlägt, können folgende Strategien angewendet werden:

• Benutzer fragen: Den Benutzer direkt nach fehlenden oder falschen Parametern fragen
• Standardwerte verwenden: Für optionale Parameter vordefinierte Standardwerte verwenden
• Intelligente Inferenz: Sinnvolle Parameterwerte basierend auf dem Kontext ableiten
• Formatkonvertierung: Das vom Benutzer bereitgestellte Format in das vom Tool benötigte Format umwandeln

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

Fehlerbehandlung und Wiederholungsmechanismus.

In der Praxis können Tool-Aufrufe fehlschlagen. Eine robuste Fehlerbehandlung und ein Wiederholungsmechanismus sind der Schlüssel zum Aufbau eines zuverlässigen KI-Agenten.

Häufige Fehler bei Tool-Aufrufen

• Netzwerkfehler: API-Aufruf-Zeitüberschreitung, Verbindungsfehler
• Parameterfehler: Parameterprüfung fehlgeschlagen, falsches Format
• Berechtigungsfehler: Ungültiger API-Schlüssel, unzureichende Berechtigungen
• Ressourcenfehler: Dienst nicht verfügbar, Aufruflimit erreicht
• Logikfehler: Interne Logikfehler im Tool

Fehlerbehandlungsstrategie

1. Abgestufte Fehlerbehandlung

Je nach Fehlertyp unterschiedliche Behandlungsstrategien anwenden:

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. Benutzerfreundliche Fehlermeldungen

Technische Fehler in für Benutzer verständliche Meldungen umwandeln:

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 "系统繁忙，请稍后重试"

Wiederholungsmechanismus

Für wiederholbare Fehler sollte eine intelligente Wiederholungsstrategie implementiert werden:

1. Exponentielle Backoff-Wiederholung

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. Circuit-Breaker-Muster

Systemlawine durch aufeinanderfolgende Fehler verhindern:

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)

Vollständiges Framework für Tool-Aufrufe

Durch Kombination aller oben genannten Komponenten können wir ein vollständiges Framework für Tool-Aufrufe erstellen: