From a82de0a50e264f8ba9912862fdf7a9b9f461d6a9 Mon Sep 17 00:00:00 2001 From: UnCLAS-Prommer Date: Fri, 25 Jul 2025 00:08:00 +0800 Subject: [PATCH 1/2] =?UTF-8?q?action=E6=96=87=E6=A1=A3?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- docs/plugins/action-components.md | 548 +++++++++++--------------- src/plugin_system/base/base_action.py | 20 +- 2 files changed, 234 insertions(+), 334 deletions(-) diff --git a/docs/plugins/action-components.md b/docs/plugins/action-components.md index d68d87076..3953c79c2 100644 --- a/docs/plugins/action-components.md +++ b/docs/plugins/action-components.md @@ -4,42 +4,183 @@ Action是给麦麦在回复之外提供额外功能的智能组件，**由麦麦的决策系统自主选择是否使用**，具有随机性和拟人化的调用特点。Action不是直接响应用户命令，而是让麦麦根据聊天情境智能地选择合适的动作，使其行为更加自然和真实。 -### 🎯 Action的特点 +### Action的特点 - 🧠 **智能激活**：麦麦根据多种条件智能判断是否使用 -- 🎲 **随机性**：增加行为的不可预测性，更接近真人交流 +- 🎲 **可随机性**：可以使用随机数激活，增加行为的不可预测性，更接近真人交流 - 🤖 **拟人化**：让麦麦的回应更自然、更有个性 - 🔄 **情境感知**：基于聊天上下文做出合适的反应 -## 🎯 两层决策机制 +--- + +## 🎯 Action组件的基本结构 +首先，所有的Action都应该继承`BaseAction`类。 + +其次，每个Action组件都应该实现以下基本信息： +```python +class ExampleAction(BaseAction): + action_name = "example_action" # 动作的唯一标识符 + action_description = "这是一个示例动作" # 动作描述 + activation_type = ActionActivationType.ALWAYS # 这里以 ALWAYS 为例 + mode_enable = ChatMode.ALL # 这里以 ALL 为例 + associated_types = ["text", "emoji", ...] # 关联类型 + parallel_action = False # 是否允许与其他Action并行执行 + action_parameters = {"param1": "参数1的说明", "param2": "参数2的说明", ...} + # Action使用场景描述 - 帮助LLM判断何时"选择"使用 + action_require = ["使用场景描述1", "使用场景描述2", ...] + + async def execute(self) -> Tuple[bool, str]: + """ + 执行Action的主要逻辑 + + Returns: + Tuple[bool, str]: (是否成功, 执行结果描述) + """ + # ---- 执行动作的逻辑 ---- + return True, "执行成功" +``` +#### associated_types: 该Action会发送的消息类型，例如文本、表情等。 + +这部分由Adapter传递给处理器。 + +以 MaiBot-Napcat-Adapter 为例，可选项目如下： +| 类型 | 说明 | 格式 | +| --- | --- | --- | +| text | 文本消息 | str | +| emoji | 表情消息 | str: 表情包的无头base64| +| image | 图片消息 | str: 图片的无头base64 | +| reply | 回复消息 | str: 回复的消息ID | +| voice | 语音消息 | str: wav格式语音的无头base64 | +| command | 命令消息 | 参见Adapter文档 | +| voiceurl | 语音URL消息 | str: wav格式语音的URL | +| music | 音乐消息 | str: 这首歌在网易云音乐的音乐id | +| videourl | 视频URL消息 | str: 视频的URL | +| file | 文件消息 | str: 文件的路径 | + +**请知悉，对于不同的处理器，其支持的消息类型可能会有所不同。在开发时请注意。** + +#### action_parameters: 该Action的参数说明。 +这是一个字典，键为参数名，值为参数说明。这个字段可以帮助LLM理解如何使用这个Action，并由LLM返回对应的参数，最后传递到 Action 的 action_data 属性中。其格式与你定义的格式完全相同 **（除非LLM哈气了，返回了错误的内容）**。 + +--- + +## 🎯 Action 调用的决策机制 Action采用**两层决策机制**来优化性能和决策质量： -### 第一层：激活控制（Activation Control） +> 设计目的：在加载许多插件的时候降低LLM决策压力，避免让麦麦在过多的选项中纠结。 -**激活决定麦麦是否"知道"这个Action的存在**，即这个Action是否进入决策候选池。**不被激活的Action麦麦永远不会选择**。 +**第一层：激活控制（Activation Control）** -> 🎯 **设计目的**：在加载许多插件的时候降低LLM决策压力，避免让麦麦在过多的选项中纠结。 +激活决定麦麦是否 **“知道”** 这个Action的存在，即这个Action是否进入决策候选池。不被激活的Action麦麦永远不会选择。 -#### 激活类型说明 +**第二层：使用决策（Usage Decision）** -| 激活类型 | 说明 | 使用场景 | -| ------------- | ------------------------------------------- | ------------------------ | -| `NEVER` | 从不激活，Action对麦麦不可见 | 临时禁用某个Action | -| `ALWAYS` | 永远激活，Action总是在麦麦的候选池中 | 核心功能，如回复、不回复 | -| `LLM_JUDGE` | 通过LLM智能判断当前情境是否需要激活此Action | 需要智能判断的复杂场景 | -| `RANDOM` | 基于随机概率决定是否激活 | 增加行为随机性的功能 | -| `KEYWORD` | 当检测到特定关键词时激活 | 明确触发条件的功能 | +在Action被激活后，使用条件决定麦麦什么时候会 **“选择”** 使用这个Action。 -#### 聊天模式控制 +### 决策参数详解 🔧 -| 模式 | 说明 | -| ------------------- | ------------------------ | -| `ChatMode.FOCUS` | 仅在专注聊天模式下可激活 | -| `ChatMode.NORMAL` | 仅在普通聊天模式下可激活 | -| `ChatMode.ALL` | 所有模式下都可激活 | +#### 第一层：ActivationType 激活类型说明 -### 第二层：使用决策（Usage Decision） +| 激活类型 | 说明 | 使用场景 | +| ----------- | ---------------------------------------- | ---------------------- | +| [`NEVER`](#never-激活) | 从不激活，Action对麦麦不可见 | 临时禁用某个Action | +| [`ALWAYS`](#always-激活) | 永远激活，Action总是在麦麦的候选池中 | 核心功能，如回复、不回复 | +| [`LLM_JUDGE`](#llm_judge-激活) | 通过LLM智能判断当前情境是否需要激活此Action | 需要智能判断的复杂场景 | +| `RANDOM` | 基于随机概率决定是否激活 | 增加行为随机性的功能 | +| `KEYWORD` | 当检测到特定关键词时激活 | 明确触发条件的功能 | + +#### `NEVER` 激活 + +`ActionActivationType.NEVER` 会使得 Action 永远不会被激活 + +```python +class DisabledAction(BaseAction): + activation_type = ActionActivationType.NEVER # 永远不激活 + + async def execute(self) -> Tuple[bool, str]: + # 这个Action永远不会被执行 + return False, "这个Action被禁用" +``` + +#### `ALWAYS` 激活 + +`ActionActivationType.ALWAYS` 会使得 Action 永远会被激活，即一直在 Action 候选池中 + +这种激活方式常用于核心功能，如回复或不回复。 + +```python +class AlwaysActivatedAction(BaseAction): + activation_type = ActionActivationType.ALWAYS # 永远激活 + + async def execute(self) -> Tuple[bool, str]: + # 执行核心功能 + return True, "执行了核心功能" +``` + +#### `LLM_JUDGE` 激活 + +`ActionActivationType.LLM_JUDGE`会使得这个 Action 根据 LLM 的判断来决定是否加入候选池。 + +而 LLM 的判断是基于代码中预设的`llm_judge_prompt`和自动提供的聊天上下文进行的。 + +因此使用此种方法需要实现`llm_judge_prompt`属性。 + +```python +class LLMJudgedAction(BaseAction): + activation_type = ActionActivationType.LLM_JUDGE # 通过LLM判断激活 + # LLM判断提示词 + llm_judge_prompt = ( + "判定是否需要使用这个动作的条件：\n" + "1. 用户希望调用XXX这个动作\n" + "...\n" + "请回答\"是\"或\"否\"。\n" + ) + + async def execute(self) -> Tuple[bool, str]: + # 根据LLM判断是否执行 + return True, "执行了LLM判断功能" +``` + +#### `RANDOM` 激活 + +`ActionActivationType.RANDOM`会使得这个 Action 根据随机概率决定是否加入候选池。 + +概率则由代码中的`random_activation_probability`控制。在内部实现中我们使用了`random.random()`来生成一个0到1之间的随机数，并与这个概率进行比较。 + +因此使用这个方法需要实现`random_activation_probability`属性。 + +```python +class SurpriseAction(BaseAction): + activation_type = ActionActivationType.RANDOM # 基于随机概率激活 + # 随机激活概率 + random_activation_probability = 0.1 # 10%概率激活 + + async def execute(self) -> Tuple[bool, str]: + # 执行惊喜动作 + return True, "发送了惊喜内容" +``` + +#### `KEYWORD` 激活 + +`ActionActivationType.KEYWORD`会使得这个 Action 在检测到特定关键词时激活。 + +关键词由代码中的`activation_keywords`定义，而`keyword_case_sensitive`则控制关键词匹配时是否区分大小写。在内部实现中，我们使用了`in`操作符来检查消息内容是否包含这些关键词。 + +因此，使用此种方法需要实现`activation_keywords`和`keyword_case_sensitive`属性。 + +```python +class GreetingAction(BaseAction): + activation_type = ActionActivationType.KEYWORD # 关键词激活 + activation_keywords = ["你好", "hello", "hi", "嗨"] # 关键词配置 + keyword_case_sensitive = False # 不区分大小写 + + async def execute(self) -> Tuple[bool, str]: + # 执行问候逻辑 + return True, "发送了问候" +``` + +#### 第二层：使用决策 **在Action被激活后，使用条件决定麦麦什么时候会"选择"使用这个Action**。 @@ -49,17 +190,16 @@ Action采用**两层决策机制**来优化性能和决策质量： - `action_parameters`：所需参数，影响Action的可执行性 - 当前聊天上下文和麦麦的决策逻辑 -### 🎬 决策流程示例 +--- -假设有一个"发送表情"Action： +### 决策流程示例 ```python class EmojiAction(BaseAction): # 第一层：激活控制 - focus_activation_type = ActionActivationType.RANDOM # 专注模式下随机激活 - normal_activation_type = ActionActivationType.KEYWORD # 普通模式下关键词激活 - activation_keywords = ["表情", "emoji", "😊"] - + activation_type = ActionActivationType.RANDOM # 随机激活 + random_activation_probability = 0.1 # 10%概率激活 + # 第二层：使用决策 action_require = [ "表达情绪时可以选择使用", @@ -72,311 +212,85 @@ class EmojiAction(BaseAction): 1. **第一层激活判断**： - - 普通模式：只有当用户消息包含"表情"、"emoji"或"😊"时，麦麦才"知道"可以使用这个Action - - 专注模式：随机激活，有概率让麦麦"看到"这个Action + - 使用随机数进行决策，当`random.random() < self.random_activation_probability`时，麦麦才"知道"可以使用这个Action 2. **第二层使用决策**： - - 即使Action被激活，麦麦还会根据 `action_require`中的条件判断是否真正选择使用 + - 即使Action被激活，麦麦还会根据 `action_require` 中的条件判断是否真正选择使用 - 例如：如果刚刚已经发过表情，根据"不要连续发送多个表情"的要求，麦麦可能不会选择这个Action -## 📋 Action必须项清单 - -每个Action类都**必须**包含以下属性： - -### 1. 激活控制必须项 +--- +## Action 内置属性说明 ```python -# 专注模式下的激活类型 -focus_activation_type = ActionActivationType.LLM_JUDGE - -# 普通模式下的激活类型 -normal_activation_type = ActionActivationType.KEYWORD - -# 启用的聊天模式 -mode_enable = ChatMode.ALL - -# 是否允许与其他Action并行执行 -parallel_action = False -``` - -### 2. 基本信息必须项 - -```python -# Action的唯一标识名称 -action_name = "my_action" - -# Action的功能描述 -action_description = "描述这个Action的具体功能和用途" -``` - -### 3. 功能定义必须项 - -```python -# Action参数定义 - 告诉LLM执行时需要什么参数 -action_parameters = { - "param1": "参数1的说明", - "param2": "参数2的说明" -} - -# Action使用场景描述 - 帮助LLM判断何时"选择"使用 -action_require = [ - "使用场景描述1", - "使用场景描述2" -] - -# 关联的消息类型 - 说明Action能处理什么类型的内容 -associated_types = ["text", "emoji", "image"] -``` - -### 4. 执行方法必须项 - -```python -async def execute(self) -> Tuple[bool, str]: - """ - 执行Action的主要逻辑 - - Returns: - Tuple[bool, str]: (是否成功, 执行结果描述) - """ - # 执行动作的代码 - success = True - message = "动作执行成功" - - return success, message -``` - -## 🔧 激活类型详解 - -### KEYWORD激活 - -当检测到特定关键词时激活Action： - -```python -class GreetingAction(BaseAction): - focus_activation_type = ActionActivationType.KEYWORD - normal_activation_type = ActionActivationType.KEYWORD - - # 关键词配置 - activation_keywords = ["你好", "hello", "hi", "嗨"] - keyword_case_sensitive = False # 不区分大小写 - - async def execute(self) -> Tuple[bool, str]: - # 执行问候逻辑 - return True, "发送了问候" -``` - -### LLM_JUDGE激活 - -通过LLM智能判断是否激活： - -```python -class HelpAction(BaseAction): - focus_activation_type = ActionActivationType.LLM_JUDGE - normal_activation_type = ActionActivationType.LLM_JUDGE - - # LLM判断提示词 - llm_judge_prompt = """ - 判定是否需要使用帮助动作的条件： - 1. 用户表达了困惑或需要帮助 - 2. 用户提出了问题但没有得到满意答案 - 3. 对话中出现了技术术语或复杂概念 - - 请回答"是"或"否"。 - """ - - async def execute(self) -> Tuple[bool, str]: - # 执行帮助逻辑 - return True, "提供了帮助" -``` - -### RANDOM激活 - -基于随机概率激活： - -```python -class SurpriseAction(BaseAction): - focus_activation_type = ActionActivationType.RANDOM - normal_activation_type = ActionActivationType.RANDOM - - # 随机激活概率 - random_activation_probability = 0.1 # 10%概率激活 - - async def execute(self) -> Tuple[bool, str]: - # 执行惊喜动作 - return True, "发送了惊喜内容" -``` - -### ALWAYS激活 - -永远激活，常用于核心功能： - -```python -class CoreAction(BaseAction): - focus_activation_type = ActionActivationType.ALWAYS - normal_activation_type = ActionActivationType.ALWAYS - - async def execute(self) -> Tuple[bool, str]: - # 执行核心功能 - return True, "执行了核心功能" -``` - -### NEVER激活 - -从不激活，用于临时禁用： - -```python -class DisabledAction(BaseAction): - focus_activation_type = ActionActivationType.NEVER - normal_activation_type = ActionActivationType.NEVER - - async def execute(self) -> Tuple[bool, str]: - # 这个方法不会被调用 - return False, "已禁用" -``` - -## 📚 BaseAction内置属性和方法 - -### 内置属性 - -```python -class MyAction(BaseAction): +class BaseAction: def __init__(self): # 消息相关属性 - self.message # 当前消息对象 - self.chat_stream # 聊天流对象 - self.user_id # 用户ID - self.user_nickname # 用户昵称 - self.platform # 平台类型 (qq, telegram等) - self.chat_id # 聊天ID - self.is_group # 是否群聊 - - # Action相关属性 - self.action_data # Action执行时的数据 - self.thinking_id # 思考ID - self.matched_groups # 匹配到的组(如果有正则匹配) -``` + self.log_prefix: str # 日志前缀 + self.group_id: str # 群组ID + self.group_name: str # 群组名称 + self.user_id: str # 用户ID + self.user_nickname: str # 用户昵称 + self.platform: str # 平台类型 (qq, telegram等) + self.chat_id: str # 聊天ID + self.chat_stream: ChatStream # 聊天流对象 + self.is_group: bool # 是否群聊 -### 内置方法 + # 消息体 + self.action_message: dict # 消息数据 + + # Action相关属性 + self.action_data: dict # Action执行时的数据 + self.thinking_id: str # 思考ID +``` +action_message为一个字典，包含的键值对如下（省略了不必要的键值对） ```python -class MyAction(BaseAction): +{ + "message_id": "1234567890", # 消息id，str + "time": 1627545600.0, # 时间戳，float + "chat_id": "abcdef123456", # 聊天ID，str + "reply_to": None, # 回复消息id，str或None + "interest_value": 0.85, # 兴趣值，float + "is_mentioned": True, # 是否被提及，bool + "chat_info_last_active_time": 1627548600.0, # 最后活跃时间，float + "processed_plain_text": None, # 处理后的文本，str或None + "additional_config": None, # Adapter传来的additional_config，dict或None + "is_emoji": False, # 是否为表情，bool + "is_picid": False, # 是否为图片ID，bool + "is_command": False # 是否为命令，bool +} +``` + +部分值的格式请自行查询数据库。 + +--- + +## Action 内置方法说明 +```python +class BaseAction: # 配置相关 def get_config(self, key: str, default=None): - """获取配置值""" - pass + """获取插件配置值，使用嵌套键访问""" - # 消息发送相关 - async def send_text(self, text: str): + async def wait_for_new_message(self, timeout: int = 1200) -> Tuple[bool, str]: + """等待新消息或超时""" + + async def send_text(self, content: str, reply_to: str = "", reply_to_platform_id: str = "", typing: bool = False) -> bool: """发送文本消息""" - pass - - async def send_emoji(self, emoji_base64: str): + + async def send_emoji(self, emoji_base64: str) -> bool: """发送表情包""" - pass - - async def send_image(self, image_base64: str): + + async def send_image(self, image_base64: str) -> bool: """发送图片""" - pass - - # 动作记录相关 - async def store_action_info(self, **kwargs): - """记录动作信息""" - pass + + async def send_custom(self, message_type: str, content: str, typing: bool = False, reply_to: str = "") -> bool: + """发送自定义类型消息""" + + async def store_action_info(self, action_build_into_prompt: bool = False, action_prompt_display: str = "", action_done: bool = True) -> None: + """存储动作信息到数据库""" + + async def send_command(self, command_name: str, args: Optional[dict] = None, display_message: str = "", storage_message: bool = True) -> bool: + """发送命令消息""" ``` - -## 🎯 完整Action示例 - -```python -from src.plugin_system import BaseAction, ActionActivationType, ChatMode -from typing import Tuple - -class ExampleAction(BaseAction): - """示例Action - 展示完整的Action结构""" - - # === 激活控制 === - focus_activation_type = ActionActivationType.LLM_JUDGE - normal_activation_type = ActionActivationType.KEYWORD - mode_enable = ChatMode.ALL - parallel_action = False - - # 关键词激活配置 - activation_keywords = ["示例", "测试", "example"] - keyword_case_sensitive = False - - # LLM判断提示词 - llm_judge_prompt = "当用户需要示例或测试功能时激活" - - # 随机激活概率（如果使用RANDOM类型） - random_activation_probability = 0.2 - - # === 基本信息 === - action_name = "example_action" - action_description = "这是一个示例Action，用于演示Action的完整结构" - - # === 功能定义 === - action_parameters = { - "content": "要处理的内容", - "type": "处理类型", - "options": "可选配置" - } - - action_require = [ - "用户需要示例功能时使用", - "适合用于测试和演示", - "不要在正式对话中频繁使用" - ] - - associated_types = ["text", "emoji"] - - async def execute(self) -> Tuple[bool, str]: - """执行示例Action""" - try: - # 获取Action参数 - content = self.action_data.get("content", "默认内容") - action_type = self.action_data.get("type", "default") - - # 获取配置 - enable_feature = self.get_config("example.enable_advanced", False) - max_length = self.get_config("example.max_length", 100) - - # 执行具体逻辑 - if action_type == "greeting": - await self.send_text(f"你好！这是示例内容：{content}") - elif action_type == "info": - await self.send_text(f"信息：{content[:max_length]}") - else: - await self.send_text("执行了示例Action") - - # 记录动作信息 - await self.store_action_info( - action_build_into_prompt=True, - action_prompt_display=f"执行了示例动作：{action_type}", - action_done=True - ) - - return True, f"示例Action执行成功，类型：{action_type}" - - except Exception as e: - return False, f"执行失败：{str(e)}" -``` - -## 🎯 最佳实践 - -### 1. Action设计原则 - -- **单一职责**：每个Action只负责一个明确的功能 -- **智能激活**：合理选择激活类型，避免过度激活 -- **清晰描述**：提供准确的`action_require`帮助LLM决策 -- **错误处理**：妥善处理执行过程中的异常情况 - -### 2. 性能优化 - -- **激活控制**：使用合适的激活类型减少不必要的LLM调用 -- **并行执行**：谨慎设置`parallel_action`，避免冲突 -- **资源管理**：及时释放占用的资源 - -### 3. 调试技巧 - -- **日志记录**：在关键位置添加日志 -- **参数验证**：检查`action_data`的有效性 -- **配置测试**：测试不同配置下的行为 +具体参数与用法参见`BaseAction`基类的定义。 \ No newline at end of file diff --git a/src/plugin_system/base/base_action.py b/src/plugin_system/base/base_action.py index 7b9cef04c..c108c5d86 100644 --- a/src/plugin_system/base/base_action.py +++ b/src/plugin_system/base/base_action.py @@ -49,12 +49,10 @@ class BaseAction(ABC): reasoning: 执行该动作的理由 cycle_timers: 计时器字典 thinking_id: 思考ID - expressor: 表达器对象 - replyer: 回复器对象 chat_stream: 聊天流对象 log_prefix: 日志前缀 - shutting_down: 是否正在关闭 plugin_config: 插件配置字典 + action_message: 消息数据 **kwargs: 其他参数 """ if plugin_config is None: @@ -414,23 +412,11 @@ class BaseAction(ABC): """ return await self.execute() - # def get_action_context(self, key: str, default=None): - # """获取action上下文信息 - - # Args: - # key: 上下文键名 - # default: 默认值 - - # Returns: - # Any: 上下文值或默认值 - # """ - # return self.api.get_action_context(key, default) - def get_config(self, key: str, default=None): - """获取插件配置值，支持嵌套键访问 + """获取插件配置值，使用嵌套键访问 Args: - key: 配置键名，支持嵌套访问如 "section.subsection.key" + key: 配置键名，使用嵌套访问如 "section.subsection.key" default: 默认值 Returns: From d4fe32b904c61d86f10749c71bb29914b9d9c4ba Mon Sep 17 00:00:00 2001 From: UnCLAS-Prommer Date: Fri, 25 Jul 2025 00:08:16 +0800 Subject: [PATCH 2/2] =?UTF-8?q?=E5=88=A0=E9=99=A4=E5=B7=B2=E7=BB=8F?= =?UTF-8?q?=E5=BC=83=E7=94=A8=E7=9A=84=E6=8F=92=E4=BB=B6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- plugins/take_picture_plugin/_manifest.json | 50 -- .../take_picture_plugin/plugin(deprecated).py | 517 ------------------ 2 files changed, 567 deletions(-) delete mode 100644 plugins/take_picture_plugin/_manifest.json delete mode 100644 plugins/take_picture_plugin/plugin(deprecated).py diff --git a/plugins/take_picture_plugin/_manifest.json b/plugins/take_picture_plugin/_manifest.json deleted file mode 100644 index 0488d1de1..000000000 --- a/plugins/take_picture_plugin/_manifest.json +++ /dev/null @@ -1,50 +0,0 @@ -{ - "manifest_version": 1, - "name": "AI拍照插件 (Take Picture Plugin)", - "version": "1.0.0", - "description": "基于AI图像生成的拍照插件，可以生成逼真的自拍照片，支持照片存储和展示功能。", - "author": { - "name": "SengokuCola", - "url": "https://github.com/SengokuCola" - }, - "license": "GPL-v3.0-or-later", - - "host_application": { - "min_version": "0.9.0" - }, - "homepage_url": "https://github.com/MaiM-with-u/maibot", - "repository_url": "https://github.com/MaiM-with-u/maibot", - "keywords": ["camera", "photo", "selfie", "ai", "image", "generation"], - "categories": ["AI Tools", "Image Processing", "Entertainment"], - - "default_locale": "zh-CN", - "locales_path": "_locales", - - "plugin_info": { - "is_built_in": false, - "plugin_type": "image_generator", - "api_dependencies": ["volcengine"], - "components": [ - { - "type": "action", - "name": "take_picture", - "description": "生成一张用手机拍摄的照片，比如自拍或者近照", - "activation_modes": ["keyword"], - "keywords": ["拍张照", "自拍", "发张照片", "看看你", "你的照片"] - }, - { - "type": "command", - "name": "show_recent_pictures", - "description": "展示最近生成的5张照片", - "pattern": "/show_pics" - } - ], - "features": [ - "AI驱动的自拍照生成", - "个性化照片风格", - "照片历史记录", - "缓存机制优化", - "火山引擎API集成" - ] - } -} \ No newline at end of file diff --git a/plugins/take_picture_plugin/plugin(deprecated).py b/plugins/take_picture_plugin/plugin(deprecated).py deleted file mode 100644 index 24e86fece..000000000 --- a/plugins/take_picture_plugin/plugin(deprecated).py +++ /dev/null @@ -1,517 +0,0 @@ -""" -拍照插件 - -功能特性： -- Action: 生成一张自拍照，prompt由人设和模板生成 -- Command: 展示最近生成的照片 - -#此插件并不完善 -#此插件并不完善 - -#此插件并不完善 - -#此插件并不完善 - -#此插件并不完善 - -#此插件并不完善 - -#此插件并不完善 - - - -包含组件： -- 拍照Action - 生成自拍照 -- 展示照片Command - 展示最近生成的照片 -""" - -from typing import List, Tuple, Type, Optional -import random -import datetime -import json -import os -import asyncio -import urllib.request -import urllib.error -import base64 -import traceback - -from src.plugin_system.base.base_plugin import BasePlugin -from src.plugin_system.base.base_action import BaseAction -from src.plugin_system.base.base_command import BaseCommand -from src.plugin_system.base.component_types import ComponentInfo, ActionActivationType, ChatMode -from src.plugin_system.base.config_types import ConfigField -from src.plugin_system import register_plugin -from src.common.logger import get_logger - -logger = get_logger("take_picture_plugin") - -# 定义数据目录常量 -DATA_DIR = os.path.join("data", "take_picture_data") -# 确保数据目录存在 -os.makedirs(DATA_DIR, exist_ok=True) -# 创建全局锁 -file_lock = asyncio.Lock() - - -class TakePictureAction(BaseAction): - """生成一张自拍照""" - - focus_activation_type = ActionActivationType.KEYWORD - normal_activation_type = ActionActivationType.KEYWORD - mode_enable = ChatMode.ALL - parallel_action = False - - action_name = "take_picture" - action_description = "生成一张用手机拍摄，比如自拍或者近照" - activation_keywords = ["拍张照", "自拍", "发张照片", "看看你", "你的照片"] - keyword_case_sensitive = False - - action_parameters = {} - - action_require = ["当用户想看你的照片时使用", "当用户让你发自拍时使用当想随手拍眼前的场景时使用"] - - associated_types = ["text", "image"] - - # 内置的Prompt模板，如果配置文件中没有定义，将使用这些模板 - DEFAULT_PROMPT_TEMPLATES = [ - "极其频繁无奇的iPhone自拍照，没有明确的主体或构图感，就是随手一拍的快照照片略带运动模糊，阳光或室内打光不均匀导致的轻微曝光过度，整体呈现出一种刻意的平庸感，就像是从口袋里拿手机时不小心拍到的一张自拍。主角是{name}，{personality}" - ] - - # 简单的请求缓存，避免短时间内重复请求 - _request_cache = {} - - async def execute(self) -> Tuple[bool, Optional[str]]: - logger.info(f"{self.log_prefix} 执行拍照动作") - - try: - # 配置验证 - http_base_url = self.api.get_config("api.base_url") - http_api_key = self.api.get_config("api.volcano_generate_api_key") - - if not (http_base_url and http_api_key): - error_msg = "抱歉，照片生成功能所需的API配置（如API地址或密钥）不完整，无法提供服务。" - await self.send_text(error_msg) - logger.error(f"{self.log_prefix} HTTP调用配置缺失: base_url 或 volcano_generate_api_key.") - return False, "API配置不完整" - - # API密钥验证 - if http_api_key == "YOUR_DOUBAO_API_KEY_HERE": - error_msg = "照片生成功能尚未配置，请设置正确的API密钥。" - await self.send_text(error_msg) - logger.error(f"{self.log_prefix} API密钥未配置") - return False, "API密钥未配置" - - # 获取全局配置信息 - bot_nickname = self.api.get_global_config("bot.nickname", "麦麦") - bot_personality = self.api.get_global_config("personality.personality_core", "") - - personality_side = self.api.get_global_config("personality.personality_side", []) - if personality_side: - bot_personality += random.choice(personality_side) - - # 准备模板变量 - template_vars = {"name": bot_nickname, "personality": bot_personality} - - logger.info(f"{self.log_prefix} 使用的全局配置: name={bot_nickname}, personality={bot_personality}") - - # 尝试从配置文件获取模板，如果没有则使用默认模板 - templates = self.api.get_config("picture.prompt_templates", self.DEFAULT_PROMPT_TEMPLATES) - if not templates: - logger.warning(f"{self.log_prefix} 未找到有效的提示词模板，使用默认模板") - templates = self.DEFAULT_PROMPT_TEMPLATES - - prompt_template = random.choice(templates) - - # 填充模板 - final_prompt = prompt_template.format(**template_vars) - - logger.info(f"{self.log_prefix} 生成的最终Prompt: {final_prompt}") - - # 从配置获取参数 - model = self.api.get_config("picture.default_model", "doubao-seedream-3-0-t2i-250415") - size = self.api.get_config("picture.default_size", "1024x1024") - watermark = self.api.get_config("picture.default_watermark", True) - guidance_scale = self.api.get_config("picture.default_guidance_scale", 2.5) - seed = self.api.get_config("picture.default_seed", 42) - - # 检查缓存 - enable_cache = self.api.get_config("storage.enable_cache", True) - if enable_cache: - cache_key = self._get_cache_key(final_prompt, model, size) - if cache_key in self._request_cache: - cached_result = self._request_cache[cache_key] - logger.info(f"{self.log_prefix} 使用缓存的图片结果") - await self.send_text("我之前拍过类似的照片，用之前的结果~") - - # 直接发送缓存的结果 - send_success = await self._send_image(cached_result) - if send_success: - await self.send_text("这是我的照片，好看吗？") - return True, "照片已发送(缓存)" - else: - # 缓存失败，清除这个缓存项并继续正常流程 - del self._request_cache[cache_key] - - await self.send_text("正在为你拍照，请稍候...") - - try: - seed = random.randint(1, 1000000) - success, result = await asyncio.to_thread( - self._make_http_image_request, - prompt=final_prompt, - model=model, - size=size, - seed=seed, - guidance_scale=guidance_scale, - watermark=watermark, - ) - except Exception as e: - logger.error(f"{self.log_prefix} (HTTP) 异步请求执行失败: {e!r}", exc_info=True) - traceback.print_exc() - success = False - result = f"照片生成服务遇到意外问题: {str(e)[:100]}" - - if success: - image_url = result - logger.info(f"{self.log_prefix} 图片URL获取成功: {image_url[:70]}... 下载并编码.") - - try: - encode_success, encode_result = await asyncio.to_thread(self._download_and_encode_base64, image_url) - except Exception as e: - logger.error(f"{self.log_prefix} (B64) 异步下载/编码失败: {e!r}", exc_info=True) - traceback.print_exc() - encode_success = False - encode_result = f"图片下载或编码时发生内部错误: {str(e)[:100]}" - - if encode_success: - base64_image_string = encode_result - # 更新缓存 - if enable_cache: - self._update_cache(final_prompt, model, size, base64_image_string) - - # 发送图片 - send_success = await self._send_image(base64_image_string) - if send_success: - # 存储到文件 - await self._store_picture_info(final_prompt, image_url) - logger.info(f"{self.log_prefix} 成功生成并存储照片: {image_url}") - await self.send_text("当当当当~这是我刚拍的照片，好看吗？") - return True, f"成功生成照片: {image_url}" - else: - await self.send_text("照片生成了，但发送失败了，可能是格式问题...") - return False, "照片发送失败" - else: - await self.send_text(f"照片下载失败: {encode_result}") - return False, encode_result - else: - await self.send_text(f"哎呀，拍照失败了: {result}") - return False, result - - except Exception as e: - logger.error(f"{self.log_prefix} 执行拍照动作失败: {e}", exc_info=True) - traceback.print_exc() - await self.send_text("呜呜，拍照的时候出了一点小问题...") - return False, str(e) - - async def _store_picture_info(self, prompt: str, image_url: str): - """将照片信息存入日志文件""" - log_file = self.api.get_config("storage.log_file", "picture_log.json") - log_path = os.path.join(DATA_DIR, log_file) - max_photos = self.api.get_config("storage.max_photos", 50) - - async with file_lock: - try: - if os.path.exists(log_path): - with open(log_path, "r", encoding="utf-8") as f: - log_data = json.load(f) - else: - log_data = [] - except (json.JSONDecodeError, FileNotFoundError): - log_data = [] - - # 添加新照片 - log_data.append( - {"prompt": prompt, "image_url": image_url, "timestamp": datetime.datetime.now().isoformat()} - ) - - # 如果超过最大数量，删除最旧的 - if len(log_data) > max_photos: - log_data = sorted(log_data, key=lambda x: x.get("timestamp", ""), reverse=True)[:max_photos] - - try: - with open(log_path, "w", encoding="utf-8") as f: - json.dump(log_data, f, ensure_ascii=False, indent=4) - except Exception as e: - logger.error(f"{self.log_prefix} 写入照片日志文件失败: {e}", exc_info=True) - - def _make_http_image_request( - self, prompt: str, model: str, size: str, seed: int, guidance_scale: float, watermark: bool - ) -> Tuple[bool, str]: - """发送HTTP请求到火山引擎豆包API生成图片""" - try: - base_url = self.api.get_config("api.base_url") - api_key = self.api.get_config("api.volcano_generate_api_key") - - # 构建请求URL和头部 - endpoint = f"{base_url.rstrip('/')}/images/generations" - headers = { - "Content-Type": "application/json", - "Authorization": f"Bearer {api_key}", - } - - # 构建请求体 - request_body = { - "model": model, - "prompt": prompt, - "response_format": "url", - "size": size, - "seed": seed, - "guidance_scale": guidance_scale, - "watermark": watermark, - "api-key": api_key, - } - - # 创建请求对象 - req = urllib.request.Request( - endpoint, - data=json.dumps(request_body).encode("utf-8"), - headers=headers, - method="POST", - ) - - # 发送请求并获取响应 - with urllib.request.urlopen(req, timeout=60) as response: - response_data = json.loads(response.read().decode("utf-8")) - - # 解析响应 - image_url = None - if ( - isinstance(response_data.get("data"), list) - and response_data["data"] - and isinstance(response_data["data"][0], dict) - ): - image_url = response_data["data"][0].get("url") - elif response_data.get("url"): - image_url = response_data.get("url") - - if image_url: - return True, image_url - else: - error_msg = response_data.get("error", {}).get("message", "未知错误") - logger.error(f"API返回错误: {error_msg}") - return False, f"API错误: {error_msg}" - - except urllib.error.HTTPError as e: - error_body = e.read().decode("utf-8") - logger.error(f"HTTP错误 {e.code}: {error_body}") - return False, f"HTTP错误 {e.code}: {error_body[:100]}..." - except Exception as e: - logger.error(f"请求异常: {e}", exc_info=True) - return False, f"请求异常: {str(e)}" - - def _download_and_encode_base64(self, image_url: str) -> Tuple[bool, str]: - """下载图片并转换为Base64编码""" - try: - with urllib.request.urlopen(image_url) as response: - image_data = response.read() - - base64_encoded = base64.b64encode(image_data).decode("utf-8") - return True, base64_encoded - except Exception as e: - logger.error(f"图片下载编码失败: {e}", exc_info=True) - return False, str(e) - - async def _send_image(self, base64_image: str) -> bool: - """发送图片""" - try: - # 使用聊天流信息确定发送目标 - chat_stream = self.api.get_service("chat_stream") - if not chat_stream: - logger.error(f"{self.log_prefix} 没有可用的聊天流发送图片") - return False - - if chat_stream.group_info: - # 群聊 - return await self.api.send_message_to_target( - message_type="image", - content=base64_image, - platform=chat_stream.platform, - target_id=str(chat_stream.group_info.group_id), - is_group=True, - display_message="发送生成的照片", - ) - else: - # 私聊 - return await self.api.send_message_to_target( - message_type="image", - content=base64_image, - platform=chat_stream.platform, - target_id=str(chat_stream.user_info.user_id), - is_group=False, - display_message="发送生成的照片", - ) - except Exception as e: - logger.error(f"{self.log_prefix} 发送图片时出错: {e}") - return False - - @classmethod - def _get_cache_key(cls, description: str, model: str, size: str) -> str: - """生成缓存键""" - return f"{description}|{model}|{size}" - - def _update_cache(self, description: str, model: str, size: str, base64_image: str): - """更新缓存""" - max_cache_size = self.api.get_config("storage.max_cache_size", 10) - cache_key = self._get_cache_key(description, model, size) - - # 添加到缓存 - self._request_cache[cache_key] = base64_image - - # 如果缓存超过最大大小，删除最旧的项 - if len(self._request_cache) > max_cache_size: - oldest_key = next(iter(self._request_cache)) - del self._request_cache[oldest_key] - - -class ShowRecentPicturesCommand(BaseCommand): - """展示最近生成的照片""" - - command_name = "show_recent_pictures" - command_description = "展示最近生成的5张照片" - command_pattern = r"^/show_pics$" - command_help = "用法: /show_pics" - command_examples = ["/show_pics"] - intercept_message = True - - async def execute(self) -> Tuple[bool, Optional[str]]: - logger.info(f"{self.log_prefix} 执行展示最近照片命令") - log_file = self.api.get_config("storage.log_file", "picture_log.json") - log_path = os.path.join(DATA_DIR, log_file) - - async with file_lock: - try: - if not os.path.exists(log_path): - await self.send_text("最近还没有拍过照片哦，快让我自拍一张吧！") - return True, "没有照片日志文件" - - with open(log_path, "r", encoding="utf-8") as f: - log_data = json.load(f) - - if not log_data: - await self.send_text("最近还没有拍过照片哦，快让我自拍一张吧！") - return True, "没有照片" - - # 获取最新的5张照片 - recent_pics = sorted(log_data, key=lambda x: x["timestamp"], reverse=True)[:5] - - # 先发送文本消息 - await self.send_text("这是我最近拍的几张照片~") - - # 逐个发送图片 - for pic in recent_pics: - # 尝试获取图片URL - image_url = pic.get("image_url") - if image_url: - try: - # 下载图片并转换为Base64 - with urllib.request.urlopen(image_url) as response: - image_data = response.read() - base64_encoded = base64.b64encode(image_data).decode("utf-8") - - # 发送图片 - await self.send_type( - message_type="image", content=base64_encoded, display_message="发送最近的照片" - ) - except Exception as e: - logger.error(f"{self.log_prefix} 下载或发送照片失败: {e}", exc_info=True) - - return True, "成功展示最近的照片" - - except json.JSONDecodeError: - await self.send_text("照片记录文件好像损坏了...") - return False, "JSON解码错误" - except Exception as e: - logger.error(f"{self.log_prefix} 展示照片失败: {e}", exc_info=True) - await self.send_text("哎呀，查找照片的时候出错了。") - return False, str(e) - - -@register_plugin -class TakePicturePlugin(BasePlugin): - """拍照插件""" - - plugin_name = "take_picture_plugin" # 内部标识符 - enable_plugin = False - dependencies = [] # 插件依赖列表 - python_dependencies = [] # Python包依赖列表 - config_file_name = "config.toml" - - # 配置节描述 - config_section_descriptions = { - "plugin": "插件基本信息配置", - "api": "API相关配置，包含火山引擎API的访问信息", - "components": "组件启用控制", - "picture": "拍照功能核心配置", - "storage": "照片存储相关配置", - } - - # 配置Schema定义 - config_schema = { - "plugin": { - "enabled": ConfigField(type=bool, default=False, description="是否启用插件"), - }, - "api": { - "base_url": ConfigField( - type=str, - default="https://ark.cn-beijing.volces.com/api/v3", - description="API基础URL", - example="https://api.example.com/v1", - ), - "volcano_generate_api_key": ConfigField( - type=str, default="YOUR_DOUBAO_API_KEY_HERE", description="火山引擎豆包API密钥", required=True - ), - }, - "components": { - "enable_take_picture_action": ConfigField(type=bool, default=True, description="是否启用拍照Action"), - "enable_show_pics_command": ConfigField(type=bool, default=True, description="是否启用展示照片Command"), - }, - "picture": { - "default_model": ConfigField( - type=str, - default="doubao-seedream-3-0-t2i-250415", - description="默认使用的文生图模型", - choices=["doubao-seedream-3-0-t2i-250415", "doubao-seedream-2-0-t2i"], - ), - "default_size": ConfigField( - type=str, - default="1024x1024", - description="默认图片尺寸", - example="1024x1024", - choices=["1024x1024", "1024x1280", "1280x1024", "1024x1536", "1536x1024"], - ), - "default_watermark": ConfigField(type=bool, default=True, description="是否默认添加水印"), - "default_guidance_scale": ConfigField( - type=float, default=2.5, description="模型指导强度，影响图片与提示的关联性", example="2.0" - ), - "default_seed": ConfigField(type=int, default=42, description="随机种子，用于复现图片"), - "prompt_templates": ConfigField( - type=list, default=TakePictureAction.DEFAULT_PROMPT_TEMPLATES, description="用于生成自拍照的prompt模板" - ), - }, - "storage": { - "max_photos": ConfigField(type=int, default=50, description="最大保存的照片数量"), - "log_file": ConfigField(type=str, default="picture_log.json", description="照片日志文件名"), - "enable_cache": ConfigField(type=bool, default=True, description="是否启用请求缓存"), - "max_cache_size": ConfigField(type=int, default=10, description="最大缓存数量"), - }, - } - - def get_plugin_components(self) -> List[Tuple[ComponentInfo, Type]]: - """返回插件包含的组件列表""" - components = [] - if self.get_config("components.enable_take_picture_action", True): - components.append((TakePictureAction.get_action_info(), TakePictureAction)) - if self.get_config("components.enable_show_pics_command", True): - components.append((ShowRecentPicturesCommand.get_command_info(), ShowRecentPicturesCommand)) - return components