better:新增log前缀映射,优化emoji的显示,加强了emoji的识别

This commit is contained in:
SengokuCola
2025-07-24 04:54:47 +08:00
parent 6c9c94d719
commit 6c91b95314
5 changed files with 159 additions and 51 deletions

View File

@@ -8,6 +8,25 @@
from src.plugin_system.apis import emoji_api from src.plugin_system.apis import emoji_api
``` ```
## 🆕 **二步走识别优化**
从最新版本开始,表情包识别系统采用了**二步走识别 + 智能缓存**的优化方案:
### **收到表情包时的识别流程**
1. **第一步**VLM视觉分析 - 生成详细描述
2. **第二步**LLM情感分析 - 基于详细描述提取核心情感标签
3. **缓存机制**将情感标签缓存到数据库详细描述保存到Images表
### **注册表情包时的优化**
- **智能复用**优先从Images表获取已有的详细描述
- **避免重复**如果表情包之前被收到过跳过VLM调用
- **性能提升**减少不必要的AI调用降低延时和成本
### **缓存策略**
- **ImageDescriptions表**:缓存最终的情感标签(用于快速显示)
- **Images表**:保存详细描述(用于注册时复用)
- **双重检查**:防止并发情况下的重复生成
## 主要功能 ## 主要功能
### 1. 表情包获取 ### 1. 表情包获取

View File

@@ -836,7 +836,7 @@ class EmojiManager:
return False return False
async def build_emoji_description(self, image_base64: str) -> Tuple[str, List[str]]: async def build_emoji_description(self, image_base64: str) -> Tuple[str, List[str]]:
"""获取表情包描述和情感列表 """获取表情包描述和情感列表,优化复用已有描述
Args: Args:
image_base64: 图片的base64编码 image_base64: 图片的base64编码
@@ -850,18 +850,35 @@ class EmojiManager:
if isinstance(image_base64, str): if isinstance(image_base64, str):
image_base64 = image_base64.encode("ascii", errors="ignore").decode("ascii") image_base64 = image_base64.encode("ascii", errors="ignore").decode("ascii")
image_bytes = base64.b64decode(image_base64) 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 image_format = Image.open(io.BytesIO(image_bytes)).format.lower() # type: ignore
# 调用AI获取描述 # 尝试从Images表获取已有的详细描述可能在收到表情包时已生成
if image_format == "gif" or image_format == "GIF": existing_description = None
image_base64 = get_image_manager().transform_gif(image_base64) # type: ignore try:
if not image_base64: from src.common.database.database_model import Images
raise RuntimeError("GIF表情包转换失败") existing_image = Images.get_or_none((Images.emoji_hash == image_hash) & (Images.type == "emoji"))
prompt = "这是一个动态图表情包,每一张图代表了动态图的某一帧,黑色背景代表透明,描述一下表情包表达的情感和内容,描述细节,从互联网梗,meme的角度去分析" if existing_image and existing_image.description:
description, _ = await self.vlm.generate_response_for_image(prompt, image_base64, "jpg") existing_description = existing_image.description
logger.info(f"[复用描述] 找到已有详细描述: {existing_description[:50]}...")
except Exception as e:
logger.debug(f"查询已有描述时出错: {e}")
# 第一步VLM视觉分析如果没有已有描述才调用
if existing_description:
description = existing_description
logger.info("[优化] 复用已有的详细描述跳过VLM调用")
else: else:
prompt = "这是一个表情包,请详细描述一下表情包所表达的情感和内容,描述细节,从互联网梗,meme的角度去分析" logger.info("[VLM分析] 生成新的详细描述")
description, _ = await self.vlm.generate_response_for_image(prompt, image_base64, image_format) if image_format == "gif" or image_format == "GIF":
image_base64 = get_image_manager().transform_gif(image_base64) # type: ignore
if not image_base64:
raise RuntimeError("GIF表情包转换失败")
prompt = "这是一个动态图表情包,每一张图代表了动态图的某一帧,黑色背景代表透明,描述一下表情包表达的情感和内容,描述细节,从互联网梗,meme的角度去分析"
description, _ = await self.vlm.generate_response_for_image(prompt, image_base64, "jpg")
else:
prompt = "这是一个表情包,请详细描述一下表情包所表达的情感和内容,描述细节,从互联网梗,meme的角度去分析"
description, _ = await self.vlm.generate_response_for_image(prompt, image_base64, image_format)
# 审核表情包 # 审核表情包
if global_config.emoji.content_filtration: if global_config.emoji.content_filtration:
@@ -877,7 +894,7 @@ class EmojiManager:
if content == "": if content == "":
return "", [] return "", []
# 分析情感含义 # 第二步LLM情感分析 - 基于详细描述生成情感标签列表
emotion_prompt = f""" emotion_prompt = f"""
请你识别这个表情包的含义和适用场景给我简短的描述每个描述不要超过15个字 请你识别这个表情包的含义和适用场景给我简短的描述每个描述不要超过15个字
这是一个基于这个表情包的描述:'{description}' 这是一个基于这个表情包的描述:'{description}'
@@ -889,12 +906,14 @@ class EmojiManager:
# 处理情感列表 # 处理情感列表
emotions = [e.strip() for e in emotions_text.split(",") if e.strip()] emotions = [e.strip() for e in emotions_text.split(",") if e.strip()]
# 根据情感标签数量随机选择喵~超过5个选3个超过2个选2个 # 根据情感标签数量随机选择 - 超过5个选3个超过2个选2个
if len(emotions) > 5: if len(emotions) > 5:
emotions = random.sample(emotions, 3) emotions = random.sample(emotions, 3)
elif len(emotions) > 2: elif len(emotions) > 2:
emotions = random.sample(emotions, 2) emotions = random.sample(emotions, 2)
logger.info(f"[注册分析] 详细描述: {description[:50]}... -> 情感标签: {emotions}")
return f"[表情包:{description}]", emotions return f"[表情包:{description}]", emotions
except Exception as e: except Exception as e:

View File

@@ -6,6 +6,7 @@ import uuid
import io import io
import asyncio import asyncio
import numpy as np import numpy as np
import jieba
from typing import Optional, Tuple from typing import Optional, Tuple
from PIL import Image from PIL import Image
@@ -94,7 +95,7 @@ class ImageManager:
logger.error(f"保存描述到数据库失败 (Peewee): {str(e)}") logger.error(f"保存描述到数据库失败 (Peewee): {str(e)}")
async def get_emoji_description(self, image_base64: str) -> str: async def get_emoji_description(self, image_base64: str) -> str:
"""获取表情包描述,带查重和保存功能""" """获取表情包描述,使用二步走识别并带缓存优化"""
try: try:
# 计算图片哈希 # 计算图片哈希
# 确保base64字符串只包含ASCII字符 # 确保base64字符串只包含ASCII字符
@@ -107,33 +108,66 @@ class ImageManager:
# 查询缓存的描述 # 查询缓存的描述
cached_description = self._get_description_from_db(image_hash, "emoji") cached_description = self._get_description_from_db(image_hash, "emoji")
if cached_description: if cached_description:
return f"[表情包,含义看起来是{cached_description}]" return f"[表情包:{cached_description}]"
# 调用AI获取描述 # === 二步走识别流程 ===
# 第一步VLM视觉分析 - 生成详细描述
if image_format in ["gif", "GIF"]: if image_format in ["gif", "GIF"]:
image_base64_processed = self.transform_gif(image_base64) image_base64_processed = self.transform_gif(image_base64)
if image_base64_processed is None: if image_base64_processed is None:
logger.warning("GIF转换失败无法获取描述") logger.warning("GIF转换失败无法获取描述")
return "[表情包(GIF处理失败)]" return "[表情包(GIF处理失败)]"
prompt = "这是一个动态图表情包,每一张图代表了动态图的某一帧,黑色背景代表透明,使用1-2个词描述一下表情包表达的情感和内容,简短一些输出一段平文本只输出1-2个词就好不要输出其他内容" vlm_prompt = "这是一个动态图表情包,每一张图代表了动态图的某一帧,黑色背景代表透明,描述一下表情包表达的情感和内容,描述细节,从互联网梗,meme的角度去分析"
description, _ = await self._llm.generate_response_for_image(prompt, image_base64_processed, "jpg") detailed_description, _ = await self._llm.generate_response_for_image(vlm_prompt, image_base64_processed, "jpg")
else: else:
prompt = "图片是一个表情包,请用使用1-2个词描述一下表情包所表达的情感和内容,简短一些输出一段平文本只输出1-2个词就好不要输出其他内容" vlm_prompt = "是一个表情包,请详细描述一下表情包所表达的情感和内容,描述细节,从互联网梗,meme的角度去分析"
description, _ = await self._llm.generate_response_for_image(prompt, image_base64, image_format) detailed_description, _ = await self._llm.generate_response_for_image(vlm_prompt, image_base64, image_format)
if description is None: if detailed_description is None:
logger.warning("AI未能生成表情包描述") logger.warning("VLM未能生成表情包详细描述")
return "[表情包(描述生成失败)]" return "[表情包(VLM描述生成失败)]"
# 第二步LLM情感分析 - 基于详细描述生成简短的情感标签
emotion_prompt = f"""
请你基于这个表情包的详细描述提取出最核心的情感含义用1-2个词概括。
详细描述:'{detailed_description}'
要求:
1. 只输出1-2个最核心的情感词汇
2. 从互联网梗、meme的角度理解
3. 输出简短精准,不要解释
4. 如果有多个词用逗号分隔
"""
# 使用较低温度确保输出稳定
emotion_llm = LLMRequest(model=global_config.model.utils, temperature=0.3, max_tokens=50, request_type="emoji")
emotion_result, _ = await emotion_llm.generate_response_async(emotion_prompt)
if emotion_result is None:
logger.warning("LLM未能生成情感标签使用详细描述的前几个词")
# 降级处理:从详细描述中提取关键词
import jieba
words = list(jieba.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"[二步走识别] 详细描述: {detailed_description[:50]}... -> 情感标签: {final_emotion}")
# 再次检查缓存,防止并发写入时重复生成 # 再次检查缓存,防止并发写入时重复生成
cached_description = self._get_description_from_db(image_hash, "emoji") cached_description = self._get_description_from_db(image_hash, "emoji")
if cached_description: if cached_description:
logger.warning(f"虽然生成了描述,但是找到缓存表情包描述: {cached_description}") logger.warning(f"虽然生成了描述,但是找到缓存表情包描述: {cached_description}")
return f"[表情包,含义看起来是{cached_description}]" return f"[表情包:{cached_description}]"
# 根据配置决定是否保存图片 # 保存表情包文件和元数据(用于可能的后续分析)
# if global_config.emoji.save_emoji:
# 生成文件名和路径
logger.debug(f"保存表情包: {image_hash}") logger.debug(f"保存表情包: {image_hash}")
current_timestamp = time.time() current_timestamp = time.time()
filename = f"{int(current_timestamp)}_{image_hash[:8]}.{image_format}" filename = f"{int(current_timestamp)}_{image_hash[:8]}.{image_format}"
@@ -146,11 +180,11 @@ class ImageManager:
with open(file_path, "wb") as f: with open(file_path, "wb") as f:
f.write(image_bytes) f.write(image_bytes)
# 保存到数据库 (Images表) # 保存到数据库 (Images表) - 包含详细描述用于可能的注册流程
try: try:
img_obj = Images.get((Images.emoji_hash == image_hash) & (Images.type == "emoji")) img_obj = Images.get((Images.emoji_hash == image_hash) & (Images.type == "emoji"))
img_obj.path = file_path img_obj.path = file_path
img_obj.description = description img_obj.description = detailed_description # 保存详细描述
img_obj.timestamp = current_timestamp img_obj.timestamp = current_timestamp
img_obj.save() img_obj.save()
except Images.DoesNotExist: # type: ignore except Images.DoesNotExist: # type: ignore
@@ -158,17 +192,17 @@ class ImageManager:
emoji_hash=image_hash, emoji_hash=image_hash,
path=file_path, path=file_path,
type="emoji", type="emoji",
description=description, description=detailed_description, # 保存详细描述
timestamp=current_timestamp, timestamp=current_timestamp,
) )
# logger.debug(f"保存表情包元数据: {file_path}")
except Exception as e: except Exception as e:
logger.error(f"保存表情包文件或元数据失败: {str(e)}") logger.error(f"保存表情包文件或元数据失败: {str(e)}")
# 保存描述到数据库 (ImageDescriptions表) # 保存最终的情感标签到缓存 (ImageDescriptions表)
self._save_description_to_db(image_hash, description, "emoji") self._save_description_to_db(image_hash, final_emotion, "emoji")
return f"[表情包:{description}]" return f"[表情包:{final_emotion}]"
except Exception as e: except Exception as e:
logger.error(f"获取表情包描述失败: {str(e)}") logger.error(f"获取表情包描述失败: {str(e)}")
return "[表情包]" return "[表情包]"

View File

@@ -349,6 +349,7 @@ MODULE_COLORS = {
"chat_stream": "\033[38;5;51m", # 亮青色 "chat_stream": "\033[38;5;51m", # 亮青色
"sender": "\033[38;5;67m", # 稍微暗一些的蓝色,不显眼 "sender": "\033[38;5;67m", # 稍微暗一些的蓝色,不显眼
"message_storage": "\033[38;5;33m", # 深蓝色 "message_storage": "\033[38;5;33m", # 深蓝色
"expressor": "\033[38;5;166m", # 橙色
# 专注聊天模块 # 专注聊天模块
"replyer": "\033[38;5;166m", # 橙色 "replyer": "\033[38;5;166m", # 橙色
"memory_activator": "\033[34m", # 绿色 "memory_activator": "\033[34m", # 绿色
@@ -408,6 +409,34 @@ MODULE_COLORS = {
"S4U_chat": "\033[92m", # 深灰色 "S4U_chat": "\033[92m", # 深灰色
} }
# 定义模块别名映射 - 将真实的logger名称映射到显示的别名
MODULE_ALIASES = {
# 示例映射
"individuality": "人格特质",
"emoji": "表情包",
"no_reply_action": "摸鱼",
"reply_action": "回复",
"action_manager": "动作",
"memory_activator": "记忆",
"tool_use": "工具",
"expressor": "表达方式",
"database_model": "数据库",
"mood": "情绪",
"memory": "记忆",
"tool_executor": "工具",
"hfc": "聊天节奏",
"chat": "所见",
"plugin_manager": "插件",
"relationship_builder": "关系",
"llm_models": "模型",
"person_info": "人物",
"chat_stream": "聊天流",
"planner": "规划器",
"replyer": "言语",
"config": "配置",
"main": "主程序",
}
RESET_COLOR = "\033[0m" RESET_COLOR = "\033[0m"
@@ -496,15 +525,18 @@ class ModuleColoredConsoleRenderer:
if self._colors and self._enable_module_colors and logger_name: if self._colors and self._enable_module_colors and logger_name:
module_color = MODULE_COLORS.get(logger_name, "") module_color = MODULE_COLORS.get(logger_name, "")
# 模块名称(带颜色) # 模块名称(带颜色和别名支持
if logger_name: if logger_name:
# 获取别名,如果没有别名则使用原名称
display_name = MODULE_ALIASES.get(logger_name, logger_name)
if self._colors and self._enable_module_colors: if self._colors and self._enable_module_colors:
if module_color: if module_color:
module_part = f"{module_color}[{logger_name}]{RESET_COLOR}" module_part = f"{module_color}[{display_name}]{RESET_COLOR}"
else: else:
module_part = f"[{logger_name}]" module_part = f"[{display_name}]"
else: else:
module_part = f"[{logger_name}]" module_part = f"[{display_name}]"
parts.append(module_part) parts.append(module_part)
# 消息内容(确保转换为字符串) # 消息内容(确保转换为字符串)
@@ -714,19 +746,7 @@ def configure_logging(
root_logger.setLevel(getattr(logging, level.upper())) root_logger.setLevel(getattr(logging, level.upper()))
def set_module_color(module_name: str, color_code: str):
"""为指定模块设置颜色
Args:
module_name: 模块名称
color_code: ANSI颜色代码例如 '\033[92m' 表示亮绿色
"""
MODULE_COLORS[module_name] = color_code
def get_module_colors():
"""获取当前模块颜色配置"""
return MODULE_COLORS.copy()
def reload_log_config(): def reload_log_config():
@@ -917,9 +937,20 @@ def show_module_colors():
for module_name, _color_code in MODULE_COLORS.items(): for module_name, _color_code in MODULE_COLORS.items():
# 临时创建一个该模块的logger来展示颜色 # 临时创建一个该模块的logger来展示颜色
demo_logger = structlog.get_logger(module_name).bind(logger_name=module_name) demo_logger = structlog.get_logger(module_name).bind(logger_name=module_name)
demo_logger.info(f"这是 {module_name} 模块的颜色效果") alias = MODULE_ALIASES.get(module_name, module_name)
if alias != module_name:
demo_logger.info(f"这是 {module_name} 模块的颜色效果 (显示为: {alias})")
else:
demo_logger.info(f"这是 {module_name} 模块的颜色效果")
print("=== 颜色展示结束 ===\n") print("=== 颜色展示结束 ===\n")
# 显示别名映射表
if MODULE_ALIASES:
print("=== 当前别名映射 ===")
for module_name, alias in MODULE_ALIASES.items():
print(f" {module_name} -> {alias}")
print("=== 别名映射结束 ===\n")
def format_json_for_logging(data, indent=2, ensure_ascii=False): def format_json_for_logging(data, indent=2, ensure_ascii=False):

View File

@@ -10,6 +10,7 @@ from src.common.logger import get_logger
# 导入API模块 - 标准Python包方式 # 导入API模块 - 标准Python包方式
from src.plugin_system.apis import emoji_api, llm_api, message_api from src.plugin_system.apis import emoji_api, llm_api, message_api
from src.plugins.built_in.core_actions.no_reply import NoReplyAction from src.plugins.built_in.core_actions.no_reply import NoReplyAction
from src.config.config import global_config
logger = get_logger("emoji") logger = get_logger("emoji")
@@ -102,7 +103,11 @@ class EmojiAction(BaseAction):
这里是可用的情感标签:{available_emotions} 这里是可用的情感标签:{available_emotions}
请直接返回最匹配的那个情感标签,不要进行任何解释或添加其他多余的文字。 请直接返回最匹配的那个情感标签,不要进行任何解释或添加其他多余的文字。
""" """
logger.info(f"{self.log_prefix} 生成的LLM Prompt: {prompt}")
if global_config.debug.enable_debug_log:
logger.info(f"{self.log_prefix} 生成的LLM Prompt: {prompt}")
else:
logger.debug(f"{self.log_prefix} 生成的LLM Prompt: {prompt}")
# 5. 调用LLM # 5. 调用LLM
models = llm_api.get_available_models() models = llm_api.get_available_models()