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

better：优化prompt和配置和无用代码

Browse Source
This commit is contained in:
SengokuCola
2025-07-24 02:37:11 +08:00
parent bc4e7db78a
commit b839f8ba6c
10 changed files with 78 additions and 175 deletions
Download Patch File Download Diff File Expand all files Collapse all files

81
src/chat/replyer/default_generator.py
View File

@@ -30,9 +30,6 @@ from src.plugin_system.base.component_types import ActionInfo
logger = get_logger("replyer")
ENABLE_S2S_MODE = True
def init_prompt():
Prompt("你正在qq群里聊天，下面是群里在聊的内容：", "chat_target_group1")
Prompt("你正在和{sender_name}聊天，这是你们之前聊的内容：", "chat_target_private1")
@@ -97,29 +94,29 @@ def init_prompt():
{relation_info_block}
{extra_info_block}
你是一个AI虚拟主播，正在直播QQ聊天，同时也在直播间回复弹幕，不过回复的时候不用过多提及这点
{identity}
{action_descriptions}
你现在的主要任务是和 {sender_name} 聊天。同时，也有其他用户会参与你们的聊天，你可以参考他们的回复内容，但是你主要还是关注你和{sender_name}的聊天内容。你现在的心情是：{mood_state}
你现在的主要任务是和 {sender_name} 聊天。同时，也有其他用户会参与你们的聊天，你可以参考他们的回复内容，但是你主要还是关注你和{sender_name}的聊天内容。
{background_dialogue_prompt}
--------------------------------
{time_block}
这是你和{sender_name}的对话，你们正在交流中：
{core_dialogue_prompt}
{reply_target_block}
对方最新发送的内容：{message_txt}
回复可以简短一些。可以参考贴吧，知乎和微博的回复风格，回复不要浮夸，不要用夸张修辞，平淡一些。
{config_expression_style}。注意不要复读你说过的话
你现在的心情是：{mood_state}
{config_expression_style}
注意不要复读你说过的话
{keywords_reaction_prompt}
请注意不要输出多余内容(包括前后缀，冒号和引号，at或 @等 )。只输出回复内容。
{moderation_prompt}
不要浮夸，不要夸张修辞，不要输出多余内容(包括前后缀，冒号和引号，括号()，表情包，at或 @等 )。只输出回复内容，现在{sender_name}正在等待你的回复。
你的回复风格不要浮夸，有逻辑和条理，请你继续回复{sender_name}。
你的发言：
不要浮夸，不要夸张修辞，不要输出多余内容(包括前后缀，冒号和引号，括号()，表情包，at或 @等 )。只输出一条回复内容就好
现在，你说：
""",
"s4u_style_prompt",
)
@@ -132,7 +129,6 @@ class DefaultReplyer:
model_configs: Optional[List[Dict[str, Any]]] = None,
request_type: str = "focus.replyer",
):
self.log_prefix = "replyer"
self.request_type = request_type
if model_configs:
@@ -196,7 +192,7 @@ class DefaultReplyer:
}
for key, value in reply_data.items():
if not value:
logger.debug(f"{self.log_prefix} 回复数据跳过{key}，生成回复时将忽略。")
logger.debug(f"回复数据跳过{key}，生成回复时将忽略。")
# 3. 构建 Prompt
with Timer("构建Prompt", {}): # 内部计时器，可选保留
@@ -217,7 +213,7 @@ class DefaultReplyer:
# 加权随机选择一个模型配置
selected_model_config = self._select_weighted_model_config()
logger.info(
f"{self.log_prefix} 使用模型配置: {selected_model_config.get('name', 'N/A')} (权重: {selected_model_config.get('weight', 1.0)})"
f"使用模型生成回复: {selected_model_config.get('name', 'N/A')} (选中概率: {selected_model_config.get('weight', 1.0)})"
)
express_model = LLMRequest(
@@ -226,9 +222,9 @@ class DefaultReplyer:
)
if global_config.debug.show_prompt:
logger.info(f"{self.log_prefix}\n{prompt}\n")
logger.info(f"\n{prompt}\n")
else:
logger.debug(f"{self.log_prefix}\n{prompt}\n")
logger.debug(f"\n{prompt}\n")
content, (reasoning_content, model_name) = await express_model.generate_response_async(prompt)
@@ -236,13 +232,13 @@ class DefaultReplyer:
except Exception as llm_e:
# 精简报错信息
logger.error(f"{self.log_prefix}LLM 生成失败: {llm_e}")
logger.error(f"LLM 生成失败: {llm_e}")
return False, None, prompt # LLM 调用失败则无法生成回复
return True, content, prompt
except Exception as e:
logger.error(f"{self.log_prefix}回复生成意外失败: {e}")
logger.error(f"回复生成意外失败: {e}")
traceback.print_exc()
return False, None, prompt
@@ -273,7 +269,7 @@ class DefaultReplyer:
reasoning_content = None
model_name = "unknown_model"
if not prompt:
logger.error(f"{self.log_prefix}Prompt 构建失败，无法生成回复。")
logger.error(f"Prompt 构建失败，无法生成回复。")
return False, None
try:
@@ -281,7 +277,7 @@ class DefaultReplyer:
# 加权随机选择一个模型配置
selected_model_config = self._select_weighted_model_config()
logger.info(
f"{self.log_prefix} 使用模型配置进行重写: {selected_model_config.get('name', 'N/A')} (权重: {selected_model_config.get('weight', 1.0)})"
f"使用模型重写回复: {selected_model_config.get('name', 'N/A')} (选中概率: {selected_model_config.get('weight', 1.0)})"
)
express_model = LLMRequest(
@@ -295,13 +291,13 @@ class DefaultReplyer:
except Exception as llm_e:
# 精简报错信息
logger.error(f"{self.log_prefix}LLM 生成失败: {llm_e}")
logger.error(f"LLM 生成失败: {llm_e}")
return False, None # LLM 调用失败则无法生成回复
return True, content
except Exception as e:
logger.error(f"{self.log_prefix}回复生成意外失败: {e}")
logger.error(f"回复生成意外失败: {e}")
traceback.print_exc()
return False, None
@@ -321,7 +317,7 @@ class DefaultReplyer:
person_info_manager = get_person_info_manager()
person_id = person_info_manager.get_person_id_by_person_name(sender)
if not person_id:
logger.warning(f"{self.log_prefix} 未找到用户 {sender} 的ID，跳过信息提取")
logger.warning(f"未找到用户 {sender} 的ID，跳过信息提取")
return f"你完全不认识{sender}，不理解ta的相关信息。"
return await relationship_fetcher.build_relation_info(person_id, points_num=5)
@@ -340,7 +336,7 @@ class DefaultReplyer:
)
if selected_expressions:
logger.debug(f"{self.log_prefix} 使用处理器选中的{len(selected_expressions)}个表达方式")
logger.debug(f"使用处理器选中的{len(selected_expressions)}个表达方式")
for expr in selected_expressions:
if isinstance(expr, dict) and "situation" in expr and "style" in expr:
expr_type = expr.get("type", "style")
@@ -349,7 +345,7 @@ class DefaultReplyer:
else:
style_habits.append(f"当{expr['situation']}时，使用 {expr['style']}")
else:
logger.debug(f"{self.log_prefix} 没有从处理器获得表达方式，将使用空的表达方式")
logger.debug(f"没有从处理器获得表达方式，将使用空的表达方式")
# 不再在replyer中进行随机选择，全部交给处理器处理
style_habits_str = "\n".join(style_habits)
@@ -431,14 +427,14 @@ class DefaultReplyer:
tool_info_str += f"- 【{tool_name}】{result_type}: {content}\n"
tool_info_str += "以上是你获取到的实时信息，请在回复时参考这些信息。"
logger.info(f"{self.log_prefix} 获取到 {len(tool_results)} 个工具结果")
logger.info(f"获取到 {len(tool_results)} 个工具结果")
return tool_info_str
else:
logger.debug(f"{self.log_prefix} 未获取到任何工具结果")
logger.debug(f"未获取到任何工具结果")
return ""
except Exception as e:
logger.error(f"{self.log_prefix} 工具信息获取失败: {e}")
logger.error(f"工具信息获取失败: {e}")
return ""
def _parse_reply_target(self, target_message: str) -> tuple:
@@ -630,31 +626,40 @@ class DefaultReplyer:
# 并行执行四个构建任务
task_results = await asyncio.gather(
self._time_and_run_task(
self.build_expression_habits(chat_talking_prompt_short, target), "build_expression_habits"
self.build_expression_habits(chat_talking_prompt_short, target), "expression_habits"
),
self._time_and_run_task(
self.build_relation_info(reply_data), "build_relation_info"
self.build_relation_info(reply_data), "relation_info"
),
self._time_and_run_task(self.build_memory_block(chat_talking_prompt_short, target), "build_memory_block"),
self._time_and_run_task(self.build_memory_block(chat_talking_prompt_short, target), "memory_block"),
self._time_and_run_task(
self.build_tool_info(chat_talking_prompt_short, reply_data, enable_tool=enable_tool), "build_tool_info"
self.build_tool_info(chat_talking_prompt_short, reply_data, enable_tool=enable_tool), "tool_info"
),
)
# 任务名称中英文映射
task_name_mapping = {
"expression_habits": "选取表达方式",
"relation_info": "感受关系",
"memory_block": "回忆",
"tool_info": "使用工具"
}
# 处理结果
timing_logs = []
results_dict = {}
for name, result, duration in task_results:
results_dict[name] = result
timing_logs.append(f"{name}: {duration:.4f}s")
chinese_name = task_name_mapping.get(name, name)
timing_logs.append(f"{chinese_name}: {duration:.1f}s")
if duration > 8:
logger.warning(f"回复生成前信息获取耗时过长: {name} 耗时: {duration:.4f}s，请使用更快的模型")
logger.info(f"回复生成前信息获取耗时: {'; '.join(timing_logs)}")
logger.warning(f"回复生成前信息获取耗时过长: {chinese_name} 耗时: {duration:.1f}s，请使用更快的模型")
logger.info(f"在回复前的步骤耗时: {'; '.join(timing_logs)}")
expression_habits_block = results_dict["build_expression_habits"]
relation_info = results_dict["build_relation_info"]
memory_block = results_dict["build_memory_block"]
tool_info = results_dict["build_tool_info"]
expression_habits_block = results_dict["expression_habits"]
relation_info = results_dict["relation_info"]
memory_block = results_dict["memory_block"]
tool_info = results_dict["tool_info"]
keywords_reaction_prompt = await self.build_keywords_reaction_prompt(target)

3
src/config/config.py
View File

@@ -31,6 +31,7 @@ from src.config.official_configs import (
LPMMKnowledgeConfig,
RelationshipConfig,
ToolConfig,
VoiceConfig,
DebugConfig,
CustomPromptConfig,
)
@@ -328,7 +329,7 @@ class Config(ConfigBase):
tool: ToolConfig
debug: DebugConfig
custom_prompt: CustomPromptConfig
voice: VoiceConfig
def load_config(config_path: str) -> Config:
"""

10
src/config/official_configs.py
View File

@@ -106,9 +106,6 @@ class ChatConfig(ConfigBase):
focus_value: float = 1.0
"""麦麦的专注思考能力，越低越容易专注，消耗token也越多"""
enable_asr: bool = False
"""是否启用语音识别"""
def get_current_talk_frequency(self, chat_stream_id: Optional[str] = None) -> float:
"""
根据当前时间和聊天流获取对应的 talk_frequency
@@ -309,6 +306,13 @@ class ToolConfig(ConfigBase):
enable_in_focus_chat: bool = True
"""是否在专注聊天中启用工具"""
@dataclass
class VoiceConfig(ConfigBase):
"""语音识别配置类"""
enable_asr: bool = False
"""是否启用语音识别"""
@dataclass

16
src/mais4u/mais4u_chat/s4u_prompt.py
View File

@@ -10,13 +10,13 @@ from datetime import datetime
import asyncio
from src.mais4u.s4u_config import s4u_config
from src.chat.message_receive.message import MessageRecvS4U
from src.person_info.relationship_manager import get_relationship_manager
from src.person_info.relationship_fetcher import relationship_fetcher_manager
from src.person_info.person_info import PersonInfoManager, get_person_info_manager
from src.chat.message_receive.chat_stream import ChatStream
from src.mais4u.mais4u_chat.super_chat_manager import get_super_chat_manager
from src.mais4u.mais4u_chat.screen_manager import screen_manager
from src.chat.express.expression_selector import expression_selector
from .s4u_mood_manager import mood_manager
from src.person_info.person_info import PersonInfoManager, get_person_info_manager
from src.mais4u.mais4u_chat.internal_manager import internal_manager
logger = get_logger("prompt")
@@ -149,9 +149,17 @@ class PromptBuilder:
relation_prompt = ""
if global_config.relationship.enable_relationship and who_chat_in_group:
relationship_manager = get_relationship_manager()
relationship_fetcher = relationship_fetcher_manager.get_fetcher(chat_stream.stream_id)
# 将 (platform, user_id, nickname) 转换为 person_id
person_ids = []
for person in who_chat_in_group:
person_id = PersonInfoManager.get_person_id(person[0], person[1])
person_ids.append(person_id)
# 使用 RelationshipFetcher 的 build_relation_info 方法，设置 points_num=3 保持与原来相同的行为
relation_info_list = await asyncio.gather(
*[relationship_manager.build_relationship_info(person) for person in who_chat_in_group]
*[relationship_fetcher.build_relation_info(person_id, points_num=3) for person_id in person_ids]
)
relation_info = "".join(relation_info_list)
if relation_info:

2
src/person_info/person_info.py
View File

@@ -41,8 +41,6 @@ person_info_default = {
"know_times": 0,
"know_since": None,
"last_know": None,
# "user_cardname": None, # This field is not in Peewee model PersonInfo
# "user_avatar": None, # This field is not in Peewee model PersonInfo
"impression": None, # Corrected from person_impression
"short_impression": None,
"info_list": None,

63
src/person_info/relationship_fetcher.py
View File

@@ -112,15 +112,6 @@ class RelationshipFetcher:
current_points = await person_info_manager.get_value(person_id, "points") or []
if isinstance(current_points, str):
try:
current_points = json.loads(current_points)
except json.JSONDecodeError:
logger.error(f"解析points JSON失败: {current_points}")
current_points = []
elif not isinstance(current_points, list):
current_points = []
# 按时间排序forgotten_points
current_points.sort(key=lambda x: x[2])
# 按权重加权随机抽取最多3个不重复的points，point[1]的值在1-10之间，权重越高被抽到概率越大
@@ -370,60 +361,6 @@ class RelationshipFetcher:
logger.error(f"{self.log_prefix} 执行信息提取时出错: {e}")
logger.error(traceback.format_exc())
def _organize_known_info(self) -> str:
"""组织已知的用户信息为字符串
Returns:
str: 格式化的用户信息字符串
"""
persons_infos_str = ""
if self.info_fetched_cache:
persons_with_known_info = [] # 有已知信息的人员
persons_with_unknown_info = [] # 有未知信息的人员
for person_id in self.info_fetched_cache:
person_known_infos = []
person_unknown_infos = []
person_name = ""
for info_type in self.info_fetched_cache[person_id]:
person_name = self.info_fetched_cache[person_id][info_type]["person_name"]
if not self.info_fetched_cache[person_id][info_type]["unknown"]:
info_content = self.info_fetched_cache[person_id][info_type]["info"]
person_known_infos.append(f"[{info_type}]：{info_content}")
else:
person_unknown_infos.append(info_type)
# 如果有已知信息，添加到已知信息列表
if person_known_infos:
known_info_str = "；".join(person_known_infos) + "；"
persons_with_known_info.append((person_name, known_info_str))
# 如果有未知信息，添加到未知信息列表
if person_unknown_infos:
persons_with_unknown_info.append((person_name, person_unknown_infos))
# 先输出有已知信息的人员
for person_name, known_info_str in persons_with_known_info:
persons_infos_str += f"你对 {person_name} 的了解：{known_info_str}\n"
# 统一处理未知信息，避免重复的警告文本
if persons_with_unknown_info:
unknown_persons_details = []
for person_name, unknown_types in persons_with_unknown_info:
unknown_types_str = "、".join(unknown_types)
unknown_persons_details.append(f"{person_name}的[{unknown_types_str}]")
if len(unknown_persons_details) == 1:
persons_infos_str += (
f"你不了解{unknown_persons_details[0]}信息，不要胡乱回答，可以直接说不知道或忘记了；\n"
)
else:
unknown_all_str = "、".join(unknown_persons_details)
persons_infos_str += f"你不了解{unknown_all_str}等信息，不要胡乱回答，可以直接说不知道或忘记了；\n"
return persons_infos_str
async def _save_info_to_cache(self, person_id: str, info_type: str, info_content: str):
# sourcery skip: use-next

54
src/person_info/relationship_manager.py
View File

@@ -55,60 +55,6 @@ class RelationshipManager:
# person_id=person_id, user_nickname=user_nickname, user_cardname=user_cardname, user_avatar=user_avatar
# )
async def build_relationship_info(self, person, is_id: bool = False) -> str:
if is_id:
person_id = person
else:
person_id = PersonInfoManager.get_person_id(person[0], person[1])
person_info_manager = get_person_info_manager()
person_name = await person_info_manager.get_value(person_id, "person_name")
if not person_name or person_name == "none":
return ""
short_impression = await person_info_manager.get_value(person_id, "short_impression")
current_points = await person_info_manager.get_value(person_id, "points") or []
# print(f"current_points: {current_points}")
if isinstance(current_points, str):
try:
current_points = json.loads(current_points)
except json.JSONDecodeError:
logger.error(f"解析points JSON失败: {current_points}")
current_points = []
elif not isinstance(current_points, list):
current_points = []
# 按时间排序forgotten_points
current_points.sort(key=lambda x: x[2])
# 按权重加权随机抽取3个points，point[1]的值在1-10之间，权重越高被抽到概率越大
if len(current_points) > 3:
# point[1] 取值范围1-10，直接作为权重
weights = [max(1, min(10, int(point[1]))) for point in current_points]
points = random.choices(current_points, weights=weights, k=3)
else:
points = current_points
# 构建points文本
points_text = "\n".join([f"{point[2]}：{point[0]}" for point in points])
nickname_str = await person_info_manager.get_value(person_id, "nickname")
platform = await person_info_manager.get_value(person_id, "platform")
if person_name == nickname_str and not short_impression:
return ""
if person_name == nickname_str:
relation_prompt = f"'{person_name}' :"
else:
relation_prompt = f"'{person_name}' ，ta在{platform}上的昵称是{nickname_str}。"
if short_impression:
relation_prompt += f"你对ta的印象是：{short_impression}。\n"
if points_text:
relation_prompt += f"你记得ta最近做的事：{points_text}"
return relation_prompt
async def update_person_impression(self, person_id, timestamp, bot_engaged_messages: List[Dict[str, Any]]):
"""更新用户印象