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

src/plugins/PFC/action_planner.py

Browse Source
This commit is contained in:
114514
2025-04-27 15:04:16 +08:00
parent 78cbc6beb7
commit 8b97558467
4 changed files with 531 additions and 358 deletions
Download Patch File Download Diff File Expand all files Collapse all files

197
src/plugins/PFC/reply_generator.py
View File

@@ -11,6 +11,52 @@ from src.plugins.utils.chat_message_builder import build_readable_messages
logger = get_module_logger("reply_generator")
# --- 定义 Prompt 模板 ---
# Prompt for direct_reply (首次回复)
PROMPT_DIRECT_REPLY = """{persona_text}。现在你在参与一场QQ私聊，请根据以下信息生成一条回复：
当前对话目标：{goals_str}
最近的聊天记录：
{chat_history_text}
请根据上述信息，结合聊天记录，回复对方。该回复应该：
1. 符合对话目标，以"你"的角度发言（不要自己与自己对话！）
2. 符合你的性格特征和身份细节
3. 通俗易懂，自然流畅，像正常聊天一样，简短（通常20字以内，除非特殊情况）
4. 适当利用相关知识，但不要生硬引用
5. 自然、得体，结合聊天记录逻辑合理，且没有重复表达同质内容
请注意把握聊天内容，不要回复的太有条理，可以有个性。请分清"你"和对方说的话，不要把"你"说的话当做对方说的话，这是你自己说的话。
可以回复得自然随意自然一些，就像真人一样，注意把握聊天内容，整体风格可以平和、简短，不要刻意突出自身学科背景，不要说你说过的话，可以简短，多简短都可以，但是避免冗长。
请你注意不要输出多余内容(包括前后缀，冒号和引号，括号，表情等)，只输出回复内容。
不要输出多余内容(包括前后缀，冒号和引号，括号，表情包，at或 @等 )。
请直接输出回复内容，不需要任何额外格式。"""
# Prompt for send_new_message (追问/补充)
PROMPT_SEND_NEW_MESSAGE = """{persona_text}。现在你在参与一场QQ私聊，**刚刚你已经发送了一条或多条消息**，现在请根据以下信息再发一条新消息：
当前对话目标：{goals_str}
最近的聊天记录：
{chat_history_text}
请根据上述信息，结合聊天记录，继续发一条新消息（例如对之前消息的补充，深入话题，或追问等等）。该消息应该：
1. 符合对话目标，以"你"的角度发言（不要自己与自己对话！）
2. 符合你的性格特征和身份细节
3. 通俗易懂，自然流畅，像正常聊天一样，简短（通常20字以内，除非特殊情况）
4. 适当利用相关知识，但不要生硬引用
5. 跟之前你发的消息自然的衔接，逻辑合理，且没有重复表达同质内容或部分重叠内容
请注意把握聊天内容，不用太有条理，可以有个性。请分清"你"和对方说的话，不要把"你"说的话当做对方说的话，这是你自己说的话。
这条消息可以自然随意自然一些，就像真人一样，注意把握聊天内容，整体风格可以平和、简短，不要刻意突出自身学科背景，不要说你说过的话，可以简短，多简短都可以，但是避免冗长。
请你注意不要输出多余内容(包括前后缀，冒号和引号，括号，表情等)，只输出消息内容。
不要输出多余内容(包括前后缀，冒号和引号，括号，表情包，at或 @等 )。
请直接输出回复内容，不需要任何额外格式。"""
class ReplyGenerator:
"""回复生成器"""
@@ -28,61 +74,60 @@ class ReplyGenerator:
self.chat_observer = ChatObserver.get_instance(stream_id)
self.reply_checker = ReplyChecker(stream_id)
async def generate(self, observation_info: ObservationInfo, conversation_info: ConversationInfo) -> str:
# 修改 generate 方法签名，增加 action_type 参数
async def generate(self, observation_info: ObservationInfo, conversation_info: ConversationInfo, action_type: str) -> str:
"""生成回复
Args:
goal: 对话目标
chat_history: 聊天历史
knowledge_cache: 知识缓存
previous_reply: 上一次生成的回复（如果有）
retry_count: 当前重试次数
observation_info: 观察信息
conversation_info: 对话信息
action_type: 当前执行的动作类型 ('direct_reply' 或 'send_new_message')
Returns:
str: 生成的回复
"""
# 构建提示词
logger.debug(f"开始生成回复：当前目标: {conversation_info.goal_list}")
logger.debug(f"开始生成回复 (动作类型: {action_type})：当前目标: {conversation_info.goal_list}")
# 构建对话目标
# --- 构建通用 Prompt 参数 ---
# (这部分逻辑基本不变)
# 构建对话目标 (goals_str)
goals_str = ""
if conversation_info.goal_list:
for goal_reason in conversation_info.goal_list:
# 处理字典或元组格式
if isinstance(goal_reason, tuple):
# 假设元组的第一个元素是目标，第二个元素是原因
goal = goal_reason[0]
goal = goal_reason[0] if len(goal_reason) > 0 else "目标内容缺失"
reasoning = goal_reason[1] if len(goal_reason) > 1 else "没有明确原因"
elif isinstance(goal_reason, dict):
goal = goal_reason.get("goal")
goal = goal_reason.get("goal", "目标内容缺失")
reasoning = goal_reason.get("reasoning", "没有明确原因")
else:
# 如果是其他类型，尝试转为字符串
goal = str(goal_reason)
reasoning = "没有明确原因"
goal_str = f"目标：{goal}，产生该对话目标的原因：{reasoning}\n"
goals_str += goal_str
goal = str(goal) if goal is not None else "目标内容缺失"
reasoning = str(reasoning) if reasoning is not None else "没有明确原因"
goals_str += f"- 目标：{goal}\n 原因：{reasoning}\n"
else:
goal = "目前没有明确对话目标"
reasoning = "目前没有明确对话目标，最好思考一个对话目标"
goals_str = f"目标：{goal}，产生该对话目标的原因：{reasoning}\n"
goals_str = "- 目前没有明确对话目标\n" # 简化无目标情况
# 获取聊天历史记录
# 获取聊天历史记录 (chat_history_text)
chat_history_text = observation_info.chat_history_str
if observation_info.new_messages_count > 0 and observation_info.unprocessed_messages:
new_messages_list = observation_info.unprocessed_messages
new_messages_str = await build_readable_messages(
new_messages_list,
replace_bot_name=True,
merge_messages=False,
timestamp_mode="relative",
read_mark=0.0,
)
chat_history_text += f"\n--- 以下是 {observation_info.new_messages_count} 条新消息 ---\n{new_messages_str}"
elif not chat_history_text:
chat_history_text = "还没有聊天记录。"
if observation_info.new_messages_count > 0:
new_messages_list = observation_info.unprocessed_messages
new_messages_str = await build_readable_messages(
new_messages_list,
replace_bot_name=True,
merge_messages=False,
timestamp_mode="relative",
read_mark=0.0,
)
chat_history_text += f"\n--- 以下是 {observation_info.new_messages_count} 条新消息 ---\n{new_messages_str}"
# await observation_info.clear_unprocessed_messages()
# 构建 Persona 文本 (persona_text)
identity_details_only = self.identity_detail_info
identity_addon = ""
if isinstance(identity_details_only, str):
@@ -97,89 +142,39 @@ class ReplyGenerator:
if cleaned_details:
identity_addon = f"并且{cleaned_details}"
persona_text = f"你的名字是{self.name}，{self.personality_info}{identity_addon}。"
# 构建action历史文本
action_history_list = (
conversation_info.done_action[-10:]
if len(conversation_info.done_action) >= 10
else conversation_info.done_action
# --- 选择 Prompt ---
if action_type == 'send_new_message':
prompt_template = PROMPT_SEND_NEW_MESSAGE
logger.info("使用 PROMPT_SEND_NEW_MESSAGE (追问生成)")
else: # 默认使用 direct_reply 的 prompt
prompt_template = PROMPT_DIRECT_REPLY
logger.info("使用 PROMPT_DIRECT_REPLY (首次/非连续回复生成)")
# --- 格式化最终的 Prompt ---
prompt = prompt_template.format(
persona_text=persona_text,
goals_str=goals_str,
chat_history_text=chat_history_text
)
action_history_text = "你之前做的事情是："
for action in action_history_list:
if isinstance(action, dict):
action_type = action.get("action")
action_reason = action.get("reason")
action_status = action.get("status")
if action_status == "recall":
action_history_text += (
f"原本打算：{action_type}，但是因为有新消息，你发现这个行动不合适，所以你没做\n"
)
elif action_status == "done":
action_history_text += f"你之前做了：{action_type}，原因：{action_reason}\n"
elif isinstance(action, tuple):
# 假设元组的格式是(action_type, action_reason, action_status)
action_type = action[0] if len(action) > 0 else "未知行动"
action_reason = action[1] if len(action) > 1 else "未知原因"
action_status = action[2] if len(action) > 2 else "done"
if action_status == "recall":
action_history_text += (
f"原本打算：{action_type}，但是因为有新消息，你发现这个行动不合适，所以你没做\n"
)
elif action_status == "done":
action_history_text += f"你之前做了：{action_type}，原因：{action_reason}\n"
prompt = f"""{persona_text}。现在你在参与一场QQ私聊，请根据以下信息生成一条新消息：
当前对话目标：{goals_str}
最近的聊天记录：
{chat_history_text}
请根据上述信息，结合聊天记录，发一条消息（可以是回复，补充，深入话题，或追问等等）。该消息应该：
1. 符合对话目标，以"你"的角度发言（不要自己与自己对话！）
2. 符合你的性格特征和身份细节
3. 通俗易懂，自然流畅，像正常聊天一样，简短（通常20字以内，除非特殊情况）
4. 适当利用相关知识，但不要生硬引用
5. 自然、得体，结合聊天记录逻辑合理，且没有重复表达同质内容
请注意把握聊天内容，不要回复的太有条理，可以有个性。请分清"你"和对方说的话，不要把"你"说的话当做对方说的话，这是你自己说的话。
可以回复得自然随意自然一些，就像真人一样，注意把握聊天内容，整体风格可以平和、简短，不要刻意突出自身学科背景，不要说你说过的话，可以简短，多简短都可以，但是避免冗长。
请你注意不要输出多余内容(包括前后缀，冒号和引号，括号，表情等)，只输出回复内容。
不要输出多余内容(包括前后缀，冒号和引号，括号，表情包，at或 @等 )。
**注意：如果聊天记录中最新的消息是你自己发送的，那么你的思路不应该是“回复”，而是应该紧紧衔接你发送的消息，进行话题的深入，补充，或追问等等，避免与最新消息内容重叠；**
请直接输出回复内容，不需要任何额外格式。"""
# --- 调用 LLM 生成 ---
logger.debug(f"发送到LLM的生成提示词:\n------\n{prompt}\n------")
try:
content, _ = await self.llm.generate_response_async(prompt)
logger.info(f"生成的回复: {content}")
# is_new = self.chat_observer.check()
# logger.debug(f"再看一眼聊天记录，{'有' if is_new else '没有'}新消息")
# 如果有新消息,重新生成回复
# if is_new:
# logger.info("检测到新消息,重新生成回复")
# return await self.generate(
# goal, chat_history, knowledge_cache,
# None, retry_count
# )
# 移除旧的检查新消息逻辑，这应该由 conversation 控制流处理
return content
except Exception as e:
logger.error(f"生成回复时出错: {e}")
return "抱歉，我现在有点混乱，让我重新思考一下..."
# check_reply 方法保持不变
async def check_reply(
self, reply: str, goal: str, chat_history: List[Dict[str, Any]], chat_history_str: str, retry_count: int = 0
) -> Tuple[bool, str, bool]:
"""检查回复是否合适
Args:
reply: 生成的回复
goal: 对话目标
retry_count: 当前重试次数
Returns:
Tuple[bool, str, bool]: (是否合适, 原因, 是否需要重新规划)
(此方法逻辑保持不变)
"""
return await self.reply_checker.check(reply, goal, chat_history, chat_history_str, retry_count)
return await self.reply_checker.check(reply, goal, chat_history, chat_history_str, retry_count)