AI调用工具类实战：从设计原则到性能优化

1次阅读

没有评论

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

在实际开发中，我们经常需要调用各种 AI 服务的 API，如文本生成、图像识别等。但直接调用 API 往往会遇到以下问题：

接口设计不规范，不同服务的 API 风格差异大
缺乏统一的错误处理机制
性能瓶颈明显，特别是在高并发场景下
代码重复率高，维护成本大

这类问题使得我们需要一个统一的工具类来封装这些调用，提高开发效率和系统稳定性。

优点：

灵活性高，可以直接控制请求细节
不依赖第三方库

缺点：

需要自己处理各种细节（认证、重试等）
代码重复率高

优点：

官方维护，稳定性有保障
通常包含最佳实践

缺点：

可能过度封装，灵活性降低
不同服务的 SDK 风格不一致

我们推荐的方式是构建自己的工具类，结合两者的优点：

底层可以使用官方 SDK
在上层提供统一的接口
可以加入自定义的逻辑（如缓存、监控等）

一个好的 AI 调用工具类应该：

提供统一的调用接口
内置合理的默认配置
支持自定义覆盖
完善的错误处理
可扩展的架构

主要方法包括：

同步调用
异步调用
批量调用

每种方法都应该处理：

参数验证
认证处理
请求构造
响应解析
错误处理

应该区分：

网络错误
API 错误
业务逻辑错误

并为每种错误提供明确的处理方式。

import requests
from typing import Dict, Any, Optional
from dataclasses import dataclass
from functools import lru_cache

@dataclass
class AIClientConfig:
    api_key: str
    base_url: str = "https://api.example.ai/v1"
    timeout: int = 30
    max_retries: int = 3

class AIClient:
    """AI 服务调用工具类"""

    def __init__(self, config: AIClientConfig):
        self.config = config
        self.session = requests.Session()

    @lru_cache(maxsize=128)
    def _get_auth_headers(self) -> Dict[str, str]:
        """获取认证头信息，使用缓存提高性能"""
        return {"Authorization": f"Bearer {self.config.api_key}",
            "Content-Type": "application/json"
        }

    def call_api(
        self, 
        endpoint: str, 
        payload: Dict[str, Any], 
        method: str = "POST"
    ) -> Dict[str, Any]:
        """
        调用 AI API

        Args:
            endpoint: API 端点路径
            payload: 请求数据
            method: HTTP 方法

        Returns:
            API 响应数据

        Raises:
            AIServiceError: 当 API 调用失败时抛出
        """url = f"{self.config.base_url}/{endpoint}"
        headers = self._get_auth_headers()

        for attempt in range(self.config.max_retries):
            try:
                response = self.session.request(
                    method,
                    url,
                    json=payload,
                    headers=headers,
                    timeout=self.config.timeout
                )
                response.raise_for_status()
                return response.json()

            except requests.exceptions.RequestException as e:
                if attempt == self.config.max_retries - 1:
                    raise AIServiceError(f"API 调用失败: {str(e)}")

    # 其他工具方法...

class AIServiceError(Exception):
    """AI 服务调用异常"""
    pass

对于批量请求，可以使用：

线程池
异步 IO
消息队列

示例代码：

from concurrent.futures import ThreadPoolExecutor

class AIClient:
    # ...

    def batch_call(
        self, 
        endpoint: str, 
        payloads: List[Dict[str, Any]],
        max_workers: int = 5
    ) -> List[Dict[str, Any]]:
        """批量调用 API"""
        with ThreadPoolExecutor(max_workers=max_workers) as executor:
            futures = [executor.submit(self.call_api, endpoint, payload)
                for payload in payloads
            ]
            return [future.result() for future in futures]

根据业务需求选择合适的缓存：