> ## Documentation Index
> Fetch the complete documentation index at: https://docs.acedata.cloud/llms.txt
> Use this file to discover all available pages before exploring further.

# SDK 任务轮询与流式响应

> Platform 集成指南 - Ace Data Cloud

Ace Data Cloud 上的服务在响应模式上分两类：

| 类型 | 典型服务 | 调用模式 |
| - | - | - |
| **同步生成** | NanoBanana / Flux / Seedream / Chat Completions（非流式）/ Google 搜索 | 一次 HTTP，结果在响应体里 |
| **流式响应** | Chat Completions（`stream: true`） | SSE，多帧逐 token 推送 |
| **异步任务** | Midjourney / Sora / Veo / Luma / Kling / Hailuo / Suno / Pixverse / Seedance | 先创建任务拿 `task_id`，再轮询 `/<provider>/tasks` |

本文重点讲后两类：**异步任务的 TaskHandle 轮询** 和 **chat 流式响应** 的细节、陷阱和跨语言差异。

## 一、TaskHandle —— 异步任务的统一抽象

三种 SDK 都把异步任务封装成 `TaskHandle`，提供同样的 4 个方法：

| 方法 | 行为 |
| - | - |
| `get()` | 拉一次最新状态（`POST /<provider>/tasks {id, action: "retrieve"}`） |
| `is_completed()` / `isCompleted()` | `get()` 一次，看 `status` 是不是 `succeeded` / `failed` |
| `wait()` | 阻塞轮询，直到 `succeeded` / `failed` 或 `max_wait` 超时 |
| `result` 属性 | 上一次 `wait()` 拿到的完整响应；调用前为 `null` |

### 创建任务的两种调用方式

每个异步资源（`images.generate` / `video.generate` / `audio.generate`）都有 `wait` 参数：

* `wait=False`（默认）：立即返回 `TaskHandle`，业务代码自己决定何时轮询。
* `wait=True`：SDK 内部直接调 `handle.wait()`，函数返回完成后的响应。**仅当你确定目标 API 一定会返回 `status: succeeded` 字段时再用**——少数 provider 没遵守这个约定，会让 `wait` 一直转到 `max_wait` 才抛 `TimeoutError`。

### 单位差异（⚠️ 必看）

`poll_interval` 和 `max_wait` 的**单位在三种语言里不一样**，跨语言迁移时是常见踩坑点：

| 语言 | `poll_interval` 单位 | `max_wait` 单位 | 默认值 |
| - | - | - | - |
| **TypeScript** | **毫秒** | **毫秒** | `pollInterval=3000`, `maxWait=600000` |
| **Python** | **秒** | **秒** | `poll_interval=3.0`, `max_wait=600.0` |
| Go | （TaskHandle 尚未在 Go SDK 暴露） | — | — |

> 把 TS 的 `{ pollInterval: 3000 }` 当成秒翻成 Python `poll_interval=3000`，会让 SDK 等 50 分钟才轮询第二次。

### 示例：Python 显式轮询 Midjourney

```python theme={null}
import os, time
from acedatacloud import AceDataCloud

client = AceDataCloud(api_token=os.environ["ACEDATACLOUD_API_TOKEN"])

# wait=False 立刻拿到 handle
handle = client.images.generate(
    provider="midjourney",
    prompt="a cinematic photo of a banana wearing a tuxedo",
    wait=False,
)
print("task_id", handle.id)

t0 = time.time()
result = handle.wait(poll_interval=3.0, max_wait=180.0)
print("elapsed_s", round(time.time() - t0, 1))
print("status", result.get("response", result).get("status"))
print("images", [it.get("image_url") for it in (result.get("response", result).get("data") or [])])
```

整段代码做的是：

1. `images.generate(..., wait=False)` 把 `prompt` 提交给 Midjourney API，立即拿到 `handle`，不阻塞。
2. `handle.wait(poll_interval=3.0, max_wait=180.0)` 内部每 3 秒 POST 一次 `/midjourney/tasks`，直到 `status` 变成 `succeeded` 或 `failed`，或总耗时超过 180 秒抛 `TimeoutError`。
3. 完成后 `result["response"]["data"]` 通常包含 4 张图（Midjourney 默认 2x2 grid）。

### 示例：TypeScript 显式轮询

```ts theme={null}
import { AceDataCloud } from '@acedatacloud/sdk';

const client = new AceDataCloud();

// wait: false 立刻拿到 handle
const handle: any = await client.images.generate({
  provider: 'midjourney',
  prompt: 'a cinematic photo of a banana wearing a tuxedo',
  wait: false,
});
console.log('task_id', handle.id);

const t0 = Date.now();
const result: any = await handle.wait({ pollInterval: 3000, maxWait: 180_000 });
console.log('elapsed_s', ((Date.now() - t0) / 1000).toFixed(1));
const response = result.response ?? result;
console.log('status', response.status);
console.log('images', (response.data ?? []).map((it: any) => it.image_url));
```

### 同步生成 vs 异步任务的取舍

如果你的 provider 本身就是同步出图（NanoBanana / Flux / Seedream），**不要传 `wait`**：

```python theme={null}
# ✅ 推荐
res = client.images.generate(provider="nano-banana", prompt="...")
url = res["data"][0]["image_url"]

# ❌ 反例：会触发 SDK 内部去轮询 /nano-banana/tasks，浪费 RTT
res = client.images.generate(provider="nano-banana", prompt="...", wait=True)
```

判断方法很简单：如果目标 API 文档里**没有** `task_id` + `/tasks` 这一对，就是同步生成；同步生成的响应里 `data` 字段已经包含最终结果。

### TaskHandle 内部协议

`TaskHandle.get()` 调用的是：

```http theme={null}
POST {API_BASE}/<provider>/tasks
Authorization: Bearer {token}
Content-Type: application/json

{"id": "<task_id>", "action": "retrieve"}
```

响应统一结构：

```json theme={null}
{
  "task_id": "...",
  "trace_id": "...",
  "response": {
    "status": "pending | running | succeeded | failed",
    "data": [...]
  }
}
```

SDK 同时兼容**没有外层 `response` 包裹**的旧版响应——直接读取顶层 `status`，所以新旧版响应切换不影响业务代码。

## 二、SSE 流式响应（chat.completions）

`chat.completions.create(stream=True)` 是目前 SDK 里唯一的流式接口（音频 / 视频流尚未支持）。三种语言的迭代风格各自原生：

| 语言 | 迭代 | 取消机制 |
| - | - | - |
| TypeScript | `for await (const chunk of stream)` | `AbortController` 传给 fetch |
| Python | `for chunk in client.openai.chat.completions.create(..., stream=True)` | 跳出循环即可（连接由 SDK 自动关闭） |
| Go | `chunks, errs := ...CreateStream(ctx, req)` → `for chunk := range chunks` | 取消 `context.Context` |

### TypeScript

```ts theme={null}
import { AceDataCloud } from '@acedatacloud/sdk';

const client = new AceDataCloud();

const stream: any = await client.openai.chat.completions.create({
  model: 'gpt-4o-mini',
  messages: [{ role: 'user', content: 'Count from 1 to 5 separated by spaces. Just the numbers.' }],
  max_tokens: 30,
  temperature: 0,
  stream: true
});

let chunks = 0;
let collected = '';
for await (const chunk of stream) {
  chunks++;
  const delta = chunk?.choices?.[0]?.delta?.content;
  if (delta) collected += delta;
}
console.log('chunks', chunks);
console.log('collected', collected);
```

真实运行结果：

```text theme={null}
total_elapsed_ms 2616
first_chunk_ms 2481
chunks 13
collected 1 2 3 4 5
```

### Python

```python theme={null}
import os
from acedatacloud import AceDataCloud

client = AceDataCloud(api_token=os.environ["ACEDATACLOUD_API_TOKEN"])

chunks = 0
collected = []
for chunk in client.openai.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": "Count from 1 to 5 separated by spaces. Just the numbers."}],
    max_tokens=30,
    temperature=0,
    stream=True,
):
    chunks += 1
    delta = (chunk.get("choices") or [{}])[0].get("delta", {}).get("content")
    if delta:
        collected.append(delta)
print("chunks", chunks)
print("collected", "".join(collected))
```

真实运行结果：

```text theme={null}
total_elapsed_ms 2111
first_chunk_ms 2104
chunks 12
collected 1 2 3 4 5
```

### Go

```go theme={null}
chunks, errs := client.OpenAI().Chat().Completions().CreateStream(ctx, adc.ChatCompletionRequest{
    Model:     "gpt-4o-mini",
    Messages:  []map[string]any{{"role": "user", "content": "Count from 1 to 5 separated by spaces. Just the numbers."}},
    MaxTokens: 30,
})
cnt := 0
collected := ""
for chunk := range chunks {
    cnt++
    if ch, ok := chunk["choices"].([]any); ok && len(ch) > 0 {
        if d, ok := ch[0].(map[string]any)["delta"].(map[string]any); ok {
            if s, ok := d["content"].(string); ok {
                collected += s
            }
        }
    }
}
if e, ok := <-errs; ok && e != nil {
    log.Println("stream_err", e)
}
fmt.Println("chunks", cnt, "collected", collected)
```

真实运行结果：

```text theme={null}
total_elapsed_ms 1816
first_chunk_ms 1633
chunks 13
collected 1 2 3 4 5
```

### 流式 chunk 的结构

每一帧 chunk 都是一个 OpenAI 兼容的 `chat.completion.chunk`：

```json theme={null}
{
  "id": "chatcmpl-...",
  "object": "chat.completion.chunk",
  "model": "gpt-4o-mini",
  "choices": [
    {
      "index": 0,
      "delta": {
        "content": " 3"
      },
      "finish_reason": null
    }
  ]
}
```

* 第一个 chunk 通常带 `delta.role: "assistant"` 但 `content` 为空。
* 中间的 chunk 每个带 `delta.content`，可以直接拼接。
* 最后一个 chunk `delta` 为空、`finish_reason` 是 `stop` / `length` / `content_filter`。

### 中途取消

| 语言 | 取消方式 |
| - | - |
| TypeScript | 在 `create()` 调用里传 `signal: abortController.signal`，调 `abortController.abort()` |
| Python | `break` 退出 `for` 循环，SDK 在 `__exit__` 关闭 HTTPx 流 |
| Go | 在 `NewClient` 时传的 `ctx` 上 `cancel()`，`chunks` channel 会立即关闭 |

提前取消已经计费的 token —— 取消时刻之前生成的 token 还是会按实际消耗扣费。

## 三、超时和重试

三种 SDK 共享同一套重试策略：

| 触发条件 | 行为 |
| - | - |
| HTTP 408 / 409 / 429 / 5xx | 默认重试 2 次，指数退避 1s → 2s → 4s |
| 网络层错误（DNS、连接被拒、TLS 失败） | 同上 |
| 401 / 403 / 404 / 422 | 不重试，直接抛对应类型化错误 |
| 流式（`stream=True`）请求 | **不重试**——首帧已经流出时无法回放 |
| 显式 `timeout` 触发 | 抛 `APITimeoutError`（Python）/ `TimeoutError`（TS）/ `context.DeadlineExceeded`（Go） |

要禁用重试：构造客户端时传 `max_retries=0` / `maxRetries: 0` / `WithMaxRetries(0)`。

异步任务（TaskHandle）的轮询本身不受 `max_retries` 影响——它的循环是业务级的而不是 HTTP 级的，靠 `max_wait` 控制总时长。

## 四、常见陷阱

1. **同步 provider 不要传 `wait`**：NanoBanana / Flux / Seedream 都是同步生成，强行 `wait=True` 会让 SDK 去轮询一个根本不会更新的 `tasks` 接口。
2. **TaskHandle 单位差异**：Python 是秒、TS 是毫秒，跨语言移植时一定换算。
3. **`wait=True` 仍可能 `TimeoutError`**：响应必须满足 `status in ('succeeded','failed')` 才会退出循环；如果 provider 用了别的字段名，需要业务代码自己 `handle.get()` 解析。
4. **流式取消**：取消前生成的 token 已经计费。
5. **同一 process 内复用 client**：SDK 自带连接池，频繁 `new AceDataCloud()` / `AceDataCloud()` 会让 TLS 握手成为瓶颈。

## 了解更多

* 📘 [TypeScript SDK 接入教程](https://platform.acedata.cloud/documents/sdk-typescript)
* 🐍 [Python SDK 接入教程](https://platform.acedata.cloud/documents/sdk-python)
* 🟦 [Go SDK 接入教程](https://platform.acedata.cloud/documents/sdk-go)
* 🔌 [SDK + X402 支付钩子](https://platform.acedata.cloud/documents/sdk-x402-payment)
* 📦 [SDK monorepo 源码](https://github.com/AceDataCloud/SDK)


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.