本文介绍如何通过 OpenAI 兼容模式的 Responses API 异步调用阿里云百炼应用(智能体、工作流)。对于耗时较长的任务,只需在请求中设置 background 为 true,API 便会立即返回一个任务 ID,用于后续的查询与管理。这种“先提交、后查询”的方式,可有效避免请求超时或长时间等待。
相关参考
同步调用:对于需要即时获取结果的实时交互场景,请参阅同步调用 API 参考。
DashScope API:如需获取更全面的功能与更高的性能,请参阅DashScope API 参考。
本文档仅适用于中国大陆版(北京地域)。
异步任务暂不支持流式输出(stream=true)。
前提条件
已获取 API Key并配置API Key到环境变量。如果通过OpenAI SDK进行调用,还需要安装SDK。
快速开始
本节提供一个完整的 Python 示例,演示如何发起一个异步任务,然后通过轮询方式持续检查任务状态,直到任务完成并获取最终结果。
这个示例覆盖了异步调用的核心流程:
创建任务:调用
create方法并设置background=True,获取任务 ID。轮询状态:在一个循环中,定期调用
retrieve方法查询任务状态。处理结果:当任务状态变为
completed、failed或cancelled时,退出循环并展示最终结果。
import asyncio
import os
from openai import AsyncOpenAI
import time
# 1. 初始化客户端
## 若没有配置环境变量,可用百炼API Key将下行替换为:api_key="sk-xxx"。但不建议在生产环境中直接将API Key硬编码到代码中,以减少API Key泄露风险。
api_key = os.getenv("DASHSCOPE_API_KEY")
app_id = 'APP_ID' # 请替换为实际的应用ID
base_url = f'https://dashscope.aliyuncs.com/api/v2/apps/agent/{app_id}/compatible-mode/v1/'
client = AsyncOpenAI(api_key=api_key, base_url=base_url)
async def main():
start_time = time.time()
# 2. 发起异步任务
print("Step 1: 正在创建异步任务...")
create_response = await client.responses.create(
input="请为我规划一个为期三天的北京旅游行程,要求包含故宫、长城。",
background=True
)
task_id = create_response.id
print(f" - 任务创建成功! Task ID: {task_id}")
print(f" - 初始状态: {create_response.status}")
# 3. 轮询任务状态直到任务完成
print("\nStep 2: 开始轮询任务状态 (每 2 秒一次)...")
while True:
retrieve_response = await client.responses.retrieve(task_id)
status = retrieve_response.status
print(f" - 当前状态: {status}")
# 检查任务是否已进入终态
if status in ['completed', 'failed', 'cancelled']:
print("\nStep 3: 任务已完成或终止。")
final_response = retrieve_response
break
# 等待 2 秒后再次查询
await asyncio.sleep(2)
# 4. 处理并打印最终结果
if final_response.status == 'completed':
print(" - 状态: Success")
# 提取并打印可读的文本结果
if final_response.output and final_response.output[0].content:
result_text = final_response.output[0].content[0].text
print("\n--- 任务输出 ---")
print(result_text)
print("-----------------")
else:
print("任务已完成,但未返回有效输出。")
else: # 'failed' or 'cancelled'
print(f" - 状态: {final_response.status.upper()}")
print("\n--- 最终响应详情 ---")
# 打印完整的JSON以供调试
print(final_response.model_dump_json(indent=2))
print("-----------------------")
end_time = time.time()
print(f"\n总耗时: {end_time - start_time:.2f} 秒")
if __name__ == "__main__":
asyncio.run(main())具体流程
以下章节详细介绍了创建、查询、取消和删除异步任务的 API 操作。
创建异步任务
将background参数设置为true来开启异步模式,创建异步任务,立即获取任务 ID。
请求示例
Python
from openai import AsyncOpenAI
import asyncio
import os
# 若没有配置环境变量,可用百炼API Key将下行替换为:api_key="sk-xxx"。但不建议在生产环境中直接将API Key硬编码到代码中,以减少API Key泄露风险。
api_key=os.getenv("DASHSCOPE_API_KEY")
app_id='APP_ID' # 替换为实际的应用 ID
base_url = f'https://dashscope.aliyuncs.com/api/v2/apps/agent/{app_id}/compatible-mode/v1/'
# 初始化异步客户端
client = AsyncOpenAI(
api_key=api_key,
base_url=base_url
)
async def main():
# 异步调用
response = await client.responses.create(
input="你是谁?",
background=True
)
# 打印完整的响应JSON对象
print(response.model_dump_json(indent=2))
# 运行异步主函数
if __name__ == "__main__":
asyncio.run(main())curl
# 若没有配置环境变量,请将 {DASHSCOPE_API_KEY} 替换为实际的阿里云百炼API Key,但不建议在生产环境中直接将API Key硬编码到代码中,以减少API Key泄露风险。
# 请将 APP_ID 替换为实际的应用ID
# 实际使用时请删除注释内容
curl --location "https://dashscope.aliyuncs.com/api/v2/apps/agent/{APP_ID}/compatible-mode/v1/responses" \
--header 'Content-Type: application/json' \
--header "Authorization: Bearer ${DASHSCOPE_API_KEY}" \
--data '{
"input": "你是谁?",
"background": true
}'请求字段说明
字段名 | 类型 | 必填 | 描述 |
| string/array | 是 | 请求的核心输入内容。可以是单个字符串,或是一个包含多轮对话历史的消息数组。 |
| boolean | 是 | 是否以异步方式执行任务。
|
响应示例
{
"id": "bcb9728b-a7f8-480c-ace0-8d61ff776857",
"created_at": 1761875532.4459603,
"error": null,
"incomplete_details": null,
"instructions": null,
"metadata": null,
"model": "",
"object": "response",
"output": [],
"parallel_tool_calls": false,
"temperature": null,
"tool_choice": "auto",
"tools": [],
"top_p": null,
"background": null,
"conversation": null,
"max_output_tokens": null,
"max_tool_calls": null,
"previous_response_id": null,
"prompt": null,
"prompt_cache_key": null,
"reasoning": null,
"safety_identifier": null,
"service_tier": null,
"status": "queued",
"text": null,
"top_logprobs": null,
"truncation": null,
"usage": null,
"user": null
}响应字段说明
字段名 | 类型 | 描述 |
| string | 异步任务的唯一标识符,用于后续查询、取消或删除操作。 |
| string | 任务的初始状态,通常为 |
| integer | 任务创建时间的Unix时间戳(秒)。 |
| string | 对象类型,固定为 |
查询任务状态
获取指定任务的当前状态和执行结果。
Endpoint:GET https://dashscope.aliyuncs.com/api/v2/apps/agent/{APP_ID}/compatible-mode/v1/responses/{RESPONSE_ID}
请将{APP_ID}替换为实际的应用ID,将{RESPONSE_ID}替换为创建任务时返回的任务ID。
请求示例
Python
import asyncio
import os
from openai import AsyncOpenAI
# 若没有配置环境变量,可用百炼API Key将下行替换为:api_key="sk-xxx"。但不建议在生产环境中直接将API Key硬编码到代码中,以减少API Key泄露风险。
api_key=os.getenv("DASHSCOPE_API_KEY")
app_id='APP_ID' # 1、替换为实际的应用 ID
base_url = f'https://dashscope.aliyuncs.com/api/v2/apps/agent/{app_id}/compatible-mode/v1/'
# 初始化客户端
client = AsyncOpenAI(api_key=api_key, base_url=base_url)
async def main():
# 查询任务状态
response = await client.responses.retrieve("6e4ba287-0e93-4d67-9955-ef1017ed3384") # 2、替换为实际的任务 ID
print(response.model_dump_json(indent=2))
if __name__ == "__main__":
asyncio.run(main())curl
# 若没有配置环境变量,请将 {DASHSCOPE_API_KEY} 替换为实际的阿里云百炼API Key,但不建议在生产环境中直接将API Key硬编码到代码中,以减少API Key泄露风险。
# 1、请将 APP_ID 替换为实际的应用ID
# 2、请将 RESPONSE_ID 替换为创建异步任务返回的ID
# 实际使用时请删除注释内容
curl "https://dashscope.aliyuncs.com/api/v2/apps/agent/{APP_ID}/compatible-mode/v1/responses/{RESPONSE_ID}" \
-H "Content-Type: application/json" \
-H "Authorization: Bearer ${DASHSCOPE_API_KEY}"
响应示例
{
"id": "bcb9728b-a7f8-480c-ace0-8d61ff776857",
"created_at": 1761875537.0,
"error": null,
"incomplete_details": null,
"instructions": null,
"metadata": null,
"model": "",
"object": "response",
"output": [
{
"id": "msg_d89e8cb9-f8fa-4092-bb86-928e21839f64",
"content": [
{
"annotations": [],
"text": "你好,我是通义千问(Qwen),是阿里巴巴集团旗下的通义实验室自主研发的超大规模语言模型。我能够回答问题、创作文字,比如写故事、写公文、写邮件、写剧本等等,还能进行逻辑推理、编程,甚至表达观点和玩游戏。我的目标是成为你值得信赖的智能助手。有什么我可以帮到你的吗?",
"type": "output_text",
"logprobs": null
}
],
"role": "assistant",
"status": "completed",
"type": "message"
}
],
"parallel_tool_calls": false,
"temperature": null,
"tool_choice": "auto",
"tools": [],
"top_p": null,
"background": null,
"conversation": null,
"max_output_tokens": null,
"max_tool_calls": null,
"previous_response_id": null,
"prompt": null,
"prompt_cache_key": null,
"reasoning": null,
"safety_identifier": null,
"service_tier": null,
"status": "completed",
"text": null,
"top_logprobs": null,
"truncation": null,
"usage": null,
"user": null
}响应字段说明
字段名 | 类型 | 描述 |
| string | 异步任务的唯一标识符。 |
| string | 任务的当前状态,详见任务生命周期。 |
| array | 任务的输出结果。当 |
取消任务
取消一个正在进行中的异步任务。此操作仅对状态为 queued 或 running 的任务有效。
Endpoint:POST https://dashscope.aliyuncs.com/api/v2/apps/agent/{APP_ID}/compatible-mode/v1/responses/{RESPONSE_ID}/cancel
请将{APP_ID}替换为实际的应用ID,将{RESPONSE_ID}替换为创建任务时返回的任务ID。
请求示例
import asyncio
import os
from openai import AsyncOpenAI
# 若没有配置环境变量,可用百炼API Key将下行替换为:api_key="sk-xxx"。但不建议在生产环境中直接将API Key硬编码到代码中,以减少API Key泄露风险。
api_key=os.getenv("DASHSCOPE_API_KEY")
app_id='APP_ID' # 1、替换为实际的应用 ID
base_url = f'https://dashscope.aliyuncs.com/api/v2/apps/agent/{app_id}/compatible-mode/v1/'
# 初始化客户端
client = AsyncOpenAI(api_key=api_key, base_url=base_url)
async def main():
# 取消任务
response = await client.responses.cancel("6e4ba287-0e93-4d67-9955-ef1017ed3384") # 2、替换为实际的任务 ID
print(response.model_dump_json(indent=2))
if __name__ == "__main__":
asyncio.run(main())# 若没有配置环境变量,请将 {DASHSCOPE_API_KEY} 替换为实际的阿里云百炼API Key,但不建议在生产环境中直接将API Key硬编码到代码中,以减少API Key泄露风险。
# 1、请将 APP_ID 替换为实际的应用ID
# 2、请将 RESPONSE_ID 替换为创建异步任务返回的ID
# 实际使用时请删除注释内容
curl -X POST "https://dashscope.aliyuncs.com/api/v2/apps/agent/{APP_ID}/compatible-mode/v1/responses/{RESPONSE_ID}/cancel" \
-H "Content-Type: application/json" \
-H "Authorization: Bearer ${DASHSCOPE_API_KEY}"
响应示例
{
"created_at": 1761877521.5800686,
"id": "568b1459-126d-4719-b58f-b41078bacb48",
"model": "",
"object": "response",
"output": [
],
"parallel_tool_calls": false,
"status": "cancelled",
"tool_choice": "auto",
"tools": [
]
}响应字段说明
字段名 | 类型 | 描述 |
| string | 被操作任务的唯一标识符。 |
| string | 任务被取消后的状态,如:
|
| string | 对象类型,固定为 |
删除任务记录
删除一个已处于终态(completed, failed, cancelled)的任务记录。此操作不可恢复。
Endpoint:DELETE /api/v2/apps/agent/{APP_ID}/compatible-mode/v1/responses/{RESPONSE_ID}
请将{APP_ID}替换为实际的应用ID,将{RESPONSE_ID}替换为创建任务时返回的任务ID。
请求示例
Python
import asyncio
import os
from openai import AsyncOpenAI
# 若没有配置环境变量,可用百炼API Key将下行替换为:api_key="sk-xxx"。
# 不建议在生产环境中直接将API Key硬编码到代码中,以减少API Key泄露风险。
api_key = os.getenv("DASHSCOPE_API_KEY")
app_id = 'APP_ID' # 1. 替换为您的应用ID
base_url = f'https://dashscope.aliyuncs.com/api/v2/apps/agent/{app_id}/compatible-mode/v1/'
client = AsyncOpenAI(api_key=api_key, base_url=base_url)
async def main():
# 2. 替换为要删除的任务ID
response_id_to_delete = "b17c64f94-7034-403e-aef5-4f48647656d5"
try:
response_wrapper = await client.responses.with_raw_response.delete(response_id_to_delete)
http_response = response_wrapper.http_response
response_json = http_response.json()
# 检查响应内容中的 "deleted" 字段是否为 true。
if response_json.get("deleted") is True:
print("任务记录已成功删除。")
print(response_json)
else:
print("删除操作未成功")
print(response_json)
except Exception as e:
print(f"\n调用API时发生严重错误: {e}")
if __name__ == "__main__":
asyncio.run(main())curl
# 若没有配置环境变量,请将 {DASHSCOPE_API_KEY} 替换为实际的阿里云百炼API Key,但不建议在生产环境中直接将API Key硬编码到代码中,以减少API Key泄露风险。
# 1、请将 APP_ID 替换为实际的应用ID
# 2、请将 RESPONSE_ID 替换为创建异步任务返回的ID
# 实际使用时请删除注释内容
curl -X DELETE "https://dashscope.aliyuncs.com/api/v2/apps/agent/{APP_ID}/compatible-mode/v1/responses/{RESPONSE_ID}" \
-H "Content-Type: application/json" \
-H "Authorization: Bearer ${DASHSCOPE_API_KEY}"响应示例
{
"deleted": true,
"id": "c8eddc0e-faf5-4228-a853-9c411194015f",
"object": "response"
}响应字段说明
字段名 | 类型 | 描述 |
| string | 被删除任务的唯一标识符。 |
| boolean |
|
| string | 对象类型,固定为 |
任务生命周期
异步任务的生命周期包含以下状态:
状态值 | 描述 |
| 任务已成功创建,正在队列中等待系统调度。 |
| 任务正在执行中。 |
| 任务已成功完成。可在响应的 |
| 任务在执行完成前被用户主动取消。 |
| 任务执行失败。可在响应的 |
错误码
如果应用调用失败并返回报错信息,请参阅错误信息进行解决。