API 调用最佳实践与性能优化

从并发控制、缓存策略、错误重试到成本优化,全面提升 API 调用效率

📅 2026-05-13 · 🏷️ 性能优化 · ⏱️ 阅读时间 10 分钟

1. 🚀 并发控制与连接管理

高效的并发控制是提升 API 调用吞吐量的关键。不当的并发设置会导致连接池耗尽、请求超时甚至触发限速。

连接池优化

大多数 HTTP 客户端默认使用连接池。建议根据你的应用场景调整连接池参数:

Python 连接池优化示例
from openai import OpenAI
import httpx

# 自定义 HTTP 客户端,优化连接池
http_client = httpx.Client(
    limits=httpx.Limits(
        max_connections=50,           # 最大连接数
        max_keepalive_connections=20    # 最大保持连接数
    ),
    timeout=httpx.Timeout(
        connect=10.0,                 # 连接超时
        read=60.0,                    # 读取超时
        write=60.0,                   # 写入超时
        pool=60.0                      # 连接池超时
    )
)

client = OpenAI(
    api_key="your-api-key-here",
    base_url="https://toenk-api.com/v1",
    http_client=http_client
)

并发请求策略

当需要发送大量请求时,采用合理的并发控制策略:

2. 💾 缓存策略

合理的缓存策略可以减少重复请求、降低延迟、节省成本。TOENK API 原生支持 Prompt Caching,命中后享受高达 75% 的折扣。

Prompt Caching(提示缓存)

当你发送的消息前缀与之前请求相同时,系统会自动命中缓存。缓存命中部分按 cache 价格计费(仅为原价的 25%)。

💡 最佳实践:将 System Prompt 和常见的指令前缀尽量保持一致,以最大化缓存命中率。例如所有请求使用统一的 System Prompt 模板。

结果缓存

对于幂等的请求(相同输入总是预期相同输出),可以本地缓存结果:

Python 本地缓存示例
import hashlib
import json
from functools import lru_cache
from openai import OpenAI

client = OpenAI(api_key="your-key", base_url="https://toenk-api.com/v1")

def _make_key(messages, model):
    return hashlib.md5(
        json.dumps({"model": model, "messages": messages}, sort_keys=True).encode()
    ).hexdigest()

# 使用 LRU 缓存,最多缓存 1000 个结果
@lru_cache(maxsize=1000)
def cached_chat(cache_key, model, messages_json):
    messages = json.loads(messages_json)
    response = client.chat.completions.create(
        model=model,
        messages=messages,
        temperature=0  # 确定性输出,适合缓存
    )
    return response.choices[0].message.content

def chat_with_cache(messages, model="deepseek-chat"):
    key = _make_key(messages, model)
    return cached_chat(key, model, json.dumps(messages, ensure_ascii=False))

3. 🔁 错误重试策略

网络是天生不可靠的。实现稳健的重试机制是生产级应用的基本要求。

指数退避 (Exponential Backoff)

首次重试等待 1-2 秒,每次重试等待时间翻倍,并加入随机抖动(jitter)避免所有客户端同时重试:

指数退避重试实现
import time
import random
from openai import RateLimitError, APIConnectionError, APIError

def retry_with_backoff(func, max_retries=5):
    for attempt in range(max_retries):
        try:
            return func()
        except (RateLimitError, APIConnectionError) as e:
            if attempt == max_retries - 1:
                raise  # 最后一次重试也失败,抛出异常
            # 指数退避 + 随机抖动
            wait = (2 ** attempt) + random.uniform(0, 1)
            print(f"第 {attempt+1} 次重试,等待 {wait:.1f}s...")
            time.sleep(wait)
        except APIError as e:
            # 4xx 错误(除 429 外)不重试
            if e.status_code < 400 or e.status_code >= 500:
                wait = (2 ** attempt) + random.uniform(0, 1)
                print(f"第 {attempt+1} 次重试,等待 {wait:.1f}s...")
                time.sleep(wait)
            elif e.status_code == 401:
                print("API Key 无效,请检查密钥")
                raise
            else:
                raise
    return None

可重试的错误场景

错误码可重试?策略
429 Too Many Requests✅ 是指数退避,尊重 Retry-After 头
500 Internal Server Error✅ 是指数退避,最多 3 次
503 Service Unavailable✅ 是延长等待时间后重试
401 Unauthorized❌ 否检查 API Key
400 Bad Request❌ 否检查请求参数
402 Payment Required❌ 否提示用户充值

4. 💰 成本优化

合理选择模型和优化调用方式,可以在保持质量的同时大幅降低成本。

模型选择策略

场景推荐模型成本
简单对话、闲聊deepseek-chat💰 极低
代码生成、调试deepseek-v4-flash💰 低
复杂推理、分析deepseek-v4-proclaude-opus-4💰💰 中等
多模态、视觉gpt-4o💰💰 中等
大规模批量处理qwen3.6-flashgpt-4o-mini💰 极低

Token 优化技巧

⚡ 成本节省提示:TOENK API 的 deepseek-chatqwen3.6-flash 是性价比极高的模型。简单任务优先使用它们,每月可节省 60-80% 的调用成本。

5. 📊 监控与日志

建立完善的监控体系,及时发现和定位问题:

Python 监控装饰器示例
import time
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("toenk_api")

def monitor_api_call(func):
    def wrapper(*args, **kwargs):
        start = time.time()
        try:
            response = func(*args, **kwargs)
            elapsed = time.time() - start
            logger.info(
                f"模型={response.model or kwargs.get('model','unknown')} "
                f"延迟={elapsed:.2f}s "
                f"输入token={response.usage.prompt_tokens} "
                f"输出token={response.usage.completion_tokens}"
            )
            return response
        except Exception as e:
            elapsed = time.time() - start
            logger.error(f"API调用失败 延迟={elapsed:.2f}s 错误={str(e)}")
            raise
    return wrapper

@monitor_api_call
def chat(client, model, messages):
    return client.chat.completions.create(
        model=model, messages=messages
    )

📝 总结

高效的 API 调用不在于单一技术的应用,而在于系统性地优化每个环节:

遵循这些最佳实践,不仅可以提升应用的稳定性和响应速度,还能显著降低运营成本。赶快在你的 TOENK API 项目中应用起来吧!