Skip to main content
应用调用

Responses API 调用

Responses API 提供 OpenAI 兼容接口，让您可以使用现有 OpenAI SDK 直接调用百炼应用（智能体、工作流），支持同步和异步两种调用模式。

适用范围

  • 地域限制：仅适用于中国大陆版（北京地域）
  • 应用类型：支持智能体应用和工作流应用
  • API Key：需要已获取并配置 DashScope API Key
  • 应用 ID：需要已创建百炼应用并获取应用 ID

前提条件

在开始使用 Responses API 之前，请确保满足以下条件：
  • API Key：已获取 DashScope API Key
  • 应用 ID：已在百炼平台创建智能体或工作流应用，并获取应用 ID
  • SDK 版本：OpenAI Python SDK >= 1.0.0
  • Python 版本：Python >= 3.7

快速开始

import os
from openai import OpenAI

# 配置认证信息
api_key = os.getenv("DASHSCOPE_API_KEY")
if not api_key:
    raise ValueError("请设置 DASHSCOPE_API_KEY 环境变量")

# 配置应用信息
app_id = os.getenv("APP_ID")  # 您的应用 ID
if not app_id:
    raise ValueError("请设置 APP_ID 环境变量")

# 构建基础 URL
# 智能体应用使用 /agent/ 路径，工作流应用使用 /workflow/ 路径
base_url = f"https://dashscope.aliyuncs.com/api/v2/apps/agent/{app_id}/compatible-mode/v1/"

# 初始化客户端
client = OpenAI(api_key=api_key, base_url=base_url)
try:
    # 发送简单请求
    response = client.responses.create(
        input="你好，请介绍一下你自己"
    )

    # 获取响应内容
    result = response.output[0].content[0].text
    print(f"回复: {result}")

except Exception as e:
    print(f"调用失败: {e}")

调用模式

同步调用

同步调用适用于需要即时响应的场景。API 会保持连接直到任务完成并返回结果。 适用场景：
  • 实时对话交互
  • 简单查询任务
  • 响应时间可预期的操作
特点：
  • 请求阻塞，直到获得完整响应
  • 支持流式输出
  • 适合短时任务（建议 < 30秒）
from openai import OpenAI
import os

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 = OpenAI(api_key=api_key, base_url=base_url)

response = client.responses.create(
    input="你是谁？",
)

# 获取文本回复
print(response.output[0].content[0].text)

异步调用

异步调用适用于耗时较长的任务。API 立即返回任务 ID，通过轮询获取最终结果。 适用场景：
  • 生成长文档或报告
  • 多步骤工具调用
  • 批量数据处理
  • 可能超时的复杂任务
特点：
  • 立即返回，不阻塞
  • 通过任务 ID 查询状态
  • 支持取消正在执行的任务
  • 不支持流式输出
from openai import AsyncOpenAI
import asyncio
import os

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():
    # 创建异步任务
    create_response = await client.responses.create(
        input="请为我规划一个为期三天的北京旅游行程",
        background=True
    )
    task_id = create_response.id
    print(f"任务ID: {task_id}")

    # 轮询查询任务状态
    while True:
        retrieve_response = await client.responses.retrieve(task_id)
        if retrieve_response.status in ['completed', 'failed', 'cancelled']:
            if retrieve_response.status == 'completed':
                print(retrieve_response.output[0].content[0].text)
            break
        await asyncio.sleep(2)

asyncio.run(main())

同步调用应用

同步调用适用于需要即时获取结果的实时交互场景，API 保持连接直到任务完成。

流程概述

  1. 初始化客户端并配置应用 ID
  2. 构建输入内容（文本、图像或文件）
  3. 发起同步请求
  4. 处理响应结果

发送文本消息

from openai import OpenAI
import os

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 = OpenAI(api_key=api_key, base_url=base_url)

# 简单字符串输入
response = client.responses.create(
    input="你是谁？",
)

# 获取文本回复
result_text = response.output[0].content[0].text
print(result_text)
参数说明：
  • input：可以是简单字符串，SDK 会自动转换为标准格式

发送多轮对话

messages = [
    {"role": "user", "content": "你是谁？"},
    {"role": "assistant", "content": "我是一个AI助手。"},
    {"role": "user", "content": "你能做什么？"}
]

response = client.responses.create(input=messages)
print(response.output[0].content[0].text)
参数说明：
  • input：消息数组，包含完整的对话历史
  • 目前需要在每次请求时传递完整对话历史，基于 pre_response_id 的上下文功能将在后续支持

发送图像

前置条件：
  • 智能体应用：需选用通义千问 VL 系列模型，文件处理方式选择"自定义处理"，并重新发布应用
  • 工作流应用：需选用通义千问 VL 系列模型，模型节点的模型入参变量填为 imageList，并重新发布应用
response = client.responses.create(
    input=[
        {
            "role": "user",
            "content": [
                {"type": "input_text", "text": "这是什么"},
                {
                    "type": "input_image",
                    "image_url": "https://dashscope.oss-cn-beijing.aliyuncs.com/images/dog_and_girl.jpeg"
                }
            ]
        }
    ]
)

print(response.output[0].content[0].text)

发送文件（仅智能体应用）

前置条件：
  • 仅智能体应用支持
  • 应用内的文件处理方式需选择"全文引用"或"切片检索"
response = client.responses.create(
    input=[
        {
            "role": "user",
            "content": [
                {"type": "input_text", "text": "总结这个文件的内容"},
                {
                    "type": "input_file",
                    "file_url": "https://dashscope.oss-cn-beijing.aliyuncs.com/audios/welcome.mp3"
                }
            ]
        }
    ]
)

print(response.output[0].content[0].text)

启用流式输出

前置条件：
  • 工作流应用：需在结束节点或流程输出节点中启用"流式输出"开关，并重新发布应用
stream = client.responses.create(
    input="用不少于100字介绍一下你自己",
    stream=True,
)

# 遍历并处理事件流
for chunk in stream:
    if hasattr(chunk, 'delta') and chunk.delta:
        print(chunk.delta, end='', flush=True)
参数说明：
  • stream=True：开启流式输出，以 Server-Sent Events (SSE) 格式返回事件流
  • 主要事件类型：response.output_text.delta（文本增量）、response.completed（响应完成）

异步调用应用

异步调用适用于耗时较长的任务（如生成报告、多步骤工具调用），通过"先提交、后查询"的方式避免请求超时。

流程概述

  1. 创建异步任务（设置 background=True），获取任务 ID
  2. 轮询查询任务状态
  3. 任务完成后获取结果
  4. （可选）取消或删除任务

创建异步任务

from openai import AsyncOpenAI
import asyncio
import os

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():
    create_response = await client.responses.create(
        input="请为我规划一个为期三天的北京旅游行程，要求包含故宫、长城。",
        background=True
    )
    task_id = create_response.id
    print(f"任务ID: {task_id}")
    print(f"初始状态: {create_response.status}")

asyncio.run(main())
参数说明：
  • background=True：开启异步模式，API 立即返回任务 ID
  • 异步任务暂不支持流式输出（stream=true）

查询任务状态

async def poll_task(task_id):
    while True:
        retrieve_response = await client.responses.retrieve(task_id)
        status = retrieve_response.status

        print(f"当前状态: {status}")

        # 检查任务是否已进入终态
        if status in ['completed', 'failed', 'cancelled']:
            if status == 'completed':
                result_text = retrieve_response.output[0].content[0].text
                print(f"\n任务结果:\n{result_text}")
            else:
                print(f"任务状态: {status}")
            break

        # 等待 2 秒后再次查询
        await asyncio.sleep(2)
任务状态说明：
  • queued：任务已创建，正在队列中等待调度
  • running：任务正在执行中
  • completed：任务成功完成，可在 output 字段获取结果
  • failed：任务执行失败，可在 output 字段查看错误信息
  • cancelled：任务被用户取消

取消任务

async def cancel_task(task_id):
    cancel_response = await client.responses.cancel(task_id)
    print(f"取消状态: {cancel_response.status}")
限制说明：
  • 只能取消处于 queued 或 running 状态的任务
  • 已处于终态（completed、failed、cancelled）的任务无法取消

删除任务记录

async def delete_task(task_id):
    response_wrapper = await client.responses.with_raw_response.delete(task_id)
    response_json = response_wrapper.http_response.json()

    if response_json.get("deleted") is True:
        print("任务记录已成功删除")
    else:
        print("删除操作未成功")
限制说明：
  • 只能删除已处于终态（completed、failed、cancelled）的任务
  • 此操作不可恢复

API 详解

请求格式

创建响应（同步/异步）

端点： POST /responses 请求参数：
参数名类型必选默认值说明
inputstring | array是-输入内容，支持字符串或消息数组
backgroundboolean否false是否使用异步模式
streamboolean否false是否启用流式输出（仅同步模式）
conversation_idstring否-会话 ID（暂不支持）
pre_response_idstring否-上一轮响应 ID（暂不支持）

响应格式

同步响应

{
    "id": "resp_xxx",
    "status": "completed",
    "output": [
        {
            "content": [
                {
                    "type": "text",
                    "text": "响应内容"
                }
            ],
            "role": "assistant"
        }
    ],
    "usage": {
        "prompt_tokens": 10,
        "completion_tokens": 20,
        "total_tokens": 30
    }
}

异步响应（创建任务）

{
    "id": "task_xxx",
    "status": "queued"
}

任务管理 API（异步模式）

查询任务状态

端点： GET /responses/{task_id} 
response = await client.responses.retrieve(task_id)

取消任务

端点： POST /responses/{task_id}/cancel 
cancel_response = await client.responses.cancel(task_id)
print(f"取消状态: {cancel_response.status}")
限制： 只能取消处于 queued 或 running 状态的任务

删除任务记录

端点： DELETE /responses/{task_id} 
async def delete_task(task_id):
    response_wrapper = await client.responses.with_raw_response.delete(task_id)
    response_json = response_wrapper.http_response.json()

    if response_json.get("deleted") is True:
        print("任务记录已成功删除")
    else:
        print("删除操作未成功")
限制： 只能删除已处于最终状态（completed、failed、cancelled）的任务

任务状态说明

状态说明可执行操作
queued任务已创建，等待执行查询、取消
running任务正在执行中查询、取消
completed任务成功完成查询、删除
failed任务执行失败查询、删除
cancelled任务被用户取消查询、删除
开始使用
应用开发
Prompt
插件
任务中心
组件
应用调用
应用评测
应用观测
权限管理
更新日志