Merge branch 'dev' of https://github.com/MoFox-Studio/MoFox_Bot into dev
This commit is contained in:
雅诺狐
@@ -1,190 +0,0 @@
|
||||
import re
|
||||
from typing import List, Tuple, Type
|
||||
|
||||
from src.plugin_system import (
|
||||
BasePlugin,
|
||||
register_plugin,
|
||||
BaseAction,
|
||||
ComponentInfo,
|
||||
ActionActivationType,
|
||||
ConfigField,
|
||||
)
|
||||
from src.common.logger import get_logger
|
||||
from .qq_emoji_list import qq_face
|
||||
from src.plugin_system.base.component_types import ChatType
|
||||
|
||||
logger = get_logger("set_emoji_like_plugin")
|
||||
|
||||
|
||||
def get_emoji_id(emoji_input: str) -> str | None:
|
||||
"""根据输入获取表情ID"""
|
||||
# 如果输入本身就是数字ID，直接返回
|
||||
if emoji_input.isdigit() or (isinstance(emoji_input, str) and emoji_input.startswith("😊")):
|
||||
if emoji_input in qq_face:
|
||||
return emoji_input
|
||||
|
||||
# 尝试从 "[表情：xxx]" 格式中提取
|
||||
match = re.search(r"\[表情：(.+?)\]", emoji_input)
|
||||
if match:
|
||||
emoji_name = match.group(1).strip()
|
||||
else:
|
||||
emoji_name = emoji_input.strip()
|
||||
|
||||
# 遍历查找
|
||||
for key, value in qq_face.items():
|
||||
# value 的格式是 "[表情：xxx]"
|
||||
if f"[表情：{emoji_name}]" == value:
|
||||
return key
|
||||
|
||||
return None
|
||||
|
||||
|
||||
# ===== Action组件 =====
|
||||
class SetEmojiLikeAction(BaseAction):
|
||||
"""设置消息表情回应"""
|
||||
|
||||
# === 基本信息（必须填写）===
|
||||
action_name = "set_emoji_like"
|
||||
action_description = "为某条已经存在的消息添加‘贴表情’回应（类似点赞），而不是发送新消息。可以在觉得某条消息非常有趣、值得赞同或者需要特殊情感回应时主动使用。"
|
||||
activation_type = ActionActivationType.ALWAYS # 消息接收时激活(?)
|
||||
chat_type_allow = ChatType.GROUP
|
||||
parallel_action = True
|
||||
|
||||
# === 功能描述（必须填写）===
|
||||
# 从 qq_face 字典中提取所有表情名称用于提示
|
||||
emoji_options = []
|
||||
for name in qq_face.values():
|
||||
match = re.search(r"\[表情：(.+?)\]", name)
|
||||
if match:
|
||||
emoji_options.append(match.group(1))
|
||||
|
||||
action_parameters = {
|
||||
"emoji": f"要回应的表情,必须从以下表情中选择: {', '.join(emoji_options)}",
|
||||
"set": "是否设置回应 (True/False)",
|
||||
}
|
||||
action_require = [
|
||||
"当需要对一个已存在消息进行‘贴表情’回应时使用",
|
||||
"这是一个对旧消息的操作，而不是发送新消息",
|
||||
"如果你想发送一个新的表情包消息，请使用 'emoji' 动作",
|
||||
]
|
||||
llm_judge_prompt = """
|
||||
判定是否需要使用贴表情动作的条件：
|
||||
1. 用户明确要求使用贴表情包
|
||||
2. 这是一个适合表达强烈情绪的场合
|
||||
3. 不要发送太多表情包，如果你已经发送过多个表情包则回答"否"
|
||||
|
||||
请回答"是"或"否"。
|
||||
"""
|
||||
associated_types = ["text"]
|
||||
|
||||
async def execute(self) -> Tuple[bool, str]:
|
||||
"""执行设置表情回应的动作"""
|
||||
message_id = None
|
||||
if self.has_action_message:
|
||||
logger.debug(str(self.action_message))
|
||||
if isinstance(self.action_message, dict):
|
||||
message_id = self.action_message.get("message_id")
|
||||
logger.info(f"获取到的消息ID: {message_id}")
|
||||
else:
|
||||
logger.error("未提供消息ID")
|
||||
await self.store_action_info(
|
||||
action_build_into_prompt=True,
|
||||
action_prompt_display=f"执行了set_emoji_like动作：{self.action_name},失败: 未提供消息ID",
|
||||
action_done=False,
|
||||
)
|
||||
return False, "未提供消息ID"
|
||||
|
||||
emoji_input = self.action_data.get("emoji")
|
||||
set_like = self.action_data.get("set", True)
|
||||
|
||||
if not emoji_input:
|
||||
logger.error("未提供表情")
|
||||
return False, "未提供表情"
|
||||
logger.info(f"设置表情回应: {emoji_input}, 是否设置: {set_like}")
|
||||
|
||||
emoji_id = get_emoji_id(emoji_input)
|
||||
if not emoji_id:
|
||||
logger.error(f"找不到表情: '{emoji_input}'。请从可用列表中选择。")
|
||||
await self.store_action_info(
|
||||
action_build_into_prompt=True,
|
||||
action_prompt_display=f"执行了set_emoji_like动作：{self.action_name},失败: 找不到表情: '{emoji_input}'",
|
||||
action_done=False,
|
||||
)
|
||||
return False, f"找不到表情: '{emoji_input}'。请从可用列表中选择。"
|
||||
|
||||
# 4. 使用适配器API发送命令
|
||||
if not message_id:
|
||||
logger.error("未提供消息ID")
|
||||
await self.store_action_info(
|
||||
action_build_into_prompt=True,
|
||||
action_prompt_display=f"执行了set_emoji_like动作：{self.action_name},失败: 未提供消息ID",
|
||||
action_done=False,
|
||||
)
|
||||
return False, "未提供消息ID"
|
||||
|
||||
try:
|
||||
# 使用适配器API发送贴表情命令
|
||||
success = await self.send_command(
|
||||
command_name="set_emoji_like", args={"message_id": message_id, "emoji_id": emoji_id, "set": set_like}, storage_message=False
|
||||
)
|
||||
if success:
|
||||
logger.info("设置表情回应成功")
|
||||
await self.store_action_info(
|
||||
action_build_into_prompt=True,
|
||||
action_prompt_display=f"执行了set_emoji_like动作,{emoji_input},设置表情回应: {emoji_id}, 是否设置: {set_like}",
|
||||
action_done=True,
|
||||
)
|
||||
return True, "成功设置表情回应"
|
||||
else:
|
||||
logger.error("设置表情回应失败")
|
||||
await self.store_action_info(
|
||||
action_build_into_prompt=True,
|
||||
action_prompt_display=f"执行了set_emoji_like动作：{self.action_name},失败",
|
||||
action_done=False,
|
||||
)
|
||||
return False, "设置表情回应失败"
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"设置表情回应失败: {e}")
|
||||
await self.store_action_info(
|
||||
action_build_into_prompt=True,
|
||||
action_prompt_display=f"执行了set_emoji_like动作：{self.action_name},失败: {e}",
|
||||
action_done=False,
|
||||
)
|
||||
return False, f"设置表情回应失败: {e}"
|
||||
|
||||
|
||||
# ===== 插件注册 =====
|
||||
@register_plugin
|
||||
class SetEmojiLikePlugin(BasePlugin):
|
||||
"""设置消息表情回应插件"""
|
||||
|
||||
# 插件基本信息
|
||||
plugin_name: str = "set_emoji_like" # 内部标识符
|
||||
enable_plugin: bool = True
|
||||
dependencies: List[str] = [] # 插件依赖列表
|
||||
python_dependencies: List[str] = [] # Python包依赖列表，现在使用内置API
|
||||
config_file_name: str = "config.toml" # 配置文件名
|
||||
|
||||
# 配置节描述
|
||||
config_section_descriptions = {"plugin": "插件基本信息", "components": "插件组件"}
|
||||
|
||||
# 配置Schema定义
|
||||
config_schema: dict = {
|
||||
"plugin": {
|
||||
"name": ConfigField(type=str, default="set_emoji_like", description="插件名称"),
|
||||
"version": ConfigField(type=str, default="1.0.0", description="插件版本"),
|
||||
"enabled": ConfigField(type=bool, default=True, description="是否启用插件"),
|
||||
"config_version": ConfigField(type=str, default="1.1", description="配置版本"),
|
||||
},
|
||||
"components": {
|
||||
"action_set_emoji_like": ConfigField(type=bool, default=True, description="是否启用设置表情回应功能"),
|
||||
},
|
||||
}
|
||||
|
||||
def get_plugin_components(self) -> List[Tuple[ComponentInfo, Type]]:
|
||||
if self.get_config("components.action_set_emoji_like"):
|
||||
return [
|
||||
(SetEmojiLikeAction.get_action_info(), SetEmojiLikeAction),
|
||||
]
|
||||
return []
|
||||
@@ -215,6 +215,7 @@ MoFox_Bot(第三方修改版)
|
||||
from src.plugin_system.apis.permission_api import permission_api
|
||||
|
||||
permission_manager = PermissionManager()
|
||||
await permission_manager.initialize()
|
||||
permission_api.set_permission_manager(permission_manager)
|
||||
logger.info("权限管理器初始化成功")
|
||||
|
||||
@@ -327,7 +328,6 @@ MoFox_Bot(第三方修改版)
|
||||
]
|
||||
|
||||
# 增强记忆系统不需要定时任务，已禁用原有记忆系统的定时任务
|
||||
logger.info("原有记忆系统定时任务已禁用 - 使用增强记忆系统")
|
||||
|
||||
await asyncio.gather(*tasks)
|
||||
|
||||
|
||||
@@ -432,23 +432,32 @@ class BaseAction(ABC):
|
||||
logger.warning(f"{log_prefix} 未找到Action组件信息: {action_name}")
|
||||
return False, f"未找到Action组件信息: {action_name}"
|
||||
|
||||
plugin_config = component_registry.get_plugin_config(component_info.plugin_name)
|
||||
# 确保获取的是Action组件
|
||||
if component_info.component_type != ComponentType.ACTION:
|
||||
logger.error(f"{log_prefix} 尝试调用的组件 '{action_name}' 不是一个Action，而是一个 '{component_info.component_type.value}'")
|
||||
return False, f"组件 '{action_name}' 不是一个有效的Action"
|
||||
|
||||
plugin_config = component_registry.get_plugin_config(component_info.plugin_name)
|
||||
# 3. 实例化被调用的Action
|
||||
action_instance = action_class(
|
||||
action_data=called_action_data,
|
||||
reasoning=f"Called by {self.action_name}",
|
||||
cycle_timers=self.cycle_timers,
|
||||
thinking_id=self.thinking_id,
|
||||
chat_stream=self.chat_stream,
|
||||
log_prefix=log_prefix,
|
||||
plugin_config=plugin_config,
|
||||
action_message=self.action_message,
|
||||
)
|
||||
action_params = {
|
||||
"action_data": called_action_data,
|
||||
"reasoning": f"Called by {self.action_name}",
|
||||
"cycle_timers": self.cycle_timers,
|
||||
"thinking_id": self.thinking_id,
|
||||
"chat_stream": self.chat_stream,
|
||||
"log_prefix": log_prefix,
|
||||
"plugin_config": plugin_config,
|
||||
"action_message": self.action_message,
|
||||
}
|
||||
action_instance = action_class(**action_params)
|
||||
|
||||
# 4. 执行Action
|
||||
logger.debug(f"{log_prefix} 开始执行...")
|
||||
result = await action_instance.execute()
|
||||
execute_result = await action_instance.execute()
|
||||
# 确保返回类型符合 (bool, str) 格式
|
||||
is_success = execute_result[0] if isinstance(execute_result, tuple) and len(execute_result) > 0 else False
|
||||
message = execute_result[1] if isinstance(execute_result, tuple) and len(execute_result) > 1 else ""
|
||||
result = (is_success, str(message))
|
||||
logger.info(f"{log_prefix} 执行完成，结果: {result}")
|
||||
return result
|
||||
|
||||
|
||||
@@ -644,7 +644,7 @@ class ChatterPlanFilter:
|
||||
# 为参数描述添加一个通用示例值
|
||||
if action_name == "set_emoji_like" and p_name == "emoji":
|
||||
# 特殊处理set_emoji_like的emoji参数
|
||||
from plugins.set_emoji_like.qq_emoji_list import qq_face
|
||||
from src.plugins.built_in.social_toolkit_plugin.qq_emoji_list import qq_face
|
||||
emoji_options = [re.search(r"\[表情：(.+?)\]", name).group(1) for name in qq_face.values() if re.search(r"\[表情：(.+?)\]", name)]
|
||||
example_value = f"<从'{', '.join(emoji_options[:10])}...'中选择一个>"
|
||||
else:
|
||||
|
||||
@@ -1,11 +0,0 @@
|
||||
{
|
||||
"manifest_version": 1,
|
||||
"name": "At User Plugin",
|
||||
"description": "一个通过名字艾特用户的插件",
|
||||
"author": "Kilo Code",
|
||||
"version": "1.0.0",
|
||||
"requirements": [],
|
||||
"license": "MIT",
|
||||
"keywords": ["at", "mention", "user"],
|
||||
"categories": ["Chat", "Utility"]
|
||||
}
|
||||
@@ -1,173 +0,0 @@
|
||||
from typing import List, Tuple, Type
|
||||
from src.plugin_system import (
|
||||
BasePlugin,
|
||||
BaseCommand,
|
||||
CommandInfo,
|
||||
register_plugin,
|
||||
BaseAction,
|
||||
ActionInfo,
|
||||
ActionActivationType,
|
||||
)
|
||||
from src.person_info.person_info import get_person_info_manager
|
||||
from src.common.logger import get_logger
|
||||
from src.plugin_system.base.component_types import ChatType
|
||||
|
||||
logger = get_logger(__name__)
|
||||
|
||||
|
||||
class AtAction(BaseAction):
|
||||
"""发送艾特消息"""
|
||||
|
||||
# === 基本信息（必须填写）===
|
||||
action_name = "at_user"
|
||||
action_description = "发送艾特消息"
|
||||
activation_type = ActionActivationType.LLM_JUDGE # 消息接收时激活(?)
|
||||
parallel_action = False
|
||||
chat_type_allow = ChatType.GROUP
|
||||
|
||||
# === 功能描述（必须填写）===
|
||||
action_parameters = {"user_name": "需要艾特用户的名字", "at_message": "艾特用户时要发送的消息"}
|
||||
action_require = [
|
||||
"当用户明确要求你去'叫'、'喊'、'提醒'或'艾特'某人时使用",
|
||||
"当你判断，为了让特定的人看到消息，需要代表用户去呼叫他/她时使用",
|
||||
"例如：'你去叫一下张三'，'提醒一下李四开会'",
|
||||
]
|
||||
llm_judge_prompt = """
|
||||
判定是否需要使用艾特用户动作的条件：
|
||||
1. 你在对话中提到了某个具体的人，并且需要提醒他/她。
|
||||
3. 上下文明确需要你艾特一个或多个人。
|
||||
|
||||
请回答"是"或"否"。
|
||||
"""
|
||||
associated_types = ["text"]
|
||||
|
||||
async def execute(self) -> Tuple[bool, str]:
|
||||
"""执行艾特用户的动作"""
|
||||
user_name = self.action_data.get("user_name")
|
||||
at_message = self.action_data.get("at_message")
|
||||
|
||||
if not user_name or not at_message:
|
||||
logger.warning("艾特用户的动作缺少必要参数。")
|
||||
await self.store_action_info(
|
||||
action_build_into_prompt=True,
|
||||
action_prompt_display=f"执行了艾特用户动作：艾特用户 {user_name} 并发送消息: {at_message},失败了,因为没有提供必要参数",
|
||||
action_done=False,
|
||||
)
|
||||
return False, "缺少必要参数"
|
||||
|
||||
user_info = await get_person_info_manager().get_person_info_by_name(user_name)
|
||||
if not user_info or not user_info.get("user_id"):
|
||||
logger.info(f"找不到名为 '{user_name}' 的用户。")
|
||||
return False, "用户不存在"
|
||||
|
||||
try:
|
||||
# 使用回复器生成艾特回复，而不是直接发送命令
|
||||
from src.chat.replyer.default_generator import DefaultReplyer
|
||||
from src.chat.message_receive.chat_stream import get_chat_manager
|
||||
|
||||
# 获取当前聊天流
|
||||
chat_manager = get_chat_manager()
|
||||
chat_stream = self.chat_stream or chat_manager.get_stream(self.chat_id)
|
||||
|
||||
if not chat_stream:
|
||||
logger.error(f"找不到聊天流: {self.chat_stream}")
|
||||
return False, "聊天流不存在"
|
||||
|
||||
# 创建回复器实例
|
||||
replyer = DefaultReplyer(chat_stream)
|
||||
|
||||
# 构建回复对象，将艾特消息作为回复目标
|
||||
reply_to = f"{user_name}:{at_message}"
|
||||
extra_info = f"你需要艾特用户 {user_name} 并回复他们说: {at_message}"
|
||||
|
||||
# 使用回复器生成回复
|
||||
success, llm_response, prompt = await replyer.generate_reply_with_context(
|
||||
reply_to=reply_to,
|
||||
extra_info=extra_info,
|
||||
enable_tool=False, # 艾特回复通常不需要工具调用
|
||||
from_plugin=False,
|
||||
)
|
||||
|
||||
if success and llm_response:
|
||||
# 获取生成的回复内容
|
||||
reply_content = llm_response.get("content", "")
|
||||
if reply_content:
|
||||
# 获取用户QQ号，发送真正的艾特消息
|
||||
user_id = user_info.get("user_id")
|
||||
|
||||
# 发送真正的艾特命令，使用回复器生成的智能内容
|
||||
await self.send_command(
|
||||
"SEND_AT_MESSAGE",
|
||||
args={"qq_id": user_id, "text": reply_content},
|
||||
display_message=f"艾特用户 {user_name} 并发送智能回复: {reply_content}",
|
||||
)
|
||||
|
||||
await self.store_action_info(
|
||||
action_build_into_prompt=True,
|
||||
action_prompt_display=f"执行了艾特用户动作：艾特用户 {user_name} 并发送智能回复: {reply_content}",
|
||||
action_done=True,
|
||||
)
|
||||
|
||||
logger.info(f"成功通过回复器生成智能内容并发送真正的艾特消息给 {user_name}: {reply_content}")
|
||||
return True, "智能艾特消息发送成功"
|
||||
else:
|
||||
logger.warning("回复器生成了空内容")
|
||||
return False, "回复内容为空"
|
||||
else:
|
||||
logger.error("回复器生成回复失败")
|
||||
return False, "回复生成失败"
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"执行艾特用户动作时发生异常: {e}", exc_info=True)
|
||||
await self.store_action_info(
|
||||
action_build_into_prompt=True,
|
||||
action_prompt_display=f"执行艾特用户动作失败：{str(e)}",
|
||||
action_done=False,
|
||||
)
|
||||
return False, f"执行失败: {str(e)}"
|
||||
|
||||
|
||||
class AtCommand(BaseCommand):
|
||||
command_name: str = "at_user"
|
||||
description: str = "通过名字艾特用户"
|
||||
command_pattern: str = r"/at\s+@?(?P<name>[\S]+)(?:\s+(?P<text>.*))?"
|
||||
|
||||
async def execute(self) -> Tuple[bool, str, bool]:
|
||||
name = self.matched_groups.get("name")
|
||||
text = self.matched_groups.get("text", "")
|
||||
|
||||
if not name:
|
||||
await self.send_text("请指定要艾特的用户名称。")
|
||||
return False, "缺少用户名称", True
|
||||
|
||||
person_info_manager = get_person_info_manager()
|
||||
user_info = await person_info_manager.get_person_info_by_name(name)
|
||||
|
||||
if not user_info or not user_info.get("user_id"):
|
||||
await self.send_text(f"找不到名为 '{name}' 的用户。")
|
||||
return False, "用户不存在", True
|
||||
|
||||
user_id = user_info.get("user_id")
|
||||
|
||||
await self.send_command(
|
||||
"SEND_AT_MESSAGE",
|
||||
args={"qq_id": user_id, "text": text},
|
||||
display_message=f"艾特用户 {name} 并发送消息: {text}",
|
||||
)
|
||||
|
||||
return True, "艾特消息已发送", True
|
||||
|
||||
|
||||
@register_plugin
|
||||
class AtUserPlugin(BasePlugin):
|
||||
plugin_name: str = "at_user_plugin"
|
||||
enable_plugin: bool = True
|
||||
dependencies: list[str] = []
|
||||
python_dependencies: list[str] = []
|
||||
config_file_name: str = "config.toml"
|
||||
config_schema: dict = {}
|
||||
|
||||
def get_plugin_components(self) -> List[Tuple[CommandInfo | ActionInfo, Type[BaseCommand] | Type[BaseAction]]]:
|
||||
return [
|
||||
(AtAction.get_action_info(), AtAction),
|
||||
]
|
||||
@@ -1,83 +0,0 @@
|
||||
{
|
||||
"manifest_version": 1,
|
||||
"name": "戳一戳插件 (Poke Plugin)",
|
||||
"version": "1.0.0",
|
||||
"description": "智能戳一戳互动插件，支持主动戳用户和自动反戳机制，提供多种反应模式包括AI回复",
|
||||
"author": {
|
||||
"name": "MaiBot-Plus开发团队",
|
||||
"url": "https://github.com/MaiBot-Plus"
|
||||
},
|
||||
"license": "GPL-v3.0-or-later",
|
||||
|
||||
"host_application": {
|
||||
"min_version": "0.8.0"
|
||||
},
|
||||
"homepage_url": "https://github.com/MoFox-Studio/MoFox_Bot",
|
||||
"repository_url": "https://github.com/MoFox-Studio/MoFox_Bot",
|
||||
"keywords": ["poke", "interaction", "fun", "social", "ai-reply", "auto-response"],
|
||||
"categories": ["Social", "Interactive", "Fun"],
|
||||
|
||||
"default_locale": "zh-CN",
|
||||
"locales_path": "_locales",
|
||||
|
||||
"plugin_info": {
|
||||
"is_built_in": false,
|
||||
"plugin_type": "interactive",
|
||||
"components": [
|
||||
{
|
||||
"type": "action",
|
||||
"name": "poke_user",
|
||||
"description": "向指定用户发送戳一戳动作",
|
||||
"parameters": {
|
||||
"user_name": "需要戳一戳的用户名称",
|
||||
"times": "戳一戳次数（可选，默认为1）"
|
||||
},
|
||||
"activation_modes": ["llm_judge", "manual"],
|
||||
"llm_conditions": [
|
||||
"用户明确要求使用戳一戳",
|
||||
"想以有趣的方式提醒或与某人互动",
|
||||
"上下文明确需要戳一个或多个人"
|
||||
]
|
||||
},
|
||||
{
|
||||
"type": "command",
|
||||
"name": "poke_back",
|
||||
"description": "检测戳一戳消息并自动反戳",
|
||||
"pattern": "(?P<poker_name>\\S+)\\s*戳了戳\\s*(?P<target_name>\\S+)",
|
||||
"features": [
|
||||
"正则表达式匹配戳一戳文本",
|
||||
"智能识别戳击目标",
|
||||
"防抖机制避免频繁反戳",
|
||||
"多种反戳模式选择"
|
||||
]
|
||||
}
|
||||
],
|
||||
"features": [
|
||||
"主动戳一戳功能 - 可指定用户和次数",
|
||||
"智能反戳机制 - 自动检测并回应戳一戳",
|
||||
"多种反戳模式 - 戳回去/AI回复/随机选择",
|
||||
"AI智能回复 - 集成回复生成器API",
|
||||
"冷却时间控制 - 防止频繁反戳",
|
||||
"灵活配置选项 - 支持自定义回复消息",
|
||||
"安全过滤机制 - 只对针对机器人的戳一戳反应"
|
||||
],
|
||||
"configuration": {
|
||||
"poke_back_mode": {
|
||||
"type": "string",
|
||||
"options": ["poke", "reply", "random"],
|
||||
"default": "poke",
|
||||
"description": "反戳模式选择"
|
||||
},
|
||||
"poke_back_cooldown": {
|
||||
"type": "integer",
|
||||
"default": 5,
|
||||
"description": "反戳冷却时间（秒）"
|
||||
},
|
||||
"enable_typo_in_reply": {
|
||||
"type": "boolean",
|
||||
"default": false,
|
||||
"description": "AI回复时是否启用错字生成"
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,307 +0,0 @@
|
||||
import asyncio
|
||||
import random
|
||||
from typing import List, Tuple, Type
|
||||
|
||||
from src.plugin_system import (
|
||||
BasePlugin,
|
||||
register_plugin,
|
||||
BaseAction,
|
||||
BaseCommand,
|
||||
ComponentInfo,
|
||||
ActionActivationType,
|
||||
ConfigField,
|
||||
)
|
||||
from src.common.logger import get_logger
|
||||
from src.person_info.person_info import get_person_info_manager
|
||||
from src.plugin_system.apis import generator_api
|
||||
|
||||
logger = get_logger("poke_plugin")
|
||||
|
||||
|
||||
# ===== Action组件 =====
|
||||
class PokeAction(BaseAction):
|
||||
"""发送戳一戳动作"""
|
||||
|
||||
# === 基本信息（必须填写）===
|
||||
action_name = "poke_user"
|
||||
action_description = "向用户发送戳一戳"
|
||||
activation_type = ActionActivationType.ALWAYS
|
||||
parallel_action = True
|
||||
|
||||
# === 功能描述（必须填写）===
|
||||
action_parameters = {
|
||||
"user_name": "需要戳一戳的用户的名字 (可选)",
|
||||
"user_id": "需要戳一戳的用户的ID (可选，优先级更高)",
|
||||
"times": "需要戳一戳的次数 (默认为 1)",
|
||||
}
|
||||
action_require = ["当需要戳某个用户时使用", "当你想提醒特定用户时使用"]
|
||||
llm_judge_prompt = """
|
||||
判定是否需要使用戳一戳动作的条件：
|
||||
1. 用户明确要求使用戳一戳。
|
||||
2. 你想以一种有趣的方式提醒或与某人互动。
|
||||
3. 上下文明确需要你戳一个或多个人。
|
||||
|
||||
请回答"是"或"否"。
|
||||
"""
|
||||
associated_types = ["text"]
|
||||
|
||||
async def execute(self) -> Tuple[bool, str]:
|
||||
"""执行戳一戳的动作"""
|
||||
user_id = self.action_data.get("user_id")
|
||||
user_name = self.action_data.get("user_name")
|
||||
|
||||
try:
|
||||
times = int(self.action_data.get("times", 1))
|
||||
except (ValueError, TypeError):
|
||||
times = 1
|
||||
|
||||
# 优先使用 user_id
|
||||
if not user_id:
|
||||
if not user_name:
|
||||
logger.warning("戳一戳动作缺少 'user_id' 或 'user_name' 参数。")
|
||||
return False, "缺少用户标识参数"
|
||||
|
||||
# 备用方案：通过 user_name 查找
|
||||
user_info = await get_person_info_manager().get_person_info_by_name(user_name)
|
||||
if not user_info or not user_info.get("user_id"):
|
||||
logger.info(f"找不到名为 '{user_name}' 的用户。")
|
||||
return False, f"找不到名为 '{user_name}' 的用户"
|
||||
user_id = user_info.get("user_id")
|
||||
|
||||
display_name = user_name or user_id
|
||||
|
||||
for i in range(times):
|
||||
logger.info(f"正在向 {display_name} ({user_id}) 发送第 {i + 1}/{times} 次戳一戳...")
|
||||
await self.send_command(
|
||||
"SEND_POKE", args={"qq_id": user_id}, display_message=f"戳了戳 {display_name} ({i + 1}/{times})"
|
||||
)
|
||||
# 添加一个小的延迟，以避免发送过快
|
||||
await asyncio.sleep(0.5)
|
||||
|
||||
success_message = f"已向 {display_name} 发送 {times} 次戳一戳。"
|
||||
await self.store_action_info(
|
||||
action_build_into_prompt=True, action_prompt_display=success_message, action_done=True
|
||||
)
|
||||
return True, success_message
|
||||
|
||||
|
||||
# ===== Command组件 =====
|
||||
class PokeBackCommand(BaseCommand):
|
||||
"""反戳命令组件"""
|
||||
|
||||
command_name = "poke_back"
|
||||
command_description = "检测到戳一戳时自动反戳回去"
|
||||
# 匹配戳一戳的正则表达式 - 匹配 "xxx戳了戳xxx" 的格式
|
||||
command_pattern = r"(?P<poker_name>\S+)\s*戳了戳\s*(?P<target_name>\S+)"
|
||||
|
||||
async def execute(self) -> Tuple[bool, str, bool]:
|
||||
"""执行反戳逻辑"""
|
||||
# 检查反戳功能是否启用
|
||||
if not self.get_config("components.command_poke_back", True):
|
||||
return False, "", False
|
||||
|
||||
# 获取匹配的用户名
|
||||
poker_name = self.matched_groups.get("poker_name", "")
|
||||
target_name = self.matched_groups.get("target_name", "")
|
||||
|
||||
if not poker_name or not target_name:
|
||||
logger.debug("戳一戳消息格式不匹配，跳过反戳")
|
||||
return False, "", False
|
||||
|
||||
# 只有当目标是机器人自己时才反戳
|
||||
if target_name not in ["我", "bot", "机器人", "麦麦"]:
|
||||
logger.debug(f"戳一戳目标不是机器人 ({target_name}), 跳过反戳")
|
||||
return False, "", False
|
||||
|
||||
# 获取戳我的用户信息
|
||||
poker_info = await get_person_info_manager().get_person_info_by_name(poker_name)
|
||||
if not poker_info or not poker_info.get("user_id"):
|
||||
logger.info(f"找不到名为 '{poker_name}' 的用户信息，无法反戳")
|
||||
return False, "", False
|
||||
|
||||
poker_id = poker_info.get("user_id")
|
||||
if not isinstance(poker_id, (int, str)):
|
||||
logger.error(f"获取到的用户ID类型不正确: {type(poker_id)}")
|
||||
return False, "", False
|
||||
|
||||
# 确保poker_id是整数类型
|
||||
try:
|
||||
poker_id = int(poker_id)
|
||||
except (ValueError, TypeError):
|
||||
logger.error(f"无法将用户ID转换为整数: {poker_id}")
|
||||
return False, "", False
|
||||
|
||||
# 检查反戳冷却时间（防止频繁反戳）
|
||||
cooldown_seconds = self.get_config("components.poke_back_cooldown", 5)
|
||||
current_time = asyncio.get_event_loop().time()
|
||||
|
||||
# 使用类变量存储上次反戳时间
|
||||
if not hasattr(PokeBackCommand, "_last_poke_back_time"):
|
||||
PokeBackCommand._last_poke_back_time = {}
|
||||
|
||||
last_time = PokeBackCommand._last_poke_back_time.get(poker_id, 0)
|
||||
if current_time - last_time < cooldown_seconds:
|
||||
logger.info(f"反戳冷却中，跳过对 {poker_name} 的反戳")
|
||||
return False, "", False
|
||||
|
||||
# 记录本次反戳时间
|
||||
PokeBackCommand._last_poke_back_time[poker_id] = current_time
|
||||
|
||||
# 执行反戳
|
||||
logger.info(f"检测到 {poker_name} 戳了我，准备反戳回去")
|
||||
|
||||
try:
|
||||
# 获取反戳模式
|
||||
poke_back_mode = self.get_config("components.poke_back_mode", "poke") # "poke", "reply", "random"
|
||||
|
||||
if poke_back_mode == "random":
|
||||
# 随机选择模式
|
||||
poke_back_mode = random.choice(["poke", "reply"])
|
||||
|
||||
if poke_back_mode == "poke":
|
||||
# 戳回去模式
|
||||
await self._poke_back(poker_id, poker_name)
|
||||
elif poke_back_mode == "reply":
|
||||
# 回复模式
|
||||
await self._reply_back(poker_name)
|
||||
else:
|
||||
logger.warning(f"未知的反戳模式: {poke_back_mode}")
|
||||
return False, "", False
|
||||
|
||||
logger.info(f"成功反戳了 {poker_name} (模式: {poke_back_mode})")
|
||||
return True, f"反戳了 {poker_name}", False # 不拦截消息继续处理
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"反戳失败: {e}")
|
||||
return False, "", False
|
||||
|
||||
async def _poke_back(self, poker_id: int, poker_name: str):
|
||||
"""执行戳一戳反击"""
|
||||
await self.send_command(
|
||||
"SEND_POKE",
|
||||
args={"qq_id": poker_id},
|
||||
display_message=f"反戳了 {poker_name}",
|
||||
storage_message=False, # 不存储到消息历史中
|
||||
)
|
||||
|
||||
# 可选：发送一个随机的反戳回复
|
||||
poke_back_messages = self.get_config(
|
||||
"components.poke_back_messages",
|
||||
[
|
||||
"哼，戳回去！",
|
||||
"戳我干嘛~",
|
||||
"反戳！",
|
||||
"你戳我，我戳你！",
|
||||
"（戳回去）",
|
||||
],
|
||||
)
|
||||
|
||||
if poke_back_messages and self.get_config("components.send_poke_back_message", False):
|
||||
reply_message = random.choice(poke_back_messages)
|
||||
await self.send_text(reply_message)
|
||||
|
||||
async def _reply_back(self, poker_name: str):
|
||||
"""生成AI回复"""
|
||||
# 构造回复上下文
|
||||
extra_info = f"{poker_name}戳了我一下，需要生成一个有趣的回应。"
|
||||
|
||||
# 获取配置，确保类型正确
|
||||
enable_typo = self.get_config("components.enable_typo_in_reply", False)
|
||||
if not isinstance(enable_typo, bool):
|
||||
enable_typo = False
|
||||
|
||||
# 使用generator_api生成回复
|
||||
success, reply_set, _ = await generator_api.generate_reply(
|
||||
chat_stream=self.message.chat_stream,
|
||||
extra_info=extra_info,
|
||||
enable_tool=False,
|
||||
enable_splitter=True,
|
||||
enable_chinese_typo=enable_typo,
|
||||
from_plugin=True,
|
||||
)
|
||||
|
||||
if success and reply_set:
|
||||
# 发送生成的回复
|
||||
for reply_item in reply_set:
|
||||
message_type, content = reply_item
|
||||
if message_type == "text":
|
||||
await self.send_text(content)
|
||||
else:
|
||||
await self.send_type(message_type, content)
|
||||
else:
|
||||
# 如果AI回复失败，发送一个默认回复
|
||||
fallback_messages = self.get_config(
|
||||
"components.fallback_reply_messages",
|
||||
[
|
||||
"被戳了！",
|
||||
"诶？",
|
||||
"做什么呢~",
|
||||
"怎么了？",
|
||||
],
|
||||
)
|
||||
|
||||
# 确保fallback_messages是列表
|
||||
if isinstance(fallback_messages, list) and fallback_messages:
|
||||
fallback_reply = random.choice(fallback_messages)
|
||||
await self.send_text(fallback_reply)
|
||||
else:
|
||||
await self.send_text("被戳了！")
|
||||
|
||||
|
||||
# ===== 插件注册 =====
|
||||
@register_plugin
|
||||
class PokePlugin(BasePlugin):
|
||||
"""戳一戳插件"""
|
||||
|
||||
# 插件基本信息
|
||||
plugin_name: str = "poke_plugin"
|
||||
enable_plugin: bool = True
|
||||
dependencies: List[str] = []
|
||||
python_dependencies: List[str] = []
|
||||
config_file_name: str = "config.toml"
|
||||
|
||||
# 配置节描述
|
||||
config_section_descriptions = {"plugin": "插件基本信息", "components": "插件组件"}
|
||||
|
||||
# 配置Schema定义
|
||||
config_schema: dict = {
|
||||
"plugin": {
|
||||
"name": ConfigField(type=str, default="poke_plugin", description="插件名称"),
|
||||
"version": ConfigField(type=str, default="1.0.0", description="插件版本"),
|
||||
"enabled": ConfigField(type=bool, default=True, description="是否启用插件"),
|
||||
"config_version": ConfigField(type=str, default="1.0", description="配置版本"),
|
||||
},
|
||||
"components": {
|
||||
"action_poke_user": ConfigField(type=bool, default=True, description="是否启用戳一戳功能"),
|
||||
"command_poke_back": ConfigField(type=bool, default=True, description="是否启用反戳功能"),
|
||||
"poke_back_mode": ConfigField(
|
||||
type=str, default="poke", description="反戳模式: poke(戳回去), reply(AI回复), random(随机)"
|
||||
),
|
||||
"poke_back_cooldown": ConfigField(type=int, default=5, description="反戳冷却时间（秒）"),
|
||||
"send_poke_back_message": ConfigField(type=bool, default=False, description="戳回去时是否发送文字回复"),
|
||||
"enable_typo_in_reply": ConfigField(type=bool, default=False, description="AI回复时是否启用错字生成"),
|
||||
"poke_back_messages": ConfigField(
|
||||
type=list,
|
||||
default=["哼，戳回去！", "戳我干嘛~", "反戳！", "你戳我，我戳你！", "（戳回去）"],
|
||||
description="戳回去时的随机回复消息列表",
|
||||
),
|
||||
"fallback_reply_messages": ConfigField(
|
||||
type=list,
|
||||
default=["被戳了！", "诶？", "做什么呢~", "怎么了？"],
|
||||
description="AI回复失败时的备用回复消息列表",
|
||||
),
|
||||
},
|
||||
}
|
||||
|
||||
def get_plugin_components(self) -> List[Tuple[ComponentInfo, Type]]:
|
||||
components = []
|
||||
|
||||
# 添加戳一戳动作组件
|
||||
if self.get_config("components.action_poke_user"):
|
||||
components.append((PokeAction.get_action_info(), PokeAction))
|
||||
|
||||
# 添加反戳命令组件
|
||||
if self.get_config("components.command_poke_back"):
|
||||
components.append((PokeBackCommand.get_command_info(), PokeBackCommand))
|
||||
|
||||
return components
|
||||
@@ -1,9 +0,0 @@
|
||||
{
|
||||
"manifest_version": 1,
|
||||
"name": "智能提醒插件",
|
||||
"version": "1.0.0",
|
||||
"description": "一个能从对话中智能识别并设置定时提醒的插件。",
|
||||
"author": {
|
||||
"name": "墨墨"
|
||||
}
|
||||
}
|
||||
@@ -1,341 +0,0 @@
|
||||
import asyncio
|
||||
from datetime import datetime
|
||||
from typing import List, Tuple, Type, Optional
|
||||
|
||||
from dateutil.parser import parse as parse_datetime
|
||||
|
||||
from src.common.logger import get_logger
|
||||
from src.manager.async_task_manager import AsyncTask, async_task_manager
|
||||
from src.person_info.person_info import get_person_info_manager
|
||||
from src.plugin_system import (
|
||||
BaseAction,
|
||||
ActionInfo,
|
||||
BasePlugin,
|
||||
register_plugin,
|
||||
ActionActivationType,
|
||||
)
|
||||
from src.plugin_system.apis import send_api, llm_api, generator_api
|
||||
from src.plugin_system.base.component_types import ComponentType
|
||||
|
||||
logger = get_logger(__name__)
|
||||
|
||||
|
||||
# ============================ AsyncTask ============================
|
||||
|
||||
class ReminderTask(AsyncTask):
|
||||
def __init__(self, delay: float, stream_id: str, group_id: Optional[str], is_group: bool, target_user_id: str, target_user_name: str, event_details: str, creator_name: str, chat_stream: "ChatStream"):
|
||||
super().__init__(task_name=f"ReminderTask_{target_user_id}_{datetime.now().timestamp()}")
|
||||
self.delay = delay
|
||||
self.stream_id = stream_id
|
||||
self.group_id = group_id
|
||||
self.is_group = is_group
|
||||
self.target_user_id = target_user_id
|
||||
self.target_user_name = target_user_name
|
||||
self.event_details = event_details
|
||||
self.creator_name = creator_name
|
||||
self.chat_stream = chat_stream
|
||||
|
||||
async def run(self):
|
||||
try:
|
||||
if self.delay > 0:
|
||||
logger.info(f"等待 {self.delay:.2f} 秒后执行提醒...")
|
||||
await asyncio.sleep(self.delay)
|
||||
|
||||
logger.info(f"执行提醒任务: 给 {self.target_user_name} 发送关于 '{self.event_details}' 的提醒")
|
||||
|
||||
extra_info = f"现在是提醒时间，请你以一种符合你人设的、俏皮的方式提醒 {self.target_user_name}。\n提醒内容: {self.event_details}\n设置提醒的人: {self.creator_name}"
|
||||
success, reply_set, _ = await generator_api.generate_reply(
|
||||
chat_stream=self.chat_stream,
|
||||
extra_info=extra_info,
|
||||
reply_message=self.chat_stream.context.get_last_message().to_dict(),
|
||||
request_type="plugin.reminder.remind_message"
|
||||
)
|
||||
|
||||
if success and reply_set:
|
||||
for i, (_, text) in enumerate(reply_set):
|
||||
if self.is_group:
|
||||
message_payload = []
|
||||
if i == 0:
|
||||
message_payload.append({"type": "at", "data": {"qq": self.target_user_id}})
|
||||
message_payload.append({"type": "text", "data": {"text": f" {text}"}})
|
||||
await send_api.adapter_command_to_stream(
|
||||
action="send_group_msg",
|
||||
params={"group_id": self.group_id, "message": message_payload},
|
||||
stream_id=self.stream_id
|
||||
)
|
||||
else:
|
||||
await send_api.text_to_stream(text=text, stream_id=self.stream_id)
|
||||
else:
|
||||
# Fallback message
|
||||
reminder_text = f"叮咚！这是 {self.creator_name} 让我准时提醒你的事情：\n\n{self.event_details}"
|
||||
if self.is_group:
|
||||
message_payload = [
|
||||
{"type": "at", "data": {"qq": self.target_user_id}},
|
||||
{"type": "text", "data": {"text": f" {reminder_text}"}}
|
||||
]
|
||||
await send_api.adapter_command_to_stream(
|
||||
action="send_group_msg",
|
||||
params={"group_id": self.group_id, "message": message_payload},
|
||||
stream_id=self.stream_id
|
||||
)
|
||||
else:
|
||||
await send_api.text_to_stream(text=reminder_text, stream_id=self.stream_id)
|
||||
|
||||
logger.info(f"提醒任务 {self.task_name} 成功完成。")
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"执行提醒任务 {self.task_name} 时出错: {e}", exc_info=True)
|
||||
|
||||
|
||||
# =============================== Actions ===============================
|
||||
|
||||
class RemindAction(BaseAction):
|
||||
"""一个能从对话中智能识别并设置定时提醒的动作。"""
|
||||
|
||||
# === 基本信息 ===
|
||||
action_name = "set_reminder"
|
||||
action_description = "根据用户的对话内容，智能地设置一个未来的提醒事项。"
|
||||
|
||||
@staticmethod
|
||||
def get_action_info() -> ActionInfo:
|
||||
return ActionInfo(
|
||||
name="set_reminder",
|
||||
component_type=ComponentType.ACTION,
|
||||
activation_type=ActionActivationType.KEYWORD,
|
||||
activation_keywords=["提醒", "叫我", "记得", "别忘了"]
|
||||
)
|
||||
|
||||
# === LLM 判断与参数提取 ===
|
||||
llm_judge_prompt = ""
|
||||
action_parameters = {}
|
||||
action_require = [
|
||||
"当用户请求在未来的某个时间点提醒他/她或别人某件事时使用",
|
||||
"适用于包含明确时间信息和事件描述的对话",
|
||||
"例如：'10分钟后提醒我收快递'、'明天早上九点喊一下李四参加晨会'"
|
||||
]
|
||||
|
||||
async def execute(self) -> Tuple[bool, str]:
|
||||
"""执行设置提醒的动作"""
|
||||
try:
|
||||
# 获取所有可用的模型配置
|
||||
available_models = llm_api.get_available_models()
|
||||
if "planner" not in available_models:
|
||||
raise ValueError("未找到 'planner' 决策模型配置，无法解析时间")
|
||||
model_to_use = available_models["planner"]
|
||||
|
||||
bot_name = self.chat_stream.user_info.user_nickname
|
||||
|
||||
prompt = f"""
|
||||
从以下用户输入中提取提醒事件的关键信息。
|
||||
用户输入: "{self.chat_stream.context.message.processed_plain_text}"
|
||||
Bot的名字是: "{bot_name}"
|
||||
|
||||
请仔细分析句子结构，以确定谁是提醒的真正目标。Bot自身不应被视为被提醒人。
|
||||
请以JSON格式返回提取的信息，包含以下字段:
|
||||
- "user_name": 需要被提醒的人的姓名。如果未指定，则默认为"自己"。
|
||||
- "remind_time": 描述提醒时间的自然语言字符串。
|
||||
- "event_details": 需要提醒的具体事件内容。
|
||||
|
||||
示例:
|
||||
- 用户输入: "提醒我十分钟后开会" -> {{"user_name": "自己", "remind_time": "十分钟后", "event_details": "开会"}}
|
||||
- 用户输入: "{bot_name}，提醒一闪一分钟后睡觉" -> {{"user_name": "一闪", "remind_time": "一分钟后", "event_details": "睡觉"}}
|
||||
|
||||
如果无法提取完整信息，请返回一个包含空字符串的JSON对象，例如：{{"user_name": "", "remind_time": "", "event_details": ""}}
|
||||
"""
|
||||
|
||||
success, response, _, _ = await llm_api.generate_with_model(
|
||||
prompt,
|
||||
model_config=model_to_use,
|
||||
request_type="plugin.reminder.parameter_extractor"
|
||||
)
|
||||
|
||||
if not success or not response:
|
||||
raise ValueError(f"LLM未能返回有效的参数: {response}")
|
||||
|
||||
import json
|
||||
import re
|
||||
try:
|
||||
# 提取JSON部分
|
||||
json_match = re.search(r"\{.*\}", response, re.DOTALL)
|
||||
if not json_match:
|
||||
raise ValueError("LLM返回的内容中不包含JSON")
|
||||
action_data = json.loads(json_match.group(0))
|
||||
except json.JSONDecodeError:
|
||||
logger.error(f"[ReminderPlugin] LLM返回的不是有效的JSON: {response}")
|
||||
return False, "LLM返回的不是有效的JSON"
|
||||
user_name = action_data.get("user_name")
|
||||
remind_time_str = action_data.get("remind_time")
|
||||
event_details = action_data.get("event_details")
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"[ReminderPlugin] 解析参数时出错: {e}", exc_info=True)
|
||||
return False, "解析参数时出错"
|
||||
|
||||
if not all([user_name, remind_time_str, event_details]):
|
||||
missing_params = [p for p, v in {"user_name": user_name, "remind_time": remind_time_str, "event_details": event_details}.items() if not v]
|
||||
error_msg = f"缺少必要的提醒参数: {', '.join(missing_params)}"
|
||||
logger.warning(f"[ReminderPlugin] LLM未能提取完整参数: {error_msg}")
|
||||
return False, error_msg
|
||||
|
||||
# 1. 解析时间
|
||||
try:
|
||||
assert isinstance(remind_time_str, str)
|
||||
# 优先尝试直接解析
|
||||
try:
|
||||
target_time = parse_datetime(remind_time_str, fuzzy=True)
|
||||
except Exception:
|
||||
# 如果直接解析失败，调用 LLM 进行转换
|
||||
logger.info(f"[ReminderPlugin] 直接解析时间 '{remind_time_str}' 失败，尝试使用 LLM 进行转换...")
|
||||
|
||||
# 获取所有可用的模型配置
|
||||
available_models = llm_api.get_available_models()
|
||||
if "planner" not in available_models:
|
||||
raise ValueError("未找到 'planner' 决策模型配置，无法解析时间")
|
||||
|
||||
# 明确使用 'planner' 模型
|
||||
model_to_use = available_models["planner"]
|
||||
|
||||
# 在执行时动态获取当前时间
|
||||
current_time_str = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
|
||||
prompt = (
|
||||
f"请将以下自然语言时间短语转换为一个未来的、标准的 'YYYY-MM-DD HH:MM:SS' 格式。"
|
||||
f"请只输出转换后的时间字符串，不要包含任何其他说明或文字。\n"
|
||||
f"作为参考，当前时间是: {current_time_str}\n"
|
||||
f"需要转换的时间短语是: '{remind_time_str}'\n"
|
||||
f"规则:\n"
|
||||
f"- 如果用户没有明确指出是上午还是下午，请根据当前时间判断。例如，如果当前是上午，用户说‘8点’，则应理解为今天的8点；如果当前是下午，用户说‘8点’，则应理解为今天的20点。\n"
|
||||
f"- 如果转换后的时间早于当前时间，则应理解为第二天的时间。\n"
|
||||
f"示例:\n"
|
||||
f"- 当前时间: 2025-09-16 10:00:00, 用户说: '8点' -> '2025-09-17 08:00:00'\n"
|
||||
f"- 当前时间: 2025-09-16 14:00:00, 用户说: '8点' -> '2025-09-16 20:00:00'\n"
|
||||
f"- 当前时间: 2025-09-16 23:00:00, 用户说: '晚上10点' -> '2025-09-17 22:00:00'"
|
||||
)
|
||||
|
||||
success, response, _, _ = await llm_api.generate_with_model(
|
||||
prompt,
|
||||
model_config=model_to_use,
|
||||
request_type="plugin.reminder.time_parser"
|
||||
)
|
||||
|
||||
if not success or not response:
|
||||
raise ValueError(f"LLM未能返回有效的时间字符串: {response}")
|
||||
|
||||
converted_time_str = response.strip()
|
||||
logger.info(f"[ReminderPlugin] LLM 转换结果: '{converted_time_str}'")
|
||||
target_time = parse_datetime(converted_time_str, fuzzy=False)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"[ReminderPlugin] 无法解析或转换时间字符串 '{remind_time_str}': {e}", exc_info=True)
|
||||
await self.send_text(f"抱歉，我无法理解您说的时间 '{remind_time_str}'，提醒设置失败。")
|
||||
return False, f"无法解析时间 '{remind_time_str}'"
|
||||
|
||||
now = datetime.now()
|
||||
if target_time <= now:
|
||||
await self.send_text("提醒时间必须是一个未来的时间点哦，提醒设置失败。")
|
||||
return False, "提醒时间必须在未来"
|
||||
|
||||
delay_seconds = (target_time - now).total_seconds()
|
||||
|
||||
# 2. 解析用户
|
||||
person_manager = get_person_info_manager()
|
||||
user_id_to_remind = None
|
||||
user_name_to_remind = ""
|
||||
|
||||
assert isinstance(user_name, str)
|
||||
|
||||
if user_name.strip() in ["自己", "我", "me"]:
|
||||
user_id_to_remind = self.user_id
|
||||
user_name_to_remind = self.user_nickname
|
||||
else:
|
||||
# 1. 精确匹配
|
||||
user_info = await person_manager.get_person_info_by_name(user_name)
|
||||
|
||||
# 2. 包含匹配
|
||||
if not user_info:
|
||||
for person_id, name in person_manager.person_name_list.items():
|
||||
if user_name in name:
|
||||
user_info = await person_manager.get_values(person_id, ["user_id", "user_nickname"])
|
||||
break
|
||||
|
||||
# 3. 模糊匹配 (此处简化为字符串相似度)
|
||||
if not user_info:
|
||||
best_match = None
|
||||
highest_similarity = 0
|
||||
for person_id, name in person_manager.person_name_list.items():
|
||||
import difflib
|
||||
similarity = difflib.SequenceMatcher(None, user_name, name).ratio()
|
||||
if similarity > highest_similarity:
|
||||
highest_similarity = similarity
|
||||
best_match = person_id
|
||||
|
||||
if best_match and highest_similarity > 0.6: # 相似度阈值
|
||||
user_info = await person_manager.get_values(best_match, ["user_id", "user_nickname"])
|
||||
|
||||
if not user_info or not user_info.get("user_id"):
|
||||
logger.warning(f"[ReminderPlugin] 找不到名为 '{user_name}' 的用户")
|
||||
await self.send_text(f"抱歉，我的联系人里找不到叫做 '{user_name}' 的人，提醒设置失败。")
|
||||
return False, f"用户 '{user_name}' 不存在"
|
||||
user_id_to_remind = user_info.get("user_id")
|
||||
user_name_to_remind = user_info.get("user_nickname") or user_name
|
||||
|
||||
# 3. 创建并调度异步任务
|
||||
try:
|
||||
assert user_id_to_remind is not None
|
||||
assert event_details is not None
|
||||
|
||||
reminder_task = ReminderTask(
|
||||
delay=delay_seconds,
|
||||
stream_id=self.chat_stream.stream_id,
|
||||
group_id=self.chat_stream.group_info.group_id if self.is_group and self.chat_stream.group_info else None,
|
||||
is_group=self.is_group,
|
||||
target_user_id=str(user_id_to_remind),
|
||||
target_user_name=str(user_name_to_remind),
|
||||
event_details=str(event_details),
|
||||
creator_name=str(self.user_nickname),
|
||||
chat_stream=self.chat_stream
|
||||
)
|
||||
await async_task_manager.add_task(reminder_task)
|
||||
|
||||
# 4. 生成并发送确认消息
|
||||
extra_info = f"你已经成功设置了一个提醒，请以一种符合你人设的、俏皮的方式回复用户。\n提醒时间: {target_time.strftime('%Y-%m-%d %H:%M:%S')}\n提醒对象: {user_name_to_remind}\n提醒内容: {event_details}"
|
||||
last_message = self.chat_stream.context.get_last_message()
|
||||
success, reply_set, _ = await generator_api.generate_reply(
|
||||
chat_stream=self.chat_stream,
|
||||
extra_info=extra_info,
|
||||
reply_message=last_message.to_dict(),
|
||||
request_type="plugin.reminder.confirm_message"
|
||||
)
|
||||
if success and reply_set:
|
||||
for _, text in reply_set:
|
||||
await self.send_text(text)
|
||||
else:
|
||||
# Fallback message
|
||||
fallback_message = f"好的，我记下了。\n将在 {target_time.strftime('%Y-%m-%d %H:%M:%S')} 提醒 {user_name_to_remind}：\n{event_details}"
|
||||
await self.send_text(fallback_message)
|
||||
|
||||
return True, "提醒设置成功"
|
||||
except Exception as e:
|
||||
logger.error(f"[ReminderPlugin] 创建提醒任务时出错: {e}", exc_info=True)
|
||||
await self.send_text("抱歉，设置提醒时发生了一点内部错误。")
|
||||
return False, "设置提醒时发生内部错误"
|
||||
|
||||
|
||||
# =============================== Plugin ===============================
|
||||
|
||||
@register_plugin
|
||||
class ReminderPlugin(BasePlugin):
|
||||
"""一个能从对话中智能识别并设置定时提醒的插件。"""
|
||||
|
||||
# --- 插件基础信息 ---
|
||||
plugin_name = "reminder_plugin"
|
||||
enable_plugin = True
|
||||
dependencies = []
|
||||
python_dependencies = []
|
||||
config_file_name = "config.toml"
|
||||
config_schema = {}
|
||||
|
||||
def get_plugin_components(self) -> List[Tuple[ActionInfo, Type[BaseAction]]]:
|
||||
"""注册插件的所有功能组件。"""
|
||||
return [
|
||||
(RemindAction.get_action_info(), RemindAction)
|
||||
]
|
||||
@@ -1,18 +0,0 @@
|
||||
{
|
||||
|
||||
"manifest_version": 1,
|
||||
"name": "Set Typing Status",
|
||||
"description": "一个在LLM生成回复时设置私聊输入状态的插件。",
|
||||
"version": "1.0.0",
|
||||
"author": {
|
||||
"name": "MoFox-Studio"
|
||||
},
|
||||
"license": "MIT",
|
||||
"homepage_url": "",
|
||||
"repository_url": "",
|
||||
"keywords": ["typing", "status", "private chat"],
|
||||
"categories": ["utility"],
|
||||
"host_application": {
|
||||
"min_version": "0.10.0"
|
||||
}
|
||||
}
|
||||
@@ -1,63 +0,0 @@
|
||||
from typing import List, Tuple, Type
|
||||
import logging
|
||||
|
||||
from src.plugin_system import (
|
||||
BasePlugin,
|
||||
register_plugin,
|
||||
ComponentInfo,
|
||||
BaseEventHandler,
|
||||
EventType,
|
||||
)
|
||||
from src.plugin_system.base.base_event import HandlerResult
|
||||
from src.plugin_system.apis import send_api
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class SetTypingStatusHandler(BaseEventHandler):
|
||||
"""在LLM处理私聊消息后设置“正在输入”状态的事件处理器。"""
|
||||
|
||||
handler_name = "set_typing_status_handler"
|
||||
handler_description = "在LLM生成回复后，将用户的聊天状态设置为“正在输入”。"
|
||||
init_subscribe = [EventType.POST_LLM]
|
||||
|
||||
async def execute(self, params: dict) -> HandlerResult:
|
||||
message = params.get("message")
|
||||
if not message or not message.is_private_message:
|
||||
return HandlerResult(success=True, continue_process=True)
|
||||
|
||||
user_id = message.message_info.user_info.user_id
|
||||
if not user_id:
|
||||
return HandlerResult(success=False, continue_process=True, message="无法获取用户ID")
|
||||
try:
|
||||
params = {"user_id": user_id, "event_type": 1}
|
||||
await send_api.adapter_command_to_stream(
|
||||
action="set_input_status",
|
||||
params=params,
|
||||
stream_id=message.stream_id,
|
||||
)
|
||||
logger.debug(f"成功为用户 {user_id} 设置“正在输入”状态。")
|
||||
return HandlerResult(success=True, continue_process=True)
|
||||
except Exception as e:
|
||||
logger.error(f"为用户 {user_id} 设置“正在输入”状态时出错: {e}")
|
||||
return HandlerResult(success=False, continue_process=True, message=str(e))
|
||||
|
||||
|
||||
@register_plugin
|
||||
class SetTypingStatusPlugin(BasePlugin):
|
||||
"""一个在LLM生成回复时设置私聊输入状态的插件。"""
|
||||
|
||||
plugin_name = "set_typing_status"
|
||||
enable_plugin = True
|
||||
dependencies = []
|
||||
python_dependencies = []
|
||||
config_file_name = ""
|
||||
|
||||
config_schema = {}
|
||||
|
||||
def get_plugin_components(self) -> List[Tuple[ComponentInfo, Type]]:
|
||||
"""注册插件的功能组件。"""
|
||||
return [(SetTypingStatusHandler.get_handler_info(), SetTypingStatusHandler)]
|
||||
|
||||
def register_plugin(self) -> bool:
|
||||
return True
|
||||
25
src/plugins/built_in/social_toolkit_plugin/_manifest.json
Normal file
25
src/plugins/built_in/social_toolkit_plugin/_manifest.json
Normal file
@@ -0,0 +1,25 @@
|
||||
{
|
||||
"manifest_version": 1,
|
||||
"name": "MoFox-Bot工具箱",
|
||||
"version": "1.0.0",
|
||||
"description": "一个集合多种实用功能的插件，旨在提升聊天体验和效率。",
|
||||
"author": {
|
||||
"name": "MoFox-Studio",
|
||||
"url": "https://github.com/MoFox-Studio"
|
||||
},
|
||||
"license": "GPL-v3.0-or-later",
|
||||
|
||||
"host_application": {
|
||||
"min_version": "0.10.0"
|
||||
},
|
||||
"keywords": ["emoji", "reaction", "like", "表情", "回应", "点赞"],
|
||||
"categories": ["Chat", "Integration"],
|
||||
|
||||
"default_locale": "zh-CN",
|
||||
"locales_path": "_locales",
|
||||
|
||||
"plugin_info": {
|
||||
"is_built_in": "true",
|
||||
"plugin_type": "functional"
|
||||
}
|
||||
}
|
||||
567
src/plugins/built_in/social_toolkit_plugin/plugin.py
Normal file
567
src/plugins/built_in/social_toolkit_plugin/plugin.py
Normal file
@@ -0,0 +1,567 @@
|
||||
import re
|
||||
from typing import List, Tuple, Type, Optional
|
||||
|
||||
from src.plugin_system import (
|
||||
BasePlugin,
|
||||
register_plugin,
|
||||
BaseAction,
|
||||
ComponentInfo,
|
||||
ActionActivationType,
|
||||
ConfigField,
|
||||
)
|
||||
from src.common.logger import get_logger
|
||||
from .qq_emoji_list import qq_face
|
||||
from src.plugin_system.base.component_types import ChatType
|
||||
from src.person_info.person_info import get_person_info_manager
|
||||
from dateutil.parser import parse as parse_datetime
|
||||
from src.manager.async_task_manager import AsyncTask, async_task_manager
|
||||
from src.plugin_system.apis import send_api, llm_api, generator_api
|
||||
from src.plugin_system.base.component_types import ComponentType
|
||||
from typing import Optional
|
||||
from src.chat.message_receive.chat_stream import ChatStream
|
||||
import asyncio
|
||||
import datetime
|
||||
|
||||
logger = get_logger("set_emoji_like_plugin")
|
||||
|
||||
# ============================ AsyncTask ============================
|
||||
|
||||
|
||||
class ReminderTask(AsyncTask):
|
||||
def __init__(
|
||||
self,
|
||||
delay: float,
|
||||
stream_id: str,
|
||||
group_id: Optional[str],
|
||||
is_group: bool,
|
||||
target_user_id: str,
|
||||
target_user_name: str,
|
||||
event_details: str,
|
||||
creator_name: str,
|
||||
chat_stream: ChatStream,
|
||||
):
|
||||
super().__init__(task_name=f"ReminderTask_{target_user_id}_{datetime.datetime.now().timestamp()}")
|
||||
self.delay = delay
|
||||
self.stream_id = stream_id
|
||||
self.group_id = group_id
|
||||
self.is_group = is_group
|
||||
self.target_user_id = target_user_id
|
||||
self.target_user_name = target_user_name
|
||||
self.event_details = event_details
|
||||
self.creator_name = creator_name
|
||||
self.chat_stream = chat_stream
|
||||
|
||||
async def run(self):
|
||||
try:
|
||||
if self.delay > 0:
|
||||
logger.info(f"等待 {self.delay:.2f} 秒后执行提醒...")
|
||||
await asyncio.sleep(self.delay)
|
||||
|
||||
logger.info(f"执行提醒任务: 给 {self.target_user_name} 发送关于 '{self.event_details}' 的提醒")
|
||||
|
||||
extra_info = f"现在是提醒时间，请你以一种符合你人设的、俏皮的方式提醒 {self.target_user_name}。\n提醒内容: {self.event_details}\n设置提醒的人: {self.creator_name}"
|
||||
last_message = self.chat_stream.context_manager.context.get_last_message()
|
||||
reply_message_dict = last_message.flatten() if last_message else None
|
||||
success, reply_set, _ = await generator_api.generate_reply(
|
||||
chat_stream=self.chat_stream,
|
||||
extra_info=extra_info,
|
||||
reply_message=reply_message_dict,
|
||||
request_type="plugin.reminder.remind_message",
|
||||
)
|
||||
|
||||
if success and reply_set:
|
||||
for i, (_, text) in enumerate(reply_set):
|
||||
if self.is_group:
|
||||
message_payload = []
|
||||
if i == 0:
|
||||
message_payload.append({"type": "at", "data": {"qq": self.target_user_id}})
|
||||
message_payload.append({"type": "text", "data": {"text": f" {text}"}})
|
||||
await send_api.adapter_command_to_stream(
|
||||
action="send_group_msg",
|
||||
params={"group_id": self.group_id, "message": message_payload},
|
||||
stream_id=self.stream_id,
|
||||
)
|
||||
else:
|
||||
await send_api.text_to_stream(text=text, stream_id=self.stream_id)
|
||||
else:
|
||||
# Fallback message
|
||||
reminder_text = f"叮咚！这是 {self.creator_name} 让我准时提醒你的事情：\n\n{self.event_details}"
|
||||
if self.is_group:
|
||||
message_payload = [
|
||||
{"type": "at", "data": {"qq": self.target_user_id}},
|
||||
{"type": "text", "data": {"text": f" {reminder_text}"}},
|
||||
]
|
||||
await send_api.adapter_command_to_stream(
|
||||
action="send_group_msg",
|
||||
params={"group_id": self.group_id, "message": message_payload},
|
||||
stream_id=self.stream_id,
|
||||
)
|
||||
else:
|
||||
await send_api.text_to_stream(text=reminder_text, stream_id=self.stream_id)
|
||||
|
||||
logger.info(f"提醒任务 {self.task_name} 成功完成。")
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"执行提醒任务 {self.task_name} 时出错: {e}", exc_info=True)
|
||||
|
||||
|
||||
# =============================== Actions ===============================
|
||||
|
||||
|
||||
def get_emoji_id(emoji_input: str) -> str | None:
|
||||
"""根据输入获取表情ID"""
|
||||
# 如果输入本身就是数字ID，直接返回
|
||||
if emoji_input.isdigit() or (isinstance(emoji_input, str) and emoji_input.startswith("😊")):
|
||||
if emoji_input in qq_face:
|
||||
return emoji_input
|
||||
|
||||
# 尝试从 "[表情：xxx]" 格式中提取
|
||||
match = re.search(r"\[表情：(.+?)\]", emoji_input)
|
||||
if match:
|
||||
emoji_name = match.group(1).strip()
|
||||
else:
|
||||
emoji_name = emoji_input.strip()
|
||||
|
||||
# 遍历查找
|
||||
for key, value in qq_face.items():
|
||||
# value 的格式是 "[表情：xxx]"
|
||||
if f"[表情：{emoji_name}]" == value:
|
||||
return key
|
||||
|
||||
return None
|
||||
|
||||
|
||||
# ===== Action组件 =====
|
||||
|
||||
|
||||
class PokeAction(BaseAction):
|
||||
"""发送戳一戳动作"""
|
||||
|
||||
# === 基本信息（必须填写）===
|
||||
action_name = "poke_user"
|
||||
action_description = "向用户发送戳一戳"
|
||||
activation_type = ActionActivationType.ALWAYS
|
||||
parallel_action = True
|
||||
|
||||
# === 功能描述（必须填写）===
|
||||
action_parameters = {
|
||||
"user_name": "需要戳一戳的用户的名字 (可选)",
|
||||
"user_id": "需要戳一戳的用户的ID (可选，优先级更高)",
|
||||
"times": "需要戳一戳的次数 (默认为 1)",
|
||||
}
|
||||
action_require = ["当需要戳某个用户时使用", "当你想提醒特定用户时使用"]
|
||||
llm_judge_prompt = """
|
||||
判定是否需要使用戳一戳动作的条件：
|
||||
1. 用户明确要求使用戳一戳。
|
||||
2. 你想以一种有趣的方式提醒或与某人互动。
|
||||
3. 上下文明确需要你戳一个或多个人。
|
||||
|
||||
请回答"是"或"否"。
|
||||
"""
|
||||
associated_types = ["text"]
|
||||
|
||||
async def execute(self) -> Tuple[bool, str]:
|
||||
"""执行戳一戳的动作"""
|
||||
user_id = self.action_data.get("user_id")
|
||||
user_name = self.action_data.get("user_name")
|
||||
|
||||
try:
|
||||
times = int(self.action_data.get("times", 1))
|
||||
except (ValueError, TypeError):
|
||||
times = 1
|
||||
|
||||
# 优先使用 user_id
|
||||
if not user_id:
|
||||
if not user_name:
|
||||
logger.warning("戳一戳动作缺少 'user_id' 或 'user_name' 参数。")
|
||||
return False, "缺少用户标识参数"
|
||||
|
||||
# 备用方案：通过 user_name 查找
|
||||
user_info = await get_person_info_manager().get_person_info_by_name(user_name)
|
||||
if not user_info or not user_info.get("user_id"):
|
||||
logger.info(f"找不到名为 '{user_name}' 的用户。")
|
||||
return False, f"找不到名为 '{user_name}' 的用户"
|
||||
user_id = user_info.get("user_id")
|
||||
|
||||
display_name = user_name or user_id
|
||||
|
||||
for i in range(times):
|
||||
logger.info(f"正在向 {display_name} ({user_id}) 发送第 {i + 1}/{times} 次戳一戳...")
|
||||
await self.send_command(
|
||||
"SEND_POKE", args={"qq_id": user_id}, display_message=f"戳了戳 {display_name} ({i + 1}/{times})"
|
||||
)
|
||||
# 添加一个小的延迟，以避免发送过快
|
||||
await asyncio.sleep(0.5)
|
||||
|
||||
success_message = f"已向 {display_name} 发送 {times} 次戳一戳。"
|
||||
await self.store_action_info(
|
||||
action_build_into_prompt=True, action_prompt_display=success_message, action_done=True
|
||||
)
|
||||
return True, success_message
|
||||
|
||||
|
||||
class SetEmojiLikeAction(BaseAction):
|
||||
"""设置消息表情回应"""
|
||||
|
||||
# === 基本信息（必须填写）===
|
||||
action_name = "set_emoji_like"
|
||||
action_description = "为某条已经存在的消息添加‘贴表情’回应（类似点赞），而不是发送新消息。可以在觉得某条消息非常有趣、值得赞同或者需要特殊情感回应时主动使用。"
|
||||
activation_type = ActionActivationType.ALWAYS # 消息接收时激活(?)
|
||||
chat_type_allow = ChatType.GROUP
|
||||
parallel_action = True
|
||||
|
||||
# === 功能描述（必须填写）===
|
||||
# 从 qq_face 字典中提取所有表情名称用于提示
|
||||
emoji_options = []
|
||||
for name in qq_face.values():
|
||||
match = re.search(r"\[表情：(.+?)\]", name)
|
||||
if match:
|
||||
emoji_options.append(match.group(1))
|
||||
|
||||
action_parameters = {
|
||||
"set": "是否设置回应 (True/False)",
|
||||
}
|
||||
action_require = [
|
||||
"当需要对一个已存在消息进行‘贴表情’回应时使用",
|
||||
"这是一个对旧消息的操作，而不是发送新消息",
|
||||
"如果你想发送一个新的表情包消息，请使用 'emoji' 动作",
|
||||
]
|
||||
llm_judge_prompt = """
|
||||
判定是否需要使用贴表情动作的条件：
|
||||
1. 用户明确要求使用贴表情包
|
||||
2. 这是一个适合表达强烈情绪的场合
|
||||
3. 不要发送太多表情包，如果你已经发送过多个表情包则回答"否"
|
||||
|
||||
请回答"是"或"否"。
|
||||
"""
|
||||
associated_types = ["text"]
|
||||
|
||||
async def execute(self) -> Tuple[bool, str]:
|
||||
"""执行设置表情回应的动作"""
|
||||
message_id = None
|
||||
set_like = self.action_data.get("set", True)
|
||||
if self.has_action_message:
|
||||
logger.debug(str(self.action_message))
|
||||
if isinstance(self.action_message, dict):
|
||||
message_id = self.action_message.get("message_id")
|
||||
logger.info(f"获取到的消息ID: {message_id}")
|
||||
else:
|
||||
logger.error("未提供消息ID")
|
||||
await self.store_action_info(
|
||||
action_build_into_prompt=True,
|
||||
action_prompt_display=f"执行了set_emoji_like动作：{self.action_name},失败: 未提供消息ID",
|
||||
action_done=False,
|
||||
)
|
||||
return False, "未提供消息ID"
|
||||
available_models = llm_api.get_available_models()
|
||||
if "utils_small" not in available_models:
|
||||
logger.error("未找到 'utils_small' 模型配置，无法选择表情")
|
||||
return False, "表情选择功能配置错误"
|
||||
|
||||
model_to_use = available_models["utils_small"]
|
||||
|
||||
# 获取最近的对话历史作为上下文
|
||||
context_text = ""
|
||||
if self.action_message:
|
||||
context_text = self.action_message.get("processed_plain_text", "")
|
||||
else:
|
||||
logger.error("无法找到动作选择的原始消息")
|
||||
return False, "无法找到动作选择的原始消息"
|
||||
|
||||
prompt = (
|
||||
f"根据以下这条消息，从列表中选择一个最合适的表情名称来回应这条消息。\n"
|
||||
f"消息内容: '{context_text}'\n"
|
||||
f"可用表情列表: {', '.join(self.emoji_options)}\n"
|
||||
f"你的任务是：只输出你选择的表情的名称，不要包含任何其他文字或标点。\n"
|
||||
f"例如，如果觉得应该用'赞'，就只输出'赞'。"
|
||||
)
|
||||
|
||||
success, response, _, _ = await llm_api.generate_with_model(
|
||||
prompt, model_config=model_to_use, request_type="plugin.set_emoji_like.select_emoji"
|
||||
)
|
||||
|
||||
if not success or not response:
|
||||
logger.error("二级LLM未能选择有效的表情。")
|
||||
return False, "无法找到合适的表情。"
|
||||
|
||||
chosen_emoji_name = response.strip()
|
||||
logger.info(f"二级LLM选择的表情是: '{chosen_emoji_name}'")
|
||||
emoji_id = get_emoji_id(chosen_emoji_name)
|
||||
|
||||
if not emoji_id:
|
||||
logger.error(f"二级LLM选择的表情 '{chosen_emoji_name}' 仍然无法匹配到有效的表情ID。")
|
||||
await self.store_action_info(
|
||||
action_build_into_prompt=True,
|
||||
action_prompt_display=f"执行了set_emoji_like动作：{self.action_name},失败: 找不到表情: '{chosen_emoji_name}'",
|
||||
action_done=False,
|
||||
)
|
||||
return False, f"找不到表情: '{chosen_emoji_name}'。"
|
||||
|
||||
# 4. 使用适配器API发送命令
|
||||
if not message_id:
|
||||
logger.error("未提供消息ID")
|
||||
await self.store_action_info(
|
||||
action_build_into_prompt=True,
|
||||
action_prompt_display=f"执行了set_emoji_like动作：{self.action_name},失败: 未提供消息ID",
|
||||
action_done=False,
|
||||
)
|
||||
return False, "未提供消息ID"
|
||||
|
||||
try:
|
||||
# 使用适配器API发送贴表情命令
|
||||
success = await self.send_command(
|
||||
command_name="set_emoji_like",
|
||||
args={"message_id": message_id, "emoji_id": emoji_id, "set": set_like},
|
||||
storage_message=False,
|
||||
)
|
||||
if success:
|
||||
logger.info("设置表情回应成功")
|
||||
await self.store_action_info(
|
||||
action_build_into_prompt=True,
|
||||
action_prompt_display=f"执行了set_emoji_like动作,{chosen_emoji_name},设置表情回应: {emoji_id}, 是否设置: {set_like}",
|
||||
action_done=True,
|
||||
)
|
||||
return True, "成功设置表情回应"
|
||||
else:
|
||||
logger.error("设置表情回应失败")
|
||||
await self.store_action_info(
|
||||
action_build_into_prompt=True,
|
||||
action_prompt_display=f"执行了set_emoji_like动作：{self.action_name},失败",
|
||||
action_done=False,
|
||||
)
|
||||
return False, "设置表情回应失败"
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"设置表情回应失败: {e}")
|
||||
await self.store_action_info(
|
||||
action_build_into_prompt=True,
|
||||
action_prompt_display=f"执行了set_emoji_like动作：{self.action_name},失败: {e}",
|
||||
action_done=False,
|
||||
)
|
||||
return False, f"设置表情回应失败: {e}"
|
||||
|
||||
|
||||
class RemindAction(BaseAction):
|
||||
"""一个能从对话中智能识别并设置定时提醒的动作。"""
|
||||
|
||||
# === 基本信息 ===
|
||||
action_name = "set_reminder"
|
||||
action_description = "根据用户的对话内容，智能地设置一个未来的提醒事项。"
|
||||
activation_type = (ActionActivationType.KEYWORD,)
|
||||
activation_keywords = ["提醒", "叫我", "记得", "别忘了"]
|
||||
chat_type_allow = ChatType.ALL
|
||||
parallel_action = True
|
||||
|
||||
# === LLM 判断与参数提取 ===
|
||||
llm_judge_prompt = ""
|
||||
action_parameters = {}
|
||||
action_require = [
|
||||
"当用户请求在未来的某个时间点提醒他/她或别人某件事时使用",
|
||||
"适用于包含明确时间信息和事件描述的对话",
|
||||
"例如：'10分钟后提醒我收快递'、'明天早上九点喊一下李四参加晨会'",
|
||||
]
|
||||
|
||||
async def execute(self) -> Tuple[bool, str]:
|
||||
"""执行设置提醒的动作"""
|
||||
user_name = self.action_data.get("user_name")
|
||||
remind_time_str = self.action_data.get("remind_time")
|
||||
event_details = self.action_data.get("event_details")
|
||||
|
||||
if not all([user_name, remind_time_str, event_details]):
|
||||
missing_params = [
|
||||
p
|
||||
for p, v in {
|
||||
"user_name": user_name,
|
||||
"remind_time": remind_time_str,
|
||||
"event_details": event_details,
|
||||
}.items()
|
||||
if not v
|
||||
]
|
||||
error_msg = f"缺少必要的提醒参数: {', '.join(missing_params)}"
|
||||
logger.warning(f"[ReminderPlugin] LLM未能提取完整参数: {error_msg}")
|
||||
return False, error_msg
|
||||
|
||||
# 1. 解析时间
|
||||
try:
|
||||
assert isinstance(remind_time_str, str)
|
||||
# 优先尝试直接解析
|
||||
try:
|
||||
target_time = parse_datetime(remind_time_str, fuzzy=True)
|
||||
except Exception:
|
||||
# 如果直接解析失败，调用 LLM 进行转换
|
||||
logger.info(f"[ReminderPlugin] 直接解析时间 '{remind_time_str}' 失败，尝试使用 LLM 进行转换...")
|
||||
|
||||
# 获取所有可用的模型配置
|
||||
available_models = llm_api.get_available_models()
|
||||
if "utils_small" not in available_models:
|
||||
raise ValueError("未找到 'utils_small' 模型配置，无法解析时间")
|
||||
|
||||
# 明确使用 'planner' 模型
|
||||
model_to_use = available_models["utils_small"]
|
||||
|
||||
# 在执行时动态获取当前时间
|
||||
current_time_str = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||||
prompt = (
|
||||
f"请将以下自然语言时间短语转换为一个未来的、标准的 'YYYY-MM-DD HH:MM:SS' 格式。"
|
||||
f"请只输出转换后的时间字符串，不要包含任何其他说明或文字。\n"
|
||||
f"作为参考，当前时间是: {current_time_str}\n"
|
||||
f"需要转换的时间短语是: '{remind_time_str}'\n"
|
||||
f"规则:\n"
|
||||
f"- 如果用户没有明确指出是上午还是下午，请根据当前时间判断。例如，如果当前是上午，用户说‘8点’，则应理解为今天的8点；如果当前是下午，用户说‘8点’，则应理解为今天的20点。\n"
|
||||
f"- 如果转换后的时间早于当前时间，则应理解为第二天的时间。\n"
|
||||
f"示例:\n"
|
||||
f"- 当前时间: 2025-09-16 10:00:00, 用户说: '8点' -> '2025-09-17 08:00:00'\n"
|
||||
f"- 当前时间: 2025-09-16 14:00:00, 用户说: '8点' -> '2025-09-16 20:00:00'\n"
|
||||
f"- 当前时间: 2025-09-16 23:00:00, 用户说: '晚上10点' -> '2025-09-17 22:00:00'"
|
||||
)
|
||||
|
||||
success, response, _, _ = await llm_api.generate_with_model(
|
||||
prompt, model_config=model_to_use, request_type="plugin.reminder.time_parser"
|
||||
)
|
||||
|
||||
if not success or not response:
|
||||
raise ValueError(f"LLM未能返回有效的时间字符串: {response}")
|
||||
|
||||
converted_time_str = response.strip()
|
||||
logger.info(f"[ReminderPlugin] LLM 转换结果: '{converted_time_str}'")
|
||||
target_time = parse_datetime(converted_time_str, fuzzy=False)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"[ReminderPlugin] 无法解析或转换时间字符串 '{remind_time_str}': {e}", exc_info=True)
|
||||
await self.send_text(f"抱歉，我无法理解您说的时间 '{remind_time_str}'，提醒设置失败。")
|
||||
return False, f"无法解析时间 '{remind_time_str}'"
|
||||
|
||||
now = datetime.datetime.now()
|
||||
if target_time <= now:
|
||||
await self.send_text("提醒时间必须是一个未来的时间点哦，提醒设置失败。")
|
||||
return False, "提醒时间必须在未来"
|
||||
|
||||
delay_seconds = (target_time - now).total_seconds()
|
||||
|
||||
# 2. 解析用户
|
||||
person_manager = get_person_info_manager()
|
||||
user_id_to_remind = None
|
||||
user_name_to_remind = ""
|
||||
|
||||
assert isinstance(user_name, str)
|
||||
|
||||
if user_name.strip() in ["自己", "我", "me"]:
|
||||
user_id_to_remind = self.user_id
|
||||
user_name_to_remind = self.user_nickname
|
||||
else:
|
||||
# 1. 精确匹配
|
||||
user_info = await person_manager.get_person_info_by_name(user_name)
|
||||
|
||||
# 2. 包含匹配
|
||||
if not user_info:
|
||||
for person_id, name in person_manager.person_name_list.items():
|
||||
if user_name in name:
|
||||
user_info = await person_manager.get_values(person_id, ["user_id", "user_nickname"])
|
||||
break
|
||||
|
||||
# 3. 模糊匹配 (此处简化为字符串相似度)
|
||||
if not user_info:
|
||||
best_match = None
|
||||
highest_similarity = 0
|
||||
for person_id, name in person_manager.person_name_list.items():
|
||||
import difflib
|
||||
|
||||
similarity = difflib.SequenceMatcher(None, user_name, name).ratio()
|
||||
if similarity > highest_similarity:
|
||||
highest_similarity = similarity
|
||||
best_match = person_id
|
||||
|
||||
if best_match and highest_similarity > 0.6: # 相似度阈值
|
||||
user_info = await person_manager.get_values(best_match, ["user_id", "user_nickname"])
|
||||
|
||||
if not user_info or not user_info.get("user_id"):
|
||||
logger.warning(f"[ReminderPlugin] 找不到名为 '{user_name}' 的用户")
|
||||
await self.send_text(f"抱歉，我的联系人里找不到叫做 '{user_name}' 的人，提醒设置失败。")
|
||||
return False, f"用户 '{user_name}' 不存在"
|
||||
user_id_to_remind = user_info.get("user_id")
|
||||
user_name_to_remind = user_info.get("user_nickname") or user_name
|
||||
|
||||
# 3. 创建并调度异步任务
|
||||
try:
|
||||
assert user_id_to_remind is not None
|
||||
assert event_details is not None
|
||||
|
||||
reminder_task = ReminderTask(
|
||||
delay=delay_seconds,
|
||||
stream_id=self.chat_stream.stream_id,
|
||||
group_id=self.chat_stream.group_info.group_id
|
||||
if self.is_group and self.chat_stream.group_info
|
||||
else None,
|
||||
is_group=self.is_group,
|
||||
target_user_id=str(user_id_to_remind),
|
||||
target_user_name=str(user_name_to_remind),
|
||||
event_details=str(event_details),
|
||||
creator_name=str(self.user_nickname),
|
||||
chat_stream=self.chat_stream,
|
||||
)
|
||||
await async_task_manager.add_task(reminder_task)
|
||||
|
||||
# 4. 生成并发送确认消息
|
||||
extra_info = f"你已经成功设置了一个提醒，请以一种符合你人设的、俏皮的方式回复用户。\n提醒时间: {target_time.strftime('%Y-%m-%d %H:%M:%S')}\n提醒对象: {user_name_to_remind}\n提醒内容: {event_details}"
|
||||
last_message = self.chat_stream.context_manager.context.get_last_message()
|
||||
reply_message_dict = last_message.flatten() if last_message else None
|
||||
success, reply_set, _ = await generator_api.generate_reply(
|
||||
chat_stream=self.chat_stream,
|
||||
extra_info=extra_info,
|
||||
reply_message=reply_message_dict,
|
||||
request_type="plugin.reminder.confirm_message",
|
||||
)
|
||||
if success and reply_set:
|
||||
for _, text in reply_set:
|
||||
await self.send_text(text)
|
||||
else:
|
||||
# Fallback message
|
||||
fallback_message = f"好的，我记下了。\n将在 {target_time.strftime('%Y-%m-%d %H:%M:%S')} 提醒 {user_name_to_remind}：\n{event_details}"
|
||||
await self.send_text(fallback_message)
|
||||
|
||||
return True, "提醒设置成功"
|
||||
except Exception as e:
|
||||
logger.error(f"[ReminderPlugin] 创建提醒任务时出错: {e}", exc_info=True)
|
||||
await self.send_text("抱歉，设置提醒时发生了一点内部错误。")
|
||||
return False, "设置提醒时发生内部错误"
|
||||
|
||||
|
||||
# ===== 插件注册 =====
|
||||
@register_plugin
|
||||
class SetEmojiLikePlugin(BasePlugin):
|
||||
"""一个集合多种实用功能的插件，旨在提升聊天体验和效率。"""
|
||||
|
||||
# 插件基本信息
|
||||
plugin_name: str = "social_toolkit_plugin" # 内部标识符
|
||||
enable_plugin: bool = True
|
||||
dependencies: List[str] = [] # 插件依赖列表
|
||||
python_dependencies: List[str] = [] # Python包依赖列表，现在使用内置API
|
||||
config_file_name: str = "config.toml" # 配置文件名
|
||||
|
||||
# 配置节描述
|
||||
config_section_descriptions = {"plugin": "插件基本信息", "components": "插件组件"}
|
||||
|
||||
# 配置Schema定义
|
||||
config_schema: dict = {
|
||||
"plugin": {
|
||||
"name": ConfigField(type=str, default="set_emoji_like", description="插件名称"),
|
||||
"version": ConfigField(type=str, default="1.0.0", description="插件版本"),
|
||||
"enabled": ConfigField(type=bool, default=True, description="是否启用插件"),
|
||||
"config_version": ConfigField(type=str, default="1.1", description="配置版本"),
|
||||
},
|
||||
"components": {
|
||||
"action_set_emoji_like": ConfigField(type=bool, default=True, description="是否启用设置表情回应功能"),
|
||||
"action_poke_enable": ConfigField(type=bool, default=True, description="是否启用戳一戳功能"),
|
||||
"action_set_reminder_enable": ConfigField(type=bool, default=True, description="是否启用定时提醒功能"),
|
||||
},
|
||||
}
|
||||
|
||||
def get_plugin_components(self) -> List[Tuple[ComponentInfo, Type]]:
|
||||
enable_components = []
|
||||
if self.get_config("components.action_set_emoji_like"):
|
||||
enable_components.append((SetEmojiLikeAction.get_action_info(), SetEmojiLikeAction))
|
||||
if self.get_config("components.action_poke_enable"):
|
||||
enable_components.append((PokeAction.get_action_info(), PokeAction))
|
||||
if self.get_config("components.action_set_reminder_enable"):
|
||||
enable_components.append((RemindAction.get_action_info(), RemindAction))
|
||||
return enable_components
|
||||
@@ -1,216 +0,0 @@
|
||||
import asyncio
|
||||
from datetime import datetime
|
||||
from typing import List, Tuple, Type
|
||||
from dateutil.parser import parse as parse_datetime
|
||||
|
||||
from src.common.logger import get_logger
|
||||
from src.manager.async_task_manager import AsyncTask, async_task_manager
|
||||
from src.person_info.person_info import get_person_info_manager
|
||||
from src.plugin_system import (
|
||||
BaseAction,
|
||||
ActionInfo,
|
||||
BasePlugin,
|
||||
register_plugin,
|
||||
ActionActivationType,
|
||||
)
|
||||
from src.plugin_system.apis import send_api
|
||||
from src.plugin_system.base.component_types import ChatType
|
||||
|
||||
logger = get_logger(__name__)
|
||||
|
||||
|
||||
# ============================ AsyncTask ============================
|
||||
|
||||
|
||||
class ReminderTask(AsyncTask):
|
||||
def __init__(
|
||||
self,
|
||||
delay: float,
|
||||
stream_id: str,
|
||||
is_group: bool,
|
||||
target_user_id: str,
|
||||
target_user_name: str,
|
||||
event_details: str,
|
||||
creator_name: str,
|
||||
):
|
||||
super().__init__(task_name=f"ReminderTask_{target_user_id}_{datetime.now().timestamp()}")
|
||||
self.delay = delay
|
||||
self.stream_id = stream_id
|
||||
self.is_group = is_group
|
||||
self.target_user_id = target_user_id
|
||||
self.target_user_name = target_user_name
|
||||
self.event_details = event_details
|
||||
self.creator_name = creator_name
|
||||
|
||||
async def run(self):
|
||||
try:
|
||||
if self.delay > 0:
|
||||
logger.info(f"等待 {self.delay:.2f} 秒后执行提醒...")
|
||||
await asyncio.sleep(self.delay)
|
||||
|
||||
logger.info(f"执行提醒任务: 给 {self.target_user_name} 发送关于 '{self.event_details}' 的提醒")
|
||||
|
||||
reminder_text = f"叮咚！这是 {self.creator_name} 让我准时提醒你的事情：\n\n{self.event_details}"
|
||||
|
||||
if self.is_group:
|
||||
# 在群聊中，构造 @ 消息段并发送
|
||||
group_id = self.stream_id.split("_")[-1] if "_" in self.stream_id else self.stream_id
|
||||
message_payload = [
|
||||
{"type": "at", "data": {"qq": self.target_user_id}},
|
||||
{"type": "text", "data": {"text": f" {reminder_text}"}},
|
||||
]
|
||||
await send_api.adapter_command_to_stream(
|
||||
action="send_group_msg",
|
||||
params={"group_id": group_id, "message": message_payload},
|
||||
stream_id=self.stream_id,
|
||||
)
|
||||
else:
|
||||
# 在私聊中，直接发送文本
|
||||
await send_api.text_to_stream(text=reminder_text, stream_id=self.stream_id)
|
||||
|
||||
logger.info(f"提醒任务 {self.task_name} 成功完成。")
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"执行提醒任务 {self.task_name} 时出错: {e}", exc_info=True)
|
||||
|
||||
|
||||
# =============================== Actions ===============================
|
||||
|
||||
|
||||
class RemindAction(BaseAction):
|
||||
"""一个能从对话中智能识别并设置定时提醒的动作。"""
|
||||
|
||||
# === 基本信息 ===
|
||||
action_name = "set_reminder"
|
||||
action_description = "根据用户的对话内容，智能地设置一个未来的提醒事项。"
|
||||
activation_type = ActionActivationType.LLM_JUDGE
|
||||
chat_type_allow = ChatType.ALL
|
||||
|
||||
# === LLM 判断与参数提取 ===
|
||||
llm_judge_prompt = """
|
||||
判断用户是否意图设置一个未来的提醒。
|
||||
- 必须包含明确的时间点或时间段（如“十分钟后”、“明天下午3点”、“周五”）。
|
||||
- 必须包含一个需要被提醒的事件。
|
||||
- 可能会包含需要提醒的特定人物。
|
||||
- 如果只是普通的聊天或询问时间，则不应触发。
|
||||
|
||||
示例:
|
||||
- "半小时后提醒我开会" -> 是
|
||||
- "明天下午三点叫张三来一下" -> 是
|
||||
- "别忘了周五把报告交了" -> 是
|
||||
- "现在几点了？" -> 否
|
||||
- "我明天下午有空" -> 否
|
||||
|
||||
请只回答"是"或"否"。
|
||||
"""
|
||||
action_parameters = {
|
||||
"user_name": "需要被提醒的人的称呼或名字，如果没有明确指定给某人，则默认为'自己'",
|
||||
"remind_time": "描述提醒时间的自然语言字符串，例如'十分钟后'或'明天下午3点'",
|
||||
"event_details": "需要提醒的具体事件内容",
|
||||
}
|
||||
action_require = [
|
||||
"当用户请求在未来的某个时间点提醒他/她或别人某件事时使用",
|
||||
"适用于包含明确时间信息和事件描述的对话",
|
||||
"例如：'10分钟后提醒我收快递'、'明天早上九点喊一下李四参加晨会'",
|
||||
]
|
||||
|
||||
async def execute(self) -> Tuple[bool, str]:
|
||||
"""执行设置提醒的动作"""
|
||||
user_name = self.action_data.get("user_name")
|
||||
remind_time_str = self.action_data.get("remind_time")
|
||||
event_details = self.action_data.get("event_details")
|
||||
|
||||
if not all([user_name, remind_time_str, event_details]):
|
||||
missing_params = [
|
||||
p
|
||||
for p, v in {
|
||||
"user_name": user_name,
|
||||
"remind_time": remind_time_str,
|
||||
"event_details": event_details,
|
||||
}.items()
|
||||
if not v
|
||||
]
|
||||
error_msg = f"缺少必要的提醒参数: {', '.join(missing_params)}"
|
||||
logger.warning(f"[ReminderPlugin] LLM未能提取完整参数: {error_msg}")
|
||||
return False, error_msg
|
||||
|
||||
# 1. 解析时间
|
||||
try:
|
||||
assert isinstance(remind_time_str, str)
|
||||
target_time = parse_datetime(remind_time_str, fuzzy=True)
|
||||
except Exception as e:
|
||||
logger.error(f"[ReminderPlugin] 无法解析时间字符串 '{remind_time_str}': {e}")
|
||||
await self.send_text(f"抱歉，我无法理解您说的时间 '{remind_time_str}'，提醒设置失败。")
|
||||
return False, f"无法解析时间 '{remind_time_str}'"
|
||||
|
||||
now = datetime.now()
|
||||
if target_time <= now:
|
||||
await self.send_text("提醒时间必须是一个未来的时间点哦，提醒设置失败。")
|
||||
return False, "提醒时间必须在未来"
|
||||
|
||||
delay_seconds = (target_time - now).total_seconds()
|
||||
|
||||
# 2. 解析用户
|
||||
person_manager = get_person_info_manager()
|
||||
user_id_to_remind = None
|
||||
user_name_to_remind = ""
|
||||
|
||||
assert isinstance(user_name, str)
|
||||
|
||||
if user_name.strip() in ["自己", "我", "me"]:
|
||||
user_id_to_remind = self.user_id
|
||||
user_name_to_remind = self.user_nickname
|
||||
else:
|
||||
user_info = await person_manager.get_person_info_by_name(user_name)
|
||||
if not user_info or not user_info.get("user_id"):
|
||||
logger.warning(f"[ReminderPlugin] 找不到名为 '{user_name}' 的用户")
|
||||
await self.send_text(f"抱歉，我的联系人里找不到叫做 '{user_name}' 的人，提醒设置失败。")
|
||||
return False, f"用户 '{user_name}' 不存在"
|
||||
user_id_to_remind = user_info.get("user_id")
|
||||
user_name_to_remind = user_name
|
||||
|
||||
# 3. 创建并调度异步任务
|
||||
try:
|
||||
assert user_id_to_remind is not None
|
||||
assert event_details is not None
|
||||
|
||||
reminder_task = ReminderTask(
|
||||
delay=delay_seconds,
|
||||
stream_id=self.chat_id,
|
||||
is_group=self.is_group,
|
||||
target_user_id=str(user_id_to_remind),
|
||||
target_user_name=str(user_name_to_remind),
|
||||
event_details=str(event_details),
|
||||
creator_name=str(self.user_nickname),
|
||||
)
|
||||
await async_task_manager.add_task(reminder_task)
|
||||
|
||||
# 4. 发送确认消息
|
||||
confirm_message = f"好的，我记下了。\n将在 {target_time.strftime('%Y-%m-%d %H:%M:%S')} 提醒 {user_name_to_remind}：\n{event_details}"
|
||||
await self.send_text(confirm_message)
|
||||
|
||||
return True, "提醒设置成功"
|
||||
except Exception as e:
|
||||
logger.error(f"[ReminderPlugin] 创建提醒任务时出错: {e}", exc_info=True)
|
||||
await self.send_text("抱歉，设置提醒时发生了一点内部错误。")
|
||||
return False, "设置提醒时发生内部错误"
|
||||
|
||||
|
||||
# =============================== Plugin ===============================
|
||||
|
||||
|
||||
@register_plugin
|
||||
class ReminderPlugin(BasePlugin):
|
||||
"""一个能从对话中智能识别并设置定时提醒的插件。"""
|
||||
|
||||
# --- 插件基础信息 ---
|
||||
plugin_name = "reminder_plugin"
|
||||
enable_plugin = True
|
||||
dependencies = []
|
||||
python_dependencies = []
|
||||
config_file_name = "config.toml"
|
||||
config_schema = {}
|
||||
|
||||
def get_plugin_components(self) -> List[Tuple[ActionInfo, Type[BaseAction]]]:
|
||||
"""注册插件的所有功能组件。"""
|
||||
return [(RemindAction.get_action_info(), RemindAction)]
|
||||
@@ -53,8 +53,8 @@ price_out = 8.0 # 输出价格（用于API调用统计，单
|
||||
#use_anti_truncation = true # [可选] 启用反截断功能。当模型输出不完整时，系统会自动重试。建议只为有需要的模型（如Gemini）开启。
|
||||
|
||||
[[models]]
|
||||
model_identifier = "deepseek-ai/DeepSeek-V3"
|
||||
name = "siliconflow-deepseek-v3"
|
||||
model_identifier = "deepseek-ai/DeepSeek-V3.1-Terminus"
|
||||
name = "siliconflow-deepseek-ai/DeepSeek-V3.1-Terminus"
|
||||
api_provider = "SiliconFlow"
|
||||
price_in = 2.0
|
||||
price_out = 8.0
|
||||
@@ -122,7 +122,7 @@ price_in = 4.0
|
||||
price_out = 16.0
|
||||
|
||||
[model_task_config.utils] # 在麦麦的一些组件中使用的模型，例如表情包模块，取名模块，关系模块，是麦麦必须的模型
|
||||
model_list = ["siliconflow-deepseek-v3"] # 使用的模型列表，每个子项对应上面的模型名称(name)
|
||||
model_list = ["siliconflow-deepseek-ai/DeepSeek-V3.1-Terminus"] # 使用的模型列表，每个子项对应上面的模型名称(name)
|
||||
temperature = 0.2 # 模型温度，新V3建议0.1-0.3
|
||||
max_tokens = 800 # 最大输出token数
|
||||
#concurrency_count = 2 # 并发请求数量，默认为1（不并发），设置为2或更高启用并发
|
||||
@@ -133,28 +133,28 @@ temperature = 0.7
|
||||
max_tokens = 800
|
||||
|
||||
[model_task_config.replyer] # 首要回复模型，还用于表达器和表达方式学习
|
||||
model_list = ["siliconflow-deepseek-v3"]
|
||||
model_list = ["siliconflow-deepseek-ai/DeepSeek-V3.1-Terminus"]
|
||||
temperature = 0.2 # 模型温度，新V3建议0.1-0.3
|
||||
max_tokens = 800
|
||||
|
||||
[model_task_config.planner] #决策：负责决定麦麦该做什么的模型
|
||||
model_list = ["siliconflow-deepseek-v3"]
|
||||
model_list = ["siliconflow-deepseek-ai/DeepSeek-V3.1-Terminus"]
|
||||
temperature = 0.3
|
||||
max_tokens = 800
|
||||
|
||||
|
||||
[model_task_config.emotion] #负责麦麦的情绪变化
|
||||
model_list = ["siliconflow-deepseek-v3"]
|
||||
model_list = ["siliconflow-deepseek-ai/DeepSeek-V3.1-Terminus"]
|
||||
temperature = 0.3
|
||||
max_tokens = 800
|
||||
|
||||
[model_task_config.mood] #负责麦麦的心情变化
|
||||
model_list = ["siliconflow-deepseek-v3"]
|
||||
model_list = ["siliconflow-deepseek-ai/DeepSeek-V3.1-Terminus"]
|
||||
temperature = 0.3
|
||||
max_tokens = 800
|
||||
|
||||
[model_task_config.maizone] # maizone模型
|
||||
model_list = ["siliconflow-deepseek-v3"]
|
||||
model_list = ["siliconflow-deepseek-ai/DeepSeek-V3.1-Terminus"]
|
||||
temperature = 0.7
|
||||
max_tokens = 800
|
||||
|
||||
@@ -181,7 +181,7 @@ temperature = 0.7
|
||||
max_tokens = 800
|
||||
|
||||
[model_task_config.schedule_generator]#日程表生成模型
|
||||
model_list = ["siliconflow-deepseek-v3"]
|
||||
model_list = ["siliconflow-deepseek-ai/DeepSeek-V3.1-Terminus"]
|
||||
temperature = 0.7
|
||||
max_tokens = 1000
|
||||
|
||||
@@ -191,12 +191,12 @@ temperature = 0.1 # 低温度确保检测结果稳定
|
||||
max_tokens = 200 # 检测结果不需要太长的输出
|
||||
|
||||
[model_task_config.monthly_plan_generator] # 月层计划生成模型
|
||||
model_list = ["siliconflow-deepseek-v3"]
|
||||
model_list = ["siliconflow-deepseek-ai/DeepSeek-V3.1-Terminus"]
|
||||
temperature = 0.7
|
||||
max_tokens = 1000
|
||||
|
||||
[model_task_config.relationship_tracker] # 用户关系追踪模型
|
||||
model_list = ["siliconflow-deepseek-v3"]
|
||||
model_list = ["siliconflow-deepseek-ai/DeepSeek-V3.1-Terminus"]
|
||||
temperature = 0.7
|
||||
max_tokens = 1000
|
||||
|
||||
@@ -209,12 +209,12 @@ model_list = ["bge-m3"]
|
||||
#------------LPMM知识库模型------------
|
||||
|
||||
[model_task_config.lpmm_entity_extract] # 实体提取模型
|
||||
model_list = ["siliconflow-deepseek-v3"]
|
||||
model_list = ["siliconflow-deepseek-ai/DeepSeek-V3.1-Terminus"]
|
||||
temperature = 0.2
|
||||
max_tokens = 800
|
||||
|
||||
[model_task_config.lpmm_rdf_build] # RDF构建模型
|
||||
model_list = ["siliconflow-deepseek-v3"]
|
||||
model_list = ["siliconflow-deepseek-ai/DeepSeek-V3.1-Terminus"]
|
||||
temperature = 0.2
|
||||
max_tokens = 800
|
||||
|
||||
|
||||
Reference in New Issue
Block a user