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

feat：添加了眨眼动作和微动作，注视动作

Browse Source
This commit is contained in:
SengokuCola
2025-07-14 02:44:27 +08:00
parent 11bef44901
commit 3332be0d12
8 changed files with 580 additions and 88 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -1,6 +1,7 @@
import asyncio
import json
import time
import random
from src.chat.message_receive.message import MessageRecv
from src.llm_models.utils_model import LLMRequest
@@ -35,6 +36,9 @@ from src.plugin_system.apis import send_api
# 重置为中性表情
await facial_expression.reset_expression()
# 执行眨眼动作
await facial_expression.perform_blink()
3. 自动表情系统：
- 当情绪值更新时，系统会自动根据mood_values选择合适的面部表情
@@ -46,7 +50,14 @@ from src.plugin_system.apis import send_api
- 每次mood更新后立即发送表情更新
- 发送消息类型为"amadus_expression_update"，格式为{"action": "表情名", "data": 1.0}
5. 表情选择逻辑：
5. 眨眼系统：
- 每4-6秒随机执行一次眨眼动作
- 眨眼包含两个阶段：先闭眼（eye_close=1.0），保持0.1-0.15秒，然后睁眼（eye_close=0.0）
- 眨眼使用override_values参数临时覆盖eye_close值，不修改原始表情状态
- 眨眼时会发送完整的表情状态，包含当前表情的所有动作
- 当eye部位已经是eye_happy_weak时，跳过眨眼动作
6. 表情选择逻辑：
- 系统会找出最强的情绪（joy, anger, sorrow, fear）
- 根据情绪强度选择相应的表情组合
- 默认情况下返回neutral表情
@@ -130,77 +141,86 @@ 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_happy_weak": {"action": "eye_happy_weak", "data": 1.0},
"eye_close": {"action": "eye_close", "data": 1.0},
"eye_normal": {"action": "eye_normal", "data": 1.0},
"eye_shift_left": {"action": "eye_shift_left", "data": 1.0},
"eye_shift_right": {"action": "eye_shift_right", "data": 1.0},
# "eye_smile": {"action": "eye_smile", "data": 1.0}, # 未定义，占位
# "eye_cry": {"action": "eye_cry", "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},
"eyebrow_happy_weak": {"action": "eyebrow_happy_weak", "data": 1.0},
"eyebrow_happy_strong": {"action": "eyebrow_happy_strong", "data": 1.0},
"eyebrow_angry_weak": {"action": "eyebrow_angry_weak", "data": 1.0},
"eyebrow_angry_strong": {"action": "eyebrow_angry_strong", "data": 1.0},
"eyebrow_sad_weak": {"action": "eyebrow_sad_weak", "data": 1.0},
"eyebrow_sad_strong": {"action": "eyebrow_sad_strong", "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},
# 嘴巴表情（注意：用户定义的是mouth，可能是mouth的拼写错误）
"mouth_default": {"action": "mouth_default", "data": 1.0},
"mouth_happy_strong": {"action": "mouth_happy_strong", "data": 1.0}, # 保持用户原始拼写
"mouth_angry_weak": {"action": "mouth_angry_weak", "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},
# "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"
"eye": "eye_happy_weak",
"eyebrow": "eyebrow_happy_weak",
"mouth": "mouth_default",
},
"very_happy": {
"eye": "eye_smile",
"eyebrow": "eyebrow_smile",
"mouth": "mouth_laugh",
"face": "face_blush"
"eye": "eye_happy_weak",
"eyebrow": "eyebrow_happy_strong",
"mouth": "mouth_happy_strong",
},
"sad": {
"eye": "eye_cry",
"eyebrow": "eyebrow_sad",
"mouth": "mouth_sad",
"face": "face_normal"
"eyebrow": "eyebrow_sad_strong",
"mouth": "mouth_default",
},
"angry": {
"eye": "eye_normal",
"eyebrow": "eyebrow_angry",
"mouth": "mouth_angry",
"face": "face_normal"
"eyebrow": "eyebrow_angry_strong",
"mouth": "mouth_angry_weak",
},
"fear": {
"eye": "eye_close",
"eyebrow": "eyebrow_normal",
"mouth": "mouth_normal",
"face": "face_normal"
"eyebrow": "eyebrow_sad_weak",
"mouth": "mouth_default",
},
"shy": {
"eye": "eye_normal",
"eyebrow": "eyebrow_normal",
"mouth": "mouth_pout",
"face": "face_blush"
"eyebrow": "eyebrow_happy_weak",
"mouth": "mouth_default",
},
"neutral": {
"eye": "eye_normal",
"eyebrow": "eyebrow_normal",
"mouth": "mouth_normal",
"face": "face_normal"
"eyebrow": "eyebrow_happy_weak",
"mouth": "mouth_default",
}
}
# 未定义的表情部位（保留备用）：
# 眼睛：eye_smile, eye_cry, eye_close, eye_normal
# 眉毛：eyebrow_smile, eyebrow_angry, eyebrow_sad, eyebrow_normal
# 嘴巴：mouth_sad, mouth_angry, mouth_laugh, mouth_pout, mouth_normal
# 脸部：face_blush, face_normal
# 初始化当前表情状态
self.last_expression = "neutral"
def select_expression_by_mood(self, mood_values: dict[str, int]) -> str:
"""根据情绪值选择合适的表情组合"""
@@ -240,25 +260,86 @@ class FacialExpression:
else:
return "neutral"
async def send_expression(self, expression_name: str):
"""发送表情组合"""
async def _send_expression_actions(self, expression_name: str, log_prefix: str = "发送面部表情", override_values: dict = None):
"""统一的表情动作发送函数 - 发送完整的表情状态
Args:
expression_name: 表情名称
log_prefix: 日志前缀
override_values: 需要覆盖的动作值，格式为 {"action_name": value}
"""
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) # 短暂延迟避免同时发送过多消息
# 按部位分组所有已定义的表情动作
expressions_by_part = {
"eye": {},
"eyebrow": {},
"mouth": {}
}
# 将所有已定义的表情按部位分组
for expression_key, expression_data in self.expressions.items():
if expression_key.startswith("eye_"):
expressions_by_part["eye"][expression_key] = expression_data
elif expression_key.startswith("eyebrow_"):
expressions_by_part["eyebrow"][expression_key] = expression_data
elif expression_key.startswith("mouth_"):
expressions_by_part["mouth"][expression_key] = expression_data
# 构建完整的表情状态
complete_expression_state = {}
# 为每个部位构建完整的表情动作状态
for part in expressions_by_part.keys():
if expressions_by_part[part]: # 如果该部位有已定义的表情
part_actions = {}
active_expression = combination.get(part) # 当前激活的表情
# 添加该部位所有已定义的表情动作
for expression_key, expression_data in expressions_by_part[part].items():
# 复制表情数据并设置激活状态
action_data = expression_data.copy()
# 检查是否有覆盖值
if override_values and expression_key in override_values:
action_data["data"] = override_values[expression_key]
else:
action_data["data"] = 1.0 if expression_key == active_expression else 0.0
part_actions[expression_key] = action_data
complete_expression_state[part] = part_actions
logger.debug(f"[{self.chat_id}] 部位 {part}: 激活 {active_expression}, 总共 {len(part_actions)} 个动作")
# 发送完整的表情状态
if complete_expression_state:
package_data = {
"expression_name": expression_name,
"actions": complete_expression_state
}
await send_api.custom_to_stream(
message_type="face_emotion",
content=package_data,
stream_id=self.chat_id,
storage_message=False,
show_log=False,
)
# 统计信息
total_actions = sum(len(part_actions) for part_actions in complete_expression_state.values())
active_actions = [f"{part}:{combination.get(part, 'none')}" for part in complete_expression_state.keys()]
logger.info(f"[{self.chat_id}] {log_prefix}: {expression_name} - 发送{total_actions}个动作，激活: {', '.join(active_actions)}")
else:
logger.warning(f"[{self.chat_id}] 表情 {expression_name} 没有有效的动作可发送")
async def send_expression(self, expression_name: str):
"""发送表情组合"""
await self._send_expression_actions(expression_name, "发送面部表情")
# 通知ChatMood需要更新amadus
# 这里需要从mood_manager获取ChatMood实例并标记
@@ -276,6 +357,78 @@ class FacialExpression:
async def reset_expression(self):
"""重置为中性表情"""
await self.send_expression("neutral")
async def perform_blink(self):
"""执行眨眼动作"""
# 检查当前表情组合中eye部位是否为eye_happy_weak
current_combination = self.expression_combinations.get(self.last_expression, {})
current_eye_expression = current_combination.get("eye")
if current_eye_expression == "eye_happy_weak":
logger.debug(f"[{self.chat_id}] 当前eye表情为{current_eye_expression}，跳过眨眼动作")
return
logger.debug(f"[{self.chat_id}] 执行眨眼动作")
# 第一阶段：闭眼
await self._send_expression_actions(
self.last_expression,
"眨眼-闭眼",
override_values={"eye_close": 1.0}
)
# 等待0.1-0.15秒
blink_duration = random.uniform(0.7, 0.12)
await asyncio.sleep(blink_duration)
# 第二阶段：睁眼
await self._send_expression_actions(
self.last_expression,
"眨眼-睁眼",
override_values={"eye_close": 0.0}
)
async def perform_shift(self):
"""执行眨眼动作"""
# 检查当前表情组合中eye部位是否为eye_happy_weak
current_combination = self.expression_combinations.get(self.last_expression, {})
current_eye_expression = current_combination.get("eye")
direction = random.choice(["left", "right"])
strength = random.randint(6, 9) / 10
time_duration = random.randint(5, 15) / 10
if current_eye_expression == "eye_happy_weak" or current_eye_expression == "eye_close":
logger.debug(f"[{self.chat_id}] 当前eye表情为{current_eye_expression}，跳过漂移动作")
return
logger.debug(f"[{self.chat_id}] 执行漂移动作，方向：{direction}，强度：{strength}，时间：{time_duration}")
if direction == "left":
override_values = {"eye_shift_left": strength}
back_values = {"eye_shift_left": 0.0}
else:
override_values = {"eye_shift_right": strength}
back_values = {"eye_shift_right": 0.0}
# 第一阶段：闭眼
await self._send_expression_actions(
self.last_expression,
"漂移",
override_values=override_values
)
# 等待0.1-0.15秒
await asyncio.sleep(time_duration)
# 第二阶段：睁眼
await self._send_expression_actions(
self.last_expression,
"回归",
override_values=back_values
)
class ChatMood:
@@ -504,51 +657,150 @@ class ChatMood:
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
# 使用统一的表情发送函数
await self.facial_expression._send_expression_actions(
self.last_expression,
"发送表情更新到amadus"
)
logger.info(f"[{self.chat_id}] 发送表情更新到amadus: {expression_data}")
self.expression_needs_update = False # 重置标记
async def perform_blink(self):
"""执行眨眼动作"""
await self.facial_expression.perform_blink()
async def perform_shift(self):
"""执行漂移动作"""
await self.facial_expression.perform_shift()
class MoodRegressionTask(AsyncTask):
def __init__(self, mood_manager: "MoodManager"):
super().__init__(task_name="MoodRegressionTask", run_interval=30)
self.mood_manager = mood_manager
self.run_count = 0
async def run(self):
logger.debug("Running mood regression task...")
self.run_count += 1
logger.info(f"[回归任务] 第{self.run_count}次检查，当前管理{len(self.mood_manager.mood_list)}个聊天的情绪状态")
now = time.time()
regression_executed = 0
for mood in self.mood_manager.mood_list:
chat_info = f"chat {mood.chat_id}"
if mood.last_change_time == 0:
logger.debug(f"[回归任务] {chat_info} 尚未有情绪变化，跳过回归")
continue
if now - mood.last_change_time > 180:
time_since_last_change = now - mood.last_change_time
if time_since_last_change > 120: # 2分钟
if mood.regression_count >= 3:
logger.debug(f"[回归任务] {chat_info} 已达到最大回归次数(3次)，停止回归")
continue
logger.info(f"chat {mood.chat_id} 开始情绪回归, 这是第 {mood.regression_count + 1} 次")
logger.info(f"[回归任务] {chat_info} 开始情绪回归 (距上次变化{int(time_since_last_change)}秒，第{mood.regression_count + 1}次回归)")
await mood.regress_mood()
regression_executed += 1
else:
remaining_time = 120 - time_since_last_change
logger.debug(f"[回归任务] {chat_info} 距离回归还需等待{int(remaining_time)}秒")
if regression_executed > 0:
logger.info(f"[回归任务] 本次执行了{regression_executed}个聊天的情绪回归")
else:
logger.debug(f"[回归任务] 本次没有符合回归条件的聊天")
class ExpressionUpdateTask(AsyncTask):
def __init__(self, mood_manager: "MoodManager"):
super().__init__(task_name="ExpressionUpdateTask", run_interval=1)
super().__init__(task_name="ExpressionUpdateTask", run_interval=0.3)
self.mood_manager = mood_manager
self.run_count = 0
self.last_log_time = 0
async def run(self):
logger.debug("Running expression update task...")
self.run_count += 1
now = time.time()
# 每60秒输出一次状态信息（避免日志太频繁）
if now - self.last_log_time > 60:
logger.info(f"[表情任务] 已运行{self.run_count}次，当前管理{len(self.mood_manager.mood_list)}个聊天的表情状态")
self.last_log_time = now
updates_sent = 0
for mood in self.mood_manager.mood_list:
await mood.send_expression_update_if_needed()
if mood.expression_needs_update:
logger.debug(f"[表情任务] chat {mood.chat_id} 检测到表情变化，发送更新")
await mood.send_expression_update_if_needed()
updates_sent += 1
if updates_sent > 0:
logger.info(f"[表情任务] 发送了{updates_sent}个表情更新")
class BlinkTask(AsyncTask):
def __init__(self, mood_manager: "MoodManager"):
# 初始随机间隔4-6秒
super().__init__(task_name="BlinkTask", run_interval=4)
self.mood_manager = mood_manager
self.run_count = 0
self.last_log_time = 0
async def run(self):
self.run_count += 1
now = time.time()
# 每60秒输出一次状态信息（避免日志太频繁）
if now - self.last_log_time > 20:
logger.debug(f"[眨眼任务] 已运行{self.run_count}次，当前管理{len(self.mood_manager.mood_list)}个聊天的眨眼状态")
self.last_log_time = now
interval_add = random.randint(0, 2)
await asyncio.sleep(interval_add)
blinks_executed = 0
for mood in self.mood_manager.mood_list:
try:
await mood.perform_blink()
blinks_executed += 1
except Exception as e:
logger.error(f"[眨眼任务] 处理chat {mood.chat_id}时出错: {e}")
if blinks_executed > 0:
logger.debug(f"[眨眼任务] 本次执行了{blinks_executed}个聊天的眨眼动作")
class ShiftTask(AsyncTask):
def __init__(self, mood_manager: "MoodManager"):
# 初始随机间隔4-6秒
super().__init__(task_name="ShiftTask", run_interval=8)
self.mood_manager = mood_manager
self.run_count = 0
self.last_log_time = 0
async def run(self):
self.run_count += 1
now = time.time()
# 每60秒输出一次状态信息（避免日志太频繁）
if now - self.last_log_time > 20:
logger.debug(f"[漂移任务] 已运行{self.run_count}次，当前管理{len(self.mood_manager.mood_list)}个聊天的漂移状态")
self.last_log_time = now
interval_add = random.randint(0, 3)
await asyncio.sleep(interval_add)
blinks_executed = 0
for mood in self.mood_manager.mood_list:
try:
await mood.perform_shift()
blinks_executed += 1
except Exception as e:
logger.error(f"[漂移任务] 处理chat {mood.chat_id}时出错: {e}")
if blinks_executed > 0:
logger.debug(f"[漂移任务] 本次执行了{blinks_executed}个聊天的漂移动作")
class MoodManager:
@@ -572,8 +824,16 @@ class MoodManager:
expression_task = ExpressionUpdateTask(self)
await async_task_manager.add_task(expression_task)
# 启动眨眼任务
blink_task = BlinkTask(self)
await async_task_manager.add_task(blink_task)
# 启动漂移任务
shift_task = ShiftTask(self)
await async_task_manager.add_task(shift_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:
@@ -617,4 +877,5 @@ class MoodManager:
init_prompt()
mood_manager = MoodManager()
"""全局情绪管理器"""