gardel/Mofox-Core
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(emoji): 重构表情包识别、描述与缓存系统

Browse Source 
本次更新全面重构了表情包处理逻辑，引入了包含“精炼描述”、“关键词”和“详细分析”的三层描述系统，以提升识别的深度与广度。同时，统一并修复了缓存机制，解决了旧缓存数据格式不兼容及新内容无法被缓存的问题，并通过增加校验逻辑增强了系统的健壮性。

注意：更新后，用户必须手动清空 emojis、images、image_descriptions 三个数据库表，并将 data/emoji_registed/ 目录下的所有文件移至 data/emoji/ 目录，然后重启程序，以确保所有表情包能被新系统正确地重新识别
This commit is contained in:
tt-P607
2025-10-21 15:57:52 +08:00
parent 82c288db6e
commit b8e49343c1
3 changed files with 131 additions and 161 deletions
Download Patch File Download Diff File Expand all files Collapse all files

64
src/chat/emoji_system/emoji_manager.py
View File

@@ -983,7 +983,12 @@ class EmojiManager:
prompt, image_base64, image_format, temperature=0.3, max_tokens=600
)
# 4. 内容审核，确保表情包符合规定
# 4. 检查VLM描述是否有效
if not description or not description.strip():
logger.warning("VLM未能生成有效的详细描述，中止处理。")
return "", []
# 5. 内容审核，确保表情包符合规定
if global_config.emoji.content_filtration:
prompt = f"""
请根据以下标准审核这个表情包：
@@ -1000,23 +1005,28 @@ class EmojiManager:
logger.warning(f"表情包审核未通过，内容: {description[:50]}...")
return "", []
# 5. 基于VLM的详细描述，调用LLM提炼情感关键词
# 6. 基于VLM的详细描述，提炼“精炼关键词”
emotions = []
emotions_text = ""
if global_config.emoji.enable_emotion_analysis:
logger.info("[情感分析] 开始提炼表情包的情感关键词")
logger.info("[情感分析] 开始提炼表情包的“精炼关键词”")
emotion_prompt = f"""
你是一个互联网“梗”学家和情感分析师。
这里有一份关于某个表情包的详细描述：
---
{description}
---
请你基于这份描述，提炼出这个表情包最核心的含义和适用场景。
请你基于这份描述，提炼出这个表情包最核心的、可用于检索的关键词。
你的任务是：
1. 分析并总结出3到5个最能代表这个表情包的关键词或短语。
2. 这些关键词应该非常凝练，比如“表达无语”、“有点小得意”、“求夸奖”、“猫猫疑惑”等。
3. 每个关键词不要超过15个字。
4. 请直接输出这些关键词，并用逗号分隔，不要添加任何其他解释。
1. **全面分析**：仔细阅读描述，理解表情包的全部细节，包括**图中文字、人物表情、动作、情绪、构图**等。
2. **提炼关键词**：总结出 5 到 8 个最能代表这个表情包的关键词或短语。
3. **关键词要求**：
- 必须包含表情包中的**核心文字**（如果有）。
- 必须描述核心的**表情和动作**（例如：“歪头杀”、“摊手”、“无奈苦笑”）。
- 必须体现核心的**情绪和氛围**（例如：“悲伤”、“喜悦”、“沙雕”、“阴阳怪气”）。
- 可以包含**核心主体或构图特点**（例如：“猫猫头”、“大头贴”、“模糊画质”）。
4. **格式要求**：请直接输出这些关键词，并用**逗号**分隔，不要添加任何其他解释或编号。
"""
emotions_text, _ = await self.llm_emotion_judge.generate_response_async(
emotion_prompt, temperature=0.6, max_tokens=150
@@ -1025,9 +1035,41 @@ class EmojiManager:
else:
logger.info("[情感分析] 表情包感情关键词二次识别已禁用，跳过此步骤")
# 6. 格式化最终的描述，并返回结果
final_description = f"表情包，关键词：[{'，'.join(emotions)}]。详细描述：{description}"
logger.info(f"[注册分析] VLM描述: {description} -> 提炼出的情感标签: {emotions}")
# 7. 基于详细描述和关键词，生成“精炼自然语言描述”
refined_description = ""
if emotions: # 只有在成功提取关键词后才进行精炼
logger.info("[自然语言精炼] 开始生成“点睛之笔”的自然语言描述")
refine_prompt = f"""
你的任务是为一张表情包写一句简洁、自然的描述，就像你在向朋友解释这张图是什么意思一样。
这里是关于这个表情包的分析信息：
# 详细描述
{description}
# 核心关键词
{emotions_text}
# 你的任务
请结合以上信息，用一句**一针见血**的自然语言，概括出这个表情包的核心内容。
# 规则 (非常重要！)
1. **必须包含图中的核心文字**。
2. **必须描述出主角的核心表情和动作**。
3. **风格要求**：简单、直接、口语化，就像一个普通人看到这张图后的第一反应。
4. **输出格式**：**请直接返回这句描述，不要添加任何前缀、标题或多余的解释。**
"""
refined_description, _ = await self.llm_emotion_judge.generate_response_async(
refine_prompt, temperature=0.7, max_tokens=100
)
refined_description = refined_description.strip()
# 8. 格式化最终的描述，并返回结果
final_description = (
f"{refined_description} Keywords: [{','.join(emotions)}] Desc: {description}"
)
logger.info(f"[注册分析] VLM描述: {description}")
logger.info(f"[注册分析] 提炼出的情感标签: {emotions}")
logger.info(f"[注册分析] 精炼后的自然语言描述: {refined_description}")
return final_description, emotions

167
src/chat/utils/utils_image.py
View File

@@ -153,152 +153,65 @@ class ImageManager:
return f"[表情包：{tag_str}]"
async def get_emoji_description(self, image_base64: str) -> str:
"""获取表情包描述，优先使用Emoji表中的缓存数据"""
"""获取表情包描述，统一使用EmojiManager中的逻辑进行处理和缓存"""
try:
# 计算图片哈希
# 确保base64字符串只包含ASCII字符
from src.chat.emoji_system.emoji_manager import get_emoji_manager
emoji_manager = get_emoji_manager()
# 1. 计算图片哈希
if isinstance(image_base64, str):
image_base64 = image_base64.encode("ascii", errors="ignore").decode("ascii")
image_bytes = base64.b64decode(image_base64)
image_hash = hashlib.md5(image_bytes).hexdigest()
image_format = Image.open(io.BytesIO(image_bytes)).format.lower() # type: ignore
# 优先使用EmojiManager查询已注册表情包的描述
try:
from src.chat.emoji_system.emoji_manager import get_emoji_manager
# 2. 优先查询已注册表情的缓存（Emoji表）
if full_description := await emoji_manager.get_emoji_description_by_hash(image_hash):
logger.info("[缓存命中] 使用已注册表情包(Emoji表)的完整描述")
refined_part = full_description.split(" Keywords:")[0]
return f"[表情包：{refined_part}]"
emoji_manager = get_emoji_manager()
tags = await emoji_manager.get_emoji_tag_by_hash(image_hash)
if tags:
tag_str = ",".join(tags)
logger.info(f"[缓存命中] 使用已注册表情包描述: {tag_str}...")
return f"[表情包：{tag_str}]"
except Exception as e:
logger.debug(f"查询EmojiManager时出错: {e}")
# 查询ImageDescriptions表的缓存描述
# 3. 查询通用图片描述缓存（ImageDescriptions表）
if cached_description := await self._get_description_from_db(image_hash, "emoji"):
logger.info(f"[缓存命中] 使用ImageDescriptions表中的描述: {cached_description}...")
return f"[表情包：{cached_description}]"
logger.info(f"[缓存命中] 使用通用图片缓存(ImageDescriptions表)中的描述")
refined_part = cached_description.split(" Keywords:")[0]
return f"[表情包：{refined_part}]"
# === 二步走识别流程 ===
# 4. 如果都未命中，则调用新逻辑生成描述
logger.info(f"[新表情识别] 表情包未注册且无缓存 (Hash: {image_hash[:8]}...)，调用新逻辑生成描述")
full_description, emotions = await emoji_manager.build_emoji_description(image_base64)
# 第一步：VLM视觉分析 - 生成详细描述
if image_format in ["gif", "GIF"]:
image_base64_processed = self.transform_gif(image_base64)
if image_base64_processed is None:
logger.warning("GIF转换失败，无法获取描述")
return "[表情包(GIF处理失败)]"
vlm_prompt = "这是一个动态图表情包，每一张图代表了动态图的某一帧，黑色背景代表透明，描述一下表情包表达的情感和内容，描述细节，从互联网梗,meme的角度去分析"
detailed_description, _ = await self.vlm.generate_response_for_image(
vlm_prompt, image_base64_processed, "jpeg", temperature=0.4, max_tokens=300
)
else:
vlm_prompt = (
"这是一个表情包，请详细描述一下表情包所表达的情感和内容，描述细节，从互联网梗,meme的角度去分析"
)
detailed_description, _ = await self.vlm.generate_response_for_image(
vlm_prompt, image_base64, image_format, temperature=0.4, max_tokens=300
)
if detailed_description is None:
logger.warning("VLM未能生成表情包详细描述")
return "[表情包(VLM描述生成失败)]"
# 第二步：LLM情感分析 - 基于详细描述生成简短的情感标签
emotion_prompt = f"""
请你基于这个表情包的详细描述，提取出最核心的情感含义，用1-2个词概括。
详细描述：'{detailed_description}'
要求：
1. 只输出1-2个最核心的情感词汇
2. 从互联网梗、meme的角度理解
3. 输出简短精准，不要解释
4. 如果有多个词用逗号分隔
"""
# 使用较低温度确保输出稳定
emotion_llm = LLMRequest(model_set=model_config.model_task_config.utils, request_type="emoji")
emotion_result, _ = await emotion_llm.generate_response_async(
emotion_prompt, temperature=0.3, max_tokens=50
)
if emotion_result is None:
logger.warning("LLM未能生成情感标签，使用详细描述的前几个词")
# 降级处理：从详细描述中提取关键词
import rjieba
words = list(rjieba.cut(detailed_description))
emotion_result = "，".join(words[:2]) if len(words) >= 2 else (words[0] if words else "表情")
# 处理情感结果，取前1-2个最重要的标签
emotions = [e.strip() for e in emotion_result.replace("，", ",").split(",") if e.strip()]
final_emotion = emotions[0] if emotions else "表情"
# 如果有第二个情感且不重复，也包含进来
if len(emotions) > 1 and emotions[1] != emotions[0]:
final_emotion = f"{emotions[0]}，{emotions[1]}"
logger.info(f"[emoji识别] 详细描述: {detailed_description}... -> 情感标签: {final_emotion}")
if cached_description := await self._get_description_from_db(image_hash, "emoji"):
logger.warning(f"虽然生成了描述，但是找到缓存表情包描述: {cached_description}")
return f"[表情包：{cached_description}]"
# 只有在开启“偷表情包”功能时，才将接收到的表情包保存到待注册目录
if not full_description:
logger.warning("未能通过新逻辑生成有效描述")
return "[表情包(描述生成失败)]"
# 4. (可选) 如果启用了“偷表情包”，则将图片和完整描述存入待注册区
if global_config.emoji.steal_emoji:
logger.debug(f"偷取表情包功能已开启，保存表情包: {image_hash}")
current_timestamp = time.time()
filename = f"{int(current_timestamp)}_{image_hash[:8]}.{image_format}"
emoji_dir = os.path.join(self.IMAGE_DIR, "emoji")
os.makedirs(emoji_dir, exist_ok=True)
file_path = os.path.join(emoji_dir, filename)
logger.debug(f"偷取表情包功能已开启，保存待注册表情包: {image_hash}")
try:
# 保存文件
image_format = (Image.open(io.BytesIO(image_bytes)).format or "jpeg").lower()
current_timestamp = time.time()
filename = f"{int(current_timestamp)}_{image_hash[:8]}.{image_format}"
emoji_dir = os.path.join(self.IMAGE_DIR, "emoji")
os.makedirs(emoji_dir, exist_ok=True)
file_path = os.path.join(emoji_dir, filename)
with open(file_path, "wb") as f:
f.write(image_bytes)
# 保存到数据库 (Images表) - 包含详细描述用于可能的注册流程
try:
from src.common.database.sqlalchemy_models import get_db_session
async with get_db_session() as session:
existing_img = (
await session.execute(
select(Images).where(and_(Images.emoji_hash == image_hash, Images.type == "emoji"))
)
).scalar()
if existing_img:
existing_img.path = file_path
existing_img.description = detailed_description # 保存详细描述
existing_img.timestamp = current_timestamp
else:
new_img = Images(
emoji_hash=image_hash,
path=file_path,
type="emoji",
description=detailed_description, # 保存详细描述
timestamp=current_timestamp,
)
session.add(new_img)
await session.commit()
except Exception as e:
logger.error(f"保存到Images表失败: {e!s}")
logger.info(f"新表情包已保存至待注册目录: {file_path}")
except Exception as e:
logger.error(f"保存表情包文件或元数据失败: {e!s}")
else:
logger.debug("偷取表情包功能已关闭，跳过保存。")
logger.error(f"保存待注册表情包文件失败: {e!s}")
# 保存最终的情感标签到缓存 (ImageDescriptions表)
await self._save_description_to_db(image_hash, final_emotion, "emoji")
# 5. 将新生成的完整描述存入通用缓存（ImageDescriptions表）
await self._save_description_to_db(image_hash, full_description, "emoji")
logger.info(f"新生成的表情包描述已存入通用缓存 (Hash: {image_hash[:8]}...)")
return f"[表情包：{final_emotion}]"
# 6. 返回新生成的描述中用于显示的“精炼描述”部分
refined_part = full_description.split(" Keywords:")[0]
return f"[表情包：{refined_part}]"
except Exception as e:
logger.error(f"获取表情包描述失败: {e!s}")
logger.error(f"获取表情包描述失败: {e!s}", exc_info=True)
return "[表情包(处理失败)]"
async def get_image_description(self, image_base64: str) -> str:

61
src/plugins/built_in/core_actions/emoji.py
View File

@@ -1,4 +1,5 @@
import random
import re
from src.chat.emoji_system.emoji_history import add_emoji_to_history, get_recent_emojis
from src.chat.emoji_system.emoji_manager import MaiEmoji, get_emoji_manager
@@ -223,7 +224,13 @@ class EmojiAction(BaseAction):
)
# 准备表情描述列表
emoji_descriptions = [desc for _, desc in all_emojis_data]
# 提取精炼描述和关键词用于LLM选择
def extract_refined_info(full_desc: str) -> str:
# 新格式: [精炼描述] Keywords: [关键词] Desc: [详细描述]
# 我们只需要 Desc: 之前的部分
return full_desc.split(" Desc:")[0].strip()
emoji_descriptions = [extract_refined_info(desc) for _, desc in all_emojis_data]
# 构建prompt让LLM选择描述
prompt = f"""
@@ -256,32 +263,40 @@ class EmojiAction(BaseAction):
chosen_emotion = chosen_description # 在描述模式下，用描述作为情感标签
logger.info(f"{self.log_prefix} LLM选择的描述: {chosen_description}")
# 简单关键词匹配
matched_emoji = next(
(
item
for item in all_emojis_data
if chosen_description.lower() in item[1].lower()
or item[1].lower() in chosen_description.lower()
),
None,
)
# 优化匹配逻辑：优先在精炼描述中精确匹配，然后进行关键词匹配
def extract_refined_info(full_desc: str) -> str:
return full_desc.split(" Desc:")[0].strip()
# 如果包含匹配失败，尝试关键词匹配
if not matched_emoji:
keywords = ["惊讶", "困惑", "呆滞", "震惊", "懵", "无语", "萌", "可爱"]
for keyword in keywords:
if keyword in chosen_description:
for item in all_emojis_data:
if any(k in item[1] for k in ["呆", "萌", "惊", "困惑", "无语"]):
matched_emoji = item
break
if matched_emoji:
break
# 1. 尝试在精炼描述中找到最匹配的表情
# 我们假设LLM返回的是精炼描述的一部分或全部
matched_emoji = None
best_match_score = 0
for item in all_emojis_data:
refined_info = extract_refined_info(item[1])
# 计算一个简单的匹配分数
score = 0
if chosen_description.lower() in refined_info.lower():
score += 2 # 包含匹配
if refined_info.lower() in chosen_description.lower():
score += 2 # 包含匹配
# 关键词匹配加分
chosen_keywords = re.findall(r'\w+', chosen_description.lower())
item_keywords = re.findall(r'\[(.*?)\]', refined_info)
if item_keywords:
item_keywords_set = {k.strip().lower() for k in item_keywords[0].split(',')}
for kw in chosen_keywords:
if kw in item_keywords_set:
score += 1
if score > best_match_score:
best_match_score = score
matched_emoji = item
if matched_emoji:
emoji_base64, emoji_description = matched_emoji
logger.info(f"{self.log_prefix} 找到匹配描述的表情包: {emoji_description}")
logger.info(f"{self.log_prefix} 找到匹配描述的表情包: {extract_refined_info(emoji_description)}")
else:
logger.warning(f"{self.log_prefix} LLM选择的描述无法匹配任何表情包, 将随机选择")
emoji_base64, emoji_description = random.choice(all_emojis_data)