DashScope 原生 SDK 与异步任务

阿里云百炼（DashScope）的 Qwen 模型可以像 OpenAI 一样经 OpenAI SDK 调（翻译模式），也可以用 DashScope 原生 SDK 走透传入口 /v1/services/{*rest}。原生 SDK 只需改一个变量：

pip install dashscope requests

import dashscope

dashscope.base_http_api_url = "http://localhost:7890/v1"   # 指到网关
dashscope.api_key = "<你的访问密钥>"

透传 vs 翻译：两路怎么选

入口	行为	适用
/v1/services/{*rest}	透传：请求体原样发 DashScope，不做协议转换	用 DashScope 原生 SDK、或需要 DashScope 特有字段/异步任务
/v1/chat/completions	翻译：OpenAI 格式 ↔ DashScope 格式互译	用 OpenAI SDK，把 Qwen 当 OpenAI 模型调
/v3/{model}/{*rest}	identity 透传（客户端协议 == 目标协议时原样转发）	需要绕过 /v1/services 前缀的客户端

语言调用

from dashscope import Generation

resp = Generation.call(
    model="qwen3.8-max",
    messages=[{"role": "user", "content": "用一句话介绍你自己"}],
)
print(resp.output.choices[0].message.content)

坑：message.content 是数组。DashScope 透传返回的 output.choices[0].message.content 是数组（[{"text": "..."}]），不是字符串，而且可能带 reasoning_content（思考过程）。照老文档把它当字符串直接 print 会打出 [{...}]。取正文要：content[0]["text"] 或 "".join(p.get("text","") for p in content)。

图像（同步接口）

from dashscope.aigc.image_generation import ImageGeneration

resp = ImageGeneration.call(
    model="wan2.6-image",
    messages=[{"role": "user", "content": [{"text": "一只在代码海洋里游泳的猫，赛博朋克风"}]}],
)
# 同步接口一次性返回，图片在 message.content 里
print(resp.output.choices[0].message.content)

视频（异步任务）

视频生成是「提交任务 → 轮询取结果」两段式。SDK 的 fetch 轮询路径与网关的任务查询路径（/v1/services/{model}/tasks/{task_id}）不同，这里用 HTTP 直调演示完整流程：

import time, requests

BASE = "http://localhost:7890/v1/services"
H = {"Authorization": "Bearer <你的访问密钥>", "Content-Type": "application/json"}

# 1) 提交异步视频任务（模型需按 direct_path + async_mode 配置，见 DashScope 快速部署）
r = requests.post(f"{BASE}/aigc/video-generation/video-synthesis", headers=H, json={
    "model": "wanx2.1-t2v",
    "input": {"prompt": "一只在代码海洋里游泳的猫，赛博朋克风"},
    "parameters": {"size": "1280*720"},
})
task_id = r.json()["output"]["task_id"]

# 2) 轮询任务状态，成功后取视频 URL
while True:
    s = requests.get(f"{BASE}/wanx2.1-t2v/tasks/{task_id}", headers=H).json()
    status = s["output"]["task_status"]
    if status == "SUCCEEDED":
        print(s["output"]["video_url"])
        break
    if status in ("FAILED", "CANCELED"):
        raise RuntimeError(s)
    time.sleep(5)

视频模型配置（direct_path + async_mode）见 接入 DashScope 上游 → 视频生成。「计费按秒」只对声明了 per_second 的模型成立，wanx2.1-t2v 这类模型按你的实际配置计费，见 定价与计费。

向量 / 重排 / ASR

# 向量
from dashscope import TextEmbedding
r = TextEmbedding.call(model="text-embedding-v4", input="你好")
print(r.output["embeddings"][0]["embedding"])

# 重排
from dashscope import Rerank
r = Rerank.call(model="gte-rerank", query="猫", documents=["一只猫", "一辆车"])
print([x["relevance_score"] for x in r.output["results"]])

向量/重排也可以走 OpenAI 协议（/v1/embeddings、/v1/rerank）——网关把 openai_embeddings/openai_rerank 与 dashscope 互译。两种接法任选，见 协议互通矩阵。

identity 透传入口 /v3/{model}/{*rest}

当客户端协议与该模型配置的目标协议相同时，可以走 /v3/{model}/{*rest}，请求体原样转发、不做转换：

curl -X POST "http://localhost:7890/v3/qwen3.8-max/services/aigc/text-generation/generation" \
  -H "Authorization: Bearer <你的访问密钥>" \
  -H "Content-Type: application/json" \
  -d '{"model":"qwen3.8-max","input":{"messages":[{"role":"user","content":"hi"}]},"parameters":{"result_format":"message"}}'

下一步

• 完整的上游配置步骤与 kind 变体：见 接入 DashScope 上游。
• DashScope 实时模型（dashscope_realtime）经 /v1/realtime 接入：见 实时会话与语音。