1
2
3
4
5
6
7
8

配置期   你的脚本 ──POST──► safety-admin:8006 ──写入──► MySQL(function_conf 表)
                              ▲
                              │ ① api 每 60s 轮询 /config/defense/api/conf
执行期   safety-api:8007 ─────┘
           │ ② 配置变更 → 重建 DAG（doMerge 展开 ref → 按 type 反射实例化 Function）
           │ ③ 请求来了：Function 用 OkHttpClient POST 你的服务
           ▼
        你的检测服务（Flask/任意语言，无状态）

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

public static final Map<RiskCheckType, Class<? extends AbstractFunction>> FUNC_MAP = ... {{
    put(RiskCheckType.dummy, DummyFunction.class);
    put(RiskCheckType.keyword, KeywordFunction.class);
    put(RiskCheckType.single_label_pred, SingleLabelPredFunction.class);
    put(RiskCheckType.multi_label_pred, MultiLablePreFunction.class);
    put(RiskCheckType.kb_search, KbSearchFunction.class);
    put(RiskCheckType.rag_answer, RagAnswerFunction.class);
    put(RiskCheckType.multi_turn_detect, MultiTurnDetectFunction.class);
    put(RiskCheckType.parallel, ParallelFunction.class);
    put(RiskCheckType.async, AsyncFunction.class);
}};

1
2
3
4
5
6

SingleLabelPredCheck.SkillsReq req = new SingleLabelPredCheck.SkillsReq();
req.setBusiness_id(context.getBusinessConf().getName());       // 业务名
req.setRequest_id(firstReq.getRequestId());                    // 请求ID
req.setSession_id(firstReq.getMessageInfo().getSessionId());   // 会话ID
req.setText_list(Lists.newArrayList(context.getCheckContent()));// 待检文本，定长1
// POST extra.url，body 即上面的 req

1

{"business_id": "test", "request_id": "xxx", "session_id": "yyy", "text_list": ["待检测文本"]}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

{
  "code": 0,              // 必须 0，否则 api 抛 inner_error
  "message": "success",
  "data": [{              // 数组非空，api 只取 data[0]
    "riskCode": 9001,     // 必须显式返回；0=正常，非0=有风险（hasRisk() 唯一判据）
    "riskMessage": "文本超长",
    "probability": 0.99,  // 置信度，供 groovy 阈值表判定，框架本身不判
    "text": "原文",        // 可选，回显
    "detail": [{"riskCode": 9001, "riskMessage": "文本超长", "probability": 0.99}]  // 可选，全标签分布
  }]
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

注册请求 POST /function/new + online
  ▼
safety-admin ──► MySQL function_conf 表（版本化状态机 edit → online）
  ▲                ▲ admin 每 60s 聚合 online 配置到内存（DefenseConfigServiceImpl.refreshConf）
  │ ① api 每 60s 轮询 POST /config/defense/api/conf {group:"default"}
DynamicConfigService.loadConf()  @Scheduled(fixedRate=60000, initialDelay=60000)
  │ ② shouldUpdate：逐业务比对 lastChangeTimestamp，变了才继续
  ▼ 全量替换内存 name2BusinessConf / nodeConfMap / functionConfMap
  │ ③ 回调监听器 → FunctionChainBuilderImpl.loadConf() 逐业务重建 DAG：
  ▼    YAML节点 → doMerge(ref展开) → 反射new Function → 构建Router → rootMap
  │ ④ Function 构造函数里：建/复用 OkHttpClient + 注册 warmup
  ▼ 执行期请求按新 DAG 流转

1
2
3

FunctionConfig copyFrom = ConfigService.functionConfMap.get(cur.getRef()); // 44: 按名字查注册的能力模板
if (copyFrom == null) { throw new Exception("function ref 不存在: " + cur.getRef()); } // 46: ref 打错字=DAG 构建失败
Map<String, Object> mergedConf = mergeProperties(copyTo.getConf(), cur.getConf());   // 51: 深合并

1
2
3
4
5

模板（注册时）          引用处（DAG 里）            合并结果
conf:                conf:                     conf:
  extra.url: A         ignoreRiskCode: [1006]      extra.url: A
  modelType: bert                                  modelType: bert
  ignoreRiskCode: []                               ignoreRiskCode: [1006]  ← 引用处赢

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

long timeout = 150;                                              // 54: 兜底的默认超时
if (this.conf().getTimeoutMilliseconds() != null)
    timeout = this.conf().getTimeoutMilliseconds();              // 56: 注册时的 timeoutMilliseconds
Object resource = FunctionResourcePool.getFunc(type(), this.fasttextConf.getUrl()); // 68: 按 type+url 查池
if (resource != null) {
    this.fasttextClient = (OkHttpClient) resource;               // 70: 命中直接复用
} else {
    Object temp = new OkHttpClient.Builder().callTimeout(timeout, MILLISECONDS).build(); // 73: 新建
    Object uniqResource = FunctionResourcePool.addFuncIfAbsent(type(), url, temp);
    if (temp == uniqResource) {                                  // 76: 确认自己是"首创者"
        ScheduleWarmup.appendTask(...)                           // 77: 注册周期性预热任务
    }
}

 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

import time
from flask import Flask, request, jsonify

app = Flask(__name__)
MAX_LEN = 10          # 演示阈值：超过 10 个字符判风险
RISK_CODE = 9001      # 自定义风险码（避开框架已有的 0/201/1001~1008 等）

@app.route('/guard/length', methods=['POST'])
def guard_length():
    start = time.time()
    req = request.get_json(force=True)
    text_list = req.get('text_list') or []
    print(f"[req] biz={req.get('business_id') or req.get('businessName')} "
          f"text_list={text_list}", flush=True)
    data = []
    for text in text_list:
        if len(text) > MAX_LEN:
            item = {"riskCode": RISK_CODE, "riskMessage": "文本超长", "probability": 0.99, "text": text}
            item["detail"] = [{"riskCode": RISK_CODE, "riskMessage": "文本超长", "probability": 0.99},
                              {"riskCode": 0, "riskMessage": "正常文本", "probability": 0.01}]
        else:
            item = {"riskCode": 0, "riskMessage": "正常文本", "probability": 1.0, "text": text}
            item["detail"] = [{"riskCode": 0, "riskMessage": "正常文本", "probability": 1.0}]
        data.append(item)
    return jsonify({"code": 0, "message": "success",
                    "cost": (time.time() - start) * 1000, "data": data})

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

1
2
3
4
5
6
7

mkdir -p /tmp/joysafety-demo && cd /tmp/joysafety-demo
python3 -m venv venv && ./venv/bin/pip install flask
./venv/bin/python text_guard.py &     # 监听 0.0.0.0:9000

# 自测（真实输出）：短文本 riskCode=0，长文本 riskCode=9001
curl -s -X POST http://127.0.0.1:9000/guard/length -H 'Content-Type: application/json' \
  -d '{"request_id":"r1","business_id":"test","session_id":"s1","text_list":["你好"]}'

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

cd /tmp/joysafety-demo
cat > Dockerfile <<'EOF'
FROM swr.cn-north-4.myhuaweicloud.com/ddn-k8s/docker.io/python:3.12-slim
WORKDIR /app
RUN pip install --no-cache-dir flask -i https://mirrors.aliyun.com/pypi/simple/
COPY text_guard.py .
CMD ["python", "text_guard.py"]
EOF
docker build -t text-guard:demo .
docker run -d --name text-guard --network joysafety_joy_safety text-guard:demo
# 从 safety-api 容器内验证连通性（真实输出：长文本返回 9001）：
docker exec safety-api sh -c 'curl -s -m 3 -X POST http://text-guard:9000/guard/length \
  -H "Content-Type: application/json" -d "{\"text_list\":[\"容器网络内的连通性测试文本\"]}"'
# 验证完清理：docker rm -f text-guard

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

curl -s -X POST http://localhost:8006/config/defense/manage/function/new \
  -H 'Content-Type: application/json' -d '{
    "name": "text_length_guard",  "group": "default",  "desc": "文本长度检测（演示）",
    "type": "single_label_pred",  "timeoutMilliseconds": 500,
    "confObj": {
      "name": "text_length_guard",  "modelType": "fasttext",
      "extra": {"url": "http://host.docker.internal:9000/guard/length"},
      "ignoreRiskCode": []
    }
  }'

1
2
3

curl -s -X POST http://localhost:8006/config/defense/manage/function/online \
  -H 'Content-Type: application/json' -d '{"id": 9}'
# → 真实返回：code=0, data.status=online（version=1）

 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

import json, urllib.request, yaml

ADMIN = "http://safety-admin:8006/config/defense/manage"
# post(path, body) 辅助函数与 02 笔记附录 A 相同，略

active = post("/dag/active", {"group": "default", "businessName": "test"})
dag = active["data"]
new_id = post("/dag/newVersion", {"id": dag["id"]})["data"]["id"]

NEW_ROUTER = """import com.jd.security.llmsec.core.check.MixedCheck
def retMap=ctx.curResult.resultMap
def next="from_user_test_bert2"
def kw = retMap.keyword
def lenGuard = retMap.text_length_guard
if (lenGuard != null && lenGuard.hasRisk()) {
    ctx.curResult = lenGuard
    return null
}
if (kw != null && (kw.bwgLabel == 1 || kw.bwgLabel == 2 )) { ctx.curResult = kw; return null } else { return next }
"""

conf_array = dag["confArray"]
for node in conf_array:
    if node.get("nodeId") == "from_user_test1":
        node["functionConf"]["conf"]["functionConfs"].append({"ref": "text_length_guard"})
        node["routerConf"]["conf"]["script"] = NEW_ROUTER

yaml_body = yaml.safe_dump({"id": new_id, "businessName": "test", "group": "default",
    "desc": dag.get("desc", "test"), "rootId": dag["rootId"], "confArray": conf_array},
    allow_unicode=True)
post("/dag/dagUpdateYaml", yaml_body, content_type="text/plain")
print(post("/dag/upgrade", {"id": new_id}))

1
2
3

docker cp update_dag_add_guard.py safety-knowledge:/work/
docker exec safety-knowledge python /work/update_dag_add_guard.py
# 真实输出：当前线上 DAG id=3 version=3；新版本 id=5；upgrade 后 version=4, status=online

1
2
3
4
5
6

start（按 fromRobot 分流）
  ├─ robot → from_robot_test1: ref keyword → stupid_end
  └─ user  → from_user_test1: parallel(keyword, text_length_guard)   ← 新能力在这里
               │ router：lenGuard 有风险 → 短路返回 9001
               │         关键词命中黑/白 → 短路返回
               └────────→ from_user_test_bert2: parallel(bert_20250916, bert_prompt_injection_20250828)

1
2
3

16:41:05 DynamicConfigService:97  | 配置更新成功，function={..., "text_length_guard":{"conf":{"extra":{"url":"http://host.docker.internal:9000/guard/length"}, ...}}, ...}
16:41:10 SingleLabelPredFunction:77 | warm up fasttext, url=http://host.docker.internal:9000/guard/length   ← 首次预热
16:42:10 SingleLabelPredFunction:77 | warm up fasttext, url=http://host.docker.internal:9000/guard/length   ← 每 60s 一次

1
2
3
4

# 路径一：不命中（2 个字符 ≤ 阈值 10）
docker exec safety-knowledge python /work/api_test.py '你好'
# 路径二：命中（长度 22 > 阈值 10）
docker exec safety-knowledge python /work/api_test.py '这段文本长度明显超过十个字符用来触发自定义能力'

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

// 不命中：cost=3856ms —— 长度检测和关键词都放行，继续走完两个 BERT 才判正常
{"code": 0, "cost": 3856,
 "data": [{"checkedContent": "你好", "riskCode": 0, "riskMessage": "0正常文本",
           "riskCheckType": "mixed", "riskCheckName": "none"}]}

// 命中：cost=28ms —— 在第一个并行节点被短路返回，根本没走到 BERT
{"code": 0, "cost": 28,
 "data": [{"checkedContent": "这段文本长度明显超过十个字符用来触发自定义能力",
   "riskCode": 9001, "riskMessage": "9001文本超长",
   "riskCheckType": "single_label_pred", "riskCheckName": "text_length_guard",
   "riskCheckResult": {"type": "single_label_pred", "riskCode": 9001,
     "srcName": "text_length_guard", "cost": 14, "probability": 0.99,
     "detail": [{"riskCode": 9001, "riskMessage": "文本超长", "probability": 0.99}, {"riskCode": 0, ...}]}}]}

1
2
3

pkill -f text_guard.py    # 模拟能力宕机
docker exec safety-knowledge python /work/api_test.py '服务挂掉时这段超过十个字符的文本会被放行吗'
# 真实输出：riskCode=0, riskCheckType=mixed —— 放行了！

1
2

{"name": "你的能力名", "modelType": "fasttext",
 "extra": {"url": "http://你的服务地址/路径"}, "ignoreRiskCode": []}