refactor(proactive_thinker): 重构上下文收集与提示词构建逻辑以支持群聊
将原有的上下文收集函数 `_gather_context` 和提示词构建函数 `_make_decision`, `_build_plan_prompt` 拆分为更细粒度的模块，以分别处理私聊和群聊场景。 主要变更： - `_gather_context`: 现在能根据聊天流是群聊还是私聊，收集不同的上下文信息，并添加 `chat_type` 字段以作区分。 - `_build_decision_prompt`: 新增函数，根据 `chat_type` 构建不同的决策提示词，使决策更贴合场景。 - `_build_plan_prompt`: 重构为调度函数，内部调用新增的 `_build_private_plan_prompt` 和 `_build_group_plan_prompt` 来生成特定场景的规划提示词。 - 整体逻辑更加清晰，增强了代码的可读性和可扩展性，为未来支持更多聊天类型奠定了基础。
This commit is contained in:
minecraft1024a
@@ -104,61 +104,31 @@ class ProactiveThinkerExecutor:
|
||||
|
||||
async def _gather_context(self, stream_id: str) -> dict[str, Any] | None:
|
||||
"""
|
||||
收集构建提示词所需的所有上下文信息
|
||||
收集构建提示词所需的所有上下文信息.
|
||||
根据聊天流是私聊还是群聊, 收集不同的上下文.
|
||||
"""
|
||||
stream = self._get_stream_from_id(stream_id)
|
||||
if not stream:
|
||||
logger.warning(f"无法找到 stream_id 为 {stream_id} 的聊天流")
|
||||
return None
|
||||
|
||||
user_info = stream.user_info
|
||||
if not user_info or not user_info.platform or not user_info.user_id:
|
||||
logger.warning(f"Stream {stream_id} 的 user_info 不完整")
|
||||
return None
|
||||
|
||||
person_id = person_api.get_person_id(user_info.platform, int(user_info.user_id))
|
||||
person_info_manager = get_person_info_manager()
|
||||
|
||||
# 获取日程
|
||||
# 1. 收集通用信息 (日程, 聊天历史, 动作历史)
|
||||
schedules = await schedule_api.ScheduleAPI.get_today_schedule()
|
||||
schedule_context = (
|
||||
"\n".join(
|
||||
[
|
||||
f"- {s.get('time_range', '未知时间')}: {s.get('activity', '未知活动')}"
|
||||
for s in schedules
|
||||
]
|
||||
)
|
||||
if schedules
|
||||
else "今天没有日程安排。"
|
||||
)
|
||||
|
||||
# 获取关系信息
|
||||
short_impression = await person_info_manager.get_value(person_id, "short_impression") or "无"
|
||||
impression = await person_info_manager.get_value(person_id, "impression") or "无"
|
||||
attitude = await person_info_manager.get_value(person_id, "attitude") or 50
|
||||
|
||||
# 获取最近聊天记录
|
||||
schedule_context = "\n".join([f"- {s.get('time_range', '未知时间')}: {s.get('activity', '未知活动')}" for s in schedules]) if schedules else "今天没有日程安排。"
|
||||
|
||||
recent_messages = await message_api.get_recent_messages(stream_id, limit=10)
|
||||
recent_chat_history = (
|
||||
await message_api.build_readable_messages_to_str(recent_messages) if recent_messages else "无"
|
||||
)
|
||||
|
||||
# 获取最近的动作历史
|
||||
action_history = await database_api.db_query(
|
||||
recent_chat_history = await message_api.build_readable_messages_to_str(recent_messages) if recent_messages else "无"
|
||||
|
||||
action_history_list = await database_api.db_query(
|
||||
database_api.MODEL_MAPPING["ActionRecords"],
|
||||
filters={"chat_id": stream_id, "action_name": "proactive_decision"},
|
||||
limit=3,
|
||||
order_by=["-time"],
|
||||
)
|
||||
action_history_context = "无"
|
||||
if isinstance(action_history, list):
|
||||
action_history_context = (
|
||||
"\n".join([f"- {a['action_data']}" for a in action_history if isinstance(a, dict)]) or "无"
|
||||
)
|
||||
action_history_context = "\n".join([f"- {a['action_data']}" for a in action_history_list if isinstance(a, dict)]) if isinstance(action_history_list, list) else "无"
|
||||
|
||||
return {
|
||||
"person_id": person_id,
|
||||
"user_info": user_info,
|
||||
# 2. 构建基础上下文
|
||||
base_context = {
|
||||
"schedule_context": schedule_context,
|
||||
"recent_chat_history": recent_chat_history,
|
||||
"action_history_context": action_history_context,
|
||||
@@ -170,21 +140,63 @@ class ProactiveThinkerExecutor:
|
||||
"current_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
|
||||
}
|
||||
|
||||
async def _make_decision(self, context: dict[str, Any], start_mode: str) -> dict[str, Any] | None:
|
||||
"""
|
||||
决策模块：判断是否应该主动发起对话，以及聊什么话题
|
||||
"""
|
||||
persona = context["persona"]
|
||||
user_info = context["user_info"]
|
||||
relationship = context["relationship"]
|
||||
# 3. 根据聊天类型补充特定上下文
|
||||
if stream.group_info: # 群聊场景
|
||||
base_context.update({
|
||||
"chat_type": "group",
|
||||
"group_info": {
|
||||
"group_name": stream.group_info.group_name,
|
||||
"group_id": stream.group_info.group_id
|
||||
}
|
||||
})
|
||||
return base_context
|
||||
elif stream.user_info: # 私聊场景
|
||||
user_info = stream.user_info
|
||||
if not user_info.platform or not user_info.user_id:
|
||||
logger.warning(f"Stream {stream_id} 的 user_info 不完整")
|
||||
return None
|
||||
|
||||
person_id = person_api.get_person_id(user_info.platform, int(user_info.user_id))
|
||||
person_info_manager = get_person_info_manager()
|
||||
|
||||
# 获取关系信息
|
||||
short_impression = await person_info_manager.get_value(person_id, "short_impression") or "无"
|
||||
impression = await person_info_manager.get_value(person_id, "impression") or "无"
|
||||
attitude = await person_info_manager.get_value(person_id, "attitude") or 50
|
||||
|
||||
base_context.update({
|
||||
"chat_type": "private",
|
||||
"person_id": person_id,
|
||||
"user_info": user_info,
|
||||
"relationship": {
|
||||
"short_impression": short_impression,
|
||||
"impression": impression,
|
||||
"attitude": attitude
|
||||
}
|
||||
})
|
||||
return base_context
|
||||
else:
|
||||
logger.warning(f"Stream {stream_id} 既没有 group_info 也没有 user_info")
|
||||
return None
|
||||
|
||||
def _build_decision_prompt(self, context: dict[str, Any], start_mode: str) -> str:
|
||||
""" 根据上下文构建决策prompt """
|
||||
chat_type = context["chat_type"]
|
||||
persona = context["persona"]
|
||||
|
||||
# 构建通用头部
|
||||
prompt = f"""
|
||||
# 角色
|
||||
你的名字是{global_config.bot.nickname}，你的人设如下：
|
||||
- 核心人设: {persona["core"]}
|
||||
- 侧面人设: {persona["side"]}
|
||||
- 身份: {persona["identity"]}
|
||||
|
||||
"""
|
||||
# 根据聊天类型构建任务和情境
|
||||
if chat_type == "private":
|
||||
user_info = context["user_info"]
|
||||
relationship = context["relationship"]
|
||||
prompt += f"""
|
||||
# 任务
|
||||
现在是 {context["current_time"]}，你需要根据当前的情境，决定是否要主动向用户 '{user_info.user_nickname}' 发起对话。
|
||||
|
||||
@@ -198,7 +210,24 @@ class ProactiveThinkerExecutor:
|
||||
- 好感度: {relationship["attitude"]}/100
|
||||
4. **最近的聊天摘要**:
|
||||
{context["recent_chat_history"]}
|
||||
"""
|
||||
elif chat_type == "group":
|
||||
group_info = context["group_info"]
|
||||
prompt += f"""
|
||||
# 任务
|
||||
现在是 {context["current_time"]}，你需要根据当前的情境，决定是否要主动向群聊 '{group_info["group_name"]}' 发起对话。
|
||||
|
||||
# 情境分析
|
||||
1. **启动模式**: {start_mode} ({"首次加入/很久未发言" if start_mode == "cold_start" else "日常唤醒"})
|
||||
2. **你的日程**:
|
||||
{context["schedule_context"]}
|
||||
3. **群聊信息**:
|
||||
- 群名称: {group_info["group_name"]}
|
||||
4. **最近的聊天摘要**:
|
||||
{context["recent_chat_history"]}
|
||||
"""
|
||||
# 构建通用尾部
|
||||
prompt += """
|
||||
# 决策指令
|
||||
请综合以上所有信息，做出决策。你的决策需要以JSON格式输出，包含以下字段：
|
||||
- `should_reply`: bool, 是否应该发起对话。
|
||||
@@ -209,22 +238,34 @@ class ProactiveThinkerExecutor:
|
||||
示例1 (应该回复):
|
||||
{{
|
||||
"should_reply": true,
|
||||
"topic": "提醒Ta今天下午有'项目会议'的日程",
|
||||
"reason": "现在是上午，Ta下午有个重要会议，我觉得应该主动提醒一下，这会显得我很贴心。"
|
||||
"topic": "提醒大家今天下午有'项目会议'的日程",
|
||||
"reason": "现在是上午，下午有个重要会议，我觉得应该主动提醒一下大家，这会显得我很贴心。"
|
||||
}}
|
||||
|
||||
示例2 (不应回复):
|
||||
{{
|
||||
"should_reply": false,
|
||||
"topic": null,
|
||||
"reason": "虽然我们的关系不错，但现在是深夜，而且Ta今天的日程都已经完成了，我没有合适的理由去打扰Ta。"
|
||||
"reason": "虽然群里很活跃，但现在是深夜，而且最近的聊天话题我也不熟悉，没有合适的理由去打扰大家。"
|
||||
}}
|
||||
---
|
||||
|
||||
请输出你的决策:
|
||||
"""
|
||||
return prompt
|
||||
|
||||
async def _make_decision(self, context: dict[str, Any], start_mode: str) -> dict[str, Any] | None:
|
||||
"""
|
||||
决策模块：判断是否应该主动发起对话，以及聊什么话题
|
||||
"""
|
||||
if context["chat_type"] not in ["private", "group"]:
|
||||
return {"should_reply": False, "reason": "未知的聊天类型"}
|
||||
|
||||
prompt = self._build_decision_prompt(context, start_mode)
|
||||
|
||||
if global_config.debug.show_prompt:
|
||||
logger.info(f"主动思考规划器原始提示词:{prompt}")
|
||||
logger.info(f"主动思考决策器原始提示词:{prompt}")
|
||||
|
||||
is_success, response, _, _ = await llm_api.generate_with_model(
|
||||
prompt=prompt, model_config=model_config.model_task_config.utils
|
||||
)
|
||||
@@ -233,31 +274,21 @@ class ProactiveThinkerExecutor:
|
||||
return {"should_reply": False, "reason": "决策模型生成失败"}
|
||||
|
||||
try:
|
||||
# 假设LLM返回JSON格式的决策结果
|
||||
if global_config.debug.show_prompt:
|
||||
logger.info(f"主动思考规划器响应:{response}")
|
||||
logger.info(f"主动思考决策器响应:{response}")
|
||||
decision = orjson.loads(response)
|
||||
return decision
|
||||
except orjson.JSONDecodeError:
|
||||
logger.error(f"决策LLM返回的JSON格式无效: {response}")
|
||||
return {"should_reply": False, "reason": "决策模型返回格式错误"}
|
||||
|
||||
def _build_plan_prompt(self, context: dict[str, Any], start_mode: str, topic: str, reason: str) -> str:
|
||||
"""
|
||||
根据启动模式和决策话题，构建最终的规划提示词
|
||||
"""
|
||||
persona = context["persona"]
|
||||
def _build_private_plan_prompt(self, context: dict[str, Any], start_mode: str, topic: str, reason: str) -> str:
|
||||
""" 构建私聊的规划Prompt """
|
||||
user_info = context["user_info"]
|
||||
relationship = context["relationship"]
|
||||
|
||||
|
||||
if start_mode == "cold_start":
|
||||
prompt = f"""
|
||||
# 角色
|
||||
你的名字是{global_config.bot.nickname}，你的人设如下：
|
||||
- 核心人设: {persona["core"]}
|
||||
- 侧面人设: {persona["side"]}
|
||||
- 身份: {persona["identity"]}
|
||||
|
||||
return f"""
|
||||
# 任务
|
||||
你需要主动向一个新朋友 '{user_info.user_nickname}' 发起对话。这是你们的第一次交流，或者很久没聊了。
|
||||
|
||||
@@ -272,16 +303,9 @@ class ProactiveThinkerExecutor:
|
||||
- 你的目标是“破冰”，让对话自然地开始。
|
||||
- 你应该围绕这个话题展开: {topic}
|
||||
- 你的语气应该符合你的人设，友好且真诚。
|
||||
- 直接输出你要说的第一句话，不要包含任何额外的前缀或解释。
|
||||
"""
|
||||
else: # wake_up
|
||||
prompt = f"""
|
||||
# 角色
|
||||
你的名字是{global_config.bot.nickname}，你的人设如下：
|
||||
- 核心人设: {persona["core"]}
|
||||
- 侧面人设: {persona["side"]}
|
||||
- 身份: {persona["identity"]}
|
||||
|
||||
return f"""
|
||||
# 任务
|
||||
现在是 {context["current_time"]}，你需要主动向你的朋友 '{user_info.user_nickname}' 发起对话。
|
||||
|
||||
@@ -303,8 +327,58 @@ class ProactiveThinkerExecutor:
|
||||
- 你决定和Ta聊聊关于“{topic}”的话题。
|
||||
- 请结合以上所有情境信息，自然地开启对话。
|
||||
- 你的语气应该符合你的人设以及你对Ta的好感度。
|
||||
- 直接输出你要说的第一句话，不要包含任何额外的前缀或解释。
|
||||
"""
|
||||
|
||||
def _build_group_plan_prompt(self, context: dict[str, Any], topic: str, reason: str) -> str:
|
||||
""" 构建群聊的规划Prompt """
|
||||
group_info = context["group_info"]
|
||||
return f"""
|
||||
# 任务
|
||||
现在是 {context["current_time"]}，你需要主动向群聊 '{group_info["group_name"]}' 发起对话。
|
||||
|
||||
# 决策上下文
|
||||
- **决策理由**: {reason}
|
||||
|
||||
# 情境分析
|
||||
1. **你的日程**:
|
||||
{context["schedule_context"]}
|
||||
2. **群聊信息**:
|
||||
- 群名称: {group_info["group_name"]}
|
||||
3. **最近的聊天摘要**:
|
||||
{context["recent_chat_history"]}
|
||||
4. **你最近的相关动作**:
|
||||
{context["action_history_context"]}
|
||||
|
||||
# 对话指引
|
||||
- 你决定和大家聊聊关于“{topic}”的话题。
|
||||
- 你的语气应该更活泼、更具包容性，以吸引更多群成员参与讨论。
|
||||
- 请结合以上所有情境信息，自然地开启对话。
|
||||
"""
|
||||
|
||||
def _build_plan_prompt(self, context: dict[str, Any], start_mode: str, topic: str, reason: str) -> str:
|
||||
"""
|
||||
根据启动模式和决策话题，构建最终的规划提示词
|
||||
"""
|
||||
persona = context["persona"]
|
||||
chat_type = context["chat_type"]
|
||||
|
||||
# 1. 构建通用角色头部
|
||||
prompt = f"""
|
||||
# 角色
|
||||
你的名字是{global_config.bot.nickname}，你的人设如下：
|
||||
- 核心人设: {persona["core"]}
|
||||
- 侧面人设: {persona["side"]}
|
||||
- 身份: {persona["identity"]}
|
||||
"""
|
||||
# 2. 根据聊天类型构建特定内容
|
||||
if chat_type == "private":
|
||||
prompt += self._build_private_plan_prompt(context, start_mode, topic, reason)
|
||||
elif chat_type == "group":
|
||||
prompt += self._build_group_plan_prompt(context, topic, reason)
|
||||
|
||||
# 3. 添加通用结尾
|
||||
prompt += "\n- 直接输出你要说的第一句话，不要包含任何额外的前缀或解释。"
|
||||
|
||||
if global_config.debug.show_prompt:
|
||||
logger.info(f"主动思考回复器原始提示词:{prompt}")
|
||||
return prompt
|
||||
|
||||
Reference in New Issue
Block a user