feat(action): 重构 Action 激活机制并添加 go_activate() 方法

引入新的 Action 激活机制,允许通过重写 go_activate() 方法来自定义激活逻辑。提供了三个工具函数:
- _random_activation(): 随机概率激活
- _keyword_match(): 关键词匹配激活
- _llm_judge_activation(): LLM 智能判断激活

主要变更:
- 在 BaseAction 中添加 go_activate() 抽象方法和相关工具函数
- 更新 ActionModifier 使用新的激活判断逻辑
- 在 hello_world_plugin 中添加新的激活方式示例
- 更新文档说明新的激活机制
- 保持向后兼容,旧的激活类型配置仍然可用

BREAKING CHANGE: Action 激活判断现在通过 go_activate() 方法进行,旧的激活类型字段已标记为废弃但仍然兼容
This commit is contained in:
Windpicker-owo
2025-10-17 20:16:15 +08:00
parent ce3fe95b37
commit f22e6365cc
7 changed files with 961 additions and 65 deletions

View File

@@ -0,0 +1,413 @@
# Action 激活机制重构指南
## 📋 概述
本文档介绍 MoFox-Bot Action 组件的新激活机制。新机制通过 `go_activate()` 方法提供更灵活、更强大的激活判断能力。
## 🎯 为什么要重构?
### 旧的激活机制的问题
1. **不够灵活**:只能使用预定义的激活类型(`ALWAYS``NEVER``RANDOM``KEYWORD``LLM_JUDGE`
2. **难以组合**:无法轻松组合多种激活条件
3. **配置复杂**:需要在类属性中配置多个字段
4. **扩展困难**:添加新的激活逻辑需要修改核心代码
### 新的激活机制的优势
1. **完全自定义**:通过重写 `go_activate()` 方法实现任意激活逻辑
2. **灵活组合**:可以轻松组合多种激活条件
3. **简洁明了**:激活逻辑集中在一个方法中
4. **易于扩展**:可以实现任何复杂的激活判断
## 🚀 快速开始
### 基本结构
```python
from src.plugin_system import BaseAction
class MyAction(BaseAction):
"""我的自定义 Action"""
action_name = "my_action"
action_description = "这是一个示例 Action"
async def go_activate(self, llm_judge_model=None) -> bool:
"""判断此 Action 是否应该被激活
Args:
chat_content: 聊天内容
llm_judge_model: LLM 判断模型(可选)
Returns:
bool: True 表示激活False 表示不激活
"""
# 在这里实现你的激活逻辑
return True
async def execute(self) -> tuple[bool, str]:
"""执行 Action 的具体逻辑"""
await self.send_text("Hello, World!")
return True, "发送成功"
```
## 🛠️ 工具函数
BaseAction 提供了三个便捷的工具函数来简化常见的激活判断:
### 1. `_random_activation(probability)` - 随机激活
```python
async def go_activate(self, llm_judge_model=None) -> bool:
"""30% 概率激活"""
return await self._random_activation(0.3)
```
**参数:**
- `probability`: 激活概率,范围 0.0 到 1.0
### 2. `_keyword_match(keywords, case_sensitive)` - 关键词匹配
```python
async def go_activate(self, llm_judge_model=None) -> bool:
"""当消息包含特定关键词时激活"""
return await self._keyword_match(
keywords=["你好", "hello", "hi"],
case_sensitive=False # 不区分大小写
)
```
**参数:**
- `keywords`: 关键词列表
- `case_sensitive`: 是否区分大小写(默认 False
### 3. `_llm_judge_activation(...)` - LLM 智能判断
```python
async def go_activate(self, llm_judge_model=None) -> bool:
"""使用 LLM 判断是否激活"""
return await self._llm_judge_activation(
judge_prompt="当用户询问天气信息时激活",
llm_judge_model=llm_judge_model
)
```
**参数:**
- `judge_prompt`: 判断提示词(核心判断逻辑)
- `llm_judge_model`: LLM 模型实例(可选,会自动创建)
- `action_description`: Action 描述(可选,默认使用类属性)
- `action_require`: 使用场景列表(可选,默认使用类属性)
## 📚 示例
### 示例 1简单的关键词激活
```python
class GreetingAction(BaseAction):
"""问候 Action - 当检测到问候语时激活"""
action_name = "greeting"
action_description = "回应用户的问候"
async def go_activate(self, llm_judge_model=None) -> bool:
"""检测到问候语时激活"""
return await self._keyword_match(
keywords=["你好", "hello", "hi", "嗨"],
case_sensitive=False
)
async def execute(self) -> tuple[bool, str]:
await self.send_text("你好!很高兴见到你!👋")
return True, "发送了问候"
```
### 示例 2LLM 智能判断激活
```python
class ComfortAction(BaseAction):
"""安慰 Action - 当用户情绪低落时激活"""
action_name = "comfort"
action_description = "提供情感支持和安慰"
action_require = ["用户情绪低落", "需要安慰"]
async def go_activate(self, llm_judge_model=None) -> bool:
"""使用 LLM 判断用户是否需要安慰"""
return await self._llm_judge_activation(
judge_prompt="""
判断用户是否表达了以下情绪或需求:
1. 感到难过、沮丧或失落
2. 表达了负面情绪
3. 需要安慰或鼓励
如果满足上述条件,回答"是",否则回答"否"。
""",
llm_judge_model=llm_judge_model
)
async def execute(self) -> tuple[bool, str]:
await self.send_text("看起来你心情不太好,希望能让你开心一点!🤗💕")
return True, "发送了安慰"
```
### 示例 3随机激活
```python
class RandomEmojiAction(BaseAction):
"""随机表情 Action - 10% 概率激活"""
action_name = "random_emoji"
action_description = "随机发送表情增加趣味性"
async def go_activate(self, llm_judge_model=None) -> bool:
"""10% 概率激活"""
return await self._random_activation(0.1)
async def execute(self) -> tuple[bool, str]:
import random
emojis = ["😊", "😂", "👍", "🎉", "🤔", "🤖"]
await self.send_text(random.choice(emojis))
return True, "发送了表情"
```
### 示例 4组合多种激活条件
```python
class FlexibleAction(BaseAction):
"""灵活的 Action - 组合多种激活条件"""
action_name = "flexible"
action_description = "展示灵活的激活逻辑"
async def go_activate(self, llm_judge_model=None) -> bool:
"""组合激活:随机 20% 概率,或者匹配关键词"""
# 策略 1: 随机激活
if await self._random_activation(0.2):
return True
# 策略 2: 关键词匹配
if await self._keyword_match(["表情", "emoji"], case_sensitive=False):
return True
# 策略 3: 所有条件都不满足
return False
async def execute(self) -> tuple[bool, str]:
await self.send_text("这是一个灵活的激活示例!✨")
return True, "执行成功"
```
### 示例 5复杂的自定义逻辑
```python
class AdvancedAction(BaseAction):
"""高级 Action - 实现复杂的激活逻辑"""
action_name = "advanced"
action_description = "高级激活逻辑示例"
async def go_activate(self, llm_judge_model=None) -> bool:
"""实现复杂的激活逻辑"""
# 1. 检查时间:只在工作时间激活
from datetime import datetime
now = datetime.now()
if now.hour < 9 or now.hour > 18:
return False
# 2. 检查消息长度:消息太短不激活
if len(chat_content) < 10:
return False
# 3. 组合关键词和 LLM 判断
has_keyword = await self._keyword_match(
["帮助", "help", "求助"],
case_sensitive=False
)
if has_keyword:
# 如果匹配到关键词,用 LLM 进一步判断
return await self._llm_judge_activation(
judge_prompt="用户是否真的需要帮助?",
llm_judge_model=llm_judge_model
)
return False
async def execute(self) -> tuple[bool, str]:
await self.send_text("我来帮助你!")
return True, "提供了帮助"
```
### 示例 6始终激活或从不激活
```python
class AlwaysActiveAction(BaseAction):
"""始终激活的 Action"""
action_name = "always_active"
action_description = "这个 Action 总是激活"
async def go_activate(self, llm_judge_model=None) -> bool:
"""始终返回 True"""
return True
async def execute(self) -> tuple[bool, str]:
await self.send_text("我总是可用!")
return True, "执行成功"
class NeverActiveAction(BaseAction):
"""从不激活的 Action可用于测试或临时禁用"""
action_name = "never_active"
action_description = "这个 Action 从不激活"
async def go_activate(self, llm_judge_model=None) -> bool:
"""始终返回 False"""
return False
async def execute(self) -> tuple[bool, str]:
# 这个方法不会被调用
return False, "未执行"
```
## 🔄 从旧的激活机制迁移
### 旧写法(已废弃但仍然兼容)
```python
class OldStyleAction(BaseAction):
action_name = "old_style"
action_description = "旧风格的 Action"
# 旧的激活配置
activation_type = ActionActivationType.KEYWORD
activation_keywords = ["你好", "hello"]
keyword_case_sensitive = False
async def execute(self) -> tuple[bool, str]:
return True, "执行成功"
```
### 新写法(推荐)
```python
class NewStyleAction(BaseAction):
action_name = "new_style"
action_description = "新风格的 Action"
async def go_activate(self, llm_judge_model=None) -> bool:
"""使用新的激活方式"""
return await self._keyword_match(
chat_content,
keywords=["你好", "hello"],
case_sensitive=False
)
async def execute(self) -> tuple[bool, str]:
return True, "执行成功"
```
### 迁移对照表
| 旧的激活类型 | 新的实现方式 |
|-------------|-------------|
| `ActionActivationType.ALWAYS` | `return True` |
| `ActionActivationType.NEVER` | `return False` |
| `ActionActivationType.RANDOM` | `return await self._random_activation(probability)` |
| `ActionActivationType.KEYWORD` | `return await self._keyword_match( keywords)` |
| `ActionActivationType.LLM_JUDGE` | `return await self._llm_judge_activation(judge_prompt, llm_judge_model)` |
## ⚠️ 注意事项
### 1. 向后兼容性
旧的激活类型配置仍然有效!如果你的 Action 没有重写 `go_activate()` 方法BaseAction 的默认实现会自动使用旧的配置字段。
### 2. 性能考虑
- `_random_activation()``_keyword_match()` 非常快速
- `_llm_judge_activation()` 需要调用 LLM会有延迟
- ActionModifier 会并行执行所有 Action 的 `go_activate()` 方法以提高性能
### 3. 日志记录
工具函数会自动记录调试日志,便于追踪激活决策过程:
```
[DEBUG] 随机激活判断: 概率=0.3, 结果=激活
[DEBUG] 匹配到关键词: ['你好', 'hello']
[DEBUG] LLM 判断结果: 响应='是', 结果=激活
```
### 4. 错误处理
- 如果 `go_activate()` 抛出异常Action 会被标记为不激活
- `_llm_judge_activation()` 在出错时默认返回 False不激活
## 🎨 最佳实践
### 1. 保持 `go_activate()` 方法简洁
```python
# ✅ 好的做法:简洁明了
async def go_activate(self, llm_judge_model=None) -> bool:
return await self._keyword_match(["帮助", "help"])
# ❌ 不好的做法:过于复杂
async def go_activate(self, llm_judge_model=None) -> bool:
# 大量复杂的逻辑...
# 应该拆分成辅助方法
```
### 2. 合理使用 LLM 判断
```python
# ✅ 好的做法:需要语义理解时使用 LLM
async def go_activate(self, llm_judge_model=None) -> bool:
# 判断用户情绪需要 LLM
return await self._llm_judge_activation(
"用户是否情绪低落?",
llm_judge_model
)
# ❌ 不好的做法:简单匹配也用 LLM浪费资源
async def go_activate(self, llm_judge_model=None) -> bool:
# 简单的关键词匹配不需要 LLM
return await self._llm_judge_activation(
"消息是否包含'你好'",
llm_judge_model
)
```
### 3. 组合条件时使用清晰的逻辑结构
```python
# ✅ 好的做法:清晰的条件组合
async def go_activate(self, llm_judge_model=None) -> bool:
# 策略 1: 快速路径 - 关键词匹配
if await self._keyword_match(["紧急", "urgent"]):
return True
# 策略 2: 随机激活
if await self._random_activation(0.1):
return True
# 策略 3: LLM 判断(最耗时,放最后)
return await self._llm_judge_activation(
"是否需要特别关注?",
llm_judge_model
)
```
## 📖 完整示例项目
查看 `plugins/hello_world_plugin/plugin.py` 获取更多实际示例。
## 🤝 贡献
如果你有更好的激活逻辑实现,欢迎分享!

View File

@@ -1,5 +1,9 @@
# ⚡ Action组件详解
> **🎉 新功能:更灵活的激活机制!**
> MoFox-Bot 现在支持通过 `go_activate()` 方法自定义 Action 激活逻辑!
> 详见:[Action 激活机制重构指南](./action-activation-guide.md)
## 📖 什么是Action
Action是给MoFox_Bot在回复之外提供额外功能的智能组件**由MoFox_Bot的决策系统自主选择是否使用**具有随机性和拟人化的调用特点。Action不是直接响应用户命令而是让MoFox_Bot根据聊天情境智能地选择合适的动作使其行为更加自然和真实。
@@ -72,11 +76,55 @@ Action采用**两层决策机制**来优化性能和决策质量:
**第一层激活控制Activation Control**
激活决定MoFox-Bot是否 **知道** 这个Action的存在即这个Action是否进入决策候选池。不被激活的ActionMoFox-Bot永远不会选择。
激活决定MoFox-Bot是否 **"知道"** 这个Action的存在即这个Action是否进入决策候选池。不被激活的ActionMoFox-Bot永远不会选择。
**第二层使用决策Usage Decision**
在Action被激活后使用条件决定MoFox-Bot什么时候会 **选择** 使用这个Action。
在Action被激活后使用条件决定MoFox-Bot什么时候会 **"选择"** 使用这个Action。
---
## 🆕 新的激活机制(推荐)
从现在开始,推荐使用 **`go_activate()` 方法** 来自定义 Action 的激活逻辑。这种方式更灵活、更强大!
### 快速示例
```python
class MyAction(BaseAction):
action_name = "my_action"
action_description = "我的自定义 Action"
async def go_activate(self, llm_judge_model=None) -> bool:
"""判断是否激活此 Action
注意:聊天内容会自动从实例属性中获取,无需手动传入
"""
# 关键词激活
if await self._keyword_match(["你好", "hello"]):
return True
# 或者随机 10% 概率激活
return await self._random_activation(0.1)
async def execute(self) -> tuple[bool, str]:
await self.send_text("Hello!")
return True, "发送成功"
```
**提供的工具函数:**
- `_random_activation(probability)` - 随机激活
- `_keyword_match(keywords)` - 关键词匹配(自动获取聊天内容)
- `_llm_judge_activation(judge_prompt, llm_judge_model)` - LLM 智能判断(自动获取聊天内容)
**📚 完整指南:** 查看 [Action 激活机制重构指南](./action-activation-guide.md) 了解详情和更多示例。
---
## 📜 旧的激活机制(已废弃但仍然兼容)
> ⚠️ **注意:** 以下激活类型配置方式已废弃,但仍然兼容。
> 推荐使用新的 `go_activate()` 方法来实现更灵活的激活逻辑。
### 决策参数详解 🔧

View File

@@ -57,16 +57,113 @@ class HelloCommand(PlusCommand):
return True, "成功发送问候", True
# ==================================================================================
# 新的激活方式示例 Actions
# ==================================================================================
class KeywordActivationExampleAction(BaseAction):
"""关键词激活示例
此示例展示如何使用关键词匹配来激活 Action。
"""
action_name = "keyword_example"
action_description = "当检测到特定关键词时发送回应"
action_require = ["用户提到了问候语"]
associated_types = ["text"]
async def go_activate(self, chat_content: str = "", llm_judge_model=None) -> bool:
"""关键词激活:检测到"你好""hello""hi"时激活"""
return await self._keyword_match(
chat_content,
keywords=["你好", "hello", "hi", ""],
case_sensitive=False # 不区分大小写
)
async def execute(self) -> tuple[bool, str]:
await self.send_text("检测到问候语,我也向你问好!👋")
return True, "发送了问候回应"
class LLMJudgeExampleAction(BaseAction):
"""LLM 判断激活示例
此示例展示如何使用 LLM 来智能判断是否激活 Action。
"""
action_name = "llm_judge_example"
action_description = "当用户表达情绪低落时提供安慰"
action_require = ["用户情绪低落", "需要情感支持"]
associated_types = ["text"]
async def go_activate(self, chat_content: str = "", llm_judge_model=None) -> bool:
"""LLM 判断激活:判断用户是否情绪低落"""
return await self._llm_judge_activation(
chat_content=chat_content,
judge_prompt="""
判断用户是否表达了以下情绪或需求:
1. 感到难过、沮丧或失落
2. 表达了负面情绪
3. 需要安慰或鼓励
如果用户表达了上述情绪或需求,回答"",否则回答""
""",
llm_judge_model=llm_judge_model
)
async def execute(self) -> tuple[bool, str]:
await self.send_text("看起来你心情不太好,希望能让你开心一点!🤗💕")
return True, "发送了安慰消息"
class CombinedActivationExampleAction(BaseAction):
"""组合激活条件示例
此示例展示如何组合多种激活条件。
"""
action_name = "combined_example"
action_description = "展示如何组合多种激活条件"
action_require = ["展示灵活的激活逻辑"]
associated_types = ["text"]
async def go_activate(self, chat_content: str = "", llm_judge_model=None) -> bool:
"""组合激活:随机 20% 概率,或者匹配特定关键词"""
# 先尝试随机激活
if await self._random_activation(0.2):
return True
# 如果随机未激活,尝试关键词匹配
if await self._keyword_match(chat_content, ["表情", "emoji", "😊"], case_sensitive=False):
return True
# 都不满足则不激活
return False
async def execute(self) -> tuple[bool, str]:
await self.send_text("这是一个组合激活条件的示例!✨")
return True, "发送了示例消息"
class RandomEmojiAction(BaseAction):
"""一个随机发送表情的动作。"""
"""一个随机发送表情的动作。
此示例展示了如何使用新的 go_activate() 方法来实现随机激活。
"""
action_name = "random_emoji"
action_description = "随机发送一个表情符号,增加聊天的趣味性。"
activation_type = ActionActivationType.RANDOM
random_activation_probability = 0.1
action_require = ["当对话气氛轻松时", "可以用来回应简单的情感表达"]
associated_types = ["text"]
async def go_activate(self, llm_judge_model=None) -> bool:
"""使用新的激活方式10% 的概率激活
注意:不需要传入 chat_content会自动从实例属性中获取
"""
return await self._random_activation(0.1)
async def execute(self) -> tuple[bool, str]:
emojis = ["😊", "😂", "👍", "🎉", "🤔", "🤖"]
await self.send_text(random.choice(emojis))

View File

@@ -197,6 +197,8 @@ class ActionModifier:
"""
根据激活类型过滤,返回需要停用的动作列表及原因
新的实现:调用每个 Action 类的 go_activate 方法来判断是否激活
Args:
actions_with_info: 带完整信息的动作字典
chat_content: 聊天内容
@@ -206,55 +208,71 @@ class ActionModifier:
"""
deactivated_actions = []
# 分类处理不同激活类型的actions
llm_judge_actions = {}
# 获取 Action 类注册表
from src.plugin_system.core.component_registry import component_registry
from src.plugin_system.base.component_types import ComponentType
actions_to_check = list(actions_with_info.items())
random.shuffle(actions_to_check)
# 创建并行任务列表
activation_tasks = []
task_action_names = []
for action_name, action_info in actions_to_check:
activation_type = action_info.activation_type or action_info.focus_activation_type
# 获取 Action 类
action_class = component_registry.get_component_class(action_name, ComponentType.ACTION)
if not action_class:
logger.warning(f"{self.log_prefix}未找到 Action 类: {action_name},默认不激活")
deactivated_actions.append((action_name, "未找到 Action 类"))
continue
if activation_type == ActionActivationType.ALWAYS:
continue # 总是激活,无需处理
# 创建一个临时实例来调用 go_activate 方法
# 注意:这里只是为了调用 go_activate不需要完整的初始化
try:
# 创建一个最小化的实例
action_instance = object.__new__(action_class)
# 设置必要的属性
action_instance.action_name = action_name
action_instance.log_prefix = self.log_prefix
# 设置聊天内容,用于激活判断
action_instance._activation_chat_content = chat_content
elif activation_type == ActionActivationType.RANDOM:
probability = action_info.random_activation_probability
probability = action_info.random_activation_probability
if random.random() >= probability:
reason = f"RANDOM类型未触发概率{probability}"
deactivated_actions.append((action_name, reason))
logger.debug(f"{self.log_prefix}未激活动作: {action_name},原因: {reason}")
# 调用 go_activate 方法(不再需要传入 chat_content
task = action_instance.go_activate(
llm_judge_model=self.llm_judge,
)
activation_tasks.append(task)
task_action_names.append(action_name)
elif activation_type == ActionActivationType.KEYWORD:
if not self._check_keyword_activation(action_name, action_info, chat_content):
keywords = action_info.activation_keywords
reason = f"关键词未匹配(关键词: {keywords}"
deactivated_actions.append((action_name, reason))
logger.debug(f"{self.log_prefix}未激活动作: {action_name},原因: {reason}")
except Exception as e:
logger.error(f"{self.log_prefix}创建 Action 实例 {action_name} 失败: {e}")
deactivated_actions.append((action_name, f"创建实例失败: {e}"))
elif activation_type == ActionActivationType.LLM_JUDGE:
llm_judge_actions[action_name] = action_info
# 并行执行所有激活判断
if activation_tasks:
logger.debug(f"{self.log_prefix}并行执行激活判断,任务数: {len(activation_tasks)}")
try:
task_results = await asyncio.gather(*activation_tasks, return_exceptions=True)
elif activation_type == ActionActivationType.NEVER:
reason = "激活类型为never"
deactivated_actions.append((action_name, reason))
logger.debug(f"{self.log_prefix}激活动作: {action_name},原因: 激活类型为never")
# 处理结果
for action_name, result in zip(task_action_names, task_results, strict=False):
if isinstance(result, Exception):
logger.error(f"{self.log_prefix}激活判断 {action_name} 时出错: {result}")
deactivated_actions.append((action_name, f"激活判断出错: {result}"))
elif not result:
# go_activate 返回 False不激活
deactivated_actions.append((action_name, "go_activate 返回 False"))
logger.debug(f"{self.log_prefix}未激活动作: {action_name},原因: go_activate 返回 False")
else:
# go_activate 返回 True激活
logger.debug(f"{self.log_prefix}激活动作: {action_name}")
else:
logger.warning(f"{self.log_prefix}未知的激活类型: {activation_type},跳过处理")
# 并行处理LLM_JUDGE类型
if llm_judge_actions:
llm_results = await self._process_llm_judge_actions_parallel(
llm_judge_actions,
chat_content,
)
for action_name, should_activate in llm_results.items():
if not should_activate:
reason = "LLM判定未激活"
deactivated_actions.append((action_name, reason))
logger.debug(f"{self.log_prefix}未激活动作: {action_name},原因: {reason}")
except Exception as e:
logger.error(f"{self.log_prefix}并行激活判断失败: {e}")
# 如果并行执行失败,为所有任务默认不激活
for action_name in task_action_names:
deactivated_actions.append((action_name, f"并行判断失败: {e}"))
return deactivated_actions

View File

@@ -1,13 +1,18 @@
# Todo: 重构Action,这里现在只剩下了报错。
import asyncio
import random
import time
from abc import ABC, abstractmethod
from typing import TYPE_CHECKING
from src.chat.message_receive.chat_stream import ChatStream
from src.common.logger import get_logger
from src.plugin_system.apis import database_api, message_api, send_api
from src.plugin_system.base.component_types import ActionActivationType, ActionInfo, ChatMode, ChatType, ComponentType
if TYPE_CHECKING:
from src.llm_models.utils_model import LLMRequest
logger = get_logger("base_action")
@@ -16,16 +21,53 @@ class BaseAction(ABC):
Action是插件的一种组件类型用于处理聊天中的动作逻辑
子类可以通过类属性定义激活条件,这些会在实例化时转换为实例属性:
==================================================================================
新的激活机制 (推荐使用)
==================================================================================
推荐通过重写 go_activate() 方法来自定义激活逻辑:
示例 1 - 关键词激活:
async def go_activate(self, llm_judge_model=None) -> bool:
return await self._keyword_match(["你好", "hello"])
示例 2 - LLM 判断激活:
async def go_activate(self, llm_judge_model=None) -> bool:
return await self._llm_judge_activation(
"当用户询问天气信息时激活",
llm_judge_model
)
示例 3 - 组合多种条件:
async def go_activate(self, llm_judge_model=None) -> bool:
# 30% 随机概率,或者匹配关键词
if await self._random_activation(0.3):
return True
return await self._keyword_match(["表情", "emoji"])
提供的工具函数:
- _random_activation(probability): 随机激活
- _keyword_match(keywords, case_sensitive): 关键词匹配(自动获取聊天内容)
- _llm_judge_activation(judge_prompt, llm_judge_model): LLM 判断(自动获取聊天内容)
注意:聊天内容会自动从实例属性中获取,无需手动传入。
==================================================================================
旧的激活机制 (已废弃,但仍然兼容)
==================================================================================
子类可以通过类属性定义激活条件(已废弃,但 go_activate() 的默认实现会使用这些):
- focus_activation_type: 专注模式激活类型
- normal_activation_type: 普通模式激活类型
- activation_keywords: 激活关键词列表
- keyword_case_sensitive: 关键词是否区分大小写
- mode_enable: 启用的聊天模式
- parallel_action: 是否允许并行执行
- random_activation_probability: 随机激活概率
- llm_judge_prompt: LLM判断提示词
==================================================================================
其他类属性
==================================================================================
- mode_enable: 启用的聊天模式
- parallel_action: 是否允许并行执行
二步Action相关属性
- is_two_step_action: 是否为二步Action
- step_one_description: 第一步的描述
@@ -559,6 +601,247 @@ class BaseAction(ABC):
# 子类需要重写此方法来实现具体的第二步逻辑
return False, f"二步Action必须实现execute_step_two方法来处理操作: {sub_action_name}"
# =============================================================================
# 新的激活机制 - go_activate 和工具函数
# =============================================================================
def _get_chat_content(self) -> str:
"""获取聊天内容用于激活判断
从实例属性中获取聊天内容。子类可以重写此方法来自定义获取逻辑。
Returns:
str: 聊天内容
"""
# 尝试从不同的实例属性中获取聊天内容
# 优先级_activation_chat_content > action_data['chat_content'] > ""
# 1. 如果有专门设置的激活用聊天内容(由 ActionModifier 设置)
if hasattr(self, '_activation_chat_content'):
return getattr(self, '_activation_chat_content', "")
# 2. 尝试从 action_data 中获取
if hasattr(self, 'action_data') and isinstance(self.action_data, dict):
return self.action_data.get('chat_content', "")
# 3. 默认返回空字符串
return ""
async def go_activate(
self,
llm_judge_model: "LLMRequest | None" = None,
) -> bool:
"""判断此 Action 是否应该被激活
这是新的激活机制的核心方法。子类可以重写此方法来实现自定义的激活逻辑,
也可以使用提供的工具函数来简化常见的激活判断。
默认实现会检查类属性中的激活类型配置,提供向后兼容支持。
聊天内容会自动从实例属性中获取,不需要手动传入。
Args:
llm_judge_model: LLM 判断模型,如果需要使用 LLM 判断
Returns:
bool: True 表示应该激活False 表示不激活
Example:
>>> # 简单的关键词激活
>>> async def go_activate(self, llm_judge_model=None) -> bool:
>>> return await self._keyword_match(["你好", "hello"])
>>>
>>> # LLM 判断激活
>>> async def go_activate(self, llm_judge_model=None) -> bool:
>>> return await self._llm_judge_activation(
>>> "当用户询问天气信息时激活",
>>> llm_judge_model
>>> )
>>>
>>> # 组合多种条件
>>> async def go_activate(self, llm_judge_model=None) -> bool:
>>> # 随机 30% 概率,或者匹配关键词
>>> if await self._random_activation(0.3):
>>> return True
>>> return await self._keyword_match(["天气"])
"""
# 默认实现:向后兼容旧的激活类型系统
activation_type = getattr(self, "activation_type", ActionActivationType.ALWAYS)
if activation_type == ActionActivationType.ALWAYS:
return True
elif activation_type == ActionActivationType.NEVER:
return False
elif activation_type == ActionActivationType.RANDOM:
probability = getattr(self, "random_activation_probability", 0.0)
return await self._random_activation(probability)
elif activation_type == ActionActivationType.KEYWORD:
keywords = getattr(self, "activation_keywords", [])
case_sensitive = getattr(self, "keyword_case_sensitive", False)
return await self._keyword_match(keywords, case_sensitive)
elif activation_type == ActionActivationType.LLM_JUDGE:
prompt = getattr(self, "llm_judge_prompt", "")
return await self._llm_judge_activation(
judge_prompt=prompt,
llm_judge_model=llm_judge_model,
)
# 未知类型,默认不激活
logger.warning(f"{self.log_prefix} 未知的激活类型: {activation_type}")
return False
async def _random_activation(self, probability: float) -> bool:
"""随机激活工具函数
Args:
probability: 激活概率,范围 0.0 到 1.0
Returns:
bool: 是否激活
"""
result = random.random() < probability
logger.debug(f"{self.log_prefix} 随机激活判断: 概率={probability}, 结果={'激活' if result else '不激活'}")
return result
async def _keyword_match(
self,
keywords: list[str],
case_sensitive: bool = False,
) -> bool:
"""关键词匹配工具函数
聊天内容会自动从实例属性中获取。
Args:
keywords: 关键词列表
case_sensitive: 是否区分大小写
Returns:
bool: 是否匹配到关键词
"""
if not keywords:
logger.warning(f"{self.log_prefix} 关键词列表为空,默认不激活")
return False
# 自动获取聊天内容
chat_content = self._get_chat_content()
search_text = chat_content
if not case_sensitive:
search_text = search_text.lower()
matched_keywords = []
for keyword in keywords:
check_keyword = keyword if case_sensitive else keyword.lower()
if check_keyword in search_text:
matched_keywords.append(keyword)
if matched_keywords:
logger.debug(f"{self.log_prefix} 匹配到关键词: {matched_keywords}")
return True
else:
logger.debug(f"{self.log_prefix} 未匹配到任何关键词: {keywords}")
return False
async def _llm_judge_activation(
self,
judge_prompt: str = "",
llm_judge_model: "LLMRequest | None" = None,
action_description: str = "",
action_require: list[str] | None = None,
) -> bool:
"""LLM 判断激活工具函数
使用 LLM 来判断是否应该激活此 Action。
会自动构建完整的判断提示词,只需要提供核心判断逻辑即可。
聊天内容会自动从实例属性中获取。
Args:
judge_prompt: 自定义判断提示词(核心判断逻辑)
llm_judge_model: LLM 判断模型实例,如果为 None 则会创建默认的小模型
action_description: Action 描述,如果不提供则使用类属性
action_require: Action 使用场景,如果不提供则使用类属性
Returns:
bool: 是否应该激活
Example:
>>> # 最简单的用法
>>> result = await self._llm_judge_activation(
>>> "当用户询问天气信息时激活"
>>> )
>>>
>>> # 提供详细信息
>>> result = await self._llm_judge_activation(
>>> judge_prompt="当用户表达情绪或需要情感支持时激活",
>>> action_description="发送安慰表情包",
>>> action_require=["用户情绪低落", "需要情感支持"]
>>> )
"""
try:
# 自动获取聊天内容
chat_content = self._get_chat_content()
# 如果没有提供 LLM 模型,创建一个默认的
if llm_judge_model is None:
from src.config.config import model_config
from src.llm_models.utils_model import LLMRequest
llm_judge_model = LLMRequest(
model_set=model_config.model_task_config.utils_small,
request_type="action.judge",
)
# 使用类属性作为默认值
if not action_description:
action_description = getattr(self, "action_description", "Action 动作")
if action_require is None:
action_require = getattr(self, "action_require", [])
# 构建完整的判断提示词
prompt = f"""你需要判断在当前聊天情况下,是否应该激活名为"{self.action_name}"的动作。
动作描述:{action_description}
"""
if action_require:
prompt += "\n动作使用场景:\n"
for req in action_require:
prompt += f"- {req}\n"
if judge_prompt:
prompt += f"\n额外判定条件:\n{judge_prompt}\n"
if chat_content:
prompt += f"\n当前聊天记录:\n{chat_content}\n"
prompt += """
请根据以上信息判断是否应该激活这个动作。
只需要回答"""",不要有其他内容。
"""
# 调用 LLM 进行判断
response, _ = await llm_judge_model.generate_response_async(prompt=prompt)
response = response.strip().lower()
should_activate = "" in response or "yes" in response or "true" in response
logger.debug(
f"{self.log_prefix} LLM 判断结果: 响应='{response}', 结果={'激活' if should_activate else '不激活'}"
)
return should_activate
except Exception as e:
logger.error(f"{self.log_prefix} LLM 判断激活时出错: {e}")
# 出错时默认不激活
return False
@abstractmethod
async def execute(self) -> tuple[bool, str]:
"""执行Action的抽象方法子类必须实现

View File

@@ -132,21 +132,30 @@ class ComponentInfo:
@dataclass
class ActionInfo(ComponentInfo):
"""动作组件信息"""
"""动作组件信息
注意:激活类型相关字段已废弃,推荐使用 Action 类的 go_activate() 方法来自定义激活逻辑。
这些字段将继续保留以提供向后兼容性BaseAction.go_activate() 的默认实现会使用这些字段。
"""
action_parameters: dict[str, str] = field(
default_factory=dict
) # 动作参数与描述,例如 {"param1": "描述1", "param2": "描述2"}
action_require: list[str] = field(default_factory=list) # 动作需求说明
associated_types: list[str] = field(default_factory=list) # 关联的消息类型
# 激活类型相关
focus_activation_type: ActionActivationType = ActionActivationType.ALWAYS
normal_activation_type: ActionActivationType = ActionActivationType.ALWAYS
activation_type: ActionActivationType = ActionActivationType.ALWAYS
random_activation_probability: float = 0.0
llm_judge_prompt: str = ""
activation_keywords: list[str] = field(default_factory=list) # 激活关键词列表
keyword_case_sensitive: bool = False
# ==================================================================================
# 激活类型相关字段(已废弃,建议使用 go_activate() 方法)
# 保留这些字段是为了向后兼容BaseAction.go_activate() 的默认实现会使用这些字段
# ==================================================================================
focus_activation_type: ActionActivationType = ActionActivationType.ALWAYS # 已废弃
normal_activation_type: ActionActivationType = ActionActivationType.ALWAYS # 已废弃
activation_type: ActionActivationType = ActionActivationType.ALWAYS # 已废弃
random_activation_probability: float = 0.0 # 已废弃,建议在 go_activate() 中使用 _random_activation()
llm_judge_prompt: str = "" # 已废弃,建议在 go_activate() 中使用 _llm_judge_activation()
activation_keywords: list[str] = field(default_factory=list) # 已废弃,建议在 go_activate() 中使用 _keyword_match()
keyword_case_sensitive: bool = False # 已废弃
# 模式和并行设置
mode_enable: ChatMode = ChatMode.ALL
parallel_action: bool = False

View File

@@ -18,8 +18,36 @@ logger = get_logger("emoji")
class EmojiAction(BaseAction):
"""表情动作 - 发送表情包"""
"""表情动作 - 发送表情包
注意:此 Action 使用旧的激活类型配置方式(已废弃但仍然兼容)。
BaseAction.go_activate() 的默认实现会自动处理这些旧配置。
推荐的新写法(迁移示例):
----------------------------------------
# 移除下面的 activation_type 相关配置,改为重写 go_activate 方法:
async def go_activate(self, chat_content: str = "", llm_judge_model=None) -> bool:
# 根据配置选择激活方式
if global_config.emoji.emoji_activate_type == "llm":
return await self._llm_judge_activation(
chat_content=chat_content,
judge_prompt=\"""
判定是否需要使用表情动作的条件:
1. 用户明确要求使用表情包
2. 这是一个适合表达情绪的场合
3. 发表情包能使当前对话更有趣
4. 不要发送太多表情包
\""",
llm_judge_model=llm_judge_model
)
else:
# 使用随机激活
return await self._random_activation(global_config.emoji.emoji_chance)
----------------------------------------
"""
# ========== 以下使用旧的激活配置(已废弃但兼容) ==========
# 激活设置
if global_config.emoji.emoji_activate_type == "llm":
activation_type = ActionActivationType.LLM_JUDGE