幂等与重试

import os
import time
import uuid

import requests

API = "https://api.tokentosea.com/v1"
AUTH = {"Authorization": f"Bearer {os.environ['TOKENTOSEA_API_KEY']}"}
FINAL = {"succeeded", "failed", "needs_review"}
ACCEPTED = {"request_already_accepted", "request_already_dispatched"}


def fetch_result(key, deadline, grace=0):
    """按幂等键查询直到终态，或 planned 超过 5 分钟；持续 grace 秒没有记录时返回 None。"""
    missing_since = planned_since = None
    while time.time() < deadline:
        try:
            r = requests.get(f"{API}/results/{key}", headers=AUTH, timeout=30)
        except requests.RequestException:
            r = None  # 查询本身失败：稍后再查
        if r is not None and r.status_code == 404:
            missing_since = missing_since or time.time()
            if time.time() - missing_since >= grace:
                return None
        elif r is not None and r.ok:
            envelope = r.json()
            if envelope["status"] in FINAL:
                return envelope
            if envelope["status"] != "planned":
                planned_since = None
            elif planned_since is None:
                planned_since = time.time()
            elif time.time() - planned_since > 300:
                return envelope  # 没有发往厂商
        time.sleep(5)
    raise TimeoutError(f"结果仍在处理，稍后用同一幂等键查询：{key}")


def call(path, body, attempts=3):
    key = str(uuid.uuid4())  # 生产环境请先把 key 与业务记录一起持久化
    deadline = time.time() + 30 * 60
    for _ in range(attempts):
        code = None
        try:
            r = requests.post(f"{API}{path}", json=body, headers={**AUTH, "Idempotency-Key": key}, timeout=(10, 1300))
        except requests.RequestException:
            r = None  # 网络错误或超时：结果未知
        if r is not None:
            if r.status_code == 200:
                return {"status": "succeeded", "output": r.json()}
            data = r.json() if "json" in r.headers.get("content-type", "") else {}
            if r.status_code == 422 and "run_id" in data:
                return data  # 原调用已失败；billing_status 为 succeeded 时已扣费
            code = data.get("error", {}).get("code")
            if r.status_code < 500 and r.status_code != 202 and code not in ACCEPTED:
                r.raise_for_status()  # 明确拒绝、没有执行：修正请求后用新键发起
        # 202、5xx、原调用已受理、网络错误：结果未知，先按原键查询
        envelope = fetch_result(key, deadline, grace=60 if code in ACCEPTED else 0)
        if envelope is not None and envelope["status"] != "planned":
            return envelope
        if envelope is not None or code in ACCEPTED:
            key = str(uuid.uuid4())  # 键已被占用，但请求从未发往厂商：换新键
        # 否则平台没有这个键的记录：下一轮用原键原样重发
    raise RuntimeError(f"多次尝试仍无结果，幂等键：{key}")


result = call("/chat/completions", {
    "model": "qwen3.8-flash",
    "messages": [{"role": "user", "content": "你好"}],
    "max_tokens": 128,
})
print(result["status"], result.get("output"))

import { randomUUID } from 'node:crypto';

const API = 'https://api.tokentosea.com/v1';
const AUTH = { Authorization: `Bearer ${process.env.TOKENTOSEA_API_KEY}` };
const FINAL = new Set(['succeeded', 'failed', 'needs_review']);
const ACCEPTED = new Set(['request_already_accepted', 'request_already_dispatched']);
const sleep = (ms) => new Promise((resolve) => setTimeout(resolve, ms));

/** 按幂等键查询直到终态，或 planned 超过 5 分钟；持续 graceMs 没有记录时返回 null。 */
async function fetchResult(key, deadline, graceMs = 0) {
  let missingSince = null;
  let plannedSince = null;
  while (Date.now() < deadline) {
    let response = null;
    try {
      response = await fetch(`${API}/results/${key}`, { headers: AUTH, signal: AbortSignal.timeout(30_000) });
    } catch {
      // 查询本身失败：稍后再查
    }
    if (response?.status === 404) {
      missingSince ??= Date.now();
      if (Date.now() - missingSince >= graceMs) return null;
    } else if (response?.ok) {
      const envelope = await response.json();
      if (FINAL.has(envelope.status)) return envelope;
      plannedSince = envelope.status === 'planned' ? plannedSince ?? Date.now() : null;
      if (plannedSince !== null && Date.now() - plannedSince > 300_000) return envelope; // 没有发往厂商
    }
    await sleep(5_000);
  }
  throw new Error(`结果仍在处理，稍后用同一幂等键查询：${key}`);
}

export async function call(path, body, attempts = 3) {
  let key = randomUUID(); // 生产环境请先把 key 与业务记录一起持久化
  const deadline = Date.now() + 30 * 60_000;
  for (let i = 0; i < attempts; i += 1) {
    let code = null;
    let response = null;
    try {
      response = await fetch(`${API}${path}`, {
        method: 'POST',
        headers: { ...AUTH, 'Content-Type': 'application/json', 'Idempotency-Key': key },
        body: JSON.stringify(body),
        signal: AbortSignal.timeout(1_300_000),
      });
    } catch {
      // 网络错误或超时：结果未知
    }
    if (response) {
      if (response.status === 200) return { status: 'succeeded', output: await response.json() };
      const payload = await response.json().catch(() => ({}));
      if (response.status === 422 && payload.run_id) return payload; // 原调用已失败；billing_status 为 succeeded 时已扣费
      code = payload.error?.code ?? null;
      if (response.status < 500 && response.status !== 202 && !ACCEPTED.has(code)) {
        // 明确拒绝、没有执行：修正请求后用新键发起
        throw new Error(`HTTP ${response.status} ${code ?? ''}`);
      }
    }
    // 202、5xx、原调用已受理、网络错误：结果未知，先按原键查询
    const envelope = await fetchResult(key, deadline, ACCEPTED.has(code) ? 60_000 : 0);
    if (envelope && envelope.status !== 'planned') return envelope;
    if (envelope || ACCEPTED.has(code)) key = randomUUID(); // 键已被占用，但请求从未发往厂商：换新键
    // 否则平台没有这个键的记录：下一轮用原键原样重发
  }
  throw new Error(`多次尝试仍无结果，幂等键：${key}`);
}

const result = await call('/chat/completions', {
  model: 'qwen3.8-flash',
  messages: [{ role: 'user', content: '你好' }],
  max_tokens: 128,
});
console.log(result.status, result.output);