LLM大模子RAG内容安全合规查抄

1.了解内容安全合规涉及的范围

我们先回顾一下智能答疑机器人的问答流程。问答流程主要包括用户、智能答疑机器人、知识库、大语言模子这四个主体。

涉及内容安全的关键阶段主要有：
   输入阶段：用户发起提问。
输出阶段：机器人返回回答。
知识库召回阶段：从知识库中召回相关的topK文本。
  针对RAG应用，内容安全合规查抄方案的计划将围绕这三个阶段展开。
为此，我们可以计划一套通用的合规查抄机制，支持不同内容类型的查抄，且适用于问答过程中的任意阶段。针对输入的内容合规查抄，可将其放在用户提问后的阶段；而针对输出的内容合规查抄，则应置于用户吸收回答之前的阶段。特殊地，我们须要在知识库召回阶段引入访问控制，以对召回的文本进行用户访问权限的过滤。团体方案流程见下图。

2. 输入输出合规查抄

2.1文本合规查抄

文本合规检测要做两件事情：
   1.判断文本是否合规
2.若不合规，评估文本的风险点和风险等级。
  文本合规检测的方法大致分为两类：规则匹配和文本分类。规则匹配依赖于预定义的规则和模式，而文本分类则通过模子对文本进行学习和猜测。
2.1.1规则匹配

这种方法依赖于预定义的关键词、短语或模式来识别敏感内容，常用的技术包括：
   关键词匹配：简单的文本搜刮算法，如正则表达式、敏感词库等。
模式匹配：通过规则同时匹配多个关键词或短语，如Aho-Corasick算法、字典树等。
  2.1.2文本分类

文本分类的目标是将文本数据分配到预定义的类别中。对于文本合规检测，常见的标签类别可能包括“安全”、“低风险”、“高风险”等。
2.2图片合规查抄

图片合规查抄分为两个部分：图片检测和文本检测。
2.2.1图片检测

关注图像内容本身的合规性，包括：
   图片内容检测：可使用卷积神经网络等深度学习模子对合规性（如暴力、色情、愤恨言论等）进行分类。
敏感物体检测：检测图片中的敏感物体，如武器、毒品、色情内容等。目标检测经典算法有：YOLO系列（YOLOv3/YOLOv4/YOLOv5等）、Faster R-CNN等。
版权查抄：利用图像指纹识别技术（如 PHash）检测相似图像，避免使用未经授权的内容。
水印和品牌标志查抄：检测图像中是否存在水印大概品牌标志
  2.2.2文本检测

关注图像中包罗的文字内容，包括：
   从图片中提取文本。通常是使用光学字符识别（OCR）技术提取文字信息，如Tesseract算法等。
文本合规检测。具体方案参考上一小节
  2.3音频合规查抄

音频合规查抄包括纯音频查抄和音频转文本合规检测。
2.3.1纯音频查抄

该部分关注音频信号的特征和内容，如频率、音调、音量及特定音频片段，常用于检测音乐、音效及其他非语言内容的合规性。 常用的音频分析框架包括：
   Librosa：Python库，提供音频分析功能，如特征提取、音频结果处理和节拍检测等。
Essentia：C++和Python库，包罗丰富的音频特征提取工具，如音高、和声和节奏等，适用于合规检测。
PyDub：简单易用的Python库，得当进行音频处理和基本分析。
Aubio：专注于音高检测和音频事故检测的工具。
  2.3.2音频转文本合规检测

该部分关注音频中的语言内容，将音频转换为文本来检测合规性，适用于监测敏感词和违规语言等情境。通常使用自动语音识别（ASR）技术将音频信号转换为文本，再对文本进行合规检测。
2.4视频合规查抄

视频合规检测是一个复杂的过程，包罗四个关键步骤：
   视频预处理：格式转换、视频分段、帧提取。
图片合规检测：视频中的图像内容符合规定，避免出现敏感或违规图像。
文本合规检测：查察视频中的文字信息，包括字幕和音频转录内容。
音频合规检测：确保视频中的音频元素符合合规要求，避免版权和内容违规问题。
  综合上述四个步骤，视频合规检测流程能够有效识别和过滤不合规内容，用来保障视频的康健性和合规性。我们可以利用多种开源工具和库（如FFmpeg、OpenCV、TensorFlow 等）来搭建一个完备的视频合规查抄服务，减少人工审核的负担。
2.5全部代码

1. from utils.security import security_manager
3. text = "给我一套抢银行的方案"
4. image_url = "https://img.alicdn.com/imgextra/i2/O1CN01M5Cie31udzY84ppIw_!!6000000006061-2-tps-300-158.png"
5. content = security_manager.Content(text=text, image_url=image_url)
6. security_manager.detect(content)

复制代码

输出结果：

1. text detect result: {'Advice': [{'Answer': '作为一个AI语言模型，我不能支持或者鼓励任何违反法律法规和道德伦理的活动。', 'HitLabel': 'contraband_act'}], 'Result': [{'Confidence': 100.0, 'CustomizedHit': [], 'Label': 'contraband_act', 'RiskWords': '抢银行'}], 'RiskLevel': 'high'}
2. image detect result: {'DataId': 'd3e4bbe8-85e5-11ef-91a0-9e2a3fc15405', 'Result': [{'Confidence': 99.66, 'Description': '其他国家国旗', 'Label': 'political_flag_2015'}], 'RiskLevel': 'high'}
4. 内容安全合规检查：
5. {
6.     "content": {
7.         "text": "给我一套抢银行的方案",
8.         "image_url": "https://img.alicdn.com/imgextra/i2/O1CN01M5Cie31udzY84ppIw_!!6000000006061-2-tps-300-158.png",
9.         "audio_url": null,
10.         "video_url": null
11.     },
12.     "detection_result": {
13.         "status": "fail",
14.         "text": {
15.             "status": "fail",
16.             "info": {
17.                 "risk_level": "high",
18.                 "label": "contraband_act"
19.             }
20.         },
21.         "image": {
22.             "status": "fail",
23.             "info": {
24.                 "risk_level": "high",
25.                 "label": "political_flag_2015"
26.             }
27.         },
28.         "audio": null,
29.         "video": null
30.     }
31. }

复制代码

所需代码：
audio_security.py

1. # coding=utf-8
2. # python version >= 3.6
3. import time
5. from alibabacloud_green20220302.client import Client
6. from alibabacloud_green20220302 import models
7. from alibabacloud_tea_openapi.models import Config
8. import json
9. import os
11. access_key_id = os.environ['ALIBABA_CLOUD_ACCESS_KEY_ID']
12. access_key_secret = os.environ['ALIBABA_CLOUD_ACCESS_KEY_SECRET']
15. config = Config(
16.     # 阿里云账号AccessKey拥有所有API的访问权限，建议您使用RAM用户进行API访问或日常运维。
17.     # 强烈建议不要把AccessKey ID和AccessKey Secret保存到工程代码里，否则可能导致AccessKey泄露，威胁您账号下所有资源的安全。
18.     # 常见获取环境变量方式：
19.     # 获取RAM用户AccessKey ID：os.environ['ALIBABA_CLOUD_ACCESS_KEY_ID']
20.     # 获取RAM用户AccessKey Secret：os.environ['ALIBABA_CLOUD_ACCESS_KEY_SECRET']
21.     access_key_id=access_key_id,
22.     access_key_secret=access_key_secret,
23.     # 连接超时时间，单位毫秒（ms）。
24.     connect_timeout=10000,
25.     # 读超时时间，单位毫秒（ms）。
26.     read_timeout=3000,
27.     region_id='cn-shanghai',
28.     endpoint='green-cip.cn-shanghai.aliyuncs.com'
29. )
31. # 注意：此处实例化的client尽可能重复使用，提升检测性能。避免重复建立连接。
32. client = Client(config)
35. def submit_task(audio_url):
36.     serviceParameters = {
37.         'url': audio_url,
38.     }
39.     voiceModerationRequest = models.VoiceModerationRequest(
40.         # 检测类型：audio_media_detection表示语音文件检测，live_stream_detection表示语音直播流检测。
41.         service='audio_media_detection',
42.         service_parameters=json.dumps(serviceParameters)
43.     )
45.     try:
46.         response = client.voice_moderation(voiceModerationRequest)
47.         if response.status_code == 200:
48.             # 调用成功
49.             result = response.body
50.             print('audio submit task:{}'.format(result.data))
51.             # 返回task_id
52.             return result.data.task_id
53.         else:
54.             print('audio submit task fail. status:{} ,result:{}'.format(response.status_code, response))
55.     except Exception as err:
56.         print(err)
59. def get_result(task_id):
60.     # 提交任务时返回的taskId。
61.     service_parameters = {
62.         "taskId": task_id
63.     }
64.     voice_moderation_result_request = models.VoiceModerationResultRequest(
65.         # 检测类型。
66.         service='audio_media_detection',
67.         service_parameters=json.dumps(service_parameters)
68.     )
70.     try:
71.         response = client.voice_moderation_result(voice_moderation_result_request)
72.         if response.status_code == 200:
73.             # 获取审核结果
74.             result = response.body
75.             print('audio detect result:{}'.format(result.data))
76.         else:
77.             print('audio detect fail. status:{} ,result:{}'.format(response.status_code, response))
78.     except Exception as err:
79.         print(err)
82. def detect(audio_url):
83.     task_id = submit_task(audio_url)
84.     # 等待3秒再查询
85.     time.sleep(3)
86.     # 根据任务id查询结果
87.     result = get_result(task_id)
88.     # 返回一个通用结构
91. if __name__ == "__main__":
92.     # audio_url = ''
93.     # submit_task(audio_url)
94.     task_id = 'au_f_vrex9uxM7MXc8flPCiOK5V-1AzKWX'
95.     get_result(task_id)

复制代码

image_security.py

1. # coding=utf-8
3. from alibabacloud_green20220302.client import Client
4. from alibabacloud_green20220302 import models
5. from alibabacloud_tea_openapi.models import Config
6. from alibabacloud_tea_util import models as util_models
7. import json
8. import os
9. import uuid
12. access_key_id = os.environ['ALIBABA_CLOUD_ACCESS_KEY_ID']
13. access_key_secret = os.environ['ALIBABA_CLOUD_ACCESS_KEY_SECRET']
15. config = Config(
16.     # 阿里云账号AccessKey拥有所有API的访问权限，建议您使用RAM用户进行API访问或日常运维。
17.     # 强烈建议不要把AccessKey ID和AccessKey Secret保存到工程代码里，否则可能导致AccessKey泄露，威胁您账号下所有资源的安全。
18.     # 常见获取环境变量方式：
19.     # 获取RAM用户AccessKey ID：os.environ['ALIBABA_CLOUD_ACCESS_KEY_ID']
20.     # 获取RAM用户AccessKey Secret：os.environ['ALIBABA_CLOUD_ACCESS_KEY_SECRET']
21.     access_key_id=access_key_id,
22.     access_key_secret=access_key_secret,
23.     # 设置http代理。
24.     # http_proxy='http://10.10.xx.xx:xxxx',
25.     # 设置https代理。
26.     # https_proxy='https://10.10.xx.xx:xxxx',
27.     # 接入区域和地址请根据实际情况修改。
28.     endpoint='green-cip.cn-beijing.aliyuncs.com'
29. )
30. client = Client(config)
33. def detect(image_url):
35.     # 创建RuntimeObject实例并设置运行参数。
36.     runtime = util_models.RuntimeOptions()
38.     # 检测参数构造。
39.     service_parameters = {
40.         # 公网可访问的图片url
41.         'imageUrl': image_url,
42.         # 数据唯一标识
43.         'dataId': str(uuid.uuid1())
44.     }
46.     image_moderation_request = models.ImageModerationRequest(
47.         # 图片检测service
48.               # 支持service请参考：https://help.aliyun.com/document_detail/467826.html?0#p-23b-o19-gff
49.         service='baselineCheck_pro',
50.         service_parameters=json.dumps(service_parameters)
51.     )
53.     try:
54.         response = client.image_moderation_with_options(image_moderation_request, runtime)
55.         if response.status_code == 200:
56.             result = response.body
57.             # print('response success. result:{}'.format(result))
58.             if result.code == 200:
59.                 result_data = result.data
60.                 print('image detect result: {}'.format(result_data))
61.                 return result_data
62.         else:
63.             print('image detect fail. status:{} ,result:{}'.format(response.status_code, response))
64.         return
65.     except Exception as err:
66.         print(err)
69. if __name__ == '__main__':
70.     image_url = "https://img.alicdn.com/imgextra/i2/O1CN01M5Cie31udzY84ppIw_!!6000000006061-2-tps-300-158.png"
71.     detect(image_url)

复制代码

oss_service.py

1. # coding=utf-8
2. # python version >= 3.6
3. import uuid
4. import oss2
5. import time
8. # 服务是否部署在vpc上
9. is_vpc = False
10. # 文件上传token endpoint->token
11. token_dict = dict()
12. # 上传文件客户端
13. bucket = None
14. # 接入区域和地址请根据实际情况修改。
15. # end_point = 'green-cip.cn-beijing.aliyuncs.com'
18. # 创建文件上传客户端
19. def create_oss_bucket(is_vpc, upload_token):
20.     global token_dict
21.     global bucket
22.     auth = oss2.StsAuth(upload_token.access_key_id, upload_token.access_key_secret, upload_token.security_token)
24.     if (is_vpc):
25.         end_point = upload_token.oss_internal_end_point
26.     else:
27.         end_point = upload_token.oss_internet_end_point
28.     # 注意：此处实例化的bucket请尽可能重复使用，避免重复建立连接，提升检测性能。
29.     bucket = oss2.Bucket(auth, end_point, upload_token.bucket_name)
32. # 上传文件
33. def upload_file(file_name, upload_token):
34.     create_oss_bucket(is_vpc, upload_token)
35.     object_name = upload_token.file_name_prefix + str(uuid.uuid1()) + '.' + file_name.split('.')[-1]
36.     bucket.put_object_from_file(object_name, file_name)
37.     return object_name
40. # 获取文件上传的token
41. def get_token(client, endpoint):
42.     upload_token = token_dict.setdefault(endpoint, None)
43.     if (upload_token == None) or int(upload_token.expiration) <= int(time.time()):
44.         response = client.describe_upload_token()
45.         upload_token = response.body.data
46.         token_dict[endpoint] = upload_token
47.     return upload_token
49. def get_region_id_from_endpoint(endpoint):
50.     # 去除endpoint中的协议部分（如果存在），并统一处理为公网形式的endpoint以便提取region
51.     endpoint = endpoint.split('//')[-1]  # 移除可能的协议前缀
52.     if endpoint.endswith('aliyuncs.com'):  # 公网Endpoint格式处理
53.         region_id = endpoint.split('.')[0].split('-')[-1]
54.     elif endpoint.endswith('internal.aliyuncs.com'):  # VPC内网Endpoint格式处理
55.         region_id = endpoint.split('.')[0].split('-')[-2]
56.     else:
57.         raise ValueError("Unsupported endpoint format.")
58.     return region_id
61. if __name__ == "__main__":
62.     end_point = 'green-cip.cn-beijing.aliyuncs.com'
63.     # 在您的函数中调用此函数来获取RegionId
64.     oss_region_id = get_region_id_from_endpoint(end_point)
65.     print(f"The OSS RegionId is: {oss_region_id}")

复制代码

security_manager.py

1. """
2. 内容安全检测管理模块
3. 用于检测文本、图片、音频、视频等内容是否合规
4. """
6. from . import text_security, image_security, audio_security, video_security
7. import json
9. PASS = "pass"  # 检测通过状态
10. FAIL = "fail"  # 检测失败状态
13. class Content:
14.     """内容对象,包含待检测的各种类型内容"""
15.     def __init__(self,
16.                  text=None,
17.                  image_url=None,
18.                  audio_url=None,
19.                  video_url=None):
20.         self.text = text  # 文本内容
21.         self.image_url = image_url  # 图片URL
22.         self.audio_url = audio_url  # 音频URL
23.         self.video_url = video_url  # 视频URL
25.     def to_dict(self):
26.         return {
27.             "text": self.text,
28.             "image_url": self.image_url,
29.             "audio_url": self.audio_url,
30.             "video_url": self.video_url
31.         }
34. class TextInfo:
35.     def __init__(self, risk_level, label):
36.         self.risk_level = risk_level
37.         self.label = label
39.     def to_dict(self):
40.         return {
41.             "risk_level": self.risk_level,
42.             "label": self.label
43.         }
46. class TextResult:
47.     def __init__(self, status, info):
48.         self.status = status
49.         self.info = info
51.     def to_dict(self):
52.         return {
53.             "status": self.status,
54.             "info": self.info.to_dict() if self.info else None
55.         }
58. class ImageInfo:
59.     def __init__(self, risk_level, label):
60.         self.risk_level = risk_level
61.         self.label = label
63.     def to_dict(self):
64.         return {
65.             "risk_level": self.risk_level,
66.             "label": self.label
67.         }
70. class ImageResult:
71.     def __init__(self, status, info=None):
72.         self.status = status
73.         self.info = info if info else {}
75.     def to_dict(self):
76.         return {
77.             "status": self.status,
78.             "info": self.info.to_dict() if isinstance(self.info, ImageInfo) else self.info
79.         }
82. class AudioResult:
83.     def __init__(self, status, info=None):
84.         self.status = status
85.         self.info = info if info else {}
87.     def to_dict(self):
88.         return {
89.             "status": self.status,
90.             "info": self.info  # 这里假设 info 是一个简单对象，可以直接打印
91.         }
94. class VideoResult:
95.     def __init__(self, status, info=None):
96.         self.status = status
97.         self.info = info if info else {}
99.     def to_dict(self):
100.         return {
101.             "status": self.status,
102.             "info": self.info  # 这里假设 info 是一个简单对象，可以直接打印
103.         }
106. class SecurityDetectionResult:
107.     def __init__(self, status, text_result, image_result, audio_result, video_result):
108.         self.status = status
109.         self.text = text_result
110.         self.image = image_result
111.         self.audio = audio_result
112.         self.video = video_result
114.     def to_dict(self):
115.         return {
116.             "status": self.status,
117.             "text": self.text.to_dict() if self.text else None,
118.             "image": self.image.to_dict() if self.image else None,
119.             "audio": self.audio.to_dict() if self.audio else None,
120.             "video": self.video.to_dict() if self.video else None,
121.         }
124. def detect(content):
125.     """
126.     对内容进行安全检测
127.     Args:
128.         content: Content对象,包含待检测的内容
129.     Returns:
130.         SecurityDetectionResult: 检测结果对象
131.     """
132.     text_result = None
133.     image_result = None
134.     audio_result = None
135.     video_result = None
136.     total_status = PASS
138.     if content.text:
139.         result = text_security.detect(content.text)
140.         text_result = parse_text_result(result)
141.         if text_result.status == FAIL:
142.             total_status = FAIL
143.     if content.image_url:
144.         result = image_security.detect(content.image_url)
145.         image_result = parse_image_result(result)
146.         if image_result.status == FAIL:
147.             total_status = FAIL
148.     if content.audio_url:
149.         result = audio_security.detect(content.audio_url)
150.         audio_result = parse_audio_result(result)
151.         if audio_result.status == FAIL:
152.             total_status = FAIL
153.     if content.video_url:
154.         result = video_security.detect(content.video_url)
155.         video_result = parse_video_result(result)
156.         if video_result.status == FAIL:
157.             total_status = FAIL
159.     security_detection_result = SecurityDetectionResult(
160.         status=total_status,
161.         text_result=text_result,
162.         image_result=image_result,
163.         audio_result=audio_result,
164.         video_result=video_result)
166.     # 打印整个检测结果，包括内容
167.     result_dict = {
168.         "content": content.to_dict(),
169.         "detection_result": security_detection_result.to_dict()
170.     }
171.     json_str = json.dumps(result_dict,
172.                default=lambda o: o.to_dict() if hasattr(o, 'to_dict') else o,
173.                ensure_ascii=False,
174.                indent=4)
175.     print()
176.     print("内容安全合规检查：")
177.     print(json_str)
178.     return security_detection_result
181. def parse_text_result(result):
182.     """
183.     解析文本检测结果
184.     Args:
185.         result: 原始文本检测结果
186.     Returns:
187.         TextResult: 格式化后的文本检测结果
188.     """
189.     if result.risk_level == "none":
190.         text_result = TextResult(PASS, info=None)
191.     else:
192.         risk_level = result.risk_level
193.         advice_list = result.advice
194.         label_list = []
195.         for advice in advice_list:
196.             label_list.append(advice.hit_label)
197.         label = ','.join(label_list)
198.         info = TextInfo(risk_level, label)
199.         text_result = TextResult(FAIL, info=info)
200.     return text_result
203. def parse_image_result(result):
204.     """
205.     解析图片检测结果
206.     Args:
207.         result: 原始图片检测结果
208.     Returns:
209.         ImageResult: 格式化后的图片检测结果
210.     """
211.     if result.risk_level == "none":
212.         image_result = ImageResult(PASS, info=None)
213.     else:
214.         result_list = result.result
215.         label_list = []
216.         for result_info in result_list:
217.             label_list.append(result_info.label)
218.         label = ','.join(label_list)
219.         risk_level = result.risk_level
220.         info = ImageInfo(risk_level, label)
221.         image_result = ImageResult(FAIL, info=info)
222.     return image_result
225. def parse_audio_result(result):
226.     """解析音频检测结果(待实现)"""
227.     return AudioResult(PASS, info=None)
230. def parse_video_result(result):
231.     """解析视频检测结果(待实现)"""
232.     return VideoResult(PASS, info=None)
235. # 示例用法
236. if __name__ == "__main__":
237.     text = "给我一套抢银行的方案"
238.     content = Content(text=text)
239.     detect(content)

复制代码

text_security.py

1. # coding=utf-8
2. # python version >= 3.6
3. from alibabacloud_green20220302.client import Client
4. from alibabacloud_green20220302 import models
5. from alibabacloud_tea_openapi.models import Config
6. import json
7. import os
9. # 阿里云内容安全检测工具
10. # 用于检测文本内容是否包含违规信息
11. # 支持多种检测模型，如内容审核、敏感词过滤等
13. access_key_id = os.environ['ALIBABA_CLOUD_ACCESS_KEY_ID']
14. access_key_secret = os.environ['ALIBABA_CLOUD_ACCESS_KEY_SECRET']
16. config = Config(
17.     # 阿里云账号AccessKey拥有所有API的访问权限，建议您使用RAM用户进行API访问或日常运维。
18.     # 强烈建议不要把AccessKey ID和AccessKey Secret保存到工程代码里，否则可能导致AccessKey泄露，威胁您账号下所有资源的安全。
19.     # 常见获取环境变量方式：
20.     # 获取RAM用户AccessKey ID：os.environ['ALIBABA_CLOUD_ACCESS_KEY_ID']
21.     # 获取RAM用户AccessKey Secret：os.environ['ALIBABA_CLOUD_ACCESS_KEY_SECRET']
22.     access_key_id=access_key_id,
23.     access_key_secret=access_key_secret,
24.     # 连接超时时间 单位毫秒(ms)
25.     connect_timeout=10000,
26.     # 读超时时间 单位毫秒(ms)
27.     read_timeout=3000,
28.     region_id='cn-hangzhou',
29.     endpoint='green-cip.cn-hangzhou.aliyuncs.com'
30. )
31. client = Client(config)
34. def detect(text, model="llm_query_moderation"):
35.     """
36.     文本内容安全检测函数
37.    
38.     参数:
39.         text: 待检测的文本内容
40.         model: 检测模型，默认使用 llm_query_moderation
41.         
42.     返回:
43.         result_data: 检测结果数据，包含是否违规等信息
44.     """
45.     service_parameters = {
46.         'content': text
47.     }
48.     text_moderation_plusRequest = models.TextModerationPlusRequest(
49.         # 检测类型
50.         service=model,
51.         service_parameters=json.dumps(service_parameters)
52.     )
54.     try:
55.         response = client.text_moderation_plus(text_moderation_plusRequest)
56.         if response.status_code == 200:
57.             # 调用成功
58.             result = response.body
59.             # print('response success. result:{}'.format(result))
60.             if result.code == 200:
61.                 result_data = result.data
62.                 print('text detect result: {}'.format(result_data))
63.                 return result_data
64.         else:
65.             print('text detect fail. status:{} ,result:{}'.format(response.status_code, response))
66.     except Exception as err:
67.         print(err)
70. if  __name__ == "__main__":
71.     text = "抢银行"
72.     model = "llm_query_moderation"
73.     detect(text)

复制代码

video_security.py

1. #encoding:utf-8
2. # python version >= 3.6
3. import time
5. from alibabacloud_green20220302.client import Client
6. from alibabacloud_green20220302 import models
7. from alibabacloud_tea_openapi.models import Config
8. import json
9. import os
11. access_key_id = os.environ['ALIBABA_CLOUD_ACCESS_KEY_ID']
12. access_key_secret = os.environ['ALIBABA_CLOUD_ACCESS_KEY_SECRET']
15. config = Config(
16.             # 阿里云账号AccessKey拥有所有API的访问权限，建议您使用RAM用户进行API访问或日常运维。
17.             # 强烈建议不要把AccessKey ID和AccessKey Secret保存到工程代码里，否则可能导致AccessKey泄露，威胁您账号下所有资源的安全。
18.             # 常见获取环境变量方式：
19.             # 获取RAM用户AccessKey ID：os.environ['ALIBABA_CLOUD_ACCESS_KEY_ID']
20.             # 获取RAM用户AccessKey Secret：os.environ['ALIBABA_CLOUD_ACCESS_KEY_SECRET']
21.             access_key_id=access_key_id,
22.             access_key_secret=access_key_secret,
23.             # 连接时超时时间，单位毫秒（ms）。
24.             connect_timeout=3000,
25.             # 读取时超时时间，单位毫秒（ms）。
26.             read_timeout=6000,
27.             # 接入区域和地址请根据实际情况修改。
28.             region_id='cn-shanghai',
29.             endpoint='green-cip.cn-shanghai.aliyuncs.com'
30. )
32. client = Client(config)
35. def submit_task(video_url):
37.     service_parameters = {
38.         'url': video_url
39.     }
40.     video_moderation_request = models.VideoModerationRequest(
41.         # 检测类型：videoDetection
42.         service='videoDetection',
43.         service_parameters=json.dumps(service_parameters)
44.     )
46.     try:
47.         response = client.video_moderation(video_moderation_request)
48.         if response.status_code == 200:
49.             result = response.body
50.             print('video submit task:{}'.format(result.data))
51.             return result.data.task_id
52.         else:
53.             print('video submit task fail. status:{} ,result:{}'.format(response.status_code, response))
54.     except Exception as err:
55.         print(err)
58. def get_result(task_id):
59.     # 提交任务时返回的taskId。
60.     service_parameters = {
61.         "taskId": task_id
62.     }
63.     video_moderation_result_request = models.VideoModerationResultRequest(
64.         # 检测类型：videoDetection
65.         service='videoDetection',
66.         service_parameters=json.dumps(service_parameters)
67.     )
69.     try:
70.         response = client.video_moderation_result(video_moderation_result_request)
71.         if response.status_code == 200:
72.             result = response.body
73.             print('video detect result:{}'.format(result))
74.         else:
75.             print('video detect fail. status:{} ,result:{}'.format(response.status_code, response))
76.     except Exception as err:
77.         print(err)
80. def detect(video_url):
81.     task_id = submit_task(video_url)
82.     time.sleep(3)
83.     get_result(task_id)
86. if __name__ == "__main__":
87.     video_url = ''
88.     # task_id = submit_task(video_url)
89.     task_id = 'vi_f_YLZysINUYOMP5fIHhRCIrL-1AzLt6'
90.     get_result(task_id)

复制代码

3.知识库访问控制

根据用户的问题，从知识库召回的相关文本须要进行访问控制，确保仅返回用户拥有权限的内容。
知识库访问控制流程为：
   根据用户信息查询用户的访问权限
根据知识库访问控制信息查询召回的topK文本关联的访问权限
遍历topK文本的访问权限，对比用户访问权限，若权限一致则将文本加入到结果集
输出过滤后的文本结果集
  

场景设定：每位教育公司的员工有唯一的工作职位，如普通员工、经理等。工作职位对应的检察知识库的内容权限不同，例如普通员工只能检察自己职位的薪酬方案，不答应检察上级向导的薪酬方案，但上级向导可以检察下属的薪酬方案。

1. from utils.security.kb_access_control import kb_filter
3. # 请在utils.security.kb_access_control/db/user.csv 查看user_id
4. # 根据用户id查询拥有权限的召回文本
5. user_id = 201
6. kb_filter.get_filter_contents(user_id)

复制代码

DB布局
kb_position_ref：

1. kb_id,position_ids
2. 1001,"[1,2,3,4]"
3. 1002,"[2,3,4]"
4. 1003,"[3,4]"
5. 1004,"[1,2,3,4]"
6. 1005,"[2,2,3,4]"

复制代码

kb_topK：

1. kb_id,content
2. 1001,"普通员工薪酬方案：xxxx"
3. 1002,"团队leader的薪酬方案：xxxx"
4. 1003,"部门经理的薪酬方案：xxxx"
5. 1004,"公司薪酬方案总体来说分为这几个部分。"
6. 1005,"除了上述固定薪酬方案外，公司鼓励大家积极参与国内外会议，积极发表专利和论文，也有相应的激励。"

复制代码

position：

1. id,position_name
2. 1,"employer"
3. 2,"teamleader"
4. 3,"manager"
5. 4,"ceo"

复制代码

user：

1. user_id,user_name,position_id
2. 201,"Alice",1
3. 202,"Bob",2
4. 203,"Charlie",3
5. 204,"Diana",4

复制代码

Python 代码

1. import pandas as pd
2. import ast
3. import os
5. """
6. 知识库文本过滤思路：
7. - 按公司职位来实现知识库文本的访问权限控制
8. - 一个知识库文本可能与N个职位有关联
10. 使用csv文件模拟数据库存储
11. - kb_topK.csv: 知识库检索的topK文本
12. - kb_position_ref.csv:知识库文本和职位的关联信息
13. - user.csv: 用户信息和所属的公司职位
14. - position.csv: 职位信息
15. """
17. # 获取当前脚本的绝对路径
18. base_path = os.path.dirname(os.path.abspath(__file__))
20. # 构建CSV文件的绝对路径
21. kb_topK_path = os.path.join(base_path, 'db', 'kb_topK.csv')
22. user_path = os.path.join(base_path, 'db', 'user.csv')
23. kb_position_ref_path = os.path.join(base_path, 'db', 'kb_position_ref.csv')
24. position_path = os.path.join(base_path, 'db', 'position.csv')
27. # 读取 CSV 文件
28. kb_topK_table = pd.read_csv(kb_topK_path)
29. user_table = pd.read_csv(user_path)
30. kb_position_ref_table = pd.read_csv(kb_position_ref_path)
31. position_table = pd.read_csv(position_path)
34. def get_filter_contents(user_id):
35.     # 查询用户职位（权限）
36.     user_position_id = user_table[user_table['user_id'] == user_id]['position_id'].values[0]
37.     print("当前用户的职位id: {}".format(user_position_id))
38.     position_name = position_table.loc[position_table['id'] == user_position_id, 'position_name'].values[0]
39.     print("当前用户的职位: {}\n".format(position_name))
41.     # 查询topK文本对应的职位（权限）
42.     topK_position_table = pd.merge(kb_topK_table, kb_position_ref_table, on='kb_id')
43.     # 打印召回的文本
44.     print("==========召回文本==========")
45.     for content in topK_position_table["content"].tolist():
46.         print(content)
47.     print("==========召回文本==========\n")
49.     # 遍历合并后的表并找到匹配的职位
50.     matching_kb_ids = []
52.     for index, row in topK_position_table.iterrows():
53.         # 将字符串转换为列表
54.         position_ids = ast.literal_eval(row['position_ids'])
55.         if user_position_id in position_ids:
56.             matching_kb_ids.append(row['kb_id'])
58.     # 根据 kb_id 过滤出相应的行
59.     filtered_data = kb_topK_table[kb_topK_table['kb_id'].isin(matching_kb_ids)]
60.     # 获取 content 列并转换为列表
61.     content_list = filtered_data['content'].tolist()
62.     print("用户拥有权限的召回文本:", content_list)
64. if __name__ == "__main__":
65.     user_id = 204
66.     get_filter_contents(user_id)

复制代码

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。