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

refactor(plugins): 将亲和力流模块重构为插件架构并移除旧代码

Browse Source 
本次重构将亲和力流（Affinity Flow）模块从核心聊天系统迁移到插件架构中，主要变更包括：

- 删除 src/chat/planner_actions/ 目录下的 plan_executor.py、plan_filter.py 和 planner.py
- 更新插件系统组件类型，将 FOCUS 聊天模式改为 GROUP 和 PRIVATE 模式
- 调整亲和力流插件中的模式引用，确保与新的聊天模式枚举保持一致
- 统一亲和力流模块的导入路径，使其完全作为插件运行

BREAKING CHANGE: 移除原有的 FOCUS 聊天模式，改为 GROUP 和 PRIVATE 模式，需要更新相关配置和代码引用
This commit is contained in:
Windpicker-owo
2025-09-23 14:13:41 +08:00
parent c08998c585
commit 68bf0972df
7 changed files with 7 additions and 1132 deletions
Download Patch File Download Diff File Expand all files Collapse all files

363
src/chat/planner_actions/plan_executor.py
View File

@@ -1,363 +0,0 @@
"""
PlanExecutor: 接收 Plan 对象并执行其中的所有动作。
集成用户关系追踪机制，自动记录交互并更新关系。
"""
import asyncio
import re
import time
from typing import Dict, List
from src.config.config import global_config
from src.chat.planner_actions.action_manager import ChatterActionManager
from src.common.data_models.info_data_model import Plan, ActionPlannerInfo
from src.common.logger import get_logger
logger = get_logger("plan_executor")
class PlanExecutor:
"""
增强版PlanExecutor，集成用户关系追踪机制。
功能：
1. 执行Plan中的所有动作
2. 自动记录用户交互并添加到关系追踪
3. 分类执行回复动作和其他动作
4. 提供完整的执行统计和监控
"""
def __init__(self, action_manager: ChatterActionManager):
"""
初始化增强版PlanExecutor。
Args:
action_manager (ChatterActionManager): 用于实际执行各种动作的管理器实例。
"""
self.action_manager = action_manager
# 执行统计
self.execution_stats = {
"total_executed": 0,
"successful_executions": 0,
"failed_executions": 0,
"reply_executions": 0,
"other_action_executions": 0,
"execution_times": [],
}
# 用户关系追踪引用
self.relationship_tracker = None
def set_relationship_tracker(self, relationship_tracker):
"""设置关系追踪器"""
self.relationship_tracker = relationship_tracker
async def execute(self, plan: Plan) -> Dict[str, any]:
"""
遍历并执行Plan对象中`decided_actions`列表里的所有动作。
Args:
plan (Plan): 包含待执行动作列表的Plan对象。
Returns:
Dict[str, any]: 执行结果统计信息
"""
if not plan.decided_actions:
logger.info("没有需要执行的动作。")
return {"executed_count": 0, "results": []}
execution_results = []
reply_actions = []
other_actions = []
# 分类动作：回复动作和其他动作
for action_info in plan.decided_actions:
if action_info.action_type in ["reply", "proactive_reply"]:
reply_actions.append(action_info)
else:
other_actions.append(action_info)
# 执行回复动作（优先执行）
if reply_actions:
reply_result = await self._execute_reply_actions(reply_actions, plan)
execution_results.extend(reply_result["results"])
self.execution_stats["reply_executions"] += len(reply_actions)
# 将其他动作放入后台任务执行，避免阻塞主流程
if other_actions:
asyncio.create_task(self._execute_other_actions(other_actions, plan))
logger.info(f"已将 {len(other_actions)} 个其他动作放入后台任务执行。")
# 注意：后台任务的结果不会立即计入本次返回的统计数据
# 更新总体统计
self.execution_stats["total_executed"] += len(plan.decided_actions)
successful_count = sum(1 for r in execution_results if r["success"])
self.execution_stats["successful_executions"] += successful_count
self.execution_stats["failed_executions"] += len(execution_results) - successful_count
logger.info(
f"规划执行完成: 总数={len(plan.decided_actions)}, 成功={successful_count}, 失败={len(execution_results) - successful_count}"
)
return {
"executed_count": len(plan.decided_actions),
"successful_count": successful_count,
"failed_count": len(execution_results) - successful_count,
"results": execution_results,
}
async def _execute_reply_actions(self, reply_actions: List[ActionPlannerInfo], plan: Plan) -> Dict[str, any]:
"""执行回复动作"""
results = []
for action_info in reply_actions:
result = await self._execute_single_reply_action(action_info, plan)
results.append(result)
return {"results": results}
async def _execute_single_reply_action(self, action_info: ActionPlannerInfo, plan: Plan) -> Dict[str, any]:
"""执行单个回复动作"""
start_time = time.time()
success = False
error_message = ""
reply_content = ""
try:
logger.info(f"执行回复动作: {action_info.action_type} (原因: {action_info.reasoning})")
# 获取用户ID - 兼容对象和字典
if hasattr(action_info.action_message, "user_info"):
user_id = action_info.action_message.user_info.user_id
else:
user_id = action_info.action_message.get("user_info", {}).get("user_id")
if user_id == str(global_config.bot.qq_account):
logger.warning("尝试回复自己，跳过此动作以防止死循环。")
return {
"action_type": action_info.action_type,
"success": False,
"error_message": "尝试回复自己，跳过此动作以防止死循环。",
"execution_time": 0,
"reasoning": action_info.reasoning,
"reply_content": "",
}
# 构建回复动作参数
action_params = {
"chat_id": plan.chat_id,
"target_message": action_info.action_message,
"reasoning": action_info.reasoning,
"action_data": action_info.action_data or {},
}
# 通过动作管理器执行回复
reply_content = await self.action_manager.execute_action(
action_name=action_info.action_type, **action_params
)
success = True
logger.info(f"回复动作 '{action_info.action_type}' 执行成功。")
except Exception as e:
error_message = str(e)
logger.error(f"执行回复动作失败: {action_info.action_type}, 错误: {error_message}")
# 记录用户关系追踪
if success and action_info.action_message:
await self._track_user_interaction(action_info, plan, reply_content)
execution_time = time.time() - start_time
self.execution_stats["execution_times"].append(execution_time)
return {
"action_type": action_info.action_type,
"success": success,
"error_message": error_message,
"execution_time": execution_time,
"reasoning": action_info.reasoning,
"reply_content": reply_content[:200] + "..." if len(reply_content) > 200 else reply_content,
}
async def _execute_other_actions(self, other_actions: List[ActionPlannerInfo], plan: Plan) -> Dict[str, any]:
"""执行其他动作"""
results = []
# 并行执行其他动作
tasks = []
for action_info in other_actions:
task = self._execute_single_other_action(action_info, plan)
tasks.append(task)
if tasks:
executed_results = await asyncio.gather(*tasks, return_exceptions=True)
for i, result in enumerate(executed_results):
if isinstance(result, Exception):
logger.error(f"执行动作 {other_actions[i].action_type} 时发生异常: {result}")
results.append(
{
"action_type": other_actions[i].action_type,
"success": False,
"error_message": str(result),
"execution_time": 0,
"reasoning": other_actions[i].reasoning,
}
)
else:
results.append(result)
return {"results": results}
async def _execute_single_other_action(self, action_info: ActionPlannerInfo, plan: Plan) -> Dict[str, any]:
"""执行单个其他动作"""
start_time = time.time()
success = False
error_message = ""
try:
logger.info(f"执行其他动作: {action_info.action_type} (原因: {action_info.reasoning})")
action_data = action_info.action_data or {}
# 针对 poke_user 动作，特殊处理
if action_info.action_type == "poke_user":
target_message = action_info.action_message
if target_message:
# 优先直接获取 user_id，这才是最可靠的信息
user_id = target_message.get("user_id")
if user_id:
action_data["user_id"] = user_id
logger.info(f"检测到戳一戳动作，目标用户ID: {user_id}")
else:
# 如果没有 user_id，再尝试用 user_nickname 作为备用方案
user_name = target_message.get("user_nickname")
if user_name:
action_data["user_name"] = user_name
logger.info(f"检测到戳一戳动作，目标用户: {user_name}")
else:
logger.warning("无法从戳一戳消息中获取用户ID或昵称。")
# 传递原始消息ID以支持引用
action_data["target_message_id"] = target_message.get("message_id")
# 构建动作参数
action_params = {
"chat_id": plan.chat_id,
"target_message": action_info.action_message,
"reasoning": action_info.reasoning,
"action_data": action_data,
}
# 通过动作管理器执行动作
await self.action_manager.execute_action(action_name=action_info.action_type, **action_params)
success = True
logger.info(f"其他动作 '{action_info.action_type}' 执行成功。")
except Exception as e:
error_message = str(e)
logger.error(f"执行其他动作失败: {action_info.action_type}, 错误: {error_message}")
execution_time = time.time() - start_time
self.execution_stats["execution_times"].append(execution_time)
return {
"action_type": action_info.action_type,
"success": success,
"error_message": error_message,
"execution_time": execution_time,
"reasoning": action_info.reasoning,
}
async def _track_user_interaction(self, action_info: ActionPlannerInfo, plan: Plan, reply_content: str):
"""追踪用户交互 - 集成回复后关系追踪"""
try:
if not action_info.action_message:
return
# 获取用户信息 - 处理对象和字典两种情况
if hasattr(action_info.action_message, "user_info"):
# 对象情况
user_info = action_info.action_message.user_info
user_id = user_info.user_id
user_name = user_info.user_nickname or user_id
user_message = action_info.action_message.content
else:
# 字典情况
user_info = action_info.action_message.get("user_info", {})
user_id = user_info.get("user_id")
user_name = user_info.get("user_nickname") or user_id
user_message = action_info.action_message.get("content", "")
if not user_id:
logger.debug("跳过追踪：缺少用户ID")
return
# 如果有设置关系追踪器，执行回复后关系追踪
if self.relationship_tracker:
# 记录基础交互信息（保持向后兼容）
self.relationship_tracker.add_interaction(
user_id=user_id,
user_name=user_name,
user_message=user_message,
bot_reply=reply_content,
reply_timestamp=time.time(),
)
# 执行新的回复后关系追踪
await self.relationship_tracker.track_reply_relationship(
user_id=user_id, user_name=user_name, bot_reply_content=reply_content, reply_timestamp=time.time()
)
logger.debug(f"已执行用户交互追踪: {user_id}")
except Exception as e:
logger.error(f"追踪用户交互时出错: {e}")
logger.debug(f"action_message类型: {type(action_info.action_message)}")
logger.debug(f"action_message内容: {action_info.action_message}")
def get_execution_stats(self) -> Dict[str, any]:
"""获取执行统计信息"""
stats = self.execution_stats.copy()
# 计算平均执行时间
if stats["execution_times"]:
avg_time = sum(stats["execution_times"]) / len(stats["execution_times"])
stats["average_execution_time"] = avg_time
stats["max_execution_time"] = max(stats["execution_times"])
stats["min_execution_time"] = min(stats["execution_times"])
else:
stats["average_execution_time"] = 0
stats["max_execution_time"] = 0
stats["min_execution_time"] = 0
# 移除执行时间列表以避免返回过大数据
stats.pop("execution_times", None)
return stats
def reset_stats(self):
"""重置统计信息"""
self.execution_stats = {
"total_executed": 0,
"successful_executions": 0,
"failed_executions": 0,
"reply_executions": 0,
"other_action_executions": 0,
"execution_times": [],
}
def get_recent_performance(self, limit: int = 10) -> List[Dict[str, any]]:
"""获取最近的执行性能"""
recent_times = self.execution_stats["execution_times"][-limit:]
if not recent_times:
return []
return [
{
"execution_index": i + 1,
"execution_time": time_val,
"timestamp": time.time() - (len(recent_times) - i) * 60, # 估算时间戳
}
for i, time_val in enumerate(recent_times)
]

519
src/chat/planner_actions/plan_filter.py
View File

@@ -1,519 +0,0 @@
"""
PlanFilter: 接收 Plan 对象，根据不同模式的逻辑进行筛选，决定最终要执行的动作。
"""
import orjson
import time
import traceback
from datetime import datetime
from typing import Any, Dict, List, Optional
from json_repair import repair_json
from src.chat.memory_system.Hippocampus import hippocampus_manager
from src.chat.utils.chat_message_builder import (
build_readable_actions,
build_readable_messages_with_id,
get_actions_by_timestamp_with_chat,
)
from src.chat.utils.prompt import global_prompt_manager
from src.common.data_models.info_data_model import ActionPlannerInfo, Plan
from src.common.logger import get_logger
from src.config.config import global_config, model_config
from src.llm_models.utils_model import LLMRequest
from src.mood.mood_manager import mood_manager
from src.plugin_system.base.component_types import ActionInfo, ChatMode
from src.schedule.schedule_manager import schedule_manager
logger = get_logger("plan_filter")
class PlanFilter:
"""
根据 Plan 中的模式和信息，筛选并决定最终的动作。
"""
def __init__(self):
self.planner_llm = LLMRequest(model_set=model_config.model_task_config.planner, request_type="planner")
self.last_obs_time_mark = 0.0
async def filter(self, reply_not_available: bool, plan: Plan) -> Plan:
"""
执行筛选逻辑，并填充 Plan 对象的 decided_actions 字段。
"""
logger.debug(f"墨墨在这里加了日志 -> filter 入口 plan: {plan}")
try:
prompt, used_message_id_list = await self._build_prompt(plan)
plan.llm_prompt = prompt
logger.debug(f"墨墨在这里加了日志 -> LLM prompt: {prompt}")
llm_content, _ = await self.planner_llm.generate_response_async(prompt=prompt)
if llm_content:
logger.debug(f"墨墨在这里加了日志 -> LLM a原始返回: {llm_content}")
try:
parsed_json = orjson.loads(repair_json(llm_content))
except orjson.JSONDecodeError:
parsed_json = {"action": "no_action", "reason": "返回内容无法解析为JSON"}
logger.debug(f"墨墨在这里加了日志 -> 解析后的 JSON: {parsed_json}")
if "reply" in plan.available_actions and reply_not_available:
# 如果reply动作不可用，但llm返回的仍然有reply，则改为no_reply
if isinstance(parsed_json, dict) and parsed_json.get("action") == "reply":
parsed_json["action"] = "no_reply"
elif isinstance(parsed_json, list):
for item in parsed_json:
if isinstance(item, dict) and item.get("action") == "reply":
item["action"] = "no_reply"
item["reason"] += " (但由于兴趣度不足，reply动作不可用，已改为no_reply)"
if isinstance(parsed_json, dict):
parsed_json = [parsed_json]
if isinstance(parsed_json, list):
final_actions = []
reply_action_added = False
# 定义回复类动作的集合，方便扩展
reply_action_types = {"reply", "proactive_reply"}
for item in parsed_json:
if not isinstance(item, dict):
continue
# 预解析 action_type 来进行判断
action_type = item.get("action", "no_action")
if action_type in reply_action_types:
if not reply_action_added:
final_actions.extend(await self._parse_single_action(item, used_message_id_list, plan))
reply_action_added = True
else:
# 非回复类动作直接添加
final_actions.extend(await self._parse_single_action(item, used_message_id_list, plan))
plan.decided_actions = self._filter_no_actions(final_actions)
except Exception as e:
logger.error(f"筛选 Plan 时出错: {e}\n{traceback.format_exc()}")
plan.decided_actions = [ActionPlannerInfo(action_type="no_action", reasoning=f"筛选时出错: {e}")]
logger.debug(f"墨墨在这里加了日志 -> filter 出口 decided_actions: {plan.decided_actions}")
return plan
async def _build_prompt(self, plan: Plan) -> tuple[str, list]:
"""
根据 Plan 对象构建提示词。
"""
try:
time_block = f"当前时间：{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"
bot_name = global_config.bot.nickname
bot_nickname = (
f",也有人叫你{','.join(global_config.bot.alias_names)}" if global_config.bot.alias_names else ""
)
bot_core_personality = global_config.personality.personality_core
identity_block = f"你的名字是{bot_name}{bot_nickname}，你{bot_core_personality}："
schedule_block = ""
if global_config.planning_system.schedule_enable:
if current_activity := schedule_manager.get_current_activity():
schedule_block = f"你当前正在：{current_activity},但注意它与群聊的聊天无关。"
mood_block = ""
if global_config.mood.enable_mood:
chat_mood = mood_manager.get_mood_by_chat_id(plan.chat_id)
mood_block = f"你现在的心情是：{chat_mood.mood_state}"
if plan.mode == ChatMode.PROACTIVE:
long_term_memory_block = await self._get_long_term_memory_context()
chat_content_block, message_id_list = build_readable_messages_with_id(
messages=[msg.flatten() for msg in plan.chat_history],
timestamp_mode="normal",
truncate=False,
show_actions=False,
)
prompt_template = await global_prompt_manager.get_prompt_async("proactive_planner_prompt")
actions_before_now = get_actions_by_timestamp_with_chat(
chat_id=plan.chat_id,
timestamp_start=time.time() - 3600,
timestamp_end=time.time(),
limit=5,
)
actions_before_now_block = build_readable_actions(actions=actions_before_now)
actions_before_now_block = f"你刚刚选择并执行过的action是：\n{actions_before_now_block}"
prompt = prompt_template.format(
time_block=time_block,
identity_block=identity_block,
schedule_block=schedule_block,
mood_block=mood_block,
long_term_memory_block=long_term_memory_block,
chat_content_block=chat_content_block or "最近没有聊天内容。",
actions_before_now_block=actions_before_now_block,
)
return prompt, message_id_list
# 构建已读/未读历史消息
read_history_block, unread_history_block, message_id_list = await self._build_read_unread_history_blocks(
plan
)
# 为了兼容性，保留原有的chat_content_block
chat_content_block, _ = build_readable_messages_with_id(
messages=[msg.flatten() for msg in plan.chat_history],
timestamp_mode="normal",
read_mark=self.last_obs_time_mark,
truncate=True,
show_actions=True,
)
actions_before_now = get_actions_by_timestamp_with_chat(
chat_id=plan.chat_id,
timestamp_start=time.time() - 3600,
timestamp_end=time.time(),
limit=5,
)
actions_before_now_block = build_readable_actions(actions=actions_before_now)
actions_before_now_block = f"你刚刚选择并执行过的action是：\n{actions_before_now_block}"
self.last_obs_time_mark = time.time()
mentioned_bonus = ""
if global_config.chat.mentioned_bot_inevitable_reply:
mentioned_bonus = "\n- 有人提到你"
if global_config.chat.at_bot_inevitable_reply:
mentioned_bonus = "\n- 有人提到你，或者at你"
if plan.mode == ChatMode.FOCUS:
no_action_block = """
动作：no_action
动作描述：不选择任何动作
{{
"action": "no_action",
"reason":"不动作的原因"
}}
动作：no_reply
动作描述：不进行回复，等待合适的回复时机
- 当你刚刚发送了消息，没有人回复时，选择no_reply
- 当你一次发送了太多消息，为了避免打扰聊天节奏，选择no_reply
{{
"action": "no_reply",
"reason":"不回复的原因"
}}
"""
else: # NORMAL Mode
no_action_block = """重要说明：
- 'reply' 表示只进行普通聊天回复，不执行任何额外动作
- 其他action表示在普通回复的基础上，执行相应的额外动作
{{
"action": "reply",
"target_message_id":"触发action的消息id",
"reason":"回复的原因"
}}"""
is_group_chat = plan.target_info.platform == "group" if plan.target_info else True
chat_context_description = "你现在正在一个群聊中"
if not is_group_chat and plan.target_info:
chat_target_name = plan.target_info.person_name or plan.target_info.user_nickname or "对方"
chat_context_description = f"你正在和 {chat_target_name} 私聊"
action_options_block = await self._build_action_options(plan.available_actions)
moderation_prompt_block = "请不要输出违法违规内容，不要输出色情，暴力，政治相关内容，如有敏感内容，请规避。"
custom_prompt_block = ""
if global_config.custom_prompt.planner_custom_prompt_content:
custom_prompt_block = global_config.custom_prompt.planner_custom_prompt_content
users_in_chat_str = "" # TODO: Re-implement user list fetching if needed
planner_prompt_template = await global_prompt_manager.get_prompt_async("planner_prompt")
prompt = planner_prompt_template.format(
schedule_block=schedule_block,
mood_block=mood_block,
time_block=time_block,
chat_context_description=chat_context_description,
read_history_block=read_history_block,
unread_history_block=unread_history_block,
actions_before_now_block=actions_before_now_block,
mentioned_bonus=mentioned_bonus,
no_action_block=no_action_block,
action_options_text=action_options_block,
moderation_prompt=moderation_prompt_block,
identity_block=identity_block,
custom_prompt_block=custom_prompt_block,
bot_name=bot_name,
users_in_chat=users_in_chat_str,
)
return prompt, message_id_list
except Exception as e:
logger.error(f"构建 Planner 提示词时出错: {e}")
logger.error(traceback.format_exc())
return "构建 Planner Prompt 时出错", []
async def _build_read_unread_history_blocks(self, plan: Plan) -> tuple[str, str, list]:
"""构建已读/未读历史消息块"""
try:
# 从message_manager获取真实的已读/未读消息
from src.chat.message_manager.message_manager import message_manager
from src.chat.utils.utils import assign_message_ids
# 获取聊天流的上下文
stream_context = message_manager.stream_contexts.get(plan.chat_id)
# 获取真正的已读和未读消息
read_messages = stream_context.history_messages # 已读消息存储在history_messages中
unread_messages = stream_context.get_unread_messages() # 获取未读消息
# 构建已读历史消息块
if read_messages:
read_content, read_ids = build_readable_messages_with_id(
messages=[msg.flatten() for msg in read_messages[-50:]], # 限制数量
timestamp_mode="normal_no_YMD",
truncate=False,
show_actions=False,
)
read_history_block = f"{read_content}"
else:
read_history_block = "暂无已读历史消息"
# 构建未读历史消息块（包含兴趣度）
if unread_messages:
# 扁平化未读消息用于计算兴趣度和格式化
flattened_unread = [msg.flatten() for msg in unread_messages]
# 尝试获取兴趣度评分（返回以真实 message_id 为键的字典）
interest_scores = await self._get_interest_scores_for_messages(flattened_unread)
# 为未读消息分配短 id（保持与 build_readable_messages_with_id 的一致结构）
message_id_list = assign_message_ids(flattened_unread)
unread_lines = []
for idx, msg in enumerate(flattened_unread):
mapped = message_id_list[idx]
synthetic_id = mapped.get("id")
original_msg_id = msg.get("message_id") or msg.get("id")
msg_time = time.strftime("%H:%M:%S", time.localtime(msg.get("time", time.time())))
msg_content = msg.get("processed_plain_text", "")
# 添加兴趣度信息
interest_score = interest_scores.get(original_msg_id, 0.0)
interest_text = f" [兴趣度: {interest_score:.3f}]" if interest_score > 0 else ""
# 在未读行中显示合成id，方便 planner 返回时使用
unread_lines.append(f"{msg_time} {synthetic_id}: {msg_content}{interest_text}")
unread_history_block = "\n".join(unread_lines)
else:
unread_history_block = "暂无未读历史消息"
return read_history_block, unread_history_block, message_id_list
except Exception as e:
logger.error(f"构建已读/未读历史消息块时出错: {e}")
return "构建已读历史消息时出错", "构建未读历史消息时出错", []
async def _get_interest_scores_for_messages(self, messages: List[dict]) -> dict[str, float]:
"""为消息获取兴趣度评分"""
interest_scores = {}
try:
from src.plugins.built_in.affinity_flow_chatter.interest_scoring import chatter_interest_scoring_system
from src.common.data_models.database_data_model import DatabaseMessages
# 转换消息格式
db_messages = []
for msg_dict in messages:
try:
db_msg = DatabaseMessages(
message_id=msg_dict.get("message_id", ""),
time=msg_dict.get("time", time.time()),
chat_id=msg_dict.get("chat_id", ""),
processed_plain_text=msg_dict.get("processed_plain_text", ""),
user_id=msg_dict.get("user_id", ""),
user_nickname=msg_dict.get("user_nickname", ""),
user_platform=msg_dict.get("platform", "qq"),
chat_info_group_id=msg_dict.get("group_id", ""),
chat_info_group_name=msg_dict.get("group_name", ""),
chat_info_group_platform=msg_dict.get("platform", "qq"),
)
db_messages.append(db_msg)
except Exception as e:
logger.warning(f"转换消息格式失败: {e}")
continue
# 计算兴趣度评分
if db_messages:
bot_nickname = global_config.bot.nickname or "麦麦"
scores = await chatter_interest_scoring_system.calculate_interest_scores(db_messages, bot_nickname)
# 构建兴趣度字典
for score in scores:
interest_scores[score.message_id] = score.total_score
except Exception as e:
logger.warning(f"获取兴趣度评分失败: {e}")
return interest_scores
async def _parse_single_action(
self, action_json: dict, message_id_list: list, plan: Plan
) -> List[ActionPlannerInfo]:
parsed_actions = []
try:
action = action_json.get("action", "no_action")
reasoning = action_json.get("reason", "未提供原因")
action_data = {k: v for k, v in action_json.items() if k not in ["action", "reason"]}
target_message_obj = None
if action not in ["no_action", "no_reply", "do_nothing", "proactive_reply"]:
if target_message_id := action_json.get("target_message_id"):
target_message_dict = self._find_message_by_id(target_message_id, message_id_list)
else:
# 如果LLM没有指定target_message_id，我们就默认选择最新的一条消息
target_message_dict = self._get_latest_message(message_id_list)
if target_message_dict:
# 直接使用字典作为action_message，避免DatabaseMessages对象创建失败
target_message_obj = target_message_dict
# 替换action_data中的临时ID为真实ID
if "target_message_id" in action_data:
real_message_id = target_message_dict.get("message_id") or target_message_dict.get("id")
if real_message_id:
action_data["target_message_id"] = real_message_id
else:
# 如果找不到目标消息，对于reply动作来说这是必需的，应该记录警告
if action == "reply":
logger.warning(
f"reply动作找不到目标消息，target_message_id: {action_json.get('target_message_id')}"
)
# 将reply动作改为no_action，避免后续执行时出错
action = "no_action"
reasoning = f"找不到目标消息进行回复。原始理由: {reasoning}"
available_action_names = list(plan.available_actions.keys())
if (
action not in ["no_action", "no_reply", "reply", "do_nothing", "proactive_reply"]
and action not in available_action_names
):
reasoning = f"LLM 返回了当前不可用的动作 '{action}'。原始理由: {reasoning}"
action = "no_action"
parsed_actions.append(
ActionPlannerInfo(
action_type=action,
reasoning=reasoning,
action_data=action_data,
action_message=target_message_obj,
available_actions=plan.available_actions,
)
)
except Exception as e:
logger.error(f"解析单个action时出错: {e}")
parsed_actions.append(
ActionPlannerInfo(
action_type="no_action",
reasoning=f"解析action时出错: {e}",
)
)
return parsed_actions
def _filter_no_actions(self, action_list: List[ActionPlannerInfo]) -> List[ActionPlannerInfo]:
non_no_actions = [a for a in action_list if a.action_type not in ["no_action", "no_reply"]]
if non_no_actions:
return non_no_actions
return action_list[:1] if action_list else []
async def _get_long_term_memory_context(self) -> str:
try:
now = datetime.now()
keywords = ["今天", "日程", "计划"]
if 5 <= now.hour < 12:
keywords.append("早上")
elif 12 <= now.hour < 18:
keywords.append("中午")
else:
keywords.append("晚上")
retrieved_memories = await hippocampus_manager.get_memory_from_topic(
valid_keywords=keywords, max_memory_num=5, max_memory_length=1
)
if not retrieved_memories:
return "最近没有什么特别的记忆。"
memory_statements = [f"关于'{topic}', 你记得'{memory_item}'。" for topic, memory_item in retrieved_memories]
return " ".join(memory_statements)
except Exception as e:
logger.error(f"获取长期记忆时出错: {e}")
return "回忆时出现了一些问题。"
async def _build_action_options(self, current_available_actions: Dict[str, ActionInfo]) -> str:
action_options_block = ""
for action_name, action_info in current_available_actions.items():
param_text = ""
if action_info.action_parameters:
param_text = "\n" + "\n".join(
f' "{p_name}":"{p_desc}"' for p_name, p_desc in action_info.action_parameters.items()
)
require_text = "\n".join(f"- {req}" for req in action_info.action_require)
using_action_prompt = await global_prompt_manager.get_prompt_async("action_prompt")
action_options_block += using_action_prompt.format(
action_name=action_name,
action_description=action_info.description,
action_parameters=param_text,
action_require=require_text,
)
return action_options_block
def _find_message_by_id(self, message_id: str, message_id_list: list) -> Optional[Dict[str, Any]]:
# 兼容多种 message_id 格式：数字、m123、buffered-xxxx
# 如果是纯数字，补上 m 前缀以兼容旧格式
candidate_ids = {message_id}
if message_id.isdigit():
candidate_ids.add(f"m{message_id}")
# 如果是 m 开头且后面是数字，尝试去掉 m 前缀的数字形式
if message_id.startswith("m") and message_id[1:].isdigit():
candidate_ids.add(message_id[1:])
# 逐项匹配 message_id_list（每项可能为 {'id':..., 'message':...}）
for item in message_id_list:
# 支持 message_id_list 中直接是字符串/ID 的情形
if isinstance(item, str):
if item in candidate_ids:
# 没有 message 对象，返回None
return None
continue
if not isinstance(item, dict):
continue
item_id = item.get("id")
# 直接匹配分配的短 id
if item_id and item_id in candidate_ids:
return item.get("message")
# 有时 message 存储里会有原始的 message_id 字段（如 buffered-xxxx）
message_obj = item.get("message")
if isinstance(message_obj, dict):
orig_mid = message_obj.get("message_id") or message_obj.get("id")
if orig_mid and orig_mid in candidate_ids:
return message_obj
# 作为兜底，尝试在 message_id_list 中找到 message.message_id 匹配
for item in message_id_list:
if isinstance(item, dict) and isinstance(item.get("message"), dict):
mid = item["message"].get("message_id") or item["message"].get("id")
if mid == message_id:
return item["message"]
return None
def _get_latest_message(self, message_id_list: list) -> Optional[Dict[str, Any]]:
if not message_id_list:
return None
return message_id_list[-1].get("message")

244
src/chat/planner_actions/planner.py
View File

@@ -1,244 +0,0 @@
"""
主规划器入口，负责协调 PlanGenerator, PlanFilter, 和 PlanExecutor。
集成兴趣度评分系统和用户关系追踪机制，实现智能化的聊天决策。
"""
from dataclasses import asdict
from typing import TYPE_CHECKING, Dict, List, Optional, Tuple
from src.plugin_system.base.component_types import ChatMode
from src.plugins.built_in.affinity_flow_chatter.plan_executor import ChatterPlanExecutor
from src.plugins.built_in.affinity_flow_chatter.plan_filter import ChatterPlanFilter
from src.plugins.built_in.affinity_flow_chatter.plan_generator import ChatterPlanGenerator
from src.plugins.built_in.affinity_flow_chatter.interest_scoring import ChatterInterestScoringSystem
from src.plugins.built_in.affinity_flow_chatter.relationship_tracker import ChatterRelationshipTracker
from src.common.logger import get_logger
from src.config.config import global_config
if TYPE_CHECKING:
from src.chat.planner_actions.action_manager import ActionManager
from src.common.data_models.message_manager_data_model import StreamContext
from src.common.data_models.info_data_model import Plan
# 导入提示词模块以确保其被初始化
from src.plugins.built_in.affinity_flow_chatter import planner_prompts # noqa
logger = get_logger("planner")
class ActionPlanner:
"""
增强版ActionPlanner，集成兴趣度评分和用户关系追踪机制。
核心功能：
1. 兴趣度评分系统：根据兴趣匹配度、关系分、提及度、时间因子对消息评分
2. 用户关系追踪：自动追踪用户交互并更新关系分
3. 智能回复决策：基于兴趣度阈值和连续不回复概率的智能决策
4. 完整的规划流程：生成→筛选→执行的完整三阶段流程
"""
def __init__(self, chat_id: str, action_manager: "ActionManager"):
"""
初始化增强版ActionPlanner。
Args:
chat_id (str): 当前聊天的 ID。
action_manager (ActionManager): 一个 ActionManager 实例。
"""
self.chat_id = chat_id
self.action_manager = action_manager
self.generator = ChatterPlanGenerator(chat_id)
self.filter = ChatterPlanFilter()
self.executor = ChatterPlanExecutor(action_manager)
# 初始化兴趣度评分系统
self.interest_scoring = ChatterInterestScoringSystem()
# 创建新的关系追踪器
self.relationship_tracker = ChatterRelationshipTracker(self.interest_scoring)
logger.info("创建新的关系追踪器实例")
# 设置执行器的关系追踪器
self.executor.set_relationship_tracker(self.relationship_tracker)
# 规划器统计
self.planner_stats = {
"total_plans": 0,
"successful_plans": 0,
"failed_plans": 0,
"replies_generated": 0,
"other_actions_executed": 0,
}
async def plan(
self, mode: ChatMode = ChatMode.FOCUS, context: "StreamContext" = None
) -> Tuple[List[Dict], Optional[Dict]]:
"""
执行完整的增强版规划流程。
Args:
mode (ChatMode): 当前的聊天模式，默认为 FOCUS。
context (StreamContext): 包含聊天流消息的上下文对象。
Returns:
Tuple[List[Dict], Optional[Dict]]: 一个元组，包含：
- final_actions_dict (List[Dict]): 最终确定的动作列表（字典格式）。
- final_target_message_dict (Optional[Dict]): 最终的目标消息（字典格式）。
"""
try:
self.planner_stats["total_plans"] += 1
return await self._enhanced_plan_flow(mode, context)
except Exception as e:
logger.error(f"规划流程出错: {e}")
self.planner_stats["failed_plans"] += 1
return [], None
async def _enhanced_plan_flow(self, mode: ChatMode, context: "StreamContext") -> Tuple[List[Dict], Optional[Dict]]:
"""执行增强版规划流程"""
try:
# 1. 生成初始 Plan
initial_plan = await self.generator.generate(mode)
unread_messages = context.get_unread_messages() if context else []
# 2. 兴趣度评分 - 只对未读消息进行评分
if unread_messages:
bot_nickname = global_config.bot.nickname
interest_scores = await self.interest_scoring.calculate_interest_scores(unread_messages, bot_nickname)
# 3. 根据兴趣度调整可用动作
if interest_scores:
latest_score = max(interest_scores, key=lambda s: s.total_score)
latest_message = next(
(msg for msg in unread_messages if msg.message_id == latest_score.message_id), None
)
should_reply, score = self.interest_scoring.should_reply(latest_score, latest_message)
reply_not_available = False
if not should_reply and "reply" in initial_plan.available_actions:
logger.info(f"兴趣度不足 ({latest_score.total_score:.2f})，移除'回复'动作。")
reply_not_available = True
# base_threshold = self.interest_scoring.reply_threshold
# 检查兴趣度是否达到非回复动作阈值
non_reply_action_interest_threshold = global_config.affinity_flow.non_reply_action_interest_threshold
if score < non_reply_action_interest_threshold:
logger.info(
f"兴趣度 {score:.3f} 低于非回复动作阈值 {non_reply_action_interest_threshold:.3f}，不执行任何动作。"
)
# 直接返回 no_action
from src.common.data_models.info_data_model import ActionPlannerInfo
no_action = ActionPlannerInfo(
action_type="no_action",
reasoning=f"兴趣度评分 {score:.3f} 未达阈值 {non_reply_action_interest_threshold:.3f}",
action_data={},
action_message=None,
)
filtered_plan = initial_plan
filtered_plan.decided_actions = [no_action]
else:
# 4. 筛选 Plan
filtered_plan = await self.filter.filter(reply_not_available, initial_plan)
# 检查filtered_plan是否有reply动作，以便记录reply action
has_reply_action = False
for decision in filtered_plan.decided_actions:
if decision.action_type == "reply":
has_reply_action = True
self.interest_scoring.record_reply_action(has_reply_action)
# 5. 使用 PlanExecutor 执行 Plan
execution_result = await self.executor.execute(filtered_plan)
# 6. 根据执行结果更新统计信息
self._update_stats_from_execution_result(execution_result)
# 7. 检查关系更新
await self.relationship_tracker.check_and_update_relationships()
# 8. 返回结果
return self._build_return_result(filtered_plan)
except Exception as e:
logger.error(f"增强版规划流程出错: {e}")
self.planner_stats["failed_plans"] += 1
return [], None
def _update_stats_from_execution_result(self, execution_result: Dict[str, any]):
"""根据执行结果更新规划器统计"""
if not execution_result:
return
successful_count = execution_result.get("successful_count", 0)
# 更新成功执行计数
self.planner_stats["successful_plans"] += successful_count
# 统计回复动作和其他动作
reply_count = 0
other_count = 0
for result in execution_result.get("results", []):
action_type = result.get("action_type", "")
if action_type in ["reply", "proactive_reply"]:
reply_count += 1
else:
other_count += 1
self.planner_stats["replies_generated"] += reply_count
self.planner_stats["other_actions_executed"] += other_count
def _build_return_result(self, plan: "Plan") -> Tuple[List[Dict], Optional[Dict]]:
"""构建返回结果"""
final_actions = plan.decided_actions or []
final_target_message = next((act.action_message for act in final_actions if act.action_message), None)
final_actions_dict = [asdict(act) for act in final_actions]
if final_target_message:
if hasattr(final_target_message, "__dataclass_fields__"):
final_target_message_dict = asdict(final_target_message)
else:
final_target_message_dict = final_target_message
else:
final_target_message_dict = None
return final_actions_dict, final_target_message_dict
def get_user_relationship(self, user_id: str) -> float:
"""获取用户关系分"""
return self.interest_scoring.get_user_relationship(user_id)
def update_interest_keywords(self, new_keywords: Dict[str, List[str]]):
"""更新兴趣关键词（已弃用，仅保留用于兼容性）"""
logger.info("传统关键词匹配已移除，此方法仅保留用于兼容性")
# 此方法已弃用，因为现在完全使用embedding匹配
def get_planner_stats(self) -> Dict[str, any]:
"""获取规划器统计"""
return self.planner_stats.copy()
def get_interest_scoring_stats(self) -> Dict[str, any]:
"""获取兴趣度评分统计"""
return {
"no_reply_count": self.interest_scoring.no_reply_count,
"max_no_reply_count": self.interest_scoring.max_no_reply_count,
"reply_threshold": self.interest_scoring.reply_threshold,
"mention_threshold": self.interest_scoring.mention_threshold,
"user_relationships": len(self.interest_scoring.user_relationships),
}
def get_relationship_stats(self) -> Dict[str, any]:
"""获取用户关系统计"""
return {
"tracking_users": len(self.relationship_tracker.tracking_users),
"relationship_history": len(self.relationship_tracker.relationship_history),
"max_tracking_users": self.relationship_tracker.max_tracking_users,
}
# 全局兴趣度评分系统实例 - 在 individuality 模块中创建

3
src/plugin_system/base/component_types.py
View File

@@ -41,7 +41,8 @@ class ActionActivationType(Enum):
class ChatMode(Enum):
"""聊天模式枚举"""
FOCUS = "focus" # Focus聊天模式
GROUP = "group" # 群聊模式
PRIVATE = "private" # 私聊模式
NORMAL = "normal" # Normal聊天模式
PROACTIVE = "proactive" # 主动思考模式
PRIORITY = "priority" # 优先级聊天模式

2
src/plugins/built_in/affinity_flow_chatter/affinity_chatter.py
View File

@@ -59,7 +59,7 @@ class AffinityChatter(BaseChatter):
unread_messages = context.get_unread_messages()
# 使用增强版规划器处理消息
actions, target_message = await self.planner.plan(mode=ChatMode.FOCUS, context=context)
actions, target_message = await self.planner.plan(mode=ChatMode.GROUP, context=context)
self.stats["plans_created"] += 1
# 执行动作（如果规划器返回了动作）

4
src/plugins/built_in/affinity_flow_chatter/plan_filter.py
View File

@@ -186,7 +186,7 @@ class ChatterPlanFilter:
if global_config.chat.at_bot_inevitable_reply:
mentioned_bonus = "\n- 有人提到你，或者at你"
if plan.mode == ChatMode.FOCUS:
if plan.mode == ChatMode.GROUP:
no_action_block = """
动作：no_action
动作描述：不选择任何动作
@@ -204,7 +204,7 @@ class ChatterPlanFilter:
"reason":"不回复的原因"
}}
"""
else: # NORMAL Mode
else: # PRIVATE Mode
no_action_block = """重要说明：
- 'reply' 表示只进行普通聊天回复，不执行任何额外动作
- 其他action表示在普通回复的基础上，执行相应的额外动作

4
src/plugins/built_in/affinity_flow_chatter/planner.py
View File

@@ -73,13 +73,13 @@ class ChatterActionPlanner:
}
async def plan(
self, mode: ChatMode = ChatMode.FOCUS, context: "StreamContext" = None
self, mode: ChatMode = ChatMode.GROUP, context: "StreamContext" = None
) -> Tuple[List[Dict], Optional[Dict]]:
"""
执行完整的增强版规划流程。
Args:
mode (ChatMode): 当前的聊天模式，默认为 FOCUS。
mode (ChatMode): 当前的聊天模式，默认为 GROUP。
context (StreamContext): 包含聊天流消息的上下文对象。
Returns: