帮助中心

开放策略平台

2026.08.04 11:36:11

Markdown 源码 LLMS.txt

1. 概述

易盾开放策略 API 允许第三方审核服务接入易盾内容安全平台，协同完成内容审核任务。

1.1 核心特性

异步回调模式：请求中携带 callback_url，审核结果通过回调上报
批量回调：一次回调可上报多条审核结果，减少网络开销
OpenAI 风格 content 数组：支持文本 + 图片混合，可扩展新类型
独立追踪：每条数据携带独立的 task_id，全链路可追溯

1.2 工作流程（异步模式）

易盾 CMA                          第三方厂商
   │                                  │
   ├─ POST {厂商 API URL} ──────────>│  1. 发送审核请求（含 callback_url + request_id + content）
   │                                  │
   │  <──────── HTTP 200 ────────────┤  2. 厂商立即响应（空结果即可）
   │                                  │
   │                                  │   3. 厂商异步处理审核...
   │                                  │
   │  <── POST {callback_url} ────────┤  4. 批量回调（request_id + data[]）
   │                                  │
   ├─ HTTP 200 ────────────────────>│ 5. 确认接收

1.3 约束

每条数据独立 task_id，回调时按 task_id 匹配
默认超时 300s 内完成所有回调，超时视为调用失败
同一 task_id 重复回调幂等处理，不会报错

1b. 异步回调接口（厂商 → CMA）

1b.1 回调接口

POST {callback_url}
Content-Type: application/json

callback_url 即请求体中收到的值，直接 POST，无需认证头。

1b.2 回调请求体

{
  "request_id": "a1b2c3d4e5f6...",
  "data": [
    {
      "task_id": "d7f3a8e2-9b4c-11e8-98d0-529269fb1459",
      "data_id": "yd_data_001",
      "status": 0,
      "result": {
        "suggestion": "pass",
        "spam_type": 100,
        "explain": "内容正常"
      }
    }
  ]
}

1b.3 回调字段说明

字段	类型	必填	说明
`request_id`	string	否	原样返回请求体中的 `request_id`，或自行生成 UUID
`data`	array	是	审核结果列表，每条对应一个 `task_id`
`data[].task_id`	string	是	与请求体中收到的值一致
`data[].data_id`	string	否	原样返回
`data[].status`	int	是	`0`-成功（result 必填），`1`-失败（error 必填）
`data[].result.suggestion`	string	条件	`"pass"` 或 `"unpass"`
`data[].result.spam_type`	int	否	违规类型编号
`data[].result.explain`	string	否	审核原因
`data[].error.code`	string	条件	错误码
`data[].error.message`	string	条件	错误描述

1b.4 回调响应

{"code": 200, "msg": "OK", "desc": null, "data": null}

响应体中 desc 和 data 字段为 null 时可忽略，仅关注 code 和 msg。

成功返回 HTTP 200。无效 task_id 时跳过该条，其余继续处理，仍返回 HTTP 200。

2. 快速开始

第一步：获取 API Key

在API key管理页面创建API key时自动生成 apiKey。目前存在多个apiKey只会使用第一个生效apiKey。

第二步：实现接口

参考以下最简实现快速验证联调：

import uuid
from flask import Flask, request, jsonify

app = Flask(__name__)

API_KEY = "your_api_key_here"

@app.route('/audit', methods=['POST'])
def audit():
    # 验证 apiKey
    auth = request.headers.get('Authorization', '')
    if auth != f'Bearer {API_KEY}':
        return jsonify({"code": 401, "message": "Unauthorized"}), 401

    data = request.json
    request_id = data.get("request_id", str(uuid.uuid4()))
    strategy_code = data.get("strategy_code")  # 可选，策略标识，用于区分业务来源
    callback_url = data.get("callback_url")

    results = []
    for req in data.get('requests', []):
        # 提取透传字段（可选，CMA 配置后才携带）
        extension_fields = req.get('extension', [])
        results.append({
            "task_id": req["task_id"],
            "data_id": req["data_id"],
            "status": 0,
            "result": {
                "suggestion": "pass",
                "spam_type": 0,
                "explain": "内容正常"
            }
        })

    # 异步回调（生产环境建议用任务队列）
    if callback_url:
        import threading, requests
        def callback():
            requests.post(callback_url, json={
                "request_id": request_id,
                "data": results
            })
        threading.Thread(target=callback, daemon=True).start()

    return jsonify({"code": 200, "message": "accepted"})


if __name__ == '__main__':
    app.run(host='0.0.0.0', port=8080)

第三步：创建开放平台策略

在开放平台页面创建开放平台策略，将您的 API 地址（如 https://your-domain.com/audit），在创建开放平台策略时设置

第四步：联调测试

在开放平台页面进行调试

3. 认证方式

所有请求均通过 HTTP Header 进行认证。

请求头

Content-Type: application/json
Authorization: Bearer {apiKey}
X-Request-Id: {请求唯一标识}
X-Yidun-Timestamp: {毫秒级时间戳}

请求头字段	类型	必填	说明
`Authorization`	string	是	Bearer Token，`apiKey` 由易盾创建开放策略时生成
`X-Request-Id`	string	是	易盾生成的请求唯一标识（UUID），用于日志追踪
`X-Yidun-Timestamp`	long	是	Unix 毫秒时间戳，用于防重放攻击

安全建议：请妥善保管 apiKey，不要暴露在客户端代码或日志中。

4. 接口规范

4.1 请求规范

基本信息

属性	说明
请求方法	`POST`
请求地址	由第三方自行提供，在易盾平台配置
Content-Type	`application/json`

请求体结构

{
  "request_id": "a1b2c3d4e5f6...",
  "strategy_code": "open@strategy",
  "callback_url": "https://{host}/open-strategy/v1/callback",
  "requests": [
    {
      "task_id": "d7f3a8e2-9b4c-11e8-98d0-529269fb1459",
      "data_id": "yd_data_001",
      "content": [
        {"type": "text", "data": "待审核文本内容"}
      ]
    },
    {
      "task_id": "b2c4e6f8-9a4c-11e8-98d0-529269fb145a",
      "data_id": "yd_data_002",
      "content": [
        {"type": "text", "data": "图片配文"},
        {"type": "image_url", "data": "https://example.com/img.jpg"}
      ],
      "extension": [
        {"key": "ip", "value": "1.2.3.4"},
        {"key": "spamtype", "value": 100}
      ]
    }
  ]
}

字段说明

字段	类型	必填	说明
`request_id`	string	否	请求标识（UUID），CMA 生成
`strategy_code`	string	否	策略标识（`"open@"` 前缀），供调用方区分业务来源
`callback_url`	string	是	异步回调地址，审核完成后回调此 URL
`requests`	array	是	审核数据列表
`requests[].task_id`	string	是	128-bit UUID，回调时必须携带，CMA 生成
`requests[].data_id`	string	是	数据源标识，回调时原样返回
`requests[].content`	array	是	内容列表（OpenAI 风格），支持混合多种类型
`requests[].content[].type`	string	是	`"text"` / `"image_url"`（后续扩展 `audio_url` 等）
`requests[].content[].data`	string	是	内容数据，文本或 URL
`requests[].extension`	array	否	透传字段列表，每条数据独立。字段由开放策略配置决定，未配置时此字段不输出
`requests[].extension[].key`	string	是	字段名（小写），如 `ip`、`extstr1`、`extension-age`
`requests[].extension[].value`	any	是	字段值，类型取决于数据源

透传字段（extension）

CMA 可在每个 requests[i] 中携带 extension 字段，将业务侧原始数据中的指定字段透传给第三方厂商。透传字段列表由开放策略配置决定，CMA 在调用厂商 API 时自动筛选对应字段并填充。

字段来源：透传字段值取自送审时携带的业务参数（如 ip、spamType、extStr1 等），与 CMA 内部的机审字段来源一致。

命名规则：

字段名统一小写，大小写不敏感匹配（IP、ip、Ip 均视为相同字段）
extension- 前缀表示从扩展参数 JSON 中提取，如 extension-age 匹配 {"age": 25} 中的 25
同一策略最多配置 10 个透传字段

行为约定：

策略未配置透传字段时，extension 不出现在请求 JSON 中
某字段在当前数据中不存在或值为空时，跳过该字段不进入 extension 数组
批量请求中，每条数据的 extension 独立计算，按索引从对应数据源的参数中取值
JSON 解析失败的 extension- 字段会被忽略，不抛错

content 类型

type	data 说明
`text`	待审核文本
`image_url`	图片 URL

4.2 厂商响应（异步模式）

厂商收到审核请求后，只需立即返回 HTTP 200，审核结果通过回调接口上报（见 1b. 异步回调接口）。

HTTP/1.1 200 OK
Content-Type: application/json

{"code": 200, "message": "accepted"}

回调格式：POST {callback_url}，请求体 {"request_id": "...", "data": [...]}，详见回调接口章节。

4.3 错误处理

审核失败通过回调上报错误项（status: 1 + error），格式见回调接口。

HTTP 错误码

HTTP 状态码	code	说明
`400`	400	请求参数错误（格式错误、缺少必填字段等）
`401`	401	认证失败（`apiKey` 无效或缺失）
`403`	403	无权限访问
`429`	429	请求频率超限（QPS 限流）
`500`	500	服务器内部错误
`503`	503	服务暂时不可用

5. 数据字典

5.1 spam_type 枚举值

值	说明	分类	对应 suggestion
`0`	正常	—	`pass`
`100`	色情	色情	`unpass`
`110`	性感低俗	色情	`unpass`
`200`	广告	广告	`unpass`
`210`	二维码	广告	`unpass`
`260`	广告法违规	广告	`unpass`
`300`	暴恐	暴恐	`unpass`
`400`	违禁	违禁	`unpass`
`500`	涉政	涉政	`unpass`
`600`	谩骂	不良信息	`unpass`
`700`	灌水	不良信息	`unpass`
`800`	恶心类	不良信息	`unpass`
`900`	其他	其他	`unpass`
`1050`	噪音	质量问题	`unpass`
`1100`	涉价值观	价值观	`unpass`

说明：若无法明确分类，可返回一级分类值（如 100 色情）或 900（其他）。

5.2 单条数据错误码

code	说明	触发场景
`INVALID_INPUT`	参数错误	文本为空、图片 URL 无效、参数格式错误
`SERVICE_UNAVAILABLE`	服务不可用	审核引擎不可用、依赖服务故障
`PROCESSING_ERROR`	处理异常	审核引擎内部错误、OCR 失败
`TIMEOUT`	处理超时	单条数据审核超时
`RATE_LIMIT_EXCEEDED`	超过频率限制	内部 QPS 限流

6. 完整示例

6.1 纯文本审核

请求

POST https://your-domain.com/audit
Content-Type: application/json
Authorization: Bearer yd_api_key_xxxx

{
  "request_id": "req_text_001",
  "strategy_code": "open@strategy",
  "callback_url": "https://{host}/open-strategy/v1/callback",
  "requests": [
    {
      "task_id": "task_text_001",
      "data_id": "data_001",
      "content": [{"type": "text", "data": "第一条评论内容"}],
      "extension": [
        {"key": "ip", "value": "1.2.3.4"},
        {"key": "spamtype", "value": 100},
        {"key": "extension-age", "value": 25}
      ]
    },
    {
      "task_id": "task_text_002",
      "data_id": "data_002",
      "content": [{"type": "text", "data": "第二条评论内容"}],
      "extension": [
        {"key": "ip", "value": "5.6.7.8"},
        {"key": "extension-age", "value": 30}
      ]
    }
  ]
}

厂商立即响应

{"code": 200, "message": "accepted"}

厂商异步回调

{
  "request_id": "req_text_001",
  "data": [
    {
      "task_id": "task_text_001",
      "data_id": "data_001",
      "status": 0,
      "result": {"suggestion": "pass", "spam_type": 0, "explain": "内容正常"}
    },
    {
      "task_id": "task_text_002",
      "data_id": "data_002",
      "status": 0,
      "result": {"suggestion": "unpass", "spam_type": 900, "explain": "检测到违规内容"}
    }
  ]
}

6.2 纯图片审核

请求

{
  "request_id": "req_img_001",
  "strategy_code": "open@strategy",
  "callback_url": "https://{host}/open-strategy/v1/callback",
  "requests": [
    {
      "task_id": "task_img_001",
      "data_id": "data_img_001",
      "content": [{"type": "image_url", "data": "https://cdn.example.com/img1.jpg"}],
      "extension": [
        {"key": "extstr1", "value": "scene_001"},
        {"key": "extension-age", "value": 18}
      ]
    },
    {
      "task_id": "task_img_002",
      "data_id": "data_img_002",
      "content": [
        {"type": "image_url", "data": "https://cdn.example.com/img2_a.jpg"},
        {"type": "image_url", "data": "https://cdn.example.com/img2_b.jpg"}
      ],
      "extension": [
        {"key": "extstr1", "value": "scene_002"},
        {"key": "ip", "value": "10.0.0.1"}
      ]
    }
  ]
}

厂商异步回调

{
  "request_id": "req_img_001",
  "data": [
    {
      "task_id": "task_img_001",
      "data_id": "data_img_001",
      "status": 0,
      "result": {"suggestion": "pass", "spam_type": 0, "explain": "图片内容正常"}
    },
    {
      "task_id": "task_img_002",
      "data_id": "data_img_002",
      "status": 0,
      "result": {"suggestion": "unpass", "spam_type": 200, "explain": "图片含违规广告"}
    }
  ]
}

6.3 图文混合审核

请求

{
  "request_id": "req_mix_001",
  "strategy_code": "open@strategy",
  "callback_url": "https://{host}/open-strategy/v1/callback",
  "requests": [
    {
      "task_id": "task_mix_001",
      "data_id": "data_mix_001",
      "content": [
        {"type": "text", "data": "今天天气真好"},
        {"type": "image_url", "data": "https://cdn.example.com/photo1.jpg"}
      ],
      "extension": [
        {"key": "ip", "value": "192.168.1.1"},
        {"key": "extension-scene", "value": "social"},
        {"key": "extension-age", "value": 22}
      ]
    },
    {
      "task_id": "task_mix_002",
      "data_id": "data_mix_002",
      "content": [
        {"type": "text", "data": "美食打卡"},
        {"type": "image_url", "data": "https://cdn.example.com/food1.jpg"},
        {"type": "image_url", "data": "https://cdn.example.com/food2.jpg"}
      ],
      "extension": [
        {"key": "ip", "value": "192.168.1.2"},
        {"key": "extension-age", "value": 35}
      ]
    }
  ]
}

厂商异步回调

{
  "request_id": "req_mix_001",
  "data": [
    {
      "task_id": "task_mix_001",
      "data_id": "data_mix_001",
      "status": 0,
      "result": {"suggestion": "pass", "spam_type": 0, "explain": "图文内容正常"}
    },
    {
      "task_id": "task_mix_002",
      "data_id": "data_mix_002",
      "status": 0,
      "result": {"suggestion": "pass", "spam_type": 0, "explain": "美食分享正常"}
    }
  ]
}

7. 接入指南

7.1 开发步骤

步骤 1：实现认证校验

接收 HTTP 请求头中的 Authorization: Bearer {apiKey}，验证 apiKey 有效性。

步骤 2：解析请求体

解析 JSON 请求体，提取 strategy_code、callback_url，遍历 requests 数组，从每个元素提取 task_id、data_id、content。

步骤 3：异步审核处理

遍历 content[]，根据 type 分流至文本或图片审核引擎

步骤 4：回调上报结果

处理完毕后 POST {callback_url}，请求体格式见回调接口
data[] 中每条必须包含 task_id（原样返回）
成功项 status: 0 + result，失败项 status: 1 + error

7.2 完整实现示例（Python）

import uuid
import threading
import requests
from flask import Flask, request, jsonify
from concurrent.futures import ThreadPoolExecutor

app = Flask(__name__)
executor = ThreadPoolExecutor(max_workers=10)
VALID_API_KEY = "your_api_key_here"


def validate_api_key(auth_header: str) -> bool:
    return auth_header == f"Bearer {VALID_API_KEY}"


def audit_text(text: str) -> tuple:
    """文本审核：返回 (suggestion, spam_type, explain)"""
    return "pass", 0, "内容正常"


def audit_image(image_url: str) -> tuple:
    """图片审核：返回 (suggestion, spam_type, explain)"""
    return "pass", 0, "图片正常"


def process_single(req: dict) -> dict:
    task_id = req["task_id"]
    data_id = req["data_id"]
    content_list = req.get("content", [])
    extension_fields = req.get("extension", [])  # CMA 透传字段（可选）

    suggestion, spam_type, explain = "pass", 0, ""
    for item in content_list:
        if item["type"] == "image_url":
            suggestion, spam_type, explain = audit_image(item["data"])
        elif item["type"] == "text" and item.get("data"):
            suggestion, spam_type, explain = audit_text(item["data"])

    return {
        "task_id": task_id,
        "data_id": data_id,
        "status": 0,
        "result": {
            "suggestion": suggestion,
            "spam_type": spam_type,
            "explain": explain,
        },
    }


def do_callback(callback_url: str, request_id: str, results: list):
    """异步回调 CMA"""
    payload = {"request_id": request_id, "data": results}
    requests.post(callback_url, json=payload, timeout=10)


@app.route("/audit", methods=["POST"])
def audit():
    auth_header = request.headers.get("Authorization", "")
    if not validate_api_key(auth_header):
        return jsonify({"code": 401, "message": "Unauthorized"}), 401

    body = request.json
    if not body or "requests" not in body:
        return jsonify({"code": 400, "message": "Bad Request"}), 400

    request_id = body.get("request_id", str(uuid.uuid4()))
    strategy_code = body.get("strategy_code")  # 可选，策略标识，用于区分业务来源
    callback_url = body.get("callback_url")

    # 并发处理
    results = list(executor.map(process_single, body["requests"]))

    # 异步回调（生产环境建议用消息队列）
    if callback_url:
        threading.Thread(
            target=do_callback, args=(callback_url, request_id, results), daemon=True
        ).start()

    return jsonify({"code": 200, "message": "accepted"})


if __name__ == "__main__":
    app.run(host="0.0.0.0", port=8080)

8. 常见问题

Q：request_id、task_id、data_id 分别有什么用？

request_id：请求标识，标识整个请求，可选，主要用于日志追踪。
task_id：任务标识，每条数据独立，回调时必须携带用于匹配。
data_id：数据标识，业务侧的数据 ID。

task_id 和 data_id 需原样返回。

Q：回调中的结果顺序是否需要与请求保持一致？

不需要。易盾通过 task_id 进行匹配，结果顺序无关。

Q：有没有回调超时限制？

cma 调试默认 300 秒。300 秒内未收到所有 `task_id` 的回调，视为调用失败。智能体使用智能体配置的超时时间。

Q：同一 task_id 多次回调会怎样？

任务处理期间重复回调幂等处理（返回 200），不会覆盖已有结果。任务完成后再次回调会返回错误。

Q：部分数据审核失败时如何处理？

回调时对应项的 status 设为 1，附带 error 对象描述失败原因。易盾侧对所有失败项统一放行。

Q：strategy_code 是什么？

strategy_code 是 CMA 发送的策略标识字段，格式为 "open@" 前缀（如 "open@xxxx"），用于告知第三方厂商当前调用来自哪个开放策略。第三方可根据该字段区分不同业务来源，实现差异化处理（如路由到不同审核引擎、使用不同阈值等）。该字段为可选字段，CMA 在策略已配置的情况下会携带。

Q：extension 透传字段是什么？如何使用？

extension 是 CMA 在每个 requests[i] 中提供的透传字段数组，包含业务方送审时携带的原始数据字段（如 ip、spamType 等）。第三方厂商可根据这些字段做辅助判断或日志追踪。

透传哪些字段由 CMA 侧在开放策略中配置（最多 10 个），厂商侧无需做任何改动
字段名统一小写，同一策略内不区分大小写
未配置透传字段时，请求 JSON 中不包含 extension 字段
某字段在当前数据中不存在或值为空时，该字段不会出现在数组中
extension- 前缀的字段来自数据源的扩展参数 JSON（如 extension-scene 对应 {"scene": "xxx"}）

9. 变更记录

版本	日期	变更内容
v4.3	2026-07-02	请求体 `requests[]` 新增可选字段 `extension`，支持透传业务侧字段（如 `ip`、`spamType`、`extension-age` 等）。透传字段由开放策略配置决定，每条数据独立，未配置时不输出
v4.2	2026-06-10	请求体新增可选字段 `strategy_code`（`"open@"` 前缀），供调用方区分业务来源
v4.1	2026-05-20	初始版本

上一篇：智能体配置攻略

对文档做个评价吧：

如果遇到产品相关问题，您可提交工单或在线咨询寻求帮助。

在线咨询电话咨询：95163223 免费试用

目录

1. 概述

1.1 核心特性

1.2 工作流程（异步模式）

1.3 约束

1b. 异步回调接口（厂商 → CMA）

1b.1 回调接口

1b.2 回调请求体

1b.3 回调字段说明

1b.4 回调响应

2. 快速开始

第一步：获取 API Key

第二步：实现接口

第三步：创建开放平台策略

第四步：联调测试

3. 认证方式

请求头

4. 接口规范

4.1 请求规范

基本信息

请求体结构

字段说明

透传字段（extension）

content 类型

4.2 厂商响应（异步模式）

4.3 错误处理

HTTP 错误码

5. 数据字典

5.1 spam_type 枚举值

5.2 单条数据错误码

6. 完整示例

6.1 纯文本审核

6.2 纯图片审核

6.3 图文混合审核

7. 接入指南

7.1 开发步骤

7.2 完整实现示例（Python）

8. 常见问题

cma 调试默认 300 秒。300 秒内未收到所有 task_id 的回调，视为调用失败。 智能体使用智能体配置的超时时间。

9. 变更记录

cma 调试默认 300 秒。300 秒内未收到所有 `task_id` 的回调，视为调用失败。智能体使用智能体配置的超时时间。