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

feat(affinity-flow): 实现回复后关系追踪系统集成

Browse Source 
- 在relationship_tracker.py中添加数据库支持的回复后关系追踪功能
- 新增UserRelationships数据库模型存储用户关系数据
- 集成全局关系追踪器到planner和interest_scoring系统
- 优化兴趣度评分系统的关系分获取逻辑，优先使用数据库存储的关系分
- 在plan_executor中执行回复后关系追踪，分析用户反应并更新关系
- 添加LLM响应清理功能确保JSON解析稳定性
- 更新模型配置模板添加relationship_tracker模型配置
This commit is contained in:
Windpicker-owo
2025-09-19 11:28:37 +08:00
parent a2225cad3a
commit 812e4d76f8
12 changed files with 715 additions and 116 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -2,6 +2,7 @@
PlanExecutor: 接收 Plan 对象并执行其中的所有动作。
集成用户关系追踪机制，自动记录交互并更新关系。
"""
import asyncio
import time
from typing import Dict, List
@@ -94,7 +95,9 @@ class PlanExecutor:
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}")
logger.info(
f"动作执行完成: 总数={len(plan.decided_actions)}, 成功={successful_count}, 失败={len(execution_results) - successful_count}"
)
return {
"executed_count": len(plan.decided_actions),
@@ -123,7 +126,7 @@ class PlanExecutor:
try:
logger.info(f"执行回复动作: {action_info.action_type}, 原因: {action_info.reasoning}")
if action_info.action_message.get("user_id","") == str(global_config.bot.qq_account):
if action_info.action_message.get("user_id", "") == str(global_config.bot.qq_account):
logger.warning("尝试回复自己，跳过此动作以防止死循环。")
return {
"action_type": action_info.action_type,
@@ -143,8 +146,7 @@ class PlanExecutor:
# 通过动作管理器执行回复
reply_content = await self.action_manager.execute_action(
action_name=action_info.action_type,
**action_params
action_name=action_info.action_type, **action_params
)
success = True
@@ -185,13 +187,15 @@ class PlanExecutor:
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,
})
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)
@@ -215,10 +219,7 @@ class PlanExecutor:
}
# 通过动作管理器执行动作
await self.action_manager.execute_action(
action_name=action_info.action_type,
**action_params
)
await self.action_manager.execute_action(action_name=action_info.action_type, **action_params)
success = True
logger.info(f"其他动作执行成功: {action_info.action_type}")
@@ -239,30 +240,49 @@ class PlanExecutor:
}
async def _track_user_interaction(self, action_info: ActionPlannerInfo, plan: Plan, reply_content: str):
"""追踪用户交互"""
"""追踪用户交互 - 集成回复后关系追踪"""
try:
if not action_info.action_message:
return
# 获取用户信息
user_id = action_info.action_message.user_id
user_name = action_info.action_message.user_nickname or user_id
user_message = action_info.action_message.content
# 获取用户信息 - 处理对象和字典两种情况
if hasattr(action_info.action_message, "user_id"):
# 对象情况
user_id = action_info.action_message.user_id
user_name = getattr(action_info.action_message, "user_nickname", user_id) or user_id
user_message = getattr(action_info.action_message, "content", "")
else:
# 字典情况
user_id = action_info.action_message.get("user_id", "")
user_name = action_info.action_message.get("user_nickname", user_id) 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()
reply_timestamp=time.time(),
)
logger.debug(f"已添加用户交互追踪: {user_id}")
# 执行新的回复后关系追踪
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]:
"""获取执行统计信息"""
@@ -308,4 +328,4 @@ class PlanExecutor:
"timestamp": time.time() - (len(recent_times) - i) * 60, # 估算时间戳
}
for i, time_val in enumerate(recent_times)
]
]

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

@@ -2,6 +2,7 @@
主规划器入口，负责协调 PlanGenerator, PlanFilter, 和 PlanExecutor。
集成兴趣度评分系统和用户关系追踪机制，实现智能化的聊天决策。
"""
from dataclasses import asdict
from typing import Dict, List, Optional, Tuple
@@ -17,10 +18,11 @@ from src.config.config import global_config
from src.plugin_system.base.component_types import ChatMode
# 导入提示词模块以确保其被初始化
from src.chat.planner_actions import planner_prompts #noqa
from src.chat.planner_actions import planner_prompts # noqa
logger = get_logger("planner")
class ActionPlanner:
"""
增强版ActionPlanner，集成兴趣度评分和用户关系追踪机制。
@@ -49,8 +51,25 @@ class ActionPlanner:
# 初始化兴趣度评分系统
self.interest_scoring = InterestScoringSystem()
# 初始化用户关系追踪器
self.relationship_tracker = UserRelationshipTracker(self.interest_scoring)
# 尝试获取全局关系追踪器，如果没有则创建新的
try:
from src.chat.affinity_flow.relationship_integration import get_relationship_tracker
global_relationship_tracker = get_relationship_tracker()
if global_relationship_tracker:
# 使用全局关系追踪器
self.relationship_tracker = global_relationship_tracker
# 设置兴趣度评分系统的关系追踪器引用
self.interest_scoring.relationship_tracker = self.relationship_tracker
logger.info("使用全局关系追踪器")
else:
# 创建新的关系追踪器
self.relationship_tracker = UserRelationshipTracker(self.interest_scoring)
logger.info("创建新的关系追踪器实例")
except Exception as e:
logger.warning(f"获取全局关系追踪器失败: {e}")
# 创建新的关系追踪器
self.relationship_tracker = UserRelationshipTracker(self.interest_scoring)
# 设置执行器的关系追踪器
self.executor.set_relationship_tracker(self.relationship_tracker)
@@ -64,7 +83,9 @@ class ActionPlanner:
"other_actions_executed": 0,
}
async def plan(self, mode: ChatMode = ChatMode.FOCUS, message_data: dict = None) -> Tuple[List[Dict], Optional[Dict]]:
async def plan(
self, mode: ChatMode = ChatMode.FOCUS, message_data: dict = None
) -> Tuple[List[Dict], Optional[Dict]]:
"""
执行完整的增强版规划流程。
@@ -86,13 +107,14 @@ class ActionPlanner:
return await self._enhanced_plan_flow(mode, unread_messages or [])
except Exception as e:
logger.error(f"规划流程出错: {e}")
self.planner_stats["failed_plans"] += 1
return [], None
async def _enhanced_plan_flow(self, mode: ChatMode, unread_messages: List[Dict]) -> Tuple[List[Dict], Optional[Dict]]:
async def _enhanced_plan_flow(
self, mode: ChatMode, unread_messages: List[Dict]
) -> Tuple[List[Dict], Optional[Dict]]:
"""执行增强版规划流程"""
try:
# 1. 生成初始 Plan
@@ -101,9 +123,7 @@ class ActionPlanner:
# 2. 兴趣度评分 - 只对未读消息进行评分
if unread_messages:
bot_nickname = global_config.bot.nickname
interest_scores = await self.interest_scoring.calculate_interest_scores(
unread_messages, bot_nickname
)
interest_scores = await self.interest_scoring.calculate_interest_scores(unread_messages, bot_nickname)
# 3. 根据兴趣度调整可用动作
if interest_scores:
@@ -123,6 +143,7 @@ class ActionPlanner:
logger.info(f"📊 最低要求: 阈值({base_threshold:.3f}) × 0.8 = {threshold_requirement:.3f}")
# 直接返回 no_action
from src.common.data_models.info_data_model import ActionPlannerInfo
no_action = ActionPlannerInfo(
action_type="no_action",
reasoning=f"兴趣度评分 {score:.3f} 未达阈值80% {threshold_requirement:.3f}",
@@ -133,7 +154,7 @@ class ActionPlanner:
filtered_plan.decided_actions = [no_action]
else:
# 4. 筛选 Plan
filtered_plan = await self.filter.filter(reply_not_available,initial_plan)
filtered_plan = await self.filter.filter(reply_not_available, initial_plan)
# 检查filtered_plan是否有reply动作，以便记录reply action
has_reply_action = False
@@ -158,42 +179,40 @@ class ActionPlanner:
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_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__'):
if hasattr(final_target_message, "__dataclass_fields__"):
final_target_message_dict = asdict(final_target_message)
else:
final_target_message_dict = final_target_message
@@ -234,4 +253,4 @@ class ActionPlanner:
}
# 全局兴趣度评分系统实例 - 在 individuality 模块中创建
# 全局兴趣度评分系统实例 - 在 individuality 模块中创建