feat:将no_reply内置

This commit is contained in:
SengokuCola
2025-08-10 22:12:56 +08:00
parent 69a855df8d
commit 9e9e79694a
6 changed files with 181 additions and 275 deletions

View File

@@ -0,0 +1,157 @@
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):
"""表情动作 - 发送表情包"""
# 激活设置
if global_config.emoji.emoji_activate_type == "llm":
activation_type = ActionActivationType.LLM_JUDGE
random_activation_probability = 0
else:
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)}"