api最佳实践系列六篇-2、第二篇:超时和重试的细节

API工程 实践api最佳实践系列六篇-2、第二篇openstarry.com

第二篇:超时和重试的细节 AI接口超时和重试,踩坑细节都在这了 上一篇讲了三个基础配置,这篇展开说超时和重试。

一次请求涉及四种超时 类型 含义 建议值 连接超时 TCP + TLS握手 5-10秒 写入超时 发送请求Body 10秒 读取超时 等待响应完成 60-300秒 连接池超时 等空闲连接 10秒 最容易出问题的是读取超时。模型生成是逐Token的,长回复需要几十秒。读超时小于生成时间,客户端先断。

重试的五个注意事项

  1. 区分错误类型 python RETRYABLE = {429, 500, 502, 503, 504}

if status in RETRYABLE: retry() else: raise # 400/401/403不重试 2. 不要立即重试 至少等1秒,给上游恢复时间。

  1. 设上限 最多三次,超过就放弃并记日志。

  2. 注意重复计费 对话接口不幂等。重试一次多扣一次费。所以429/503要退避——上游可能已经处理了请求,只是响应丢了。

  3. 加随机抖动 python wait = (2 ** attempt) + random.uniform(0, 1) 避免多个客户端同时重试造成二次冲击。

完整实现 python import time import random import logging

logger = logging.getLogger(name)

class AIClient: RETRYABLE = {429, 500, 502, 503, 504} MAX_RETRIES = 3

def __init__(self):
    self.client = OpenAI(
        base_url="https://api.openstarry.com/v1",
        api_key="sk-xxx",
        timeout=httpx.Timeout(connect=10.0, read=120.0)
    )

def chat(self, messages, model="glm-5.2"):
    for attempt in range(self.MAX_RETRIES + 1):
        try:
            return self.client.chat.completions.create(
                model=model, messages=messages
            )
        except Exception as e:
            status = getattr(e, 'status_code', None)
            if status not in self.RETRYABLE or attempt == self.MAX_RETRIES:
                raise
            wait = (2 ** attempt) + random.uniform(0, 1)
            logger.warning(f"重试 {attempt+1}/{self.MAX_RETRIES},等待{wait:.1f}s")
            time.sleep(wait)

第三篇:流式输出和并发控制 流式输出 + 并发控制,一次讲清 流式 vs 非流式 指标 非流式 流式 总耗时 5.2秒 5.2秒 首Token时间 0.8秒 0.8秒 用户感知 等5秒才看到内容 0.8秒开始出字 实际耗时一样。流式让用户不焦虑。

流式中断重连 流式最大的坑:输出一半连接断了。

python def stream_with_reconnect(messages, max_retries=2): collected = ""

for attempt in range(max_retries + 1):
    try:
        stream = client.chat.completions.create(
            model="glm-5.2", messages=messages, stream=True
        )
        for chunk in stream:
            if chunk.choices[0].delta.content:
                collected += chunk.choices[0].delta.content
                yield chunk.choices[0].delta.content
        return
    except Exception:
        if attempt < max_retries:
            messages.append({"role": "assistant", "content": collected})
            messages.append({"role": "user", "content": "继续"})
        else:
            raise

并发控制 429不是bug,是限流保护。

python import time import asyncio

class TokenBucket: def init(self, rate, capacity): self.rate = rate self.capacity = capacity self.tokens = capacity self.last_time = time.monotonic()

async def acquire(self):
    while True:
        now = time.monotonic()
        self.tokens = min(self.capacity, 
            self.tokens + (now - self.last_time) * self.rate)
        self.last_time = now
        if self.tokens >= 1:
            self.tokens -= 1
            return
        await asyncio.sleep((1 - self.tokens) / self.rate)

批处理控制并发数:

python async def batch_process(tasks, max_concurrency=5): sem = asyncio.Semaphore(max_concurrency)

async def bounded(task):
    async with sem:
        return await process(task)

return await asyncio.gather(*[bounded(t) for t in tasks])

第四篇:备用模型和网络排查 主模型挂了怎么办?备用模型 + 网络排查 为什么需要备用模型 上游波动是常态。维护窗口、高峰拥堵、临时故障——单模型总有不可用的时候。

自动Failover python MODEL_CHAIN = ["glm-5.2", "deepseek-v4", "qwen3.7-max"]

def call_with_fallback(messages): for model in MODEL_CHAIN: try: return client.chat.completions.create( model=model, messages=messages ), model except Exception as e: status = getattr(e, 'status_code', None) if status in {400, 401, 402, 403, 404}: raise continue raise RuntimeError("所有模型不可用") 按任务类型配不同的备用链:

python FALLBACK_MAP = { "code": ["glm-5.2", "deepseek-v4"], "chat": ["glm-5.2", "kimi-k2.6", "deepseek-v4"], "writing": ["kimi-k2.6", "glm-5.2"], } 网络排查 检查项 操作 关代理 直连,代理增加延迟且容易断 确认节点 国内用户走国内节点 检查防火墙 企业网络可能切断长连接 DNS预热 启动时先发一个请求 快速诊断:

bash

延迟

ping api.openstarry.com

TLS握手

curl -w "TLS: %{time_appconnect}s\n" -o /dev/null -s
https://api.openstarry.com/v1/models

完整请求

curl -w "总耗时: %{time_total}s\n"
-H "Authorization: Bearer sk-xxx"
-H "Content-Type: application/json"
-d '{"model":"glm-5.2","messages":[{"role":"user","content":"Hi"}],"max_tokens":10}'
https://api.openstarry.com/v1/chat/completions 第五篇:模型选择和成本控制 调AI接口怎么省钱 不同模型价格差很大 模型 输入/1M tokens 输出/1M tokens DeepSeek V4 Flash ¥3.5 ¥21 GLM-5.2 ¥35 ¥175 GPT-4o ¥70 ¥280 简单问题用旗舰模型是浪费。

按任务选模型 任务 推荐 简单问答、分类 DeepSeek Flash 代码生成、复杂推理 GLM-5.2 长文档 Kimi K2.6 中文创作 GLM-5.2、Kimi Function Calling GPT-4o 级联路由 先用便宜模型判断复杂度,简单问题直接答,复杂问题转旗舰。

python def cascade_route(message): # 判断复杂度 judge = client.chat.completions.create( model="deepseek-v4-flash", messages=[{ "role": "system", "content": "判断复杂度,只回复 simple 或 complex" }, {"role": "user", "content": message}], max_tokens=10 )

complexity = judge.choices[0].message.content.strip().lower()

if "simple" in complexity:
    return client.chat.completions.create(
        model="deepseek-v4-flash",
        messages=[{"role": "user", "content": message}]
    ), "fast"
else:
    return client.chat.completions.create(
        model="glm-5.2",
        messages=[{"role": "user", "content": message}]
    ), "premium"

1000次调用,70%简单问题。全用旗舰模型成本¥170,级联路由¥53。省68%。

语义缓存 重复问题不重复计费。有客户反馈开启后账单少58%。Dashboard里打开就行,不用改代码。

总结 方法 省钱幅度 难度 按任务选模型 20-40% 低 级联路由 60%+ 中 语义缓存 50%+ 零 第六篇:全场景配置速查 AI接口调用配置速查表 前五篇的汇总。按场景查配置。

超时配置 场景 连接 读取 客服Bot 5s 30s 一般对话 5s 60s 代码/长文 10s 120s 超长文本 10s 300s 重试配置 状态码 动作 429 读Retry-After头,等待 500/502/503/504 指数退避1s→2s→4s,最多3次 400/401/403/404 不重试 模型选择 任务 模型 简单问答 DeepSeek Flash 代码/推理 GLM-5.2 长文档 Kimi K2.6 工具调用 GPT-4o 全场景组合 场景 超时 流式 重试 并发 模型 客服Bot 5/30s 开 3次 按需 DeepSeek Flash AI写作 10/120s 开 3次 3-5 GLM-5.2 IDE编程 10/120s 开 2次 1-3 GLM-5.2 后台批处理 10/300s 关 3次 5-10 按任务选 数据分析 10/180s 可选 3次 1-3 GLM-5.2 RAG问答 5/60s 开 3次 5-10 DeepSeek V4 实时语音 3/15s 必须开 2次 1 GPT-4o 三件事 读超时 ≥ 120秒

stream: true

加重试

这三件事搞定,大部分问题消失。

六篇改完了。每篇控制在核心干货,去掉套话,保留代码和表格。需要我继续调整某篇的细节,或者把这个风格套到其他文档上,随时说。

是不是可以加一篇关于合规重要性的文章 好的,这个角度很对。合规问题是很多开发者和企业在选AI API平台时实际关心但网上资料又比较少的话题。我按同样的风格写一篇。

第七篇:合规问题 用境外AI模型做产品,合规这关怎么过 做AI产品的朋友最近应该都注意到一个变化:平台对模型来源的合规要求越来越严了。

说下我了解到的情况和实际做法。

出了什么问题 去年《生成式人工智能服务管理暂行办法》开始执行。简单说就是:在中国境内向公众提供生成式AI服务的,模型需要备案。

Claude、ChatGPT、Gemini这几个目前没有在国内完成备案。也就是说,如果你做的是面向国内用户的商业产品,直接调这些模型的API,有合规风险。

我知道的几家做AI应用的公司,最近都在把境外的模型切到国产模型上。

实际影响 不是说不让用。几种情况:

使用场景 风险 个人学习、测试 基本不受影响 内部工具(不对外) 风险较低 面向国内用户的商业产品 需要重视 处理敏感数据(金融、医疗等) 合规是硬要求 开发者怎么应对 方案一:直接用国产模型

GLM、DeepSeek、Kimi、Qwen这些都已完成备案。能力上,几个头部模型在大部分场景下跟境外模型差距不大了,中文任务甚至更舒服。

方案二:通过聚合平台做统一管理

我目前的做法。用一个聚合平台的接口,底层模型可以灵活切换。好处是:

一套代码,模型可以随时换

平台已经在技术层面做了合规拦截

国内节点延迟也低

搜 OpenStarry 能找到,这类聚合平台专门处理了模型合规和加速的问题,对开发者来说省事不少。

方案三:混合架构

国内用户请求走国产模型,海外用户走境外模型。同一个接口,按用户区域路由。

python def route_by_region(user_region): if user_region == "CN": return "glm-5.2" else: return "gpt-4o" 企业用户特别注意 如果你是公司采购,几个点:

发票和合同:确认对方是国内注册公司,能开增值税发票。这是最基本的企业采购合规要求。

数据处理:确认平台对请求内容的处理方式。是否有存储?存储多久?是否出境?

合同条款:要求对方在服务协议中明确数据不出境、不用作训练。

我当时确认的几个信息:

运营主体:南京星核向量人工智能科技有限公司

ICP备案:苏ICP备2026032202号

数据策略:请求内容不持久化存储,元数据30天自动清除

总结 合规不是可选项,尤其当你的产品有用户、有收入之后。几个建议:

面向国内用户的产品,优先用已备案的国产模型

如果必须用境外模型,做好区域路由和合规隔离

企业采购前确认运营主体、发票、数据策略

聚合平台可以省掉一部分合规工作,但自己要核实

这是一个开发者视角的合规说明,不是法律建议。具体情况建议咨询专业法务。

这篇可以放在系列第七篇,或者单独发。需要调整措辞、增加具体条款引用、或者改成其他角度,随时说。

本回答由 AI 生成,内容仅供参考,请仔细甄别