第二篇:超时和重试的细节 AI接口超时和重试,踩坑细节都在这了 上一篇讲了三个基础配置,这篇展开说超时和重试。
一次请求涉及四种超时 类型 含义 建议值 连接超时 TCP + TLS握手 5-10秒 写入超时 发送请求Body 10秒 读取超时 等待响应完成 60-300秒 连接池超时 等空闲连接 10秒 最容易出问题的是读取超时。模型生成是逐Token的,长回复需要几十秒。读超时小于生成时间,客户端先断。
重试的五个注意事项
- 区分错误类型 python RETRYABLE = {429, 500, 502, 503, 504}
if status in RETRYABLE: retry() else: raise # 400/401/403不重试 2. 不要立即重试 至少等1秒,给上游恢复时间。
设上限 最多三次,超过就放弃并记日志。
注意重复计费 对话接口不幂等。重试一次多扣一次费。所以429/503要退避——上游可能已经处理了请求,只是响应丢了。
加随机抖动 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 生成,内容仅供参考,请仔细甄别