# 开发指南
> 从原型到上线的工程实践
> 覆盖重试、超时、幂等、流式解析与成本控制
> 示例以 Python 与 JavaScript 为主，其他语言思路一致
## 选接口：同步还是异步
| 类型 | 典型接口 | 返回方式 | 适合 |
|---|---|---|---|
| 同步 | 文本对话、embedding | 一次返回或流式返回 | 交互式问答、批处理 |
| 异步 | 图像、视频、音乐 | 提交拿 task_id，再轮询 | 生成类长任务 |
判断标准很简单：**结果能不能在一分钟内出来**。能就用同步，不能就用异步。
## 超时与重试
同步接口建议的超时：
| 场景 | 超时 | 重试 |
|---|---|---|
| 非流式对话 | 60 秒 | 最多 2 次 |
| 流式对话 | 首 token 30 秒，总时长 5 分钟 | 首 token 超时才重试 |
| embedding | 30 秒 | 最多 3 次 |
重试只对这几类错误有意义：
- `408`、`429`、`500`、`502`、`503`
- 网络层超时
**不要重试** `400`、`401`、`402`、`403`，这些重试多少次都是一样的结果，只会白花时间。
```python
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())
```
退避用指数加随机抖动，避免多个客户端在同一时刻一起重试，把上游再打垮一次。
## 异步任务的正确轮询
```python
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 的解析坑主要集中在「一次读到的数据不完整」：
```javascript
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` 分别处理
- 成本有监控或告警