一、概述
JoySafety 是一个大模型内容安全防护系统，提供敏感词检测、模型分类、知识库搜索等检测能力，并通过 DAG 策略编排检测流程。
核心架构：
1
| 原子能力(Atomic Capability) → 业务(Business) → DAG 策略 → 审核接口(Defense API)
|
快速上线三步走：
- 上线原子能力 → 注册模型/检测能力
- 上线业务 → 创建业务并启用
- 上线 DAG 策略 → 编排检测流程
二、快速上线三步走
2.1 上线原子能力
原子能力是检测的最小单元，如敏感词匹配、BERT 分类器、FastText 分类器等。
请求格式
1
2
3
4
5
6
7
8
9
10
| curl -X POST http://{admin_host}:8006/config/defense/manage/function/new \
-H 'Content-Type: application/json' \
-d '{
"name": "原子能力名称",
"group": "default",
"desc": "描述",
"type": "single_label_pred", # 能力类型
"timeoutMilliseconds": 2000,
"confObj": { ... }
}'
|
能力类型（type）枚举
| type | 说明 | 示例 |
|---|
keyword | 敏感词/正则匹配 | 黑白名单检测 |
single_label_pred | 单标签分类模型 | BERT、FastText |
kb_search | 知识库语义搜索 | 红线知识检索 |
rag_answer | RAG 代答 | 基于知识库的回答 |
multi_turn_detect | 多轮对话检测 | 对话上下文安全检测 |
各类能力注册示例
1. keyword（敏感词）
1
2
3
4
5
6
7
8
9
10
11
12
13
| curl -X POST http://localhost:8006/config/defense/manage/function/new \
-H 'Content-Type: application/json' \
-d '{
"name": "keyword",
"group": "default",
"desc": "敏感词识别",
"type": "keyword",
"timeoutMilliseconds": 1000,
"confObj": {
"name": "敏感词服务",
"url": "http://safety-keywords:8005/keyword/query"
}
}'
|
2. BERT 分类器（single_label_pred）
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
| curl -X POST http://localhost:8006/config/defense/manage/function/new \
-H 'Content-Type: application/json' \
-d '{
"name": "bert_20250916",
"group": "default",
"desc": "BERT通用安全分类器",
"type": "single_label_pred",
"timeoutMilliseconds": 2000,
"confObj": {
"extra": {
"url": "http://safety-bert:8003/v1/bert/bert_20250916"
},
"name": "bert_20250916",
"modelType": "bert",
"ignoreRiskCode": []
}
}'
|
3. FastText 分类器（single_label_pred）
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
| curl -X POST http://localhost:8006/config/defense/manage/function/new \
-H 'Content-Type: application/json' \
-d '{
"name": "fast20250710",
"group": "default",
"desc": "FastText快速分类器",
"type": "single_label_pred",
"timeoutMilliseconds": 100,
"confObj": {
"extra": {
"url": "http://safety-fasttext:8002/fasttext/fast20250710"
},
"name": "fast20250710",
"modelType": "fasttext",
"ignoreRiskCode": []
}
}'
|
4. kb_search（知识库搜索）
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
| curl -X POST http://localhost:8006/config/defense/manage/function/new \
-H 'Content-Type: application/json' \
-d '{
"name": "kb_search",
"group": "default",
"desc": "红线知识检索",
"type": "kb_search",
"timeoutMilliseconds": 2000,
"confObj": {
"topK": 5,
"threshold": 0.8,
"collection": "red_kg_prod_hnsw_bge512_20250826",
"url": "http://safety-knowledge:8004/knowledge/search"
}
}'
|
5. rag_answer（RAG 代答）
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
| curl -X POST http://localhost:8006/config/defense/manage/function/new \
-H 'Content-Type: application/json' \
-d '{
"name": "rag_answer",
"group": "default",
"desc": "红线代答",
"type": "rag_answer",
"timeoutMilliseconds": 5000,
"confObj": {
"topK": 5,
"threshold": 0.8,
"collection": "hnsw_bge512_20250826",
"url": "http://safety-knowledge:8004/knowledge/answer"
}
}'
|
6. multi_turn_detect（多轮对话检测）
1
2
3
4
5
6
7
8
9
10
11
12
13
| curl -X POST http://localhost:8006/config/defense/manage/function/new \
-H 'Content-Type: application/json' \
-d '{
"name": "multi_turn_detect",
"group": "default",
"desc": "多轮对话检测",
"type": "multi_turn_detect",
"timeoutMilliseconds": 3000,
"confObj": {
"maxTurns": 20,
"url": "http://safety-knowledge:8004/dialog"
}
}'
|
上线原子能力
创建成功后会返回 id，用该 id 调用上线接口：
1
2
3
4
5
6
7
| # 获取返回的 id
# { "data": { "id": 1 } }
# 上线
curl -X POST http://localhost:8006/config/defense/manage/function/online \
-H 'Content-Type: application/json' \
-d '{"id": 1}'
|
一键初始化脚本
也可使用 Python 脚本批量注册并上线：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
| import json
import requests
ADMIN_HOST = "http://localhost:8006"
skills = [
{
"name": "keyword",
"group": "default",
"desc": "敏感词识别",
"type": "keyword",
"timeoutMilliseconds": 1000,
"confObj": {
"name": "敏感词服务",
"url": "http://safety-keywords:8005/keyword/query"
}
},
{
"name": "fast20250710",
"group": "default",
"desc": "FastText分类器",
"type": "single_label_pred",
"timeoutMilliseconds": 100,
"confObj": {
"extra": {"url": "http://safety-fasttext:8002/fasttext/fast20250710"},
"name": "fast20250710",
"modelType": "fasttext",
"ignoreRiskCode": []
}
},
{
"name": "bert_20250916",
"group": "default",
"desc": "BERT通用分类器",
"type": "single_label_pred",
"timeoutMilliseconds": 2000,
"confObj": {
"extra": {"url": "http://safety-bert:8003/v1/bert/bert_20250916"},
"name": "bert_20250916",
"modelType": "bert",
"ignoreRiskCode": []
}
},
{
"name": "bert_prompt_injection_20250828",
"group": "default",
"desc": "BERT提示词注入分类器",
"type": "single_label_pred",
"timeoutMilliseconds": 2000,
"confObj": {
"extra": {"url": "http://safety-bert:8003/v1/bert/bert_prompt_injection_20250828"},
"name": "bert_prompt_injection_20250828",
"modelType": "bert",
"ignoreRiskCode": []
}
}
]
for skill in skills:
resp = requests.post(f"{ADMIN_HOST}/config/defense/manage/function/new",
data=json.dumps(skill),
headers={'Content-Type': 'application/json'}).json()
func_id = resp['data']['id']
requests.post(f"{ADMIN_HOST}/config/defense/manage/function/online",
data=json.dumps({"id": func_id}),
headers={'Content-Type': 'application/json'})
print(f"上线原子能力: {skill['name']}, id={func_id}")
|
2.2 上线业务
业务是检测流程的入口，每个业务有唯一的 accessKey。
创建业务
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
| curl -X POST http://localhost:8006/config/defense/manage/business/new \
-H 'Content-Type: application/json' \
-d '{
"name": "my_business",
"group": "default",
"desc": "我的业务",
"type": "toC",
"secretKey": "your-secret-key-here",
"qpsLimit": 20,
"robotCheckConfObj": {
"checkNum": 1,
"unit": "slice_single",
"timeoutMilliseconds": 5000
},
"userCheckConfObj": {
"checkNum": 1,
"unit": "message_single",
"timeoutMilliseconds": 5000
}
}'
|
参数说明：
| 参数 | 必填 | 说明 |
|---|
| name | 是 | 业务名称 |
| group | 是 | 业务分组 |
| type | 是 | 业务类型：toB（商户）、toC（C端用户）、toE（内部用户） |
| secretKey | 是 | 业务的密钥，用于 API 签名验证 |
| qpsLimit | 是 | QPS 限制 |
| robotCheckConfObj | 是 | 机器人输出检测配置 |
| userCheckConfObj | 是 | 用户输入检测配置 |
上线业务
1
2
3
| curl -X POST http://localhost:8006/config/defense/manage/business/online \
-H 'Content-Type: application/json' \
-d '{"id": 业务id}'
|
2.3 上线 DAG 策略
DAG 策略编排多个原子能力，形成完整的检测流程。
默认 test 业务的审核流程
系统初始化时（strategy_init.py）为 test 业务加载了以下检测流程：
1
2
3
4
5
6
7
8
9
10
11
| 用户输入 → 关键词检测 (parallel, timeout=1000ms)
├── 命中黑名单(1) → 拦截，结束
├── 命中白名单(2) → 放行，结束
└── 未命中/灰名单 → BERT 双模型并行 (parallel, timeout=2000ms)
├── 两个模型都无风险 → 正常，结束
├── 通用模型高置信度(概率≥阈值) → 按通用模型结果拦截/放行
├── 通用模型正常概率极低(<0.1) → 按通用模型结果拦截/放行
├── 注入模型高置信度(概率≥阈值) → 按注入模型结果拦截/放行
└── 都不够确信 → 正常，结束
机器人输出 → 关键词检测 → 无脑结束
|
说明：
- 用户输入走两层：先关键词快速过滤，不确定再进 BERT 精判
- 机器人输出只走关键词检测
- 默认策略没有 FastText 分类器，所有关键词未命中的请求都会进入 BERT 层（耗时约 2 秒）
策略执行流程（含 FastText）
1
2
3
4
5
6
7
8
9
10
11
12
| 用户输入 → 关键词 + FastText 并行 (parallel, timeout=200ms)
├── 关键词黑白名单 → 直接结束
├── FastText 高置信度正常(>0.9) → 放行，结束
├── FastText 高置信度恶意(>0.95) → 拦截，结束
└── 不确定 → BERT 双模型并行 (parallel, timeout=2000ms)
├── 两个模型都无风险 → 正常，结束
├── 通用模型高置信度(概率≥阈值) → 按通用模型结果拦截/放行
├── 通用模型正常概率极低(<0.1) → 按通用模型结果拦截/放行
├── 注入模型高置信度(概率≥阈值) → 按注入模型结果拦截/放行
└── 都不够确信 → 正常，结束
机器人输出 → 关键词检测 → 无脑结束
|
策略 YAML 模板（含 FastText，推荐）
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
| businessName: my_business
group: default
desc: 我的业务策略
rootId: start
confArray:
- nodeId: start
functionConf:
type: dummy
name: dummy
timeoutMilliseconds: 5
conf: {}
routerConf:
type: groovy
name: groovy
conf:
script: |-
def from_robot=ctx.curReq[0].fromRobot()
if (from_robot) {
return 'from_robot_path'
} else {
return 'from_user_path'
}
ignoreError: true
# ========== 用户输入检测 ==========
- nodeId: from_user_path
functionConf:
type: parallel
name: parallel_from_user
timeoutMilliseconds: 200
conf:
functionConfs:
- ref: keyword
- ref: fast20250710
routerConf:
type: groovy
name: groovy
conf:
script: |-
import com.jd.security.llmsec.core.check.MixedCheck
def retMap=ctx.curResult.resultMap
def kw = retMap.keyword
def fast = retMap.fast20250710
# 关键词命中黑白名单 → 直接决策
if (kw != null && (kw.bwgLabel == 1 || kw.bwgLabel == 2)) {
ctx.curResult = kw
return null # 结束
}
# FastText 高置信度正常 → 放行
if (fast != null && !fast.hasRisk() && fast.probability > 0.9) {
ctx.curResult = fast
return null # 结束
}
# FastText 高置信度恶意 → 拦截
if (fast != null && fast.hasRisk() && fast.probability > 0.95) {
ctx.curResult = fast
return null # 结束
}
# 不确定 → 进 BERT 层
return 'from_user_bert'
ignoreError: true
- nodeId: from_user_bert
functionConf:
type: parallel
name: parallel_bert
timeoutMilliseconds: 2000
conf:
functionConfs:
- ref: bert_20250916
conf:
ignoreRiskCode:
- 1006
- ref: bert_prompt_injection_20250828
conf:
ignoreRiskCode:
- 1006
routerConf:
type: groovy
name: groovy
conf:
script: |-
import com.jd.security.llmsec.core.check.MixedCheck
def retMap=ctx.curResult.resultMap
def bert_common = retMap.bert_20250916
def bert_inject = retMap.bert_prompt_injection_20250828
# 两个模型都无风险 → 正常
if ((bert_common == null || !bert_common.hasRisk()) && (bert_inject == null || !bert_inject.hasRisk())) {
ctx.curResult = MixedCheck.noRisk()
return null
}
# 设置阈值
def thresholds_common = [0:0.9, 201:1.0, 1001:0.9, 1002:0.9, 1003:0.85, 1004:0.996, 1005:0.97, 1006:0.95, 1007:0.9]
def thresholds_inject = [0:0.8, 201:0.99]
# 通用模型高置信度命中
if (bert_common.hasRisk() && bert_common.probability >= thresholds_common[bert_common.riskCode]) {
ctx.curResult = bert_common
}
# 通用模型正常概率极低
else if(bert_common.detail.find { it.riskCode == 0 }?.probability < 0.1 && bert_common.riskCode != 201) {
ctx.curResult = bert_common
}
# 注入模型高置信度命中
else if(bert_inject.probability >= thresholds_inject[bert_inject.riskCode]) {
ctx.curResult = bert_inject
}
# 都不够确信 → 视为正常
else {
ctx.curResult = MixedCheck.noRisk()
}
return null
ignoreError: true
# ========== 机器人输出检测 ==========
- nodeId: from_robot_path
functionConf:
ref: keyword
routerConf:
type: stupid_end
name: stupid_end
ignoreError: true
|
上线 DAG 策略
1
2
3
4
5
6
7
8
9
10
11
| # 1. 创建 DAG 策略
curl -X POST http://localhost:8006/config/defense/manage/dag/newDagWithYaml \
-H 'Content-Type: text/plain' \
-d '上面的YAML内容'
# 返回示例：{ "id": 5, "version": 1, "status": "edit" }
# 2. 上线 DAG 策略
curl -X POST http://localhost:8006/config/defense/manage/dag/online \
-H 'Content-Type: application/json' \
-d '{"id": 5}'
|
更新 DAG 策略
如果已有线上 DAG，需要先创建新版本再升级：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
| # 1. 获取当前线上 DAG id
curl -X POST http://localhost:8006/config/defense/manage/dag/active \
-H 'Content-Type: application/json' \
-d '{"group": "default", "name": "my_business"}'
# 2. 基于线上 id 创建新版本
curl -X POST http://localhost:8006/config/defense/manage/dag/newVersion \
-H 'Content-Type: application/json' \
-d '{"id": 线上id}'
# 返回新 id
# 3. 更新新版本的内容
curl -X POST http://localhost:8006/config/defense/manage/dag/dagUpdateYaml \
-H 'Content-Type: text/plain' \
-d 'id: 新id
businessName: my_business
...（其他YAML内容，需带id字段）'
# 4. 升级上线
curl -X POST http://localhost:8006/config/defense/manage/dag/upgrade \
-H 'Content-Type: application/json' \
-d '{"id": 新id}'
|
更新流程说明
1
2
3
4
| dagUpdateYaml（id=线上id）→ 直接修改原记录，id 不变，仅限 edit 状态
newVersion（id=线上id）→ 创建新版本，返回新 id（推荐流程）
update / dagUpdateYaml（id=新id）→ 修改新版本内容
upgrade（id=新id）→ 新版本上线，旧版本自动下线
|
附：敏感词管理
敏感词（关键词）检测是最基础的过滤手段，使用 AC 自动机（NFA）算法，毫秒级响应。
1. 敏感词字段说明
| 字段 | 必填 | 说明 |
|---|
| businessScene | 是 | 业务场景（对应业务名称），all 表示全局生效 |
| group | 是 | 业务分组 |
| word | 是 | 敏感词内容 |
| className | 是 | 类别名称，如"辱骂"、“涉黄” |
| classNo | 是 | 类别编码 |
| matchType | 是 | 匹配方式（见下表） |
| handleStrategy | 是 | 处置策略：black(黑名单)、white(白名单)、grey(灰名单) |
| status | 是 | online(有效)、offline(无效) |
2. 匹配方式（matchType）
| matchType | 说明 | 示例 |
|---|
contain | 包含指定词即命中 | word=“王八蛋”，内容含"王八蛋"即命中 |
equal | 完全相等才命中 | word=“你好”，内容必须完全等于"你好" |
regex | 正则表达式匹配 | word=“杀|死”，匹配其中任意一个 |
exclude | 排除匹配（在白名单中排除某些内容） | word=“好”，排除"好"字所在的匹配 |
3. 处置策略（handleStrategy）
| handleStrategy | 说明 | DAG 中的行为 |
|---|
black | 黑名单（bwgLabel=1） | 命中后直接拦截，from_user_test1 节点 router 返回 null 结束 |
white | 白名单（bwgLabel=2） | 命中后直接放行，from_user_test1 节点 router 返回 null 结束 |
grey | 灰名单（bwgLabel=3） | 命中后不确定，继续进入 BERT 层精判 |
4. 新增敏感词
1
2
3
4
5
6
7
8
9
10
11
12
13
14
| curl -X POST http://localhost:8006/data/sensitive_words/upsert \
-H 'Content-Type: application/json' \
-d '{
"businessScene": "test",
"group": "default",
"word": "王八蛋",
"className": "辱骂",
"classNo": 1006,
"matchType": "contain",
"handleStrategy": "black",
"source": "手动录入",
"status": "online",
"desc": "辱骂词"
}'
|
5. 查询敏感词
1
2
3
4
5
6
7
8
9
10
11
12
13
| # 按词模糊查询
curl -X POST http://localhost:8006/data/sensitive_words/queryByWord \
-H 'Content-Type: application/json' \
-d '{
"word": "杀",
"status": "online",
"businessScene": "test"
}'
# 按 ID 查询
curl -X POST http://localhost:8006/data/sensitive_words/getById \
-H 'Content-Type: application/json' \
-d '{"id": 1}'
|
6. 上线/下线敏感词
1
2
3
4
5
6
7
8
9
| # 上线
curl -X POST http://localhost:8006/data/sensitive_words/online \
-H 'Content-Type: application/json' \
-d '{"id": 1}'
# 下线
curl -X POST http://localhost:8006/data/sensitive_words/offline \
-H 'Content-Type: application/json' \
-d '{"id": 1}'
|
7. 默认演示数据
系统提供 4 条演示敏感词：
| 敏感词 | matchType | handleStrategy | 类别 |
|---|
| 王八蛋 | contain | black | 辱骂 |
| 小箩筣av | contain | black | 涉黄 |
| 成人a在线 | contain | black | 涉黄 |
| 我艹 | contain | black | 辱骂 |
注意： 生产环境中需要根据实际业务场景补充完整的敏感词库，演示数据不足以覆盖所有违规内容。
8. 敏感词在 DAG 中的执行流程
1
2
3
4
5
6
7
| 用户输入
↓
keyword 服务 (AC 自动机)
├── 命中黑名单(bwgLabel=1) → router 直接拦截，不进 BERT
├── 命中白名单(bwgLabel=2) → router 直接放行，不进 BERT
├── 命中灰名单(bwgLabel=3) → 不确定，进 BERT 层精判
└── 未命中 → 进 BERT 层精判
|
三、使用审核接口
3.1 签名生成
调用审核接口需要对请求进行 HMAC-SHA1 签名。
签名算法
1
2
| signText = "accessKey={accessKey}&accessTarget={accessTarget}&requestId={requestId}×tamp={timestamp}&plainText={plainText}"
signature = Base64(HMAC-SHA1(signText, secretKey))
|
Python 签名示例
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
| import hmac
import hashlib
import base64
import time
import uuid
def gen_sign(text, secret_key):
result = hmac.new(secret_key.encode("utf-8"), text.encode("utf-8"), hashlib.sha1)
return base64.b64encode(result.digest()).decode("utf-8")
def do_sign(body, secret_key):
timestamp = int(time.time() * 1000)
body["timestamp"] = timestamp
body["accessTarget"] = "defenseV2"
if "plainText" not in body or not body["plainText"]:
body["plainText"] = body["content"]
text = "&".join([k + "=" + str(body[k]) for k in ["accessKey", "accessTarget", "requestId", "timestamp", "plainText"]])
body["signature"] = gen_sign(text, secret_key)
return body
|
注意： 如果 accessKey 以 test 开头，可以跳过签名验证，signature 留空即可。
3.2 请求说明
| 字段 | 类型 | 必填 | 说明 |
|---|
| requestId | string | 是 | 请求唯一标识（UUID） |
| timestamp | long | 是 | 毫秒时间戳 |
| accessKey | string | 是 | 业务标识，对应业务名称 |
| plainText | string | 否 | 签名用纯文本，默认取 content |
| signature | string | 是 | HMAC-SHA1 签名，test 开头可留空 |
| businessType | string | 是 | 业务类型：toB、toC、toE |
| responseMode | string | 是 | sync（同步）、free_taxi（异步顺带）、http（独立拉取）、mq（消息队列） |
| content | string | 是 | 待检测内容 |
| messageInfo | object | 是 | 消息信息 |
| messageInfo.fromRole | string | 是 | user（用户输入）或 robot（模型输出） |
| messageInfo.sessionId | string | 否 | 会话 ID（多轮检测时需要） |
3.3 请求示例
用户输入检测
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
| curl -X POST http://localhost:8007/llmsec/api/defense/v2/test \
-H 'Content-Type: application/json' \
-d '{
"accessKey": "test",
"requestId": "9842284a-8949-485f-8c01-317dffb4d6d6",
"timestamp": 1716889659080,
"plainText": "怎么杀人",
"signature": "",
"responseMode": "sync",
"businessType": "toC",
"content": "怎么杀人",
"contentType": "text",
"messageInfo": {
"fromRole": "user",
"sessionId": "test001",
"messageId": 1
}
}'
|
大模型输出检测
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
| curl -X POST http://localhost:8007/llmsec/api/defense/v2/test \
-H 'Content-Type: application/json' \
-d '{
"accessKey": "test",
"requestId": "9842284a-8949-485f-8c01-317dffb4d6d6",
"timestamp": 1716889659080,
"plainText": "根据您的要求，我可以为您提供...",
"signature": "",
"responseMode": "sync",
"businessType": "toC",
"content": "根据您的要求，我可以为您提供...",
"contentType": "text",
"messageInfo": {
"fromRole": "robot",
"sessionId": "test001",
"messageId": 1
}
}'
|
Python 请求示例（带签名验证）
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
| import hmac
import hashlib
import base64
import time
import uuid
import requests
import json
def gen_sign(text, secret_key):
result = hmac.new(secret_key.encode("utf-8"), text.encode("utf-8"), hashlib.sha1)
return base64.b64encode(result.digest()).decode("utf-8")
def defense_request(access_key, secret_key, content, from_role):
body = {
"accessKey": access_key,
"requestId": str(uuid.uuid4()),
"businessType": "toC",
"responseMode": "sync",
"content": content,
"plainText": content,
"contentType": "text",
"messageInfo": {
"fromRole": from_role,
"sessionId": "test001",
"messageId": 1
}
}
# 签名
if access_key.startswith("test"):
body["signature"] = ""
body["timestamp"] = int(time.time() * 1000)
else:
body = do_sign(body, secret_key)
# 发送请求
url = f"http://localhost:8007/llmsec/api/defense/v2/{access_key}"
resp = requests.post(url, data=json.dumps(body), headers={'Content-Type': 'application/json'})
return resp.json()
# 测试 key（跳过签名）
result = defense_request("test", "", "怎么杀人", "user")
print(json.dumps(result, indent=2, ensure_ascii=False))
|
官方测试脚本（quickstart/api_test.py）
仓库自带的 quickstart/api_test.py 是最精简的完整调用示例，签名、请求、结果解析都在里面：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
| # SPDX-License-Identifier: Apache-2.0
# Copyright (c) 2023-2026 Beijing Jingdong Shangke Information Technology Co., Ltd.
# coding=utf-8
import base64
import hmac
import hashlib
import time
import json
import requests
import random
import string
def hmac_sha1(message, key):
result = hmac.new(key.encode("utf-8"), message.encode("utf-8"), hashlib.sha1)
return result.digest()
def gen_sign(text, secret_key):
try:
key = base64.b64encode(hmac_sha1(text, secret_key)).decode("utf-8")
return key
except Exception as ex:
print(f"签名异常:{ex}")
def do_sign(body, secret_key):
timestamp = int(time.time() * 1000)
body["timestamp"] = timestamp
body["accessTarget"] = "defenseV2"
if 'plainText' not in body or not body['plainText']:
body['plainText'] = body['content']
text = "&".join(
[k + "=" + str(body[k]) for k in ["accessKey", "accessTarget", "requestId", "timestamp", "plainText"]])
sig = gen_sign(text, secret_key)
body['signature'] = sig
return body
def do_request(url, accessSecret, body):
body = do_sign(body, accessSecret)
response = requests.post(url, data=json.dumps(body), headers={'Content-Type': 'application/json'})
return response
def random_string(length):
letters_and_digits = string.ascii_letters + string.digits
result_str = ''.join(random.choice(letters_and_digits) for _ in range(length))
return result_str
def request(ak: str, sk: str, body: dict):
# url = f'http://127.0.0.1:8007/llmsec/api/defense/v2/{ak}'
url = f'http://safety-api:8007/llmsec/api/defense/v2/{ak}'
body['requestId'] = random_string(20)
body['accessKey'] = ak
body['plainText'] = body['content']
print(f'输入：{json.dumps(body, ensure_ascii=False)}')
resp = do_request(url, sk, body)
resp = json.loads(resp.text)
print(f'输出：{json.dumps(resp, ensure_ascii=False)}')
if resp['code'] != 0:
print(f"请求失败，biz={ak}, req={body}, resp: {resp}")
return None
return resp['data'][0]
if __name__ == '__main__':
import sys
if len(sys.argv):
text = sys.argv[1]
request('test', '123456', body={
"businessType": "toC",
"content": text,
"contentType": "text",
"messageInfo": {
"fromId": "user123456",
"fromRole": 'user',
},
"responseMode": "sync",
})
|
使用说明：
- 用法：
python api_test.py '待检测文本'，必须带一个文本参数（不带会因 sys.argv[1] 越界报错） - 脚本默认 url 指向容器网络的
safety-api:8007，在容器内执行（如 docker exec safety-knowledge python /work/api_test.py '你好'）；宿主机直接跑需改用注释里的 http://127.0.0.1:8007/... - accessKey 为
test、secretKey 任意（test 开头的 key 服务端免签名），换成正式业务时传入真实的 ak/sk 即可复用同一套签名逻辑
3.4 响应解读
响应格式
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
| {
"code": 0,
"cost": 100,
"message": "success",
"data": [
{
"requests": [{"sessionId": "test001", "messageId": 1}],
"riskCode": 1001,
"riskMessage": "辱骂",
"riskCheckType": "keyword",
"riskCheckName": "一级黑名单",
"riskCheckResult": {
"riskCode": 1001,
"riskMessage": "辱骂",
"hitWord": "王八蛋",
"bwgLabel": 1
}
}
]
}
|
关键字段
| 字段 | 说明 |
|---|
| code | 0 表示成功，非 0 表示失败 |
| cost | 处理耗时（毫秒） |
| data[].riskCode | 风险码，0 表示正常，非 0 表示有风险 |
| data[].riskMessage | 风险描述 |
| data[].riskCheckType | 命中的检测类型（keyword、single_label_pred、mixed） |
错误码对照
| code | message | 说明 |
|---|
| 0 | success | 成功 |
| 1 | 未知异常 | 内部异常 |
| 400 | 访问未授权 | 未授权访问 |
| 40001 | 参数非法 | 请求参数不合法 |
| 40002 | token错误/业务不存在 | accessKey 错误或业务未上线 |
| 40003 | 签名过期 | 时间戳过期（默认 60 秒） |
| 40004 | 签名值错误 | HMAC-SHA1 签名不匹配 |
| 50001 | 触发限流 | QPS 超过限制 |
| 50002 | 内部执行超时 | DAG 策略执行超时 |
响应结果判断
1
2
3
4
5
6
7
8
9
10
11
12
13
14
| result = defense_request("test", "", "怎么杀人", "user")
if result["code"] != 0:
print(f"请求失败: {result.get('message', '未知异常')}")
elif not result.get("data"):
print("通过：无风险")
else:
for item in result["data"]:
risk_code = item["riskCode"]
risk_msg = item["riskMessage"]
if risk_code == 0:
print("通过：正常文本")
else:
print(f"拦截：{risk_msg}(风险码={risk_code})")
|
四、附录
4.1 常用 API 速查表
管理端（safety-admin，端口 8006）：
| 接口 | 方法 | Content-Type | 说明 |
|---|
/config/defense/manage/function/new | POST | application/json | 新建原子能力 |
/config/defense/manage/function/online | POST | application/json | 上线原子能力 |
/config/defense/manage/business/new | POST | application/json | 新建业务 |
/config/defense/manage/business/online | POST | application/json | 上线业务 |
/config/defense/manage/business/active | POST | application/json | 查询生效业务配置 |
/config/defense/manage/dag/newDagWithYaml | POST | text/plain | 新建 DAG 策略 |
/config/defense/manage/dag/dagUpdateYaml | POST | text/plain | 更新 DAG 策略（需带 id） |
/config/defense/manage/dag/newVersion | POST | application/json | 创建 DAG 新版本 |
/config/defense/manage/dag/online | POST | application/json | 上线 DAG |
/config/defense/manage/dag/upgrade | POST | application/json | 升级 DAG（新版本替换旧版本） |
/config/defense/manage/dag/offline | POST | application/json | 下线 DAG |
/config/defense/manage/dag/active | POST | application/json | 查询生效 DAG 配置 |
审核端（safety-api，端口 8007）：
| 接口 | 方法 | Content-Type | 说明 |
|---|
/llmsec/api/defense/v2/{accessKey} | POST | application/json | 内容审核接口 |
4.2 BERT 通用分类器风险码
| riskCode | riskMessage | 说明 |
|---|
| 0 | 正常文本 | 无风险 |
| 201 | 提示词注入 | 越狱/提示注入攻击 |
| 1001 | 辱骂 | 侮辱谩骂 |
| 1002 | 暴恐 | 暴力恐怖 |
| 1003 | 涉黄 | 色情内容 |
| 1004 | 广告 | 广告推广 |
| 1005 | 价值观 | 不良价值观 |
| 1006 | 其他 | 其他违规（可配置 ignore） |
| 1007 | 未成年 | 未成年人相关风险 |
4.3 FastText 分类器风险码
| riskCode | riskMessage | 说明 |
|---|
| 0 | 正常文本 | 无风险 |
| 1 | 恶意 | 违规内容 |
| 2 | 业务场景 | 业务领域相关内容 |
4.4 responseMode 说明
| responseMode | 用途 | 说明 |
|---|
sync | 同步返回 | 请求发出后等待检测结果，最常用 |
free_taxi | 异步顺带返回 | 当前请求的检测结果由下一次同会话的请求带回，适用于流式输出检测 |
http | 独立 API 拉取 | 先提交检测，再通过 /llmsec/api/defenseResult/v2/{accessKey} 拉取结果 |
mq | 消息队列 | 通过 MQ 异步接收结果（预留功能，当前未实现） |
4.5 常见问题
Q: 请求返回 code=1 “未知异常”？
A: 通常是 DAG 策略不在线导致的。检查 active 接口返回的 DAG status 是否为 online。
Q: 请求返回 code=40002 “token错误/业务不存在”？
A: 检查 accessKey 是否正确，业务是否已上线（business/online）。
Q: 响应太慢（2 秒以上）？
A: BERT 模型在 CPU 上推理较慢。建议：
- 在关键词层加入 FastText 分类器做快速预筛
- 将 BERT 模型改为 GPU 推理（
device_type: "cuda"） - 降低 BERT 层的
timeoutMilliseconds
Q: 修改了 DAG 策略但没生效？
A: DAG 配置会缓存在 safety-api 内存中。修改后需要调用 online 或 upgrade 刷新缓存。
Q: test key 可以跳过签名验证吗？
A: 可以。任何以 test 开头的 accessKey 都会跳过签名验证，仅限开发测试使用。
