Refactor antiprompt flowchart and fix indentation
Updated the antipromptinjector flowchart for improved clarity and logic. Fixed indentation in expression_selector.py to correct code structure. Removed obsolete db_migration_plan.md documentation.
This commit is contained in:
@@ -1,18 +1,18 @@
|
||||
```mermaid
|
||||
flowchart TD
|
||||
A([消息进入]) --> B{LLM反注入拦截
|
||||
是否启动?}
|
||||
B -- 否 --> Z([进入消息流])
|
||||
B -- 是 --> C{黑白名单判断}
|
||||
C -- 黑名单 --> D([丢弃消息])
|
||||
C -- 白名单 --> Z
|
||||
C -- 不在名单中 --> E{模式选择}
|
||||
E -- 仅正则表达模式 --> F[正则验证] --> G{验证通过?}
|
||||
G -- 是 --> Z
|
||||
G -- 否 --> D
|
||||
E -- 二次判定模式 --> H[正则验证] --> I{正则结果}
|
||||
I -- 正常 --> Z
|
||||
I -- 可疑 --> J[LLM二次判定] --> K{判定结果}
|
||||
K -- 安全 --> Z
|
||||
K -- 注入风险 --> D
|
||||
```
|
||||
A[消息进入系统] --> B{LLM反注入是否启动?}
|
||||
B -->|是| C{黑白名单检测}
|
||||
B -->|否| Y
|
||||
C -->|白名单| Y{继续进行消息处理}
|
||||
C -->|无记录| D{是否命中规则集}
|
||||
C -->|黑名单| X{丢弃消息}
|
||||
D -->|否| E{是否启动LLM二次分析}
|
||||
D -->|是| G{处理模式}
|
||||
E -->|是| F{提交LLM处理}
|
||||
E -->|否| Y
|
||||
F -->|LLM判定高危| G
|
||||
F -->|LLM判定无害| Y
|
||||
G -->|严格模式| X
|
||||
G -->|宽松模式| H{消息加盾}
|
||||
H --> Y
|
||||
```
|
||||
|
||||
@@ -140,47 +140,47 @@ class ExpressionSelector:
|
||||
(Expression.chat_id.in_(related_chat_ids)) & (Expression.type == "grammar")
|
||||
))
|
||||
|
||||
style_exprs = [
|
||||
{
|
||||
"situation": expr.situation,
|
||||
"style": expr.style,
|
||||
"count": expr.count,
|
||||
"last_active_time": expr.last_active_time,
|
||||
"source_id": expr.chat_id,
|
||||
"type": "style",
|
||||
"create_date": expr.create_date if expr.create_date is not None else expr.last_active_time,
|
||||
}
|
||||
for expr in style_query.scalars()
|
||||
]
|
||||
style_exprs = [
|
||||
{
|
||||
"situation": expr.situation,
|
||||
"style": expr.style,
|
||||
"count": expr.count,
|
||||
"last_active_time": expr.last_active_time,
|
||||
"source_id": expr.chat_id,
|
||||
"type": "style",
|
||||
"create_date": expr.create_date if expr.create_date is not None else expr.last_active_time,
|
||||
}
|
||||
for expr in style_query.scalars()
|
||||
]
|
||||
|
||||
grammar_exprs = [
|
||||
{
|
||||
"situation": expr.situation,
|
||||
"style": expr.style,
|
||||
"count": expr.count,
|
||||
"last_active_time": expr.last_active_time,
|
||||
"source_id": expr.chat_id,
|
||||
"type": "grammar",
|
||||
"create_date": expr.create_date if expr.create_date is not None else expr.last_active_time,
|
||||
}
|
||||
for expr in grammar_query.scalars()
|
||||
]
|
||||
grammar_exprs = [
|
||||
{
|
||||
"situation": expr.situation,
|
||||
"style": expr.style,
|
||||
"count": expr.count,
|
||||
"last_active_time": expr.last_active_time,
|
||||
"source_id": expr.chat_id,
|
||||
"type": "grammar",
|
||||
"create_date": expr.create_date if expr.create_date is not None else expr.last_active_time,
|
||||
}
|
||||
for expr in grammar_query.scalars()
|
||||
]
|
||||
|
||||
style_num = int(total_num * style_percentage)
|
||||
grammar_num = int(total_num * grammar_percentage)
|
||||
# 按权重抽样(使用count作为权重)
|
||||
if style_exprs:
|
||||
style_weights = [expr.get("count", 1) for expr in style_exprs]
|
||||
selected_style = weighted_sample(style_exprs, style_weights, style_num)
|
||||
else:
|
||||
selected_style = []
|
||||
if grammar_exprs:
|
||||
grammar_weights = [expr.get("count", 1) for expr in grammar_exprs]
|
||||
selected_grammar = weighted_sample(grammar_exprs, grammar_weights, grammar_num)
|
||||
else:
|
||||
selected_grammar = []
|
||||
|
||||
return selected_style, selected_grammar
|
||||
style_num = int(total_num * style_percentage)
|
||||
grammar_num = int(total_num * grammar_percentage)
|
||||
# 按权重抽样(使用count作为权重)
|
||||
if style_exprs:
|
||||
style_weights = [expr.get("count", 1) for expr in style_exprs]
|
||||
selected_style = weighted_sample(style_exprs, style_weights, style_num)
|
||||
else:
|
||||
selected_style = []
|
||||
if grammar_exprs:
|
||||
grammar_weights = [expr.get("count", 1) for expr in grammar_exprs]
|
||||
selected_grammar = weighted_sample(grammar_exprs, grammar_weights, grammar_num)
|
||||
else:
|
||||
selected_grammar = []
|
||||
|
||||
return selected_style, selected_grammar
|
||||
|
||||
def update_expressions_count_batch(self, expressions_to_update: List[Dict[str, Any]], increment: float = 0.1):
|
||||
"""对一批表达方式更新count值,按chat_id+type分组后一次性写入数据库"""
|
||||
|
||||
Reference in New Issue
Block a user