DashScope Python SDK 调用 Qwen3 — 阿里云原生 API 接入代码示例

安装

pip install dashscope

验证安装：

import dashscope
print(dashscope.__version__)

初始化

import os
import dashscope

# 方式一：环境变量（推荐）
dashscope.api_key = os.getenv("DASHSCOPE_API_KEY")

# 方式二：直接设置（方便测试但不推荐生产）
# dashscope.api_key = "sk-xxxxxxxxxxxxx"

最简对话

import os
import dashscope
from dashscope import Generation

dashscope.api_key = os.getenv("DASHSCOPE_API_KEY")

response = Generation.call(
    model="qwen3-flash",
    messages=[
        {"role": "system", "content": "你是一个有帮助的助手。"},
        {"role": "user", "content": "介绍一下 Qwen3 的创新点。"}
    ],
    result_format="message"
)

# 获取输出
if response.status_code == 200:
    print(response.output.choices[0].message.content)
else:
    print(f"错误：{response.code} - {response.message}")

思考模式控制

开启思考（适合复杂问题）

response = Generation.call(
    model="qwen3-max",
    messages=[
        {"role": "user", "content": "证明勾股定理。"}
    ],
    result_format="message",
    enable_thinking=True  # 显式开启思考模式
)

# 思考模式下，SDK 会分离思考链和最终答案
if response.status_code == 200:
    msg = response.output.choices[0].message
    print("=== 思考过程 ===")
    print(msg.reasoning_content)  # 思考链
    print("\n=== 最终答案 ===")
    print(msg.content)            # 最终答案

关闭思考（适合简单问题）

response = Generation.call(
    model="qwen3-plus",
    messages=[
        {"role": "user", "content": "1 + 1 = ?"}
    ],
    result_format="message",
    enable_thinking=False  # 显式关闭思考
)

qwen3-max 的自动思考

qwen3-max 是云端旗舰模型，默认会根据问题复杂度自动决定是否思考。你也可以显式覆盖：

# 强制 qwen3-max 思考
response = Generation.call(
    model="qwen3-max",
    messages=[{"role": "user", "content": "分析这段代码的性能问题。"}],
    enable_thinking=True
)

# 强制 qwen3-max 不思考（低延迟）
response = Generation.call(
    model="qwen3-max",
    messages=[{"role": "user", "content": "翻译这句话。"}],
    enable_thinking=False
)

流式输出

response = Generation.call(
    model="qwen3-flash",
    messages=[{"role": "user", "content": "介绍一下 Qwen3。"}],
    result_format="message",
    stream=True
)

for chunk in response:
    if chunk.status_code == 200:
        # 流式中思考内容和答案在不同字段
        if chunk.output.choices[0].message.reasoning_content:
            # 这是思考过程（只在开启思考时出现）
            print(chunk.output.choices[0].message.reasoning_content, end="")
        if chunk.output.choices[0].message.content:
            # 这是最终答案
            print(chunk.output.choices[0].message.content, end="")
    else:
        print(f"\n错误：{chunk.code}")
        break

多轮对话

DashScope 支持多轮对话——你只需要把历史消息传进去：

import os
import dashscope
from dashscope import Generation

dashscope.api_key = os.getenv("DASHSCOPE_API_KEY")

# 维护对话历史
messages = [
    {"role": "system", "content": "你是一个有帮助的助手。"},
]

# 第一轮
messages.append({"role": "user", "content": "我叫小明。"})
response = Generation.call(
    model="qwen3-flash",
    messages=messages,
    result_format="message"
)
reply = response.output.choices[0].message.content
print(f"助手：{reply}")
messages.append({"role": "assistant", "content": reply})

# 第二轮
messages.append({"role": "user", "content": "我刚才说我叫什么？"})
response = Generation.call(
    model="qwen3-flash",
    messages=messages,
    result_format="message"
)
reply = response.output.choices[0].message.content
print(f"助手：{reply}")

注意：每次调用都传递完整的历史消息。上下文长度有限制（qwen3-flash 是 128K），对话太长会超出限制。

多模态（视觉理解）

用 qwen3-vl-plus 模型调用图像理解：

import os
import dashscope
from dashscope import MultiModalConversation

dashscope.api_key = os.getenv("DASHSCOPE_API_KEY")

messages = [
    {
        "role": "user",
        "content": [
            {"image": "file:///path/to/your/image.jpg"},  # 本地图片
            {"text": "这张图片里有什么？"}
        ]
    }
]

response = MultiModalConversation.call(
    model="qwen3-vl-plus",
    messages=messages
)

if response.status_code == 200:
    print(response.output.choices[0].message.content[0]["text"])
else:
    print(f"错误：{response.code} - {response.message}")

也支持 URL 图片：

{"image": "https://example.com/image.jpg"}

更多生成参数

response = Generation.call(
    model="qwen3-plus",
    messages=[{"role": "user", "content": "写一首关于春天的诗。"}],
    result_format="message",
    # 以下是可选参数
    max_tokens=1024,          # 最多生成的 token 数
    temperature=0.7,           # 温度，控制创造性（0-1）
    top_p=0.9,                 # 核采样参数
    top_k=50,                  # Top-K 采样
    repetition_penalty=1.05,   # 重复惩罚
    enable_thinking=False      # 关闭思考模式
)

错误处理

常见错误码

带重试的调用

import os
import time
import dashscope
from dashscope import Generation

def call_with_retry(max_retries=3):
    for attempt in range(max_retries):
        response = Generation.call(
            model="qwen3-flash",
            messages=[{"role": "user", "content": "你好"}],
            result_format="message"
        )
        if response.status_code == 200:
            return response.output.choices[0].message.content
        elif response.code == "Throttling":
            # 限流，指数退避
            wait_time = 2 ** attempt
            print(f"限流，等待 {wait_time}s 后重试...")
            time.sleep(wait_time)
        else:
            print(f"错误 {response.code}：{response.message}")
            return None
    return None

异步调用

DashScope 也支持异步调用（asyncio）：

import asyncio
import os
import dashscope
from dashscope import Generation

dashscope.api_key = os.getenv("DASHSCOPE_API_KEY")

async def main():
    responses = await Generation.acall(
        model="qwen3-flash",
        messages=[{"role": "user", "content": "你好"}],
        result_format="message"
    )
    print(responses.output.choices[0].message.content)

asyncio.run(main())

批量并发调用

import os
import asyncio
import dashscope
from dashscope import Generation

dashscope.api_key = os.getenv("DASHSCOPE_API_KEY")

async def batch_call():
    tasks = [
        Generation.acall(
            model="qwen3-flash",
            messages=[{"role": "user", "content": f"{i} 的平方是多少？"}],
            result_format="message"
        )
        for i in range(1, 6)
    ]
    
    responses = await asyncio.gather(*tasks)
    
    for i, resp in enumerate(responses, 1):
        if resp.status_code == 200:
            print(f"{i} 的平方：{resp.output.choices[0].message.content}")

asyncio.run(batch_call())

相关阅读

• 01-阿里云百炼控制台：API Key 获取流程
• 03-OpenAI 兼容接口：用 OpenAI SDK 调百炼
• 04-API Key 管理与计费：安全和用量管理