开发指南

  • 从原型到上线的工程实践
  • 覆盖重试、超时、幂等、流式解析与成本控制
  • 示例以 Python 与 JavaScript 为主，其他语言思路一致

选接口：同步还是异步

类型典型接口返回方式适合
同步文本对话、embedding一次返回或流式返回交互式问答、批处理
异步图像、视频、音乐提交拿 task_id，再轮询生成类长任务

判断标准很简单：结果能不能在一分钟内出来。能就用同步，不能就用异步。

超时与重试

同步接口建议的超时：

场景超时重试
非流式对话60 秒最多 2 次
流式对话首 token 30 秒，总时长 5 分钟首 token 超时才重试
embedding30 秒最多 3 次

重试只对这几类错误有意义：

  • 408、429、500、502、503
  • 网络层超时

不要重试 400、401、402、403，这些重试多少次都是一样的结果，只会白花时间。

import time, random, requests

RETRYABLE = {408, 429, 500, 502, 503}

def call_with_retry(url, headers, payload, tries=3):
    for attempt in range(tries):
        try:
            res = requests.post(url, headers=headers, json=payload, timeout=60)
            if res.status_code in RETRYABLE and attempt < tries - 1:
                time.sleep((2 ** attempt) + random.random())
                continue
            return res
        except (requests.Timeout, requests.ConnectionError):
            if attempt == tries - 1:
                raise
            time.sleep((2 ** attempt) + random.random())

退避用指数加随机抖动，避免多个客户端在同一时刻一起重试，把上游再打垮一次。

异步任务的正确轮询

import time

def wait_task(task_id, interval=5, timeout=600):
    deadline = time.time() + timeout
    while time.time() < deadline:
        r = requests.get(
            f"https://api.oairoute.com/v1/tasks/{task_id}",
            headers={"Authorization": "Bearer YOUR_OAIRoute_KEY"},
            timeout=30,
        )
        body = r.json()
        data = body.get("data") or {}

        status = data.get("status")
        if status == "completed":
            return data["result"]
        if status in ("failed", "cancelled"):
            raise RuntimeError(data.get("error", {}).get("message"))

        time.sleep(interval)
    raise TimeoutError(f"任务 {task_id} 超时")

几个要点：

  • 间隔不要小于 5 秒，太密没有意义且浪费额度
  • 设一个总超时，避免无限等待
  • 终态有两个：成功和失败，都要处理

幂等

同一个请求重复发送，不应该产生两次扣费或两个任务。

服务端已有的幂等能力：

场景机制
创建任务传入 Idempotency-Key 头，相同值在 24 小时内只建一个任务
回调接收按 task_id 去重

客户端也要做一层：记录已经提交过的业务主键，重试前先查。

流式解析

SSE 的解析坑主要集中在「一次读到的数据不完整」：

const res = await fetch(url, { method: "POST", headers, body });
const reader = res.body.getReader();
const decoder = new TextDecoder();
let buf = "";

while (true) {
  const { value, done } = await reader.read();
  if (done) break;

  buf += decoder.decode(value, { stream: true });

  // 按空行切分完整的 SSE 事件
  const parts = buf.split("\n\n");
  buf = parts.pop();          // 最后一段可能不完整，留到下次

  for (const part of parts) {
    for (const line of part.split("\n")) {
      if (!line.startsWith("data: ")) continue;
      const payload = line.slice(6);
      if (payload === "[DONE]") return;
      const chunk = JSON.parse(payload);
      process.stdout.write(chunk.choices[0]?.delta?.content ?? "");
    }
  }
}

关键是把剩余的不完整片段留在 buf 里，等下一次数据补齐再解析。

成本控制

用 max_tokens 卡住上限。 不设的话，偶发的超长输出会显著推高成本。

把固定内容前置以命中缓存。 系统提示、产品手册放在最前面且保持不变，见「上下文缓存」相关页面。

按任务选模型。 分类、抽取、改写用不了旗舰模型，小模型便宜一个数量级。

批量时限制并发。 并发过高会触发 429，重试又增加成本。用信号量把并发压在 5 到 10。

上线前检查清单

  • 密钥从环境变量读取，不在代码里硬编码
  • 所有请求都设了超时
  • 只对可重试的错误重试，且带退避
  • 异步任务有总超时，不会永久阻塞
  • 结果文件下载后转存到自己的存储
  • 错误结构按 error.code 与 error.type 分别处理
  • 成本有监控或告警