MaiBot/src/plugins/built_in/emoji_plugin/emoji.py

153 lines
6.3 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters!

This file contains ambiguous Unicode characters that may be confused with others in your current locale. If your use case is intentional and legitimate, you can safely ignore this warning. Use the Escape button to highlight these characters.

import random
from typing import Tuple
# 导入新插件系统
from src.plugin_system import BaseAction, ActionActivationType, ChatMode
# 导入依赖的系统组件
from src.common.logger import get_logger
# 导入API模块 - 标准Python包方式
from src.plugin_system.apis import emoji_api, llm_api, message_api
# NoReplyAction已集成到heartFC_chat.py中不再需要导入
from src.config.config import global_config
logger = get_logger("emoji")
class EmojiAction(BaseAction):
"""表情动作 - 发送表情包"""
activation_type = ActionActivationType.RANDOM
random_activation_probability = global_config.emoji.emoji_chance
mode_enable = ChatMode.ALL
parallel_action = True
# 动作基本信息
action_name = "emoji"
action_description = "发送表情包辅助表达情绪"
# LLM判断提示词
llm_judge_prompt = """
判定是否需要使用表情动作的条件:
1. 用户明确要求使用表情包
2. 这是一个适合表达强烈情绪的场合
3. 不要发送太多表情包,如果你已经发送过多个表情包则回答""
请回答""""
"""
# 动作参数定义
action_parameters = {}
# 动作使用场景
action_require = [
"发送表情包辅助表达情绪",
"表达情绪时可以选择使用",
"不要连续发送,如果你已经发过[表情包],就不要选择此动作",
]
# 关联类型
associated_types = ["emoji"]
async def execute(self) -> Tuple[bool, str]:
# sourcery skip: assign-if-exp, introduce-default-else, swap-if-else-branches, use-named-expression
"""执行表情动作"""
logger.info(f"{self.log_prefix} 决定发送表情")
try:
# 1. 获取发送表情的原因
reason = self.action_data.get("reason", "表达当前情绪")
logger.info(f"{self.log_prefix} 发送表情原因: {reason}")
# 2. 随机获取20个表情包
sampled_emojis = await emoji_api.get_random(30)
if not sampled_emojis:
logger.warning(f"{self.log_prefix} 无法获取随机表情包")
return False, "无法获取随机表情包"
# 3. 准备情感数据
emotion_map = {}
for b64, desc, emo in sampled_emojis:
if emo not in emotion_map:
emotion_map[emo] = []
emotion_map[emo].append((b64, desc))
available_emotions = list(emotion_map.keys())
if not available_emotions:
logger.warning(f"{self.log_prefix} 获取到的表情包均无情感标签, 将随机发送")
emoji_base64, emoji_description, _ = random.choice(sampled_emojis)
else:
# 获取最近的5条消息内容用于判断
recent_messages = message_api.get_recent_messages(chat_id=self.chat_id, limit=5)
messages_text = ""
if recent_messages:
# 使用message_api构建可读的消息字符串
messages_text = message_api.build_readable_messages(
messages=recent_messages,
timestamp_mode="normal_no_YMD",
truncate=False,
show_actions=False,
)
# 4. 构建prompt让LLM选择情感
prompt = f"""
你是一个正在进行聊天的网友,你需要根据一个理由和最近的聊天记录,从一个情感标签列表中选择最匹配的一个。
这是最近的聊天记录:
{messages_text}
这是理由:“{reason}
这里是可用的情感标签:{available_emotions}
请直接返回最匹配的那个情感标签,不要进行任何解释或添加其他多余的文字。
"""
if global_config.debug.show_prompt:
logger.info(f"{self.log_prefix} 生成的LLM Prompt: {prompt}")
else:
logger.debug(f"{self.log_prefix} 生成的LLM Prompt: {prompt}")
# 5. 调用LLM
models = llm_api.get_available_models()
chat_model_config = models.get("utils_small") # 使用字典访问方式
if not chat_model_config:
logger.error(f"{self.log_prefix} 未找到'utils_small'模型配置无法调用LLM")
return False, "未找到'utils_small'模型配置"
success, chosen_emotion, _, _ = await llm_api.generate_with_model(
prompt, model_config=chat_model_config, request_type="emoji"
)
if not success:
logger.error(f"{self.log_prefix} LLM调用失败: {chosen_emotion}")
return False, f"LLM调用失败: {chosen_emotion}"
chosen_emotion = chosen_emotion.strip().replace('"', "").replace("'", "")
logger.info(f"{self.log_prefix} LLM选择的情感: {chosen_emotion}")
# 6. 根据选择的情感匹配表情包
if chosen_emotion in emotion_map:
emoji_base64, emoji_description = random.choice(emotion_map[chosen_emotion])
logger.info(f"{self.log_prefix} 找到匹配情感 '{chosen_emotion}' 的表情包: {emoji_description}")
else:
logger.warning(
f"{self.log_prefix} LLM选择的情感 '{chosen_emotion}' 不在可用列表中, 将随机选择一个表情包"
)
emoji_base64, emoji_description, _ = random.choice(sampled_emojis)
# 7. 发送表情包
success = await self.send_emoji(emoji_base64)
if not success:
logger.error(f"{self.log_prefix} 表情包发送失败")
return False, "表情包发送失败"
# no_reply计数器现在由heartFC_chat.py统一管理无需在此重置
return True, f"发送表情包: {emoji_description}"
except Exception as e:
logger.error(f"{self.log_prefix} 表情动作执行失败: {e}", exc_info=True)
return False, f"表情发送失败: {str(e)}"