从零解析skill调用第三方API的实现原理与最佳实践

8次阅读

共计 2454 个字符，预计需要花费 7 分钟才能阅读完成。

当我们开发一个 skill（技能）时，比如聊天机器人、智能助手等，经常会遇到需要调用外部服务的情况。这些外部服务通过 API（应用程序接口）提供各种功能，比如天气查询、翻译服务、支付接口等。通过集成这些第三方 API，我们可以快速实现复杂功能，而不需要从头开发所有功能。

举个例子，如果你正在开发一个旅行助手 skill，你可能需要调用航班查询 API、酒店预订 API、天气 API 等。这样你就能在短时间内构建一个功能丰富的应用，而不需要自己收集和处理所有这些数据。

在调用第三方 API 时，身份认证是第一步。最常见的两种认证方式是 API Key 和 OAuth2.0：

API Key：最简单的方式，一般是一个长字符串，随请求一起发送
优点：实现简单
缺点：安全性较低，Key 泄露后可能被滥用
OAuth2.0：更安全的认证流程
优点：支持权限控制，可以撤销访问
缺点：实现复杂度高

一个典型的 API 调用流程如下：

应用向 API 服务发送请求
API 服务验证请求的认证信息
API 服务处理请求
API 服务返回响应
应用处理响应数据

API 调用可能会失败，常见的错误处理模式包括：

重试机制：对临时性错误自动重试
回退机制：主 API 不可用时切换到备用 API
熔断机制：连续失败时暂时停止调用，避免雪崩效应

下面是一个完整的 Python 示例，展示如何调用天气 API 并处理各种异常情况：

import requests
from requests.exceptions import RequestException
import time

# API 配置
API_URL = "https://api.weatherapi.com/v1/current.json"
API_KEY = "your_api_key_here"
MAX_RETRIES = 3
RETRY_DELAY = 1  # 秒

def get_weather(city):
    params = {
        'key': API_KEY,
        'q': city
    }

    for attempt in range(MAX_RETRIES):
        try:
            # 发送请求
            response = requests.get(API_URL, params=params, timeout=5)

            # 检查响应状态码
            response.raise_for_status()

            # 解析 JSON 响应
            data = response.json()
            return data

        except RequestException as e:
            print(f"请求失败 ( 尝试 {attempt + 1}/{MAX_RETRIES}): {str(e)}")
            if attempt < MAX_RETRIES - 1:
                time.sleep(RETRY_DELAY)
            else:
                raise Exception(f"获取天气数据失败: {str(e)}")

# 使用示例
try:
    weather_data = get_weather("Beijing")
    print(f"当前温度: {weather_data['current']['temp_c']}°C")
except Exception as e:
    print(f"错误: {str(e)}")

这段代码展示了几个关键点：

使用 API Key 进行认证
实现了重试机制
处理了各种网络和 API 错误
设置了合理的超时时间

频繁创建和关闭 HTTP 连接会影响性能。使用连接池可以显著提高性能：

import requests
from requests.adapters import HTTPAdapter

# 创建 Session 并配置连接池
session = requests.Session()
adapter = HTTPAdapter(pool_connections=10, pool_maxsize=100, max_retries=3)
session.mount('http://', adapter)
session.mount('https://', adapter)

# 使用 Session 发送请求
response = session.get('https://api.example.com/data')

第三方 API 通常有调用频率限制。我们需要实现限流和熔断机制：

限流：控制请求速率不超过 API 限制
熔断：当错误率超过阈值时，暂时停止调用

在处理 API 响应时，要注意保护敏感数据：

import re

def sanitize_response(response):
    # 脱敏 API Key
    if 'api_key' in response:
        response['api_key'] = '***REDACTED***'

    # 脱敏信用卡号
    if 'credit_card' in response:
        response['credit_card'] = re.sub(r'\d', '*', response['credit_card'])

    return response