功能概述关键词审核是最简单、最快速的审核方式通过维护一个关键词列表对输入或输出进行子字符串匹配。当内容包含列表中的任何关键词时判定为违规。实现细节数据结构文件: api/core/moderation/keywords/keywords.pyclassKeywordsModeration(Moderation):name:strkeywords# 策略标识# 从基类继承# self.app_id - 应用 ID# self.tenant_id - 租户 ID# self.config - 配置字典配置验证方法:validate_config()classmethoddefvalidate_config(cls,tenant_id:str,config:dict):# 1. 验证基础结构cls._validate_inputs_and_outputs_config(config,True)# 2. 验证关键词字段ifnotconfig.get(keywords):raiseValueError(keywords is required)# 3. 验证关键词字符数iflen(config.get(keywords,[]))10000:raiseValueError(keywords length must be less than 10000)# 4. 验证关键词行数最多100行keywords_row_lenconfig[keywords].split(\n)iflen(keywords_row_len)100:raiseValueError(the number of rows for the keywords must be less than 100)限制约束:约束项限制说明关键词总字符数≤ 10,000内存和性能考虑关键词行数≤ 100防止过大的列表输入/输出配置至少启用一个必须配置某种审核预设响应≤ 100 字符UI 限制核心审核逻辑方法:moderation_for_inputs()defmoderation_for_inputs(self,inputs:dict,query:str)-ModerationInputsResult:# 1. 初始化结果flaggedFalsepreset_response# 2. 获取配置ifself.configisNone:raiseValueError(The config is not set.)# 3. 检查输入审核是否启用ifself.config[inputs_config][enabled]:preset_responseself.config[inputs_config][preset_response]# 4. 合并输入内容包括查询ifquery:inputs[query__]query# 5. 解析关键词列表过滤空行keywords_list[keywordforkeywordinself.config[keywords].split(\n)ifkeyword]# 6. 执行匹配flaggedself._is_violated(inputs,keywords_list)# 7. 返回结果returnModerationInputsResult(flaggedflagged,actionModerationAction.DIRECT_OUTPUT,preset_responsepreset_response)匹配逻辑:遍历inputs字典的每个值对每个值调用_check_keywords_in_value()进行不区分大小写的子字符串查找任何一个值匹配即返回True短路求值关键字符串匹配方法:_check_keywords_in_value()def_check_keywords_in_value(self,keywords_list:Sequence[str],value:Any)-bool:# 对每个关键词检查returnany(keyword.lower()# 转换关键词为小写instr(value).lower()# 转换值为字符串并小写forkeywordinkeywords_list)匹配特性:✅ 子字符串匹配“性感” 会匹配 “很性感”✅ 不区分大小写“SPAM” 会匹配 “spam”✅ 支持任意类型值自动转换为字符串❌ 无法处理变形、转义、符号隐藏等对抗手段性能分析时间复杂度假设n 关键词数量m 单个内容长度k 输入字典大小匹配时间: O(n × m × k)对每个输入值 (k) 对每个关键词 (n) 执行子字符串查找 (O(m))性能特点操作耗时说明关键词加载O(1)在init时完成单次匹配O(n×m)n关键词数m内容长整体审核O(k×n×m)k输入字段数优化建议缓存关键词集合:# 当前实现在每次调用时重新分割keywords_list[kforkinself.config[keywords].split(\n)ifk]# 优化在 __init__ 时缓存classKeywordsModeration(Moderation):def__init__(self,app_id,tenant_id,configNone):super().__init__(app_id,tenant_id,config)ifconfig:self._keywords_cache[kforkinconfig.get(keywords,).split(\n)ifk]使用 Trie 树:# 对于超大关键词列表10,000可使用 Trie 树# 减少匹配时间到 O(m) 而非 O(n×m)frompyahocorasickimportAutomaton trieAutomaton()forkeywordinkeywords_list:trie.add_word(keyword.lower())trie.make_automaton()使用场景适用场景简单过滤- 敏感词/垃圾词过滤keywords: 刷单 返利 外挂行业合规- 金融/医疗领域的禁用词keywords: 保证收益 包治百病 秘方品牌保护- 竞争对手名称keywords: ChatGPT4 DeepSeek不适用场景❌ 语义理解“这个不错” vs “这个不错” 表达不同意思❌ 多语言混合中英混合的复杂内容❌ 对抗变形“s-p-a-m”、“spm” 等❌ 长文本精细化审核仅适合简单过滤与其他策略的对比特性KeywordsOpenAI自定义API响应速度⚡⚡⚡ 快⚡ 慢⚡⚡ 中准确度⭐⭐ 低⭐⭐⭐⭐⭐ 高⭐⭐⭐ 中高成本 无 高 根据实现定制性⭐ 低⭐ 低⭐⭐⭐⭐ 高多语言⭐⭐ 差⭐⭐⭐⭐⭐ 优⭐⭐⭐ 良测试用例文件: api/tests/unit_tests/core/moderation/test_content_moderation.py测试类测试类场景TestKeywordsModeration关键词审核功能测试关键测试deftest_moderation_for_inputs_no_violation(self):输入无违规内容inputs{user_input:This is a clean message}resultkeywords_moderation.moderation_for_inputs(inputs,)assertresult.flaggedisFalsedeftest_moderation_for_inputs_with_violation(self):输入包含敏感词inputs{user_input:This message contains spam}resultkeywords_moderation.moderation_for_inputs(inputs,)assertresult.flaggedisTrueassertresult.actionModerationAction.DIRECT_OUTPUTdeftest_moderation_for_outputs_with_violation(self):输出包含敏感词textThis response contains spam contentresultkeywords_moderation.moderation_for_outputs(text)assertresult.flaggedisTrue配置方式配置数据结构关键词审核的配置存储在应用的sensitive_word_avoidance字段中采用 JSON 格式{enabled:true,type:keywords,config:{keywords:敏感词1\n敏感词2\n敏感词3,inputs_config:{enabled:true,preset_response:您的输入包含敏感词请重新输入},outputs_config:{enabled:true,preset_response:抱歉我无法回答这个问题}}}字段说明:字段类型必填说明enabledboolean是是否启用审核功能typestring是审核类型关键词审核为keywordsconfig.keywordsstring是关键词列表每行一个最多100行总长度≤10000字符config.inputs_configobject是输入审核配置config.inputs_config.enabledboolean是是否启用输入审核config.inputs_config.preset_responsestring否输入违规时的预设响应≤100字符config.outputs_configobject是输出审核配置config.outputs_config.enabledboolean是是否启用输出审核config.outputs_config.preset_responsestring否输出违规时的预设响应≤100字符通过 UI 配置前端步骤 1: 打开功能面板打开应用编辑界面在右侧找到功能(Features) 按钮/面板在功能列表中找到内容审核卡片步骤 2: 选择审核类型在功能卡片上点击启用开关或设置按钮打开审核配置弹窗选择“关键词”(Keywords) 作为审核提供者系统会显示关键词配置表单步骤 3: 配置关键词列表在关键词文本框中输入敏感词敏感词1 敏感词2 敏感词3限制说明:✅ 每行一个关键词自动换行✅ 最多支持 100 行✅ 总字符数不超过 10,000✅ 每个关键词最长 100 字符⚠️ 超出部分会被自动截断步骤 4: 配置审核范围可选择审核的内容类型输入审核(Input Moderation):☑️ 启用输入审核 预设响应您的输入包含不当内容请重新输入作用检查用户输入是否包含关键词输出审核(Output Moderation):☑️ 启用输出审核 预设响应抱歉我无法回答这个问题作用检查 AI 生成内容是否包含关键词注意: 至少需要启用输入或输出审核之一步骤 5: 保存配置点击保存按钮系统会验证关键词列表格式检查字符数和行数限制确认至少启用一种审核类型保存配置到数据库通过 API 配置后端1. 获取当前配置请求示例:curl-XGEThttps://api.dify.ai/v1/apps/{app_id}/model-config\-HAuthorization: Bearer {api_key}响应示例:{sensitive_word_avoidance:{enabled:true,type:keywords,config:{keywords:spam\nscam\nfake,inputs_config:{enabled:true,preset_response:Invalid input},outputs_config:{enabled:false}}}}2. 更新配置请求示例:curl-XPOSThttps://api.dify.ai/v1/apps/{app_id}/model-config\-HAuthorization: Bearer {api_key}\-HContent-Type: application/json\-d{ sensitive_word_avoidance: { enabled: true, type: keywords, config: { keywords: 敏感词1\n敏感词2\n敏感词3, inputs_config: { enabled: true, preset_response: 输入包含敏感内容 }, outputs_config: { enabled: true, preset_response: 无法提供此内容 } } } }3. 配置验证后端会自动进行以下验证文件: api/core/moderation/keywords/keywords.pyclassmethoddefvalidate_config(cls,tenant_id:str,config:dict):# 验证基础结构cls._validate_inputs_and_outputs_config(config,True)# 验证关键词字段存在ifnotconfig.get(keywords):raiseValueError(keywords is required)# 验证关键词总长度iflen(config.get(keywords,[]))10000:raiseValueError(keywords length must be less than 10000)# 验证关键词行数keywords_row_lenconfig[keywords].split(\n)iflen(keywords_row_len)100:raiseValueError(the number of rows for the keywords must be less than 100)验证错误示例:// 错误关键词为空{error:keywords is required}// 错误超过字符限制{error:keywords length must be less than 10000}// 错误超过行数限制{error:the number of rows for the keywords must be less than 100}// 错误未启用任何审核{error:At least one of inputs_config or outputs_config must be enabled}存储位置配置最终存储在数据库中表:app_model_configs字段:sensitive_word_avoidance(LongText)格式: JSON 字符串SELECTid,app_id,JSON_EXTRACT(sensitive_word_avoidance,$.type)asmoderation_type,JSON_EXTRACT(sensitive_word_avoidance,$.enabled)asis_enabledFROMapp_model_configsWHEREJSON_EXTRACT(sensitive_word_avoidance,$.type)keywords;配置加载流程1. 用户保存配置 ↓ 2. 前端调用 API: POST /apps/{app_id}/model-config ↓ 3. AppModelConfigService.validate_configuration() ↓ 4. KeywordsModeration.validate_config() # 验证配置 ↓ 5. 保存到 app_model_configs.sensitive_word_avoidance ↓ 6. 下次请求时ModerationFactory 加载配置 ↓ 7. 创建 KeywordsModeration 实例 ↓ 8. 执行审核逻辑配置示例示例 1: 仅审核输入适用场景防止用户输入敏感问题{enabled:true,type:keywords,config:{keywords:政治\n色情\n赌博\n暴力,inputs_config:{enabled:true,preset_response:您的输入包含敏感词请重新输入合适的问题},outputs_config:{enabled:false}}}示例 2: 仅审核输出适用场景防止 AI 生成不当内容{enabled:true,type:keywords,config:{keywords:竞品A\n竞品B\n内部机密,inputs_config:{enabled:false},outputs_config:{enabled:true,preset_response:抱歉我无法提供相关信息}}}示例 3: 双向审核适用场景全方位内容安全保护{enabled:true,type:keywords,config:{keywords:垃圾邮件\nspam\n刷单\n返利,inputs_config:{enabled:true,preset_response:检测到违规内容},outputs_config:{enabled:true,preset_response:系统无法处理此请求}}}示例 4: 多语言关键词{enabled:true,type:keywords,config:{keywords:spam\nスパム\n垃圾邮件\nmalware\nマルウェア\n恶意软件,inputs_config:{enabled:true,preset_response:Invalid content detected},outputs_config:{enabled:true,preset_response:无法提供此信息}}}常见问题Q: 如何处理多语言关键词A: 关键词列表支持任何语言的 UTF-8 文本直接添加即可spam 垃圾邮件 スパムQ: 能否支持正则表达式A: 当前实现不支持但可以扩展importre keywords_patterns[re.compile(pattern)forpatterninconfig[keywords].split(\n)]def_check_keywords_in_value(self,keywords,value):value_strstr(value).lower()returnany(pattern.search(value_str)forpatterninkeywords)Q: 如何处理繁简体变换A: 可使用 OpenCC 库进行繁简转换fromopenccimportOpenCC ccOpenCC(s2t)# 简体转繁体textcc.convert(value)Q: 关键词修改后何时生效A: 下次创建ModerationFactory时加载新配置。已有的审核实例不受影响。扩展建议1. 缓存优化classKeywordsModeration(Moderation):_keywords_cache:dict[str,set[str]]{}def__init__(self,app_id,tenant_id,configNone):super().__init__(app_id,tenant_id,config)self._cache_keyf{tenant_id}:{app_id}ifself._cache_keynotinself._keywords_cacheandconfig:self._keywords_cache[self._cache_key]set(k.lower()forkinconfig.get(keywords,).split(\n)ifk)2. 分布式匹配# 使用 Elasticsearch 进行大规模关键词匹配def_is_violated_distributed(self,inputs,keywords_list):content .join(str(v)forvininputs.values())responsees.search(indexkeywords,query{match:{content:content}})returnlen(response[hits][hits])03. 动态权重# 不同关键词的风险等级HIGH_RISK[非法,诈骗]# 立即拒绝MEDIUM_RISK[广告,推销]# 需要审核LOW_RISK[可能,了解]# 只记录def_is_violated(self,inputs,keywords_map):forrisk_level,keywordsinkeywords_map.items():ifself._check_keywords_in_value(keywords,value):ifrisk_levelHIGH_RISK:returnTrue# 立即返回最后更新: 2026-01-18