【必读】AIGCaaS API 异步调用指南 - AIGC SaaS API 平台 - 专注 Stable Diffusion，LangChain 知识库等 AIGC SaaS API 能力建设

Aug 09,2023

AIGCaaS 平台是一个 AIGC 领域的 SaaS API 平台，平台以开放形式提供大量的 AIGC API，开发者可以无需考虑 GPU 显卡等情况下，进行对应的 API 的快速使用。本文档将会用户引导用户如何进行 API 的调用。

## 参数分类

在 AIGCaaS 平台，分为系统参数和 API 参数。

### 系统参数

所谓系统参数指的是和你使用具体的 API 无关的参数，主要用户进行系统鉴权使用，这部分参数主要包括：

关于参数说明：

1. 请求级安全防护：

**Nonce** 参数是防重放配置，本质是随机字符串（最大长度64位，推荐20位，且10秒内不可重复；

**Timestamp** 参数是发起请求时间戳，与服务器时间差小于10秒；

**Token** 签名结果`sha256(Timestamp, SecretKey, Nonce)`；

所以可以认为，在10秒内，每个用户的 Nonce 是唯一的，所以 Token 也是唯一的（出现重复的概率极低，可以趋近于无穷小），此时即便你的 API 请求被别人抓包所获取，他也没办法多次盗用请求，因为每个请求只能被服务器认可一次，如果：

- 10秒内重复请求，会因为 Nonce 参数重复而拒绝响应；
   - 10秒外再发起重复请求，会因为 Timestamp 参数过期而拒绝响应；

2. 账号信息安全防护：

**SecretID** 密钥信息，密钥配置中的 secret_id 配置；

**Token** 签名结果`sha256(Timestamp, SecretKey, Nonce)`；

在服务端获取到的 SecretId与SecretKey两个信息，分别用于参数传输与参与签名，通过`sha256`算法签名后的结果极难通过撞库等方法进行逆向，所以相对拉说也是进行账号信息防护；

综上所述，`SecretID`与`Token`组合是用于身份信息验证的使用；`Nonce`与`Timestamp`，`Token`组合，是用户请求防盗使用；

### API 参数

所谓的 API 参数，指的是具体的应用 API 所需要的参数信息，以Stable Diffusion为例，其中部分的参数：

## 请求方式

对比同步请求不同的是，异步请求要分为两个步骤进行：

- 发起请求，下达任务：以 Stable Diffusion 应用为例，通过对 API 接口调用，下发任务，告诉系统我要画图，我的参数是什么，系统会返回一个 RequestId
- 查询结果或通知结果：
  - 如果在请求头中配置了Aigcaas-Async-Callback参数，系统将会以通知方式告知这个参数的 URL，结果是什么；
  - 如果在请求头中没有配置Aigcaas-Async-Callback参数，需要用户带着 RequestID 信息，调用 `/v3/async`接口查询结果；

### 任务下发

AIGCaaS 平台所接入的 AIGC 应用 API，其地址格式为：

```
https://api.aigcaas.cn/v3/application/{应用名}/action/{具体方法}
```

例如某接口**请求地址**：https://api.aigcaas.cn/v3/application/{应用名}/action/{具体方法}

则可以根据该地址进行 API 调用：

**请求方法**：POST

**入参**：

以 Python 语言为例：

```python
import time
import json
import hashlib
import random
import requests

# 签名对象
getSha256 = lambda content: hashlib.sha256(content.encode("utf-8")).hexdigest()

# 密钥信息获取与帮助文档：https://www.aigcaas.cn/article/16.html

secret_id = '<请替换为自己的密钥信息: SecretId>'  # 密钥信息
secret_key = '<请替换为自己的密钥信息: SecretKey>'  # 密钥信息
application_name = 'analog'  # 应用名称
api_name = 'analog'  # 接口名称

# 获取响应
response = requests.request("POST", url, headers=headers, data=json.dumps(data))
print(response.text)
```

### 响应结果

在 AIGCaaS 平台中，平台级响应结果主要体现在 Response Header 中，API 的响应结果主要体现在 Response Body 中。

例如，某次失败的请求，响应结果：

**Headers**

```
X-Forwarded-For: 172.17.0.9
Host: api.aigcaas.cn
X-Nginx-Proxy: true
Connection: upgrade
Content-Length: 22
Sec-Ch-Ua: ".Not/A)Brand";v="99", "Google Chrome";v="103", "Chromium";v="103"
Sec-Ch-Ua-Mobile: ?0
User-Agent: Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/103.0.0.0 Safari/537.36
Secretid: x.temp.id.cibqLMAKuyKkUBABVPbDdpJlGbuqwNtZhnNTiyjHxuiDfPOHCz
Accept: application/json, text/plain, */*
Content-Type: application/json
Sec-Ch-Ua-Platform: "macOS"
Origin: https://www.aigcaas.cn
Sec-Fetch-Site: same-site
Sec-Fetch-Mode: cors
Sec-Fetch-Dest: empty
Referer: https://www.aigcaas.cn/
Accept-Encoding: gzip, deflate, br
Accept-Language: zh-CN,zh;q=0.9,en;q=0.8
Aigcaas-Request-Id: 354ea732-6402-4614-a83b-81cbbd37f148
Aigcaas-Timestamp: 1684898592
Aigcaas-Backed-Status: False
```

其中，平台级结果为`Aigcaas-XXXX`，例如：

```
Aigcaas-Request-Id: 354ea732-6402-4614-a83b-81cbbd37f148
Aigcaas-Timestamp: 1684898592
Aigcaas-Statue: False
```

对应的是请求 ID（用于错误追踪等）以及接收到请求的时间戳等。值得注意的是，`Aigcaas-Statue`参数决定当前请求是否成功，如果为 True，表示成功，并进行计费，如果为 False 表示失败，不进行计费。

- 成功的定义：成功获取到了结果；

- 失败的定义：没有成功获得到结果，出现了错误信息；

- 特殊的情况：由于安全限制等原因，导致模型已经响应结果，但是被安全隔离，尽管没有获得到预期结果，但是也算是成功请求；

**Body**

```
{
  "error": "APIError",
  "message": "API Server Error, the system failed to obtain a correct return of the model.",
  "help": "https://www.aigcaas.cn/article/35.html"
}
```

该响应结果，为最终的结果。通常情况下，包括两个层面：

1. 平台级错误：主要有`status`，`message`以及`help`三个参数，用于表示状态，信息以及帮助文档；
2. 模型自定义响应结果：此处不做规约；

### 异步请求

AIGCaaS 平台支持的异步请求有两个模式：

- **轮训模式**：

- 发起请求（路由格式为 `/v3/application/{应用名}/action/{具体方法}`），告诉 AIGCaaS 系统说，我要进行异步请求（实际上是在请求头中增加`Aigcaas-Async: True`参数）

- AIGCaaS 会根据请求信息，将请求发送到对应的模型服务

- AIGCaaS 系统自动启动异步模式，并迅速将 RequestID 等信息返回给用户（RequestID信息是响应头中`Aigcaas-Request-Id`）

- 用户不断发起查看异步结果（路由为 `/v3/async`）的请求，直到获得到最终的结果（判断标准是响应的 HTTP Code 不为 202），在进行轮训结果的过程中，如果没有结果信息，系统会提示`There are currently no results, please try again later.`，如果对应的模型/功能存在实时进度查看能力，可以通过每次轮训结果中响应头中的`Aigcaas-Async-Progress`参数进行判定；

- **通知模式**：

- 发起请求（路由格式为 `/v3/application/{应用名}/action/{具体方法}`），告诉 AIGCaaS 系统说，我要进行异步请求（实际上是在请求头中增加`Aigcaas-Async: True`参数），同时告诉他你的通知地址（实际上是在请求头中增加`Aigcaas-Async-Callback: https://www.xxx.com`参数）

- AIGCaaS 会根据请求信息，将请求发送到对应的模型服务

- AIGCaaS 系统自动启动异步模式，并迅速将 RequestID 等信息返回给用户（RequestID信息是响应头中`Aigcaas-Request-Id`）

- AIGCaaS 完成推理处理，会将结果以 POST 请求方式，发送到你指定的通知地址（即请求头中的`Aigcaas-Async-Callback: https://www.xxx.com`参数，为了判断是否是 AIGCaaS 平台发送，里面会有一个 Token 信息最为权限鉴定）

当然，即便是第二种通知模式，也是支持用户主动通过轮训的方式进行结果的获取。

#### 轮询模式

其核心流程就是：

1. 客户端发起请求（路由格式为 `/v3/application/{应用名}/action/{具体方法}`），告诉 AIGCaaS 系统说，我要进行异步请求，并在请求头中，增加参数进行相关的声明
   - `'Aigcaas-Async': 'True'`【必须】表示异步调用
   - `'Aigcaas-Async-Callback': 'https://www.xxx.com'`【非必需】通知地址（会在后文通知模式中重点讲解）
2. 获取请求结果，用户不断发起查看异步结果（路由为 `/v3/async`）的请求，直到获得到最终的结果（判断标准是响应的 HTTP Code 不为 202），在进行轮训结果的过程中，如果没有结果信息，系统会提示`There are currently no results, please try again later.`，如果对应的模型/功能存在实时进度查看能力，可以通过每次轮训结果中响应头中的`Aigcaas-Async-Progress`参数进行判定；

以 Python 语言为例，进行详细的案例介绍：

1. 发起请求，以`胶片质感扩散生成模型`为例，复制网页右侧的示例代码：

```python
import time
import json
import hashlib
import random
import requests

# 签名对象
getSha256 = lambda content: hashlib.sha256(content.encode("utf-8")).hexdigest()

# 密钥信息获取与帮助文档：https://www.aigcaas.cn/article/16.html

# 请求地址
url = "https://api.aigcaas.cn/v3/application/%s/action/%s" % (application_name, api_name)
# 构建请求头
nonce = str(random.randint(1, 10000))
timestamp = str(int(time.time()))
token = getSha256(("%s%s%s" % (timestamp, secret_key, nonce)))
headers = {
    'SecretID': secret_id,
    'Nonce': nonce,
    'Token': token,
    'Timestamp': timestamp,
    'Aigcaas-Async': 'True',  # 异步调用配置，相关教程：https://www.aigcaas.cn/article/18.html
    # 'Aigcaas-Async-Callback': 'https://www.xxxx.com',  # 异步调用结果接收配置，相关教程：https://www.aigcaas.cn/article/18.html
    'Content-Type': 'application/json'
}
# 构建请求 body
data = {
    "text": ""
}

# 获取响应
response = requests.request("POST", url, headers=headers, data=json.dumps(data))
print(response.text)
```

值得注意的是，在请求头中增加`'Aigcaas-Async': 'True'`，即可表示本次请求为异步请求。

2. 需要对代码中的部分参数进行更新，主要是密钥信息的替换，密钥的管理和创建等可以在页面 https://aigcaas.anyenv.cn/home/secret/list 中操作：

```
secret_id = '<请替换为自己的密钥信息: SecretId>'  # 密钥信息
secret_key = '<请替换为自己的密钥信息: SecretKey>'  # 密钥信息
```

通过对代码进行运行，发起请求可以看到结果：

![image-20230515215556308](https://www.images.wiki/4k5h4gxAcbDxeSls6Eib.png)

此时，后端系统会很快地给出响应信息：`{"Aigcaas-Request-Id":"5d31f517-bbfa-4a2e-a49d-024ee9127607"}`，当然，我们更推荐开发者直接获取相应头中的`Aigcaas-Request-Id`：

![image-20230515220936016](https://www.images.wiki/EdZ4EC8jFbwBvsjBhs7t.png)

3. 接下来，只需带着该请求 ID，以 GET方法访问`https://api.aigcaas.cn/v3/async`接口即可。该接口的文档如下：

**请求地址**：`https://api.aigcaas.cn/v3/async`

**请求方法**：GET

**签名方法**：与正常获取请求的签名一致，都是`sha256(时间戳，密钥信息，防重放数字)`

**请求头**：

| 参数名    | 类型   | 位置   | 解释                                                      |
| --------- | ------ | ------ | --------------------------------------------------------- |
| Nonce     | string | header | **防重放配置** 随机字符串（最大长度20位）且10秒内不可重复 |
| Timestamp | string | header | **发起请求时间戳** 与服务器时间差遭遇10秒                 |
| SecretID  | string | header | **密钥信息** 密钥配置中的 secret_id 配置                  |
| Token     | string | header | **签名结果** sha256(Timestamp, SecretKey, Nonce)          |

**请求结果**：

一般情况下，请求结果是会告诉使用者当前任务是否完成：

- 如果已经完成，返回对应的结果信息（无论成功失败，都会返回对应的信息），此时的 HTTP 状态为 200

- 如果没有完成，返回对应的提示信息`There are currently no results, please try again later.`，针对部分支持进度查看的模型，可以在相应头中获取`Aigcaas-Async-Progress`参数以便确定任务执行进度

> 此处需要额外注意的两个事情：
  >
  > 1. 发起异步的请求中，nonce，timestamp等参数不可以复用前次请求，需要单独生成；
  > 2. 异步请求结果存放有效期24小时，24小时内可以多次调用，获取结果；
  > 3. 值得注意的是，在获取结果/进度的请求中，  RequestID 的来源是相应头中的`Aigcaas-Request-Id`参数，再次发起请求时，参数名字为`Aigcaas-Request-Id`或`RequestID`;

这一部分对应的测试代码，以 Python 语言为例可以是：

```python
url = "https://api.aigcaas.cn/v3/async"
nonce = str(random.randint(1, 10000))
timestamp = str(int(time.time()))
token = getSha256(("%s%s%s" % (timestamp, secret_key, nonce)))
headers = {
   'SecretID': secret_id,
   'RequestID': "5d31f517-bbfa-4a2e-a49d-024ee9127607",
   'Nonce': nonce,
   'Token': token,
   'Timestamp': timestamp,
   'Content-Type': 'application/json'
}
response = requests.request("GET", url, headers=headers, data=json.dumps({}))
print(response.content.decode("utf-8"))
```

综上所述，以 Python 语言为例，一个完整的案例代码可以是：

```python
import time
import json
import hashlib
import random
import requests

# 签名对象
getSha256 = lambda content: hashlib.sha256(content.encode("utf-8")).hexdigest()

# 密钥信息获取与帮助文档：https://www.aigcaas.cn/article/16.html

secret_id = 'zCDTYBIwwiSuAKp'  # 密钥信息
secret_key = 'qTFLBIuPBGuNWyp'  # 密钥信息
application_name = 'analog'  # 应用名称
api_name = 'analog'  # 接口名称

# 获取响应
response = requests.request("POST", url, headers=headers, data=json.dumps(data))
request_id = response.headers["Aigcaas-Request-Id"]
# print(response.content.decode("utf-8"))

while True:
    time.sleep(1)
    url = "https://api.aigcaas.cn/v3/async"
    # 构建请求头
    nonce = str(random.randint(1, 10000))
    timestamp = str(int(time.time()))
    token = getSha256(("%s%s%s" % (timestamp, secret_key, nonce)))
    headers = {
        'SecretID': secret_id,
        'RequestID': request_id,
        'Nonce': nonce,
        'Token': token,
        'Timestamp': timestamp,
        'Content-Type': 'application/json'
    }
    response = requests.request("GET", url, headers=headers, data=json.dumps({}))
    response_attr = json.loads(response.content.decode("utf-8"))
    if response_attr.get('status') == 'Success':
        print(response_attr.get('data'))
        break
```

对该代码进行运行，并查看结果：

![image-20230515220406183](https://www.images.wiki/z8vrkGbgqbfjAr9b9cbC.png)

结果对应的图片信息：

![image-20230515220429001](https://www.images.wiki/64Cq7GaSbCeCGwb76D6a.png)

另外，值得重点关注的两个信息：

1. 在没有获得到结果时，异步请求的状态码为202，返回结果为：`{"message": "There are currently no results, please try again later.", "help": "https://www.aigcaas.cn/article/38.html"}`
2. 可以通过休眠+轮训方式，进行结果的持续获取，当获得到的状态码不为202，即可停止，例如：

```python
while True:
    time.sleep(1)
    url = "https://api.aigcaas.cn/v3/async"
    # 构建请求头
    nonce = str(random.randint(1, 10000))
    timestamp = str(int(time.time()))
    token = getSha256(("%s%s%s" % (timestamp, secret_key, nonce)))
    headers = {
        'SecretID': secret_id,
        'Aigcaas-Request-Id': response.headers["Aigcaas-Request-Id"], # 请求输出的 RequestID
        'Nonce': nonce,
        'Token': token,
        'Timestamp': timestamp,
        'Content-Type': 'application/json'
    }
    async_response = requests.request("GET", url, headers=headers, data=json.dumps({}))
    print(async_response.status_code, async_response.content)
    if async_response.status_code != 202: break
```

#### 通知模式

在轮询模式下，客户端首次发起请求，告知系统“我要通过异步请求的方式调用接口”，即客户端对 `/v3/application/{应用名}/action/{具体方法}`接口发起请求，此时在请求头中增加了`Aigcaas-Async`参数用来声明异步调用，除此之外，还可以在请求头中增加参数`Aigcaas-Async-Callback`表示结果完成后的通知/回调地址。例如，在请求头中，增加：

```
Aigcaas-Async-Callback: https://www.abc.com/get_result
```

当 AIGCaaS 平台完成异步调用，获得到最终结果后，会将结果：

1. 以 POST 的请求方法，发送到 `https://www.abc.com/get_result`
2. Request Body 内容为模型服务的响应结果，可能是正确响应，也可能是报错信息；
3. Reqeust Headers：
   - `Nonce`: 随即参数，用于签名校验
   - `Token`: 签名字符串，sha256(Timestamp, SecretKey, Nonce)  ，其中`SecretKey`是发起请求所使用密钥对应的`SecretKey`
   - `Aigcaas-Source-Request-Id`:  任务的原始 ID
   - `Aigcaas-Request-Id`: 当前的请求 ID
   - `Aigcaas-Timestamp`: 时间戳，用户签名校验
   - `Aigcaas-Status`:  结果状态，`True`表示正常返回，`False`表示异常返回，也是通过该参数进行计费判断
   - `Content-Type`: 内容类型（统一为`application/json`）

所以，回调的 HTTP/HTTPS 服务，可以根据上面的内容进行签名信息判断，以确定该请求是否从 AIGCaaS 平台所发出。

当然，即便采用了通知模式，开发者在系统进行异步处理的过程中，仍然可以进行轮询操作，以获取更为明确的任务执行进度（注意，并不是所有的异步任务都具备进度查询能力）。

Tags :

关于 AIGCaaS

导航

应用与接口

平台协议

联系我们