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

feat(prompt): 添加主人认证提示词功能 (#47)

Browse Source
This commit is contained in:
Eric-Terminal
2025-10-25 15:55:12 +08:00
committed by Windpicker-owo
parent d13bf18cc0
commit 864ec3304c
6 changed files with 145 additions and 103 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
.gitignore vendored
View File

@@ -343,4 +343,4 @@ rust_video/Cargo.lock
.claude/settings.local.json
package-lock.json
package.json
src/chat/planner_actions/新建 文本文档.txt
src/chat/planner_actions/新建 文本文档.txt

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

@@ -38,10 +38,14 @@ from src.mais4u.mai_think import mai_thinking_manager
from src.mood.mood_manager import mood_manager
from src.person_info.person_info import get_person_info_manager
from src.plugin_system.apis import llm_api
from src.plugin_system.apis.permission_api import permission_api
from src.plugin_system.base.component_types import ActionInfo, EventType
logger = get_logger("replyer")
# 用于存储后台任务的集合，防止被垃圾回收
_background_tasks: set[asyncio.Task] = set()
def init_prompt():
Prompt("你正在qq群里聊天，下面是群里在聊的内容：", "chat_target_group1")
@@ -58,6 +62,7 @@ def init_prompt():
{time_block}
{chat_info}
{identity}
{auth_role_prompt_block}
你正在{chat_target_2},{reply_target_block}
对这句话，你想表达，原句：{raw_reply},原因是：{reason}。你现在要思考怎么组织回复
@@ -113,6 +118,7 @@ def init_prompt():
{relation_info_block}
{extra_info_block}
{auth_role_prompt_block}
{action_descriptions}
@@ -163,6 +169,60 @@ If you need to use the search tool, please directly call the function "lpmm_sear
name="lpmm_get_knowledge_prompt",
)
# normal 版 prompt 模板（0.9之前的简化模式）
logger.debug("[Prompt模式调试] 正在注册normal_style_prompt模板")
Prompt(
"""
{chat_scene}
**重要：消息针对性判断**
在回应之前，首先分析消息的针对性：
1. **直接针对你**：@你、回复你、明确询问你 → 必须回应
2. **间接相关**：涉及你感兴趣的话题但未直接问你 → 谨慎参与
3. **他人对话**：与你无关的私人交流 → 通常不参与
4. **重复内容**：他人已充分回答的问题 → 避免重复
{expression_habits_block}
{tool_info_block}
{knowledge_prompt}
{memory_block}
{relation_info_block}
{extra_info_block}
{cross_context_block}
{identity}
如果有人说你是人机，你可以用一种阴阳怪气的口吻来回应
{schedule_block}
{action_descriptions}
下面是群里最近的聊天内容：
--------------------------------
{time_block}
{chat_info}
--------------------------------
{reply_target_block}
你现在的心情是：{mood_state}
{config_expression_style}
注意不要复读你前面发过的内容，意思相近也不行。
{keywords_reaction_prompt}
请注意不要输出多余内容(包括前后缀，冒号和引号，at或 @等 )。只输出回复内容。
{moderation_prompt}
你的核心任务是针对 {reply_target_block} 中提到的内容，{relation_info_block}生成一段紧密相关且能推动对话的回复。你的回复应该：
1. 明确回应目标消息，而不是宽泛地评论。
2. 可以分享你的看法、提出相关问题，或者开个合适的玩笑。
3. 目的是让对话更有趣、更深入。
最终请输出一条简短、完整且口语化的回复。
*你叫{bot_name}，也有人叫你{bot_nickname}*
现在，你说：
""",
"normal_style_prompt",
)
logger.debug("[Prompt模式调试] normal_style_prompt模板注册完成")
class DefaultReplyer:
@@ -197,18 +257,21 @@ class DefaultReplyer:
self.tool_executor = ToolExecutor(chat_id=self.chat_stream.stream_id)
def _should_block_self_message(self, reply_message: Optional[Dict[str, Any]]) -> bool:
"""判定是否应阻断当前待处理消息（自消息且无外部触发）"""
try:
bot_id = str(global_config.bot.qq_account)
uid = str(reply_message.get("user_id"))
if uid != bot_id:
return False
return True
async def _build_auth_role_prompt(self) -> str:
"""根据主人配置生成额外提示词"""
master_config = getattr(global_config.permission, "master_prompt", None)
if not master_config or not master_config.enable:
return ""
platform, user_id = self.chat_stream.platform, self.chat_stream.user_info.user_id
try:
is_master = await permission_api.is_master(platform, user_id)
hint = master_config.master_hint if is_master else master_config.non_master_hint
return hint.strip()
except Exception as e:
logger.warning(f"[SelfGuard] 判定异常，回退为不阻断: {e}")
return False
logger.warning(f"检测主人身份失败: {e}")
return ""
async def generate_reply_with_context(
self,
@@ -368,6 +431,7 @@ class DefaultReplyer:
Returns:
Tuple[bool, Optional[str]]: (是否成功, 重写后的回复内容)
"""
prompt = None
try:
with Timer("构建Prompt", {}): # 内部计时器，可选保留
prompt = await self.build_prompt_rewrite_context(
@@ -976,29 +1040,24 @@ class DefaultReplyer:
unread_messages = []
bot_id = str(global_config.bot.qq_account)
# 第一次遍历：按 is_read 字段分离
for msg_dict in message_list_before_now:
try:
msg_user_id = str(msg_dict.get("user_id"))
if msg_dict.get("is_read", False):
read_messages.append(msg_dict)
else:
unread_messages.append(msg_dict)
except Exception as e:
logger.error(f"处理消息记录时出错: {msg_dict}, 错误: {e}")
msg_user_id = str(msg_dict.get("user_id", ""))
if msg_dict.get("is_read", False):
read_messages.append(msg_dict)
else:
unread_messages.append(msg_dict)
# 如果没有is_read字段，使用原有的逻辑
if not read_messages and not unread_messages:
# 使用原有的核心对话逻辑
core_dialogue_list = []
for msg_dict in message_list_before_now:
try:
msg_user_id = str(msg_dict.get("user_id"))
reply_to = msg_dict.get("reply_to", "")
_platform, reply_to_user_id = self._parse_reply_target(reply_to)
if (msg_user_id == bot_id and reply_to_user_id == target_user_id) or msg_user_id == target_user_id:
core_dialogue_list.append(msg_dict)
except Exception as e:
logger.error(f"处理消息记录时出错: {msg_dict}, 错误: {e}")
msg_user_id = str(msg_dict.get("user_id", ""))
reply_to = msg_dict.get("reply_to", "")
_platform, reply_to_user_id = self._parse_reply_target(reply_to)
if (msg_user_id == bot_id and reply_to_user_id == target_user_id) or msg_user_id == target_user_id:
core_dialogue_list.append(msg_dict)
read_messages = [msg for msg in message_list_before_now if msg not in core_dialogue_list]
unread_messages = core_dialogue_list
@@ -1526,6 +1585,8 @@ class DefaultReplyer:
else:
chat_scene_prompt = f"你正在和 {sender} 私下聊天，你需要理解你们的对话并做出自然的回应。"
auth_role_prompt_block = await self._build_auth_role_prompt()
# 使用新的统一Prompt系统 - 创建PromptParameters
prompt_parameters = PromptParameters(
chat_scene=chat_scene_prompt,
@@ -1560,6 +1621,7 @@ class DefaultReplyer:
safety_guidelines_block=safety_guidelines_block,
reply_target_block=reply_target_block,
mood_prompt=mood_prompt,
auth_role_prompt_block=auth_role_prompt_block,
action_descriptions=action_descriptions,
bot_name=global_config.bot.nickname,
bot_nickname=",".join(global_config.bot.alias_names) if global_config.bot.alias_names else "",
@@ -1705,6 +1767,8 @@ class DefaultReplyer:
await global_prompt_manager.format_prompt("chat_target_private1", sender_name=chat_target_name)
await global_prompt_manager.format_prompt("chat_target_private2", sender_name=chat_target_name)
auth_role_prompt_block = await self._build_auth_role_prompt()
# 使用新的统一Prompt系统 - Expressor模式，创建PromptParameters
prompt_parameters = PromptParameters(
chat_id=chat_id,
@@ -1721,6 +1785,7 @@ class DefaultReplyer:
mood_prompt=mood_prompt,
keywords_reaction_prompt=keywords_reaction_prompt,
moderation_prompt_block=moderation_prompt_block,
auth_role_prompt_block=auth_role_prompt_block,
# 添加已构建的表达习惯和关系信息
expression_habits_block=expression_habits_block,
relation_info_block=relation_info,
@@ -2011,7 +2076,7 @@ class DefaultReplyer:
# 异步存储聊天历史（完全非阻塞）
memory_system = get_memory_system()
asyncio.create_task(
task = asyncio.create_task(
memory_system.process_conversation_memory(
context={
"conversation_text": chat_history,
@@ -2021,6 +2086,9 @@ class DefaultReplyer:
}
)
)
# 将任务添加到全局集合以防止被垃圾回收
_background_tasks.add(task)
task.add_done_callback(_background_tasks.discard)
logger.debug(f"已启动记忆存储任务，用户: {memory_user_display or memory_user_id}")

105
src/chat/utils/prompt.py
View File

@@ -24,80 +24,6 @@ install(extra_lines=3)
logger = get_logger("unified_prompt")
@dataclass
class PromptParameters:
"""统一提示词参数系统"""
# 基础参数
chat_id: str = ""
is_group_chat: bool = False
sender: str = ""
target: str = ""
reply_to: str = ""
extra_info: str = ""
prompt_mode: Literal["s4u", "normal", "minimal"] = "s4u"
bot_name: str = ""
bot_nickname: str = ""
# 功能开关
enable_tool: bool = True
enable_memory: bool = True
enable_expression: bool = True
enable_relation: bool = True
enable_cross_context: bool = True
enable_knowledge: bool = True
# 性能控制
max_context_messages: int = 50
# 调试选项
debug_mode: bool = False
# 聊天历史和上下文
chat_target_info: dict[str, Any] | None = None
message_list_before_now_long: list[dict[str, Any]] = field(default_factory=list)
message_list_before_short: list[dict[str, Any]] = field(default_factory=list)
chat_talking_prompt_short: str = ""
target_user_info: dict[str, Any] | None = None
# 已构建的内容块
expression_habits_block: str = ""
relation_info_block: str = ""
memory_block: str = ""
tool_info_block: str = ""
knowledge_prompt: str = ""
cross_context_block: str = ""
notice_block: str = ""
# 其他内容块
keywords_reaction_prompt: str = ""
extra_info_block: str = ""
time_block: str = ""
identity_block: str = ""
schedule_block: str = ""
moderation_prompt_block: str = ""
safety_guidelines_block: str = ""
reply_target_block: str = ""
mood_prompt: str = ""
action_descriptions: str = ""
# 可用动作信息
available_actions: dict[str, Any] | None = None
# 动态生成的聊天场景提示
chat_scene: str = ""
def validate(self) -> list[str]:
"""参数验证"""
errors = []
if not self.chat_id:
errors.append("chat_id不能为空")
if self.prompt_mode not in ["s4u", "normal", "minimal"]:
errors.append("prompt_mode必须是's4u'、'normal'或'minimal'")
if self.max_context_messages <= 0:
errors.append("max_context_messages必须大于0")
return errors
class PromptContext:
"""提示词上下文管理器"""
@@ -891,10 +817,39 @@ class Prompt:
"moderation_prompt": self.parameters.moderation_prompt_block or context_data.get("moderation_prompt", ""),
"safety_guidelines_block": self.parameters.safety_guidelines_block
or context_data.get("safety_guidelines_block", ""),
"auth_role_prompt_block": self.parameters.auth_role_prompt_block
or context_data.get("auth_role_prompt_block", ""),
"chat_scene": self.parameters.chat_scene
or "你正在一个QQ群里聊天，你需要理解整个群的聊天动态和话题走向，并做出自然的回应。",
}
def _prepare_normal_params(self, context_data: dict[str, Any]) -> dict[str, Any]:
"""准备Normal模式的参数"""
return {
**context_data,
"expression_habits_block": context_data.get("expression_habits_block", ""),
"tool_info_block": context_data.get("tool_info_block", ""),
"knowledge_prompt": context_data.get("knowledge_prompt", ""),
"memory_block": context_data.get("memory_block", ""),
"relation_info_block": context_data.get("relation_info_block", ""),
"extra_info_block": self.parameters.extra_info_block or context_data.get("extra_info_block", ""),
"cross_context_block": context_data.get("cross_context_block", ""),
"identity": self.parameters.identity_block or context_data.get("identity", ""),
"action_descriptions": self.parameters.action_descriptions or context_data.get("action_descriptions", ""),
"schedule_block": self.parameters.schedule_block or context_data.get("schedule_block", ""),
"time_block": context_data.get("time_block", ""),
"chat_info": context_data.get("chat_info", ""),
"reply_target_block": context_data.get("reply_target_block", ""),
"config_expression_style": global_config.personality.reply_style,
"mood_state": self.parameters.mood_prompt or context_data.get("mood_state", ""),
"keywords_reaction_prompt": self.parameters.keywords_reaction_prompt
or context_data.get("keywords_reaction_prompt", ""),
"moderation_prompt": self.parameters.moderation_prompt_block or context_data.get("moderation_prompt", ""),
"safety_guidelines_block": self.parameters.safety_guidelines_block
or context_data.get("safety_guidelines_block", ""),
"chat_scene": self.parameters.chat_scene
or "你正在一个QQ群里聊天，你需要理解整个群的聊天动态和话题走向，并做出自然的回应。",
}
def _prepare_default_params(self, context_data: dict[str, Any]) -> dict[str, Any]:
"""准备默认模式的参数"""
@@ -916,6 +871,8 @@ class Prompt:
"moderation_prompt": self.parameters.moderation_prompt_block or context_data.get("moderation_prompt", ""),
"safety_guidelines_block": self.parameters.safety_guidelines_block
or context_data.get("safety_guidelines_block", ""),
"auth_role_prompt_block": self.parameters.auth_role_prompt_block
or context_data.get("auth_role_prompt_block", ""),
"bot_name": self.parameters.bot_name,
"bot_nickname": self.parameters.bot_nickname,
}

1
src/chat/utils/prompt_params.py
View File

@@ -52,6 +52,7 @@ class PromptParameters:
# 其他内容块
keywords_reaction_prompt: str = ""
extra_info_block: str = ""
auth_role_prompt_block: str = ""
time_block: str = ""
identity_block: str = ""
schedule_block: str = ""

11
src/config/official_configs.py
View File

@@ -657,6 +657,14 @@ class CommandConfig(ValidatedConfigBase):
command_prefixes: list[str] = Field(default_factory=lambda: ["/", "!", ".", "#"], description="支持的命令前缀列表")
class MasterPromptConfig(ValidatedConfigBase):
"""主人身份提示词配置"""
enable: bool = Field(default=False, description="是否启用主人提示词注入功能")
master_hint: str = Field(default="", description="对主人注入的额外提示词内容")
non_master_hint: str = Field(default="", description="对非主人注入的额外提示词内容")
class PermissionConfig(ValidatedConfigBase):
"""权限系统配置类"""
@@ -664,6 +672,9 @@ class PermissionConfig(ValidatedConfigBase):
master_users: list[list[str]] = Field(
default_factory=list, description="Master用户列表，格式: [[platform, user_id], ...]"
)
master_prompt: MasterPromptConfig = Field(
default_factory=MasterPromptConfig, description="主人身份提示词配置"
)
class AffinityFlowConfig(ValidatedConfigBase):

7
template/bot_config_template.toml
View File

@@ -1,5 +1,5 @@
[inner]
version = "7.4.6"
version = "7.3.7"
#----以下是给开发人员阅读的，如果你只是部署了MoFox-Bot，不需要阅读----
#如果你想要修改配置文件，请递增version的值
@@ -49,6 +49,11 @@ batch_action_storage_enabled = true # 是否启用批量保存动作记录（开
# 示例：[["qq", "123456"], ["telegram", "user789"]]
master_users = []# ["qq", "123456789"], # 示例：QQ平台的Master用户
[permission.master_prompt] # 主人身份提示词配置
enable = false # 是否启用主人/非主人提示注入
master_hint = "你正在与自己的主人交流，注意展现亲切与尊重。" # 主人提示词
non_master_hint = "对方并非你的主人，请保持常规互动风格。" # 非主人提示词
[bot]
platform = "qq"
qq_account = 1145141919810 # MoFox-Bot的QQ账号