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

1. 背景痛点：为什么需要封装 AI 工具调用？

直接调用 AI 服务 API 时，开发者常遇到这些问题：

• 接口碎片化 ：不同厂商的 API 设计差异大（如 OpenAI 用messages 列表，Claude 用 prompt 字符串）
• 版本管理混乱：v1/v2/beta 接口混用导致兼容性问题
• 错误处理缺失：网络波动或服务限流时直接崩溃
• 性能瓶颈：同步请求导致 I / O 阻塞，无法发挥硬件性能

2. 技术选型：HTTP 客户端库对比

推荐选择：
– 简单场景：httpx（同步 / 异步双模式）
– 高并发生产环境：aiohttp

3. 核心实现

3.1 统一接口封装（含类型注解）

from typing import Literal, TypedDict
import httpx

class AIRequest(TypedDict):
    provider: Literal['openai', 'claude', 'cohere']
    endpoint: str
    payload: dict
    timeout: float = 30.0

class AIResponse(TypedDict):
    status_code: int
    data: dict
    latency: float

def call_ai(request: AIRequest) -> AIResponse:
    """统一处理不同 AI 厂商的 API 调用"""
    # 实际项目中应使用依赖注入配置 client
    with httpx.Client() as client:
        resp = client.post(url=f"{BASE_URLS[request['provider']]}/{request['endpoint']}",
            json=request['payload'],
            timeout=request['timeout']
        )
        return {
            'status_code': resp.status_code,
            'data': resp.json(),
            'latency': resp.elapsed.total_seconds()}

3.2 异步批处理实现

import asyncio
from collections import defaultdict

async def batch_call(requests: list[AIRequest]) -> dict[str, list[AIResponse]]:
    """并发处理多个 AI 请求，按服务商分组"""
    async with httpx.AsyncClient() as client:
        tasks = [_single_call(client, req) for req in requests]
        results = await asyncio.gather(*tasks, return_exceptions=True)

        # 分组处理结果
        output = defaultdict(list)
        for req, resp in zip(requests, results):
            output[req['provider']].append(resp if not isinstance(resp, Exception) 
                else {'error': str(resp)}
            )
        return output

async def _single_call(client: httpx.AsyncClient, request: AIRequest):
    """单次请求包装器"""
    try:
        resp = await client.post(f"{BASE_URLS[request['provider']]}/{request['endpoint']}",
            json=request['payload'],
            timeout=request['timeout']
        )
        resp.raise_for_status()
        return resp.json()
    except Exception as e:
        return e

3.3 错误重试机制

import random
from math import exp

async def call_with_retry(
    request: AIRequest, 
    max_retries: int = 3
) -> AIResponse:
    """带指数退避的重试机制"""
    base_delay = 1.0  # 初始延迟 1 秒
    for attempt in range(max_retries + 1):
        try:
            return await _single_call(request)
        except (httpx.NetworkError, httpx.HTTPStatusError) as e:
            if attempt == max_retries:
                raise

            # 指数退避 + 随机抖动
            delay = base_delay * exp(attempt) + random.uniform(0, 0.5)
            await asyncio.sleep(delay)

4. 性能优化实战

4.1 连接池配置

# aiohttp 最佳实践
conn = aiohttp.TCPConnector(
    limit=100,  # 最大连接数
    limit_per_host=20,  # 单主机连接数
    enable_cleanup_closed=True,  # 自动清理关闭的连接
    force_close=False  # 禁用强制关闭（保持长连接）)

4.2 结果缓存

from datetime import timedelta
from aiocache import cached, SimpleMemoryCache

@cached(ttl=timedelta(minutes=5), cache=SimpleMemoryCache)
async def get_cached_response(prompt: str) -> dict:
    """缓存相同 prompt 的响应结果"""
    return await call_ai({
        'provider': 'openai',
        'endpoint': 'chat/completions',
        'payload': {'messages': [{'role': 'user', 'content': prompt}]}
    })

4.3 性能测试数据

测试环境：
– 机器：AWS c5.2xlarge (8vCPU/16GB)
– Python 3.10 + aiohttp 3.8

5. 避坑指南

5.1 API 密钥安全

错误做法：

# 直接硬编码在代码中（会被 Git 扫描到）API_KEY = "sk-xxxxxx"

正确方案：
1. 环境变量（推荐）：

export OPENAI_KEY="sk-xxxxxx"

import os
os.environ["OPENAI_KEY"]

2. 密钥管理服务（如 AWS Secrets Manager）

5.2 速率限制处理

def check_rate_limit(headers: dict) -> bool:
    """通过响应头识别限流"""
    return int(headers.get('x-ratelimit-remaining', 1)) <= 0

# 使用示例
resp = await client.get(...)
if check_rate_limit(resp.headers):
    await asyncio.sleep(float(resp.headers['retry-after']))

5.3 大模型提示工程

• 结构化输出：要求模型返回 JSON 格式

  请用 JSON 格式回答，包含 title 和 summary 字段

• 温度系数：创造性任务用 0.7，确定性任务用 0.2
• 停止序列：避免多余输出

  payload = {"stop": ["\n###", "<|endoftext|>"]
  }

6. 扩展为中间件架构

from typing import Protocol

class AIMiddleware(Protocol):
    async def __call__(self, request: AIRequest, next_middleware) -> AIResponse:
        ...

class RetryMiddleware:
    def __init__(self, max_retries: int = 3):
        self.max_retries = max_retries

    async def __call__(self, request, next_middleware):
        for attempt in range(self.max_retries + 1):
            try:
                return await next_middleware(request)
            except Exception:
                if attempt == self.max_retries:
                    raise
                await asyncio.sleep(2 ** attempt)

# 使用示例
middleware_stack = [LoggingMiddleware(),
    RetryMiddleware(),
    CacheMiddleware()]

# 通过闭包构建调用链
handler = build_middleware_chain(middleware_stack, final_handler)

总结

通过本文介绍的技术方案，我们实现了：
1. 统一接入层：屏蔽不同 AI 服务的接口差异
2. 高性能处理：异步 IO 达到每秒 800+ 请求
3. 生产级健壮性：自动重试、限流处理和密钥安全
4. 可扩展架构：中间件模式支持灵活扩展

实际项目中，建议根据业务需求选择合适的技术组合。例如简单内部工具可以直接用 httpx 同步调用，而面向用户的生产系统推荐aiohttp+ 连接池 + 中间件架构。