1a24233b86
本次更新对系统核心处理流程和插件功能进行了重要升级，主要包含以下两方面：
1. **消息处理异步化**:
- 在 `main.py` 中引入了 `asyncio.create_task` 机制，将每条消息的处理过程包装成一个独立的后台任务。
- 这解决了长时间运行的AI或插件操作可能阻塞主事件循环的问题，显著提升了机器人的响应速度和系统稳定性。
- 为后台任务添加了完成回调，现在可以详细地记录每个消息处理任务的成功、失败或取消状态及其耗时，便于监控和调试。
2. **`set_emoji_like` 插件智能化**:
- 为 `set_emoji_like` 插件增加了LLM驱动的表情选择功能。当动作指令未指定具体表情时，插件会自动构建包含聊天上下文、情绪和人设的提示，请求LLM选择一个最合适的表情进行回应。
- 为支持此功能，对AFC规划器的提示词进行了优化，为LLM提供了更清晰的参数示例和规则，提高了动作生成的准确性。
此外，为了统一日志规范，将 `[所见]` 消息接收日志集中到 `bot.py` 中，确保在任何过滤逻辑执行前记录所有收到的消息，并移除了插件中重复的日志。
274 lines
11 KiB
Python
274 lines
11 KiB
Python
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 src.plugin_system.apis import send_api
|
||
from .qq_emoji_list import qq_face
|
||
from src.plugin_system.base.component_types import ChatType
|
||
from src.plugin_system.apis import llm_api
|
||
from src.config.config import model_config, global_config
|
||
from src.chat.utils.chat_message_builder import build_readable_messages
|
||
|
||
logger = get_logger("set_emoji_like_plugin")
|
||
|
||
|
||
async 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.info("未提供表情，将由LLM决定")
|
||
try:
|
||
emoji_input = await self.ask_llm_for_emoji()
|
||
if not emoji_input:
|
||
logger.error("LLM未能选择表情")
|
||
return False, "LLM未能选择表情"
|
||
except Exception as e:
|
||
logger.error(f"请求LLM选择表情时出错: {e}")
|
||
return False, f"请求LLM选择表情时出错: {e}"
|
||
|
||
logger.info(f"设置表情回应: {emoji_input}, 是否设置: {set_like}")
|
||
|
||
emoji_id = await 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发送贴表情命令
|
||
response = await send_api.adapter_command_to_stream(
|
||
action="set_msg_emoji_like",
|
||
params={"message_id": message_id, "emoji_id": emoji_id, "set": set_like},
|
||
stream_id=self.chat_stream.stream_id if self.chat_stream else None,
|
||
timeout=30.0,
|
||
storage_message=False,
|
||
)
|
||
|
||
if response["status"] == "ok":
|
||
logger.info(f"设置表情回应成功: {response}")
|
||
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, f"成功设置表情回应: {response.get('message', '成功')}"
|
||
else:
|
||
error_msg = response.get("message", "未知错误")
|
||
logger.error(f"设置表情回应失败: {error_msg}")
|
||
await self.store_action_info(
|
||
action_build_into_prompt=True,
|
||
action_prompt_display=f"执行了set_emoji_like动作：{self.action_name},失败: {error_msg}",
|
||
action_done=False,
|
||
)
|
||
return False, f"设置表情回应失败: {error_msg}"
|
||
|
||
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}"
|
||
|
||
async def ask_llm_for_emoji(self) -> str | None:
|
||
"""构建Prompt并请求LLM选择一个表情"""
|
||
from src.mood.mood_manager import mood_manager
|
||
from src.individuality.individuality import get_individuality
|
||
from src.chat.message_manager.message_manager import message_manager
|
||
|
||
# 1. 获取上下文信息
|
||
stream_context = message_manager.stream_contexts.get(self.chat_stream.stream_id)
|
||
if not stream_context:
|
||
logger.error(f"无法为 stream_id '{self.chat_stream.stream_id}' 找到 StreamContext")
|
||
return None
|
||
|
||
history_messages = stream_context.get_latest_messages(20)
|
||
chat_context = build_readable_messages(
|
||
[msg.flatten() for msg in history_messages],
|
||
replace_bot_name=True,
|
||
timestamp_mode="normal_no_YMD",
|
||
truncate=True,
|
||
)
|
||
|
||
target_message_content = self.action_message.get("processed_plain_text", "")
|
||
mood = mood_manager.get_mood_by_chat_id(self.chat_stream.stream_id).mood_state
|
||
identity = await get_individuality().get_personality_block()
|
||
|
||
# 2. 构建Prompt
|
||
emoji_options_str = ", ".join(self.emoji_options)
|
||
bot_name = global_config.bot.nickname or "爱莉希雅"
|
||
prompt = f"""
|
||
# 指令：选择一个最合适的表情来回应消息
|
||
|
||
## 场景描述
|
||
你的名字是“{bot_name}”。
|
||
{identity}
|
||
你现在的心情是：{mood}
|
||
|
||
## 聊天上下文
|
||
下面是最近的聊天记录：
|
||
{chat_context}
|
||
|
||
## 你的任务
|
||
你需要针对下面的这条消息，选择一个最合适的表情来“贴”在上面，以表达你的心情和回应。
|
||
目标消息："{target_message_content}"
|
||
|
||
## 表情选项
|
||
请从以下表情中，选择一个最能代表你此刻心情的表情。你只能选择一个，并直接返回它的【名称】。
|
||
{emoji_options_str}
|
||
|
||
## 输出要求
|
||
直接输出你选择的表情【名称】，不要添加任何多余的文字、解释或标点符号。
|
||
|
||
你选择的表情名称是：
|
||
"""
|
||
|
||
# 3. 调用LLM
|
||
success, response, _, _ = await llm_api.generate_with_model(
|
||
prompt, model_config.model_task_config.tool_executor
|
||
)
|
||
|
||
if success and response:
|
||
# 清理LLM返回的可能存在的额外字符
|
||
cleaned_response = re.sub(r"[\[\]\'\"]", "", response).strip()
|
||
logger.info(f"LLM选择了表情: '{cleaned_response}'")
|
||
return cleaned_response
|
||
|
||
return None
|
||
|
||
# ===== 插件注册 =====
|
||
@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 []
|