Losita/mai-bot
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix：更新s4u表情系统

Browse Source
This commit is contained in:
SengokuCola
2025-07-13 13:00:03 +08:00
parent c6ec4c53db
commit 5fd4caf23b
5 changed files with 292 additions and 245 deletions
Download Patch File Download Diff File Expand all files Collapse all files

323
src/mais4u/mais4u_chat/s4u_mood_manager.py
View File

@@ -11,21 +11,50 @@ from src.chat.utils.prompt_builder import Prompt, global_prompt_manager
from src.manager.async_task_manager import AsyncTask, async_task_manager
from src.plugin_system.apis import send_api
"""
面部表情系统使用说明：
1. 预定义的面部表情：
- happy: 高兴表情（眼睛微笑 + 眉毛微笑 + 嘴巴大笑）
- very_happy: 非常高兴（高兴表情 + 脸红）
- sad: 悲伤表情（眼睛哭泣 + 眉毛忧伤 + 嘴巴悲伤）
- angry: 生气表情（眉毛生气 + 嘴巴生气）
- fear: 恐惧表情（眼睛闭上）
- shy: 害羞表情（嘴巴嘟起 + 脸红）
- neutral: 中性表情（无表情）
2. 使用方法：
# 获取面部表情管理器
facial_expression = mood_manager.get_facial_expression_by_chat_id(chat_id)
# 发送指定表情
await facial_expression.send_expression("happy")
# 根据情绪值自动选择表情
await facial_expression.send_expression_by_mood(mood_values)
# 重置为中性表情
await facial_expression.reset_expression()
3. 自动表情系统：
- 当情绪值更新时，系统会自动根据mood_values选择合适的面部表情
- 只有当新表情与当前表情不同时才会发送，避免重复发送
- 支持joy >= 8时显示very_happy，joy >= 6时显示happy等梯度表情
4. amadus表情更新系统：
- 每1秒检查一次表情是否有变化，如有变化则发送到amadus
- 每次mood更新后立即发送表情更新
- 发送消息类型为"amadus_expression_update"，格式为{"action": "表情名", "data": 1.0}
5. 表情选择逻辑：
- 系统会找出最强的情绪（joy, anger, sorrow, fear）
- 根据情绪强度选择相应的表情组合
- 默认情况下返回neutral表情
"""
logger = get_logger("mood")
async def send_joy_action(chat_id: str):
action_content = {"action": "Joy_eye", "data": 1.0}
await send_api.custom_to_stream(message_type="face_emotion", content=action_content, stream_id=chat_id)
logger.info(f"[{chat_id}] 已发送 Joy 动作: {action_content}")
await asyncio.sleep(5.0)
end_action_content = {"action": "Joy_eye", "data": 0.0}
await send_api.custom_to_stream(message_type="face_emotion", content=end_action_content, stream_id=chat_id)
logger.info(f"[{chat_id}] 已发送 Joy 结束动作: {end_action_content}")
def init_prompt():
Prompt(
"""
@@ -64,13 +93,12 @@ def init_prompt():
喜(Joy): {joy}
怒(Anger): {anger}
哀(Sorrow): {sorrow}
乐(Pleasure): {pleasure}
惧(Fear): {fear}
现在，发送了消息，引起了你的注意，你对其进行了阅读和思考。请基于对话内容，评估你新的情绪状态。
请以JSON格式输出你新的情绪状态，包含“喜怒哀乐惧”五个维度，每个维度的取值范围为1-10。
键值请使用英文: "joy", "anger", "sorrow", "pleasure", "fear".
例如: {{"joy": 5, "anger": 1, "sorrow": 1, "pleasure": 5, "fear": 1}}
请以JSON格式输出你新的情绪状态，包含"喜怒哀惧"四个维度，每个维度的取值范围为1-10。
键值请使用英文: "joy", "anger", "sorrow", "fear".
例如: {{"joy": 5, "anger": 1, "sorrow": 1, "fear": 1}}
不要输出任何其他内容，只输出JSON。
""",
"change_mood_numerical_prompt",
@@ -86,24 +114,175 @@ def init_prompt():
喜(Joy): {joy}
怒(Anger): {anger}
哀(Sorrow): {sorrow}
乐(Pleasure): {pleasure}
惧(Fear): {fear}
距离你上次关注直播间消息已经过去了一段时间，你冷静了下来。请基于此，评估你现在的情绪状态。
请以JSON格式输出你新的情绪状态，包含“喜怒哀乐惧”五个维度，每个维度的取值范围为1-10。
键值请使用英文: "joy", "anger", "sorrow", "pleasure", "fear".
例如: {{"joy": 5, "anger": 1, "sorrow": 1, "pleasure": 5, "fear": 1}}
请以JSON格式输出你新的情绪状态，包含"喜怒哀惧"四个维度，每个维度的取值范围为1-10。
键值请使用英文: "joy", "anger", "sorrow", "fear".
例如: {{"joy": 5, "anger": 1, "sorrow": 1, "fear": 1}}
不要输出任何其他内容，只输出JSON。
""",
"regress_mood_numerical_prompt",
)
class FacialExpression:
def __init__(self, chat_id: str):
self.chat_id: str = chat_id
# 预定义面部表情动作
self.expressions = {
# 眼睛表情
"eye_smile": {"action": "eye_smile", "data": 1.0},
"eye_cry": {"action": "eye_cry", "data": 1.0},
"eye_close": {"action": "eye_close", "data": 1.0},
"eye_normal": {"action": "eye_normal", "data": 1.0},
# 眉毛表情
"eyebrow_smile": {"action": "eyebrow_smile", "data": 1.0},
"eyebrow_angry": {"action": "eyebrow_angry", "data": 1.0},
"eyebrow_sad": {"action": "eyebrow_sad", "data": 1.0},
"eyebrow_normal": {"action": "eyebrow_normal", "data": 1.0},
# 嘴巴表情
"mouth_sad": {"action": "mouth_sad", "data": 1.0},
"mouth_angry": {"action": "mouth_angry", "data": 1.0},
"mouth_laugh": {"action": "mouth_laugh", "data": 1.0},
"mouth_pout": {"action": "mouth_pout", "data": 1.0},
"mouth_normal": {"action": "mouth_normal", "data": 1.0},
# 脸部表情
"face_blush": {"action": "face_blush", "data": 1.0},
"face_normal": {"action": "face_normal", "data": 1.0},
}
# 表情组合模板
self.expression_combinations = {
"happy": {
"eye": "eye_smile",
"eyebrow": "eyebrow_smile",
"mouth": "mouth_laugh",
"face": "face_normal"
},
"very_happy": {
"eye": "eye_smile",
"eyebrow": "eyebrow_smile",
"mouth": "mouth_laugh",
"face": "face_blush"
},
"sad": {
"eye": "eye_cry",
"eyebrow": "eyebrow_sad",
"mouth": "mouth_sad",
"face": "face_normal"
},
"angry": {
"eye": "eye_normal",
"eyebrow": "eyebrow_angry",
"mouth": "mouth_angry",
"face": "face_normal"
},
"fear": {
"eye": "eye_close",
"eyebrow": "eyebrow_normal",
"mouth": "mouth_normal",
"face": "face_normal"
},
"shy": {
"eye": "eye_normal",
"eyebrow": "eyebrow_normal",
"mouth": "mouth_pout",
"face": "face_blush"
},
"neutral": {
"eye": "eye_normal",
"eyebrow": "eyebrow_normal",
"mouth": "mouth_normal",
"face": "face_normal"
}
}
def select_expression_by_mood(self, mood_values: dict[str, int]) -> str:
"""根据情绪值选择合适的表情组合"""
joy = mood_values.get("joy", 5)
anger = mood_values.get("anger", 1)
sorrow = mood_values.get("sorrow", 1)
fear = mood_values.get("fear", 1)
# 找出最强的情绪
emotions = {
"joy": joy,
"anger": anger,
"sorrow": sorrow,
"fear": fear
}
# 获取最强情绪
dominant_emotion = max(emotions, key=emotions.get)
dominant_value = emotions[dominant_emotion]
# 根据情绪强度和类型选择表情
if dominant_emotion == "joy":
if joy >= 8:
return "very_happy"
elif joy >= 6:
return "happy"
elif joy >= 4:
return "shy"
else:
return "neutral"
elif dominant_emotion == "anger" and anger >= 6:
return "angry"
elif dominant_emotion == "sorrow" and sorrow >= 6:
return "sad"
elif dominant_emotion == "fear" and fear >= 6:
return "fear"
else:
return "neutral"
async def send_expression(self, expression_name: str):
"""发送表情组合"""
if expression_name not in self.expression_combinations:
logger.warning(f"[{self.chat_id}] 未知表情: {expression_name}")
return
combination = self.expression_combinations[expression_name]
# 依次发送各部位表情
for part, expression_key in combination.items():
if expression_key in self.expressions:
expression_data = self.expressions[expression_key]
await send_api.custom_to_stream(
message_type="facial_expression",
content=expression_data,
stream_id=self.chat_id
)
logger.info(f"[{self.chat_id}] 发送面部表情 {part}: {expression_data}")
await asyncio.sleep(0.1) # 短暂延迟避免同时发送过多消息
# 通知ChatMood需要更新amadus
# 这里需要从mood_manager获取ChatMood实例并标记
chat_mood = mood_manager.get_mood_by_chat_id(self.chat_id)
if chat_mood.last_expression != expression_name:
chat_mood.last_expression = expression_name
chat_mood.expression_needs_update = True
async def send_expression_by_mood(self, mood_values: dict[str, int]):
"""根据情绪值发送相应的面部表情"""
expression_name = self.select_expression_by_mood(mood_values)
logger.info(f"[{self.chat_id}] 根据情绪值选择表情: {expression_name}, 情绪值: {mood_values}")
await self.send_expression(expression_name)
async def reset_expression(self):
"""重置为中性表情"""
await self.send_expression("neutral")
class ChatMood:
def __init__(self, chat_id: str):
self.chat_id: str = chat_id
self.mood_state: str = "感觉很平静"
self.mood_values: dict[str, int] = {"joy": 5, "anger": 1, "sorrow": 1, "pleasure": 5, "fear": 1}
self.mood_values: dict[str, int] = {"joy": 5, "anger": 1, "sorrow": 1, "fear": 1}
self.regression_count: int = 0
@@ -119,6 +298,15 @@ class ChatMood:
)
self.last_change_time = 0
# 添加面部表情系统
self.facial_expression = FacialExpression(chat_id)
self.last_expression = "neutral" # 记录上一次的表情
self.expression_needs_update = False # 标记表情是否需要更新
# 设置初始中性表情
asyncio.create_task(self.facial_expression.reset_expression())
self.expression_needs_update = True # 初始化时也标记需要更新
def _parse_numerical_mood(self, response: str) -> dict[str, int] | None:
try:
@@ -131,7 +319,7 @@ class ChatMood:
data = json.loads(response)
# Validate
required_keys = {"joy", "anger", "sorrow", "pleasure", "fear"}
required_keys = {"joy", "anger", "sorrow", "fear"}
if not required_keys.issubset(data.keys()):
logger.warning(f"Numerical mood response missing keys: {response}")
return None
@@ -203,7 +391,6 @@ class ChatMood:
joy=self.mood_values["joy"],
anger=self.mood_values["anger"],
sorrow=self.mood_values["sorrow"],
pleasure=self.mood_values["pleasure"],
fear=self.mood_values["fear"],
)
logger.info(f"numerical mood prompt: {prompt}")
@@ -221,9 +408,16 @@ class ChatMood:
self.mood_state = text_mood_response
if numerical_mood_response:
old_mood_values = self.mood_values.copy()
self.mood_values = numerical_mood_response
if self.mood_values.get("joy", 0) > 5:
asyncio.create_task(send_joy_action(self.chat_id))
# 发送面部表情
new_expression = self.facial_expression.select_expression_by_mood(self.mood_values)
if new_expression != self.last_expression:
# 立即发送表情
asyncio.create_task(self.facial_expression.send_expression(new_expression))
self.last_expression = new_expression
self.expression_needs_update = True # 标记表情已更新
self.last_change_time = message_time
@@ -277,7 +471,6 @@ class ChatMood:
joy=self.mood_values["joy"],
anger=self.mood_values["anger"],
sorrow=self.mood_values["sorrow"],
pleasure=self.mood_values["pleasure"],
fear=self.mood_values["fear"],
)
logger.debug(f"numerical regress prompt: {prompt}")
@@ -295,12 +488,37 @@ class ChatMood:
self.mood_state = text_mood_response
if numerical_mood_response:
old_mood_values = self.mood_values.copy()
self.mood_values = numerical_mood_response
if self.mood_values.get("joy", 0) > 5:
asyncio.create_task(send_joy_action(self.chat_id))
# 发送面部表情
new_expression = self.facial_expression.select_expression_by_mood(self.mood_values)
if new_expression != self.last_expression:
# 立即发送表情
asyncio.create_task(self.facial_expression.send_expression(new_expression))
self.last_expression = new_expression
self.expression_needs_update = True # 标记表情已更新
self.regression_count += 1
async def send_expression_update_if_needed(self):
"""如果表情有变化，发送更新到amadus"""
if self.expression_needs_update:
# 发送当前表情状态到amadus，使用简洁的action/data格式
expression_data = {
"action": self.last_expression,
"data": 1.0
}
await send_api.custom_to_stream(
message_type="amadus_expression_update",
content=expression_data,
stream_id=self.chat_id
)
logger.info(f"[{self.chat_id}] 发送表情更新到amadus: {expression_data}")
self.expression_needs_update = False # 重置标记
class MoodRegressionTask(AsyncTask):
def __init__(self, mood_manager: "MoodManager"):
@@ -322,6 +540,17 @@ class MoodRegressionTask(AsyncTask):
await mood.regress_mood()
class ExpressionUpdateTask(AsyncTask):
def __init__(self, mood_manager: "MoodManager"):
super().__init__(task_name="ExpressionUpdateTask", run_interval=1)
self.mood_manager = mood_manager
async def run(self):
logger.debug("Running expression update task...")
for mood in self.mood_manager.mood_list:
await mood.send_expression_update_if_needed()
class MoodManager:
def __init__(self):
self.mood_list: list[ChatMood] = []
@@ -333,11 +562,18 @@ class MoodManager:
if self.task_started:
return
logger.info("启动情绪回归任务...")
task = MoodRegressionTask(self)
await async_task_manager.add_task(task)
logger.info("启动情绪管理任务...")
# 启动情绪回归任务
regression_task = MoodRegressionTask(self)
await async_task_manager.add_task(regression_task)
# 启动表情更新任务
expression_task = ExpressionUpdateTask(self)
await async_task_manager.add_task(expression_task)
self.task_started = True
logger.info("情绪回归任务已启动")
logger.info("情绪管理任务已启动（包含情绪回归和表情更新）")
def get_mood_by_chat_id(self, chat_id: str) -> ChatMood:
for mood in self.mood_list:
@@ -352,9 +588,30 @@ class MoodManager:
for mood in self.mood_list:
if mood.chat_id == chat_id:
mood.mood_state = "感觉很平静"
mood.mood_values = {"joy": 5, "anger": 1, "sorrow": 1, "fear": 1}
mood.regression_count = 0
# 重置面部表情为中性
asyncio.create_task(mood.facial_expression.reset_expression())
mood.last_expression = "neutral"
mood.expression_needs_update = True # 标记表情需要更新
return
self.mood_list.append(ChatMood(chat_id))
# 如果没有找到现有的mood，创建新的
new_mood = ChatMood(chat_id)
self.mood_list.append(new_mood)
asyncio.create_task(new_mood.facial_expression.reset_expression())
new_mood.expression_needs_update = True # 标记表情需要更新
def get_facial_expression_by_chat_id(self, chat_id: str) -> FacialExpression:
"""获取聊天对应的面部表情管理器"""
for mood in self.mood_list:
if mood.chat_id == chat_id:
return mood.facial_expression
# 如果没有找到，创建新的
new_mood = ChatMood(chat_id)
self.mood_list.append(new_mood)
return new_mood.facial_expression
init_prompt()