BruceSun
/
MaiBot
mirror of https://github.com/Mai-with-u/MaiBot.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

remove：移除问题跟踪和记忆整理

pull/1348/head
SengokuCola 2025-11-09 13:59:15 +08:00
parent 94405856ff
commit 82004567a6
4 changed files with 13 additions and 756 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

216
src/memory_system/Memory_chest.py
View File

@ -14,11 +14,13 @@ from src.plugin_system.apis.message_api import get_raw_msg_by_timestamp_with_cha
from json_repair import repair_json
from src.memory_system.questions import global_conflict_tracker
from .memory_utils import (
find_best_matching_memory,
check_title_exists_fuzzy,
get_all_titles,
find_most_similar_memory_by_chat_id,
compute_merge_similarity_threshold
)
@ -82,218 +84,6 @@ class MemoryChest:
except Exception as e:
logger.error(f"[记忆管理] 按年龄权重删除记忆时出错: {e}")
return False
def _compute_merge_similarity_threshold(self) -> float:
"""
根据当前记忆数量占比动态计算合并相似度阈值。
规则：占比越高，阈值越低。
- < 60%: 0.80（更严格，避免早期误合并）
- < 80%: 0.70
- < 100%: 0.60
- < 120%: 0.50
- >= 120%: 0.45（最宽松，加速收敛）
"""
try:
current_count = MemoryChestModel.select().count()
max_count = max(1, int(global_config.memory.max_memory_number))
percentage = current_count / max_count
if percentage < 0.6:
return 0.70
elif percentage < 0.8:
return 0.60
elif percentage < 1.0:
return 0.50
elif percentage < 1.5:
return 0.40
elif percentage < 2:
return 0.30
else:
return 0.25
except Exception:
# 发生异常时使用保守阈值
return 0.70
async def build_running_content(self, chat_id: str = None) -> str:
"""
构建记忆仓库的运行内容
Args:
message_str: 消息内容
chat_id: 聊天ID，用于提取对应的运行内容
Returns:
str: 构建后的运行内容
"""
# 检查是否需要更新：基于消息数量和最新消息时间差的智能更新机制
#
# 更新机制说明：
# 1. 消息数量 > 100：直接触发更新（高频消息场景）
# 2. 消息数量 > 70 且最新消息时间差 > 30秒：触发更新（中高频消息场景）
# 3. 消息数量 > 50 且最新消息时间差 > 60秒：触发更新（中频消息场景）
# 4. 消息数量 > 30 且最新消息时间差 > 300秒：触发更新（低频消息场景）
#
# 设计理念：
# - 消息越密集，时间阈值越短，确保及时更新记忆
# - 消息越稀疏，时间阈值越长，避免频繁无意义的更新
# - 通过最新消息时间差判断消息活跃度，而非简单的总时间差
# - 平衡更新频率与性能，在保证记忆及时性的同时减少计算开销
if chat_id not in self.running_content_list:
self.running_content_list[chat_id] = {
"content": "",
"last_update_time": time.time(),
"create_time": time.time()
}
should_update = True
if chat_id and chat_id in self.running_content_list:
last_update_time = self.running_content_list[chat_id]["last_update_time"]
current_time = time.time()
# 使用message_api获取消息数量
message_list = get_raw_msg_by_timestamp_with_chat(
timestamp_start=last_update_time,
timestamp_end=current_time,
chat_id=chat_id,
limit=global_config.chat.max_context_size * 2,
)
new_messages_count = len(message_list)
# 获取最新消息的时间戳
latest_message_time = last_update_time
if message_list:
# 假设消息列表按时间排序，取最后一条消息的时间戳
latest_message = message_list[-1]
if hasattr(latest_message, 'timestamp'):
latest_message_time = latest_message.timestamp
elif isinstance(latest_message, dict) and 'timestamp' in latest_message:
latest_message_time = latest_message['timestamp']
# 计算最新消息时间与现在时间的差（秒）
latest_message_time_diff = current_time - latest_message_time
# 智能更新条件判断 - 按优先级从高到低检查
should_update = False
update_reason = ""
if global_config.memory.memory_build_frequency > 0:
if new_messages_count > 100/global_config.memory.memory_build_frequency:
# 条件1：消息数量 > 100，直接触发更新
# 适用场景：群聊刷屏、高频讨论等消息密集场景
# 无需时间限制，确保重要信息不被遗漏
should_update = True
update_reason = f"消息数量 {new_messages_count} > 100，直接触发更新"
elif new_messages_count > 70/global_config.memory.memory_build_frequency and latest_message_time_diff > 30:
# 条件2：消息数量 > 70 且最新消息时间差 > 30秒
# 适用场景：中高频讨论，但需要确保消息流已稳定
# 30秒的时间差确保不是正在进行的实时对话
should_update = True
update_reason = f"消息数量 {new_messages_count} > 70 且最新消息时间差 {latest_message_time_diff:.1f}s > 30s"
elif new_messages_count > 50/global_config.memory.memory_build_frequency and latest_message_time_diff > 60:
# 条件3：消息数量 > 50 且最新消息时间差 > 60秒
# 适用场景：中等频率讨论，等待1分钟确保对话告一段落
# 平衡及时性与稳定性
should_update = True
update_reason = f"消息数量 {new_messages_count} > 50 且最新消息时间差 {latest_message_time_diff:.1f}s > 60s"
elif new_messages_count > 30/global_config.memory.memory_build_frequency and latest_message_time_diff > 300:
# 条件4：消息数量 > 30 且最新消息时间差 > 300秒（5分钟）
# 适用场景：低频但有一定信息量的讨论
# 5分钟的时间差确保对话完全结束，避免频繁更新
should_update = True
update_reason = f"消息数量 {new_messages_count} > 30 且最新消息时间差 {latest_message_time_diff:.1f}s > 300s"
logger.debug(f"chat_id {chat_id} 更新检查: {update_reason if should_update else f'消息数量 {new_messages_count}，最新消息时间差 {latest_message_time_diff:.1f}s，不满足更新条件'}")
if should_update:
# 如果有chat_id，先提取对应的running_content
message_str = build_readable_messages(
message_list,
replace_bot_name=True,
timestamp_mode="relative",
read_mark=0.0,
show_actions=False,
remove_emoji_stickers=True,
)
# 随机从格式示例列表中选取若干行用于提示
format_candidates = [
"[概念] 是 [概念的含义(简短描述，不超过十个字)]",
"[概念] 不是 [对概念的负面含义(简短描述，不超过十个字)]",
"[概念1] 与 [概念2] 是 [概念1和概念2的关联(简短描述，不超过二十个字)]",
"[概念1] 包含 [概念2] 和 [概念3]",
"[概念1] 属于 [概念2]",
"[概念1] 的例子是 [例子1] 和 [例子2]",
"[概念] 的特征是 [特征1]、[特征2]",
"[概念1] 导致 [概念2]",
"[概念1] 需要 [条件1] 和 [条件2]",
"[概念1] 的用途是 [用途1] 和 [用途2]",
"[概念1] 与 [概念2] 的区别是 [区别点]",
"[概念] 的别名是 [别名]",
"[概念1] 包括但不限于 [概念2]、[概念3]",
"[概念] 的反义是 [反义概念]",
"[概念] 的组成有 [部分1]、[部分2]",
"[概念] 出现于 [时间或场景]",
"[概念] 的方法有 [方法1]、[方法2]",
]
selected_count = random.randint(3, 6)
selected_lines = random.sample(format_candidates, selected_count)
format_section = "\n".join(selected_lines) + "\n......(不要包含中括号)"
prompt = f"""
以下是一段你参与的聊天记录，请你在其中总结出记忆：
<聊天记录>
{message_str}
</聊天记录>
聊天记录中可能包含有效信息，也可能信息密度很低，请你根据聊天记录中的信息，总结出记忆内容
--------------------------------
对[图片]的处理：
1.除非与文本有关，不要将[图片]的内容整合到记忆中
2.如果图片与某个概念相关，将图片中的关键内容也整合到记忆中，不要写入图片原文，例如：
聊天记录（与图片有关）：
用户说：[图片1：这是一个黄色的龙形状玩偶，被一只手拿着。]
用户说：这个玩偶看起来很可爱，是我新买的奶龙
总结的记忆内容：
黄色的龙形状玩偶 是 奶龙
聊天记录（概念与图片无关）：
用户说：[图片1：这是一个台电脑，屏幕上显示了某种游戏。]
用户说：使命召唤今天发售了新一代，有没有人玩
总结的记忆内容：
使命召唤新一代 是 最新发售的游戏
请主要关注概念和知识或者时效性较强的信息！！，而不是聊天的琐事
1.不要关注诸如某个用户做了什么，说了什么，不要关注某个用户的行为，而是关注其中的概念性信息
2.概念要求精确，不啰嗦，像科普读物或教育课本那样
3.记忆为一段纯文本，逻辑清晰，指出概念的含义，并说明关系
记忆内容的格式，你必须仿照下面的格式，但不一定全部使用:
{format_section}
请仿照上述格式输出，每个知识点一句话。输出成一段平文本
现在请你输出,不要输出其他内容，注意一定要直白，白话，口语化不要浮夸，修辞。：
"""
if global_config.debug.show_prompt:
logger.info(f"记忆仓库构建运行内容 prompt: {prompt}")
else:
logger.debug(f"记忆仓库构建运行内容 prompt: {prompt}")
running_content, (reasoning_content, model_name, tool_calls) = await self.LLMRequest_build.generate_response_async(prompt)
print(f"prompt: {prompt}\n记忆仓库构建运行内容: {running_content}")
# 直接保存：每次构建后立即入库，并刷新时间戳窗口
if chat_id and running_content:
await self._save_to_database_and_clear(chat_id, running_content)
return running_content
async def get_answer_by_question(self, chat_id: str = "", question: str = "") -> str:
"""
@ -521,7 +311,7 @@ class MemoryChest:
return [], []
# 动态计算相似度阈值（占比越高阈值越低）
dynamic_threshold = self._compute_merge_similarity_threshold()
dynamic_threshold = compute_merge_similarity_threshold()
# 使用相似度匹配查找最相似的记忆（基于动态阈值）
similar_memory = find_most_similar_memory_by_chat_id(

13
src/memory_system/curious.py
View File

@ -62,11 +62,7 @@ class CuriousDetector:
show_actions=True,
)
# 检查是否已经有问题在跟踪中
existing_questions = global_conflict_tracker.get_questions_by_chat_id(self.chat_id)
if len(existing_questions) > 0:
logger.debug(f"当前已有{len(existing_questions)}个问题在跟踪中，跳过检测")
return None
# 问题跟踪功能已移除，不再检查已有问题
# 构建检测提示词
prompt = f"""你是一个严谨的聊天内容分析器。请分析以下聊天记录，检测是否存在需要提问的内容。
@ -154,11 +150,10 @@ class CuriousDetector:
if not question or not question.strip():
return False
# 记录问题到冲突追踪器，并开始跟踪
await global_conflict_tracker.track_conflict(
question=question.strip(),
# 记录问题到冲突追踪器
await global_conflict_tracker.record_conflict(
conflict_content=question.strip(),
context=context,
start_following=False,
chat_id=self.chat_id
)

241
src/memory_system/memory_management_task.py
View File

@ -1,241 +0,0 @@
# -*- coding: utf-8 -*-
import asyncio
import random
import time
from typing import List
from src.manager.async_task_manager import AsyncTask
from src.memory_system.Memory_chest import global_memory_chest
from src.common.logger import get_logger
from src.common.database.database_model import MemoryChest as MemoryChestModel, MemoryConflict
from src.config.config import global_config
logger = get_logger("memory")
class MemoryManagementTask(AsyncTask):
"""记忆管理定时任务
根据Memory_chest中的记忆数量与MAX_MEMORY_NUMBER的比例来决定执行频率：
- 小于50%：每600秒执行一次
- 大于等于50%：每300秒执行一次
每次执行时随机选择一个title，执行choose_merge_target和merge_memory，
然后删除原始记忆
"""
def __init__(self):
super().__init__(
task_name="Memory Management Task",
wait_before_start=10, # 启动后等待10秒再开始
run_interval=300 # 默认300秒间隔，会根据记忆数量动态调整
)
self.max_memory_number = global_config.memory.max_memory_number
async def start_task(self, abort_flag: asyncio.Event):
"""重写start_task方法，支持动态调整执行间隔"""
if self.wait_before_start > 0:
# 等待指定时间后开始任务
await asyncio.sleep(self.wait_before_start)
while not abort_flag.is_set():
await self.run()
# 动态调整执行间隔
current_interval = self._calculate_interval()
logger.info(f"[记忆管理] 下次执行间隔: {current_interval}秒")
if current_interval > 0:
await asyncio.sleep(current_interval)
else:
break
def _calculate_interval(self) -> int:
"""根据当前记忆数量计算执行间隔"""
try:
current_count = self._get_memory_count()
percentage = current_count / self.max_memory_number
if percentage < 0.6:
# 小于50%，每600秒执行一次
return 3600
elif percentage < 1:
# 大于等于50%，每300秒执行一次
return 1800
elif percentage < 1.5:
# 大于等于100%，每120秒执行一次
return 600
elif percentage < 1.8:
return 120
else:
return 30
except Exception as e:
logger.error(f"[记忆管理] 计算执行间隔时出错: {e}")
return 300 # 默认300秒
def _get_memory_count(self) -> int:
"""获取当前记忆数量"""
try:
count = MemoryChestModel.select().count()
logger.debug(f"[记忆管理] 当前记忆数量: {count}")
return count
except Exception as e:
logger.error(f"[记忆管理] 获取记忆数量时出错: {e}")
return 0
async def run(self):
"""执行记忆管理任务"""
try:
# 获取当前记忆数量
current_count = self._get_memory_count()
percentage = current_count / self.max_memory_number
logger.info(f"当前记忆数量: {current_count}/{self.max_memory_number} ({percentage:.1%})")
# 当占比 > 1.6 时，持续删除直到占比 <= 1.6（越老/越新更易被删）
if percentage > 2:
logger.info("记忆过多，开始遗忘记忆")
while True:
if percentage <= 1.8:
break
removed = global_memory_chest.remove_one_memory_by_age_weight()
if not removed:
logger.warning("没有可删除的记忆，停止连续删除")
break
# 重新计算占比
current_count = self._get_memory_count()
percentage = current_count / self.max_memory_number
logger.info(f"遗忘进度: 当前 {current_count}/{self.max_memory_number} ({percentage:.1%})")
logger.info("遗忘记忆结束")
# 如果记忆数量为0，跳过执行
if current_count < 10:
return
# 随机选择一个记忆标题和chat_id
selected_title, selected_chat_id = self._get_random_memory_title()
if not selected_title:
logger.warning("无法获取随机记忆标题，跳过执行")
return
# 执行choose_merge_target获取相关记忆（标题与内容）
related_titles, related_contents = await global_memory_chest.choose_merge_target(selected_title, selected_chat_id)
if not related_titles or not related_contents:
logger.info("无合适合并内容，跳过本次合并")
return
logger.info(f"{selected_chat_id} 为 [{selected_title}] 找到 {len(related_contents)} 条相关记忆:{related_titles}")
# 执行merge_memory合并记忆
merged_title, merged_content = await global_memory_chest.merge_memory(related_contents,selected_chat_id)
if not merged_title or not merged_content:
logger.warning("[记忆管理] 记忆合并失败，跳过删除")
return
logger.info(f"记忆合并成功，新标题: {merged_title}")
# 删除原始记忆（包括选中的标题和相关的记忆标题）
titles_to_delete = [selected_title] + related_titles
deleted_count = self._delete_original_memories(titles_to_delete)
logger.info(f"已删除 {deleted_count} 条原始记忆")
except Exception as e:
logger.error(f"[记忆管理] 执行记忆管理任务时发生错误: {e}", exc_info=True)
def _get_random_memory_title(self) -> tuple[str, str]:
"""随机获取一个记忆标题和对应的chat_id"""
try:
# 获取所有记忆记录
all_memories = MemoryChestModel.select()
if not all_memories:
return "", ""
# 随机选择一个记忆
selected_memory = random.choice(list(all_memories))
return selected_memory.title, selected_memory.chat_id or ""
except Exception as e:
logger.error(f"[记忆管理] 获取随机记忆标题时发生错误: {e}")
return "", ""
def _delete_original_memories(self, related_titles: List[str]) -> int:
"""按标题删除原始记忆"""
try:
deleted_count = 0
# 删除相关记忆（通过标题匹配）
for title in related_titles:
try:
# 通过标题查找并删除对应的记忆
memories_to_delete = MemoryChestModel.select().where(MemoryChestModel.title == title)
for memory in memories_to_delete:
MemoryChestModel.delete().where(MemoryChestModel.id == memory.id).execute()
deleted_count += 1
logger.debug(f"[记忆管理] 删除相关记忆: {memory.title}")
except Exception as e:
logger.error(f"[记忆管理] 删除相关记忆时出错: {e}")
continue
return deleted_count
except Exception as e:
logger.error(f"[记忆管理] 删除原始记忆时发生错误: {e}")
return 0
class MemoryConflictCleanupTask(AsyncTask):
"""记忆冲突清理定时任务
定期清理 memory_conflicts 表中 create_time 较早（7天前）且 answer 为空的项目
默认每小时执行一次
"""
def __init__(self, cleanup_days: int = 7, run_interval: int = 3600):
"""
初始化清理任务
Args:
cleanup_days: 清理多少天前的记录，默认7天
run_interval: 执行间隔（秒），默认3600秒（1小时）
"""
super().__init__(
task_name="Memory Conflict Cleanup Task",
wait_before_start=60, # 启动后等待60秒再开始
run_interval=run_interval
)
self.cleanup_days = cleanup_days
async def run(self):
"""执行清理任务"""
try:
current_time = time.time()
# 计算7天前的时间戳
cutoff_time = current_time - (self.cleanup_days * 24 * 60 * 60)
logger.info(f"[冲突清理] 开始清理 {self.cleanup_days} 天前且 answer 为空的冲突记录（截止时间: {cutoff_time}）")
# 查询需要清理的记录：create_time < cutoff_time 且 answer 为空
# answer 为空的条件：answer IS NULL 或 answer == ''
query = MemoryConflict.select().where(
(MemoryConflict.create_time < cutoff_time) &
((MemoryConflict.answer.is_null()) | (MemoryConflict.answer == ''))
)
# 先统计要删除的数量
deleted_count = query.count()
# 批量删除
if deleted_count > 0:
deleted = MemoryConflict.delete().where(
(MemoryConflict.create_time < cutoff_time) &
((MemoryConflict.answer.is_null()) | (MemoryConflict.answer == ''))
).execute()
deleted_count = deleted
if deleted_count > 0:
logger.info(f"[冲突清理] 成功清理 {deleted_count} 条过期且未回答的冲突记录")
else:
logger.debug("[冲突清理] 没有需要清理的记录")
except Exception as e:
logger.error(f"[冲突清理] 执行清理任务时发生错误: {e}", exc_info=True)

299
src/memory_system/questions.py
View File

@ -1,120 +1,12 @@
import time
import asyncio
from src.common.logger import get_logger
from src.common.database.database_model import MemoryConflict
from src.chat.utils.chat_message_builder import (
get_raw_msg_by_timestamp_with_chat,
build_readable_messages,
)
from src.llm_models.utils_model import LLMRequest
from src.config.config import model_config, global_config
from typing import List
from src.config.config import model_config
from src.memory_system.memory_utils import parse_md_json
logger = get_logger("conflict_tracker")
class QuestionTracker:
"""
用于跟踪一个问题在后续聊天中的解答情况
"""
def __init__(self, question: str, chat_id: str, context: str = "") -> None:
self.question = question
self.chat_id = chat_id
now = time.time()
self.context = context
self.start_time = now
self.last_read_time = now
self.last_judge_time = now # 上次判定的时间
self.judge_debounce_interval = 10.0 # 判定防抖间隔：10秒
self.consecutive_end_count = 0 # 连续END计数
self.active = True
# 将 LLM 实例作为类属性，使用 utils 模型
self.llm_request = LLMRequest(model_set=model_config.model_task_config.utils, request_type="conflict.judge")
def stop(self) -> None:
self.active = False
def should_judge_now(self) -> bool:
"""
检查是否应该进行判定（防抖检查）
Returns:
bool: 是否可以判定
"""
now = time.time()
# 检查是否已经过了10秒的防抖间隔
return (now - self.last_judge_time) >= self.judge_debounce_interval
def __eq__(self, other) -> bool:
"""比较两个追踪器是否相等（基于问题内容和聊天ID）"""
if not isinstance(other, QuestionTracker):
return False
return self.question == other.question and self.chat_id == other.chat_id
def __hash__(self) -> int:
"""为对象提供哈希值，支持集合操作"""
return hash((self.question, self.chat_id))
async def judge_answer(self, conversation_text: str,chat_len: int) -> tuple[bool, str, str]:
"""
使用模型判定问题是否已得到解答。
Returns:
tuple[bool, str, str]: (是否结束跟踪, 结束原因或答案, 判定类型)
- True: 结束跟踪（已解答、话题转向等）
- False: 继续跟踪
判定类型: "ANSWERED", "END", "CONTINUE"
"""
end_prompt = ""
if chat_len > 20:
end_prompt = "\n- 如果最新20条聊天记录内容与问题无关，话题已转向其他方向，请只输出：END"
prompt = f"""你是一个严谨的判定器。下面给出聊天记录以及一个问题。
任务：判断在这段聊天中，该问题是否已经得到明确解答。
**你必须严格按照聊天记录的内容，不要添加额外的信息**
输出规则：
- 如果聊天记录内容的信息已解答问题，请只输出：YES: <简短答案>{end_prompt}
- 如果问题尚未解答但聊天仍在相关话题上，请只输出：NO
**问题**
{self.question}
**聊天记录**
{conversation_text}
"""
if global_config.debug.show_prompt:
logger.info(f"判定提示词: {prompt}")
else:
logger.debug("已发送判定提示词")
result_text, _ = await self.llm_request.generate_response_async(prompt, temperature=0.5)
logger.info(f"判定结果: {prompt}\n{result_text}")
# 更新上次判定时间
self.last_judge_time = time.time()
if not result_text:
return False, "", "CONTINUE"
text = result_text.strip()
if text.upper().startswith("YES:"):
answer = text[4:].strip()
return True, answer, "ANSWERED"
if text.upper().startswith("YES"):
# 兼容仅输出 YES 或 YES <answer>
answer = text[3:].strip().lstrip(":").strip()
return True, answer, "ANSWERED"
if text.upper().startswith("END"):
# 聊天内容与问题无关，放弃该问题思考
return True, "话题已转向其他方向，放弃该问题思考", "END"
return False, "", "CONTINUE"
class ConflictTracker:
"""
记忆整合冲突追踪器
@ -122,31 +14,19 @@ class ConflictTracker:
用于记录和存储记忆整合过程中的冲突内容
"""
def __init__(self):
self.question_tracker_list:List[QuestionTracker] = []
self.LLMRequest_tracker = LLMRequest(
model_set=model_config.model_task_config.utils,
request_type="conflict_tracker",
)
def get_questions_by_chat_id(self, chat_id: str) -> List[QuestionTracker]:
return [tracker for tracker in self.question_tracker_list if tracker.chat_id == chat_id]
async def track_conflict(self, question: str, context: str = "",start_following: bool = False,chat_id: str = "") -> bool:
"""
跟踪冲突内容
"""
tracker = QuestionTracker(question.strip(), chat_id, context)
self.question_tracker_list.append(tracker)
asyncio.create_task(self._follow_and_record(tracker, question.strip()))
return True
async def record_conflict(self, conflict_content: str, context: str = "",start_following: bool = False,chat_id: str = "") -> bool:
async def record_conflict(self, conflict_content: str, context: str = "", chat_id: str = "") -> bool:
"""
记录冲突内容
Args:k
Args:
conflict_content: 冲突内容
context: 上下文
chat_id: 聊天ID
Returns:
bool: 是否成功记录
@ -155,15 +35,7 @@ class ConflictTracker:
if not conflict_content or conflict_content.strip() == "":
return False
# 若需要跟随后续消息以判断是否得到解答，则进入跟踪流程
if start_following and chat_id:
tracker = QuestionTracker(conflict_content.strip(), chat_id, context)
self.question_tracker_list.append(tracker)
# 后台启动跟踪任务，避免阻塞
asyncio.create_task(self._follow_and_record(tracker, conflict_content.strip()))
return True
# 默认：直接记录，不进行跟踪
# 直接记录，不进行跟踪
MemoryConflict.create(
conflict_content=conflict_content,
create_time=time.time(),
@ -179,164 +51,6 @@ class ConflictTracker:
logger.error(f"记录冲突内容时出错: {e}")
return False
async def _follow_and_record(self, tracker: QuestionTracker, original_question: str) -> None:
"""
后台任务：跟踪问题是否被解答，并写入数据库。
"""
try:
max_duration = 10 * 60 # 30 分钟
max_messages = 50 # 最多 100 条消息
poll_interval = 2.0 # 秒
logger.info(f"开始跟踪问题: {original_question}")
while tracker.active:
now_ts = time.time()
# 终止条件：时长达到上限
if now_ts - tracker.start_time >= max_duration:
logger.info("问题跟踪达到10分钟上限，判定为未解答")
break
# 统计最近一段是否有新消息（不过滤机器人，过滤命令）
recent_msgs = get_raw_msg_by_timestamp_with_chat(
chat_id=tracker.chat_id,
timestamp_start=tracker.last_read_time,
timestamp_end=now_ts,
limit=30,
limit_mode="latest",
filter_bot=False,
filter_command=True,
)
if len(recent_msgs) > 0:
tracker.last_read_time = now_ts
# 统计从开始到现在的总消息数（用于触发100条上限）
all_msgs = get_raw_msg_by_timestamp_with_chat(
chat_id=tracker.chat_id,
timestamp_start=tracker.start_time,
timestamp_end=now_ts,
limit=0,
limit_mode="latest",
filter_bot=False,
filter_command=True,
)
# 检查是否应该进行判定（防抖检查）
if not tracker.should_judge_now():
logger.debug(f"判定防抖中，跳过本次判定: {tracker.question}")
await asyncio.sleep(poll_interval)
continue
# 构建可读聊天文本
chat_text = build_readable_messages(
all_msgs,
replace_bot_name=True,
timestamp_mode="relative",
read_mark=0.0,
truncate=False,
show_actions=False,
show_pic=False,
remove_emoji_stickers=True,
)
chat_len = len(all_msgs)
# 让小模型判断是否有答案
answered, answer_text, judge_type = await tracker.judge_answer(chat_text,chat_len)
if judge_type == "ANSWERED":
# 问题已解答，直接结束跟踪
logger.info("问题已得到解答，结束跟踪并写入答案")
await self.add_or_update_conflict(
conflict_content=tracker.question,
create_time=tracker.start_time,
update_time=time.time(),
answer=answer_text or "",
chat_id=tracker.chat_id,
)
return
elif judge_type == "END":
# 话题转向，增加END计数
tracker.consecutive_end_count += 1
logger.info(f"话题已转向，连续END次数: {tracker.consecutive_end_count}")
if tracker.consecutive_end_count >= 2:
# 连续两次END，结束跟踪
logger.info("连续两次END，结束跟踪")
break
else:
# 第一次END，重置计数器并继续跟踪
logger.info("第一次END，继续跟踪")
continue
elif judge_type == "CONTINUE":
# 继续跟踪，重置END计数器
tracker.consecutive_end_count = 0
continue
if len(all_msgs) >= max_messages:
logger.info("问题跟踪达到100条消息上限，判定为未解答")
logger.info(f"追踪结束：{tracker.question}")
break
# 无新消息时稍作等待
await asyncio.sleep(poll_interval)
# 未获取到答案，检查是否需要删除记录
# 查找现有的冲突记录
existing_conflict = MemoryConflict.get_or_none(
MemoryConflict.conflict_content == original_question,
MemoryConflict.chat_id == tracker.chat_id
)
if existing_conflict:
# 检查raise_time是否大于3且没有答案
current_raise_time = getattr(existing_conflict, "raise_time", 0) or 0
if current_raise_time > 0 and not existing_conflict.answer:
# 删除该条目
await self.delete_conflict(original_question, tracker.chat_id)
logger.info(f"追踪结束后删除条目(raise_time={current_raise_time}且无答案): {original_question}")
else:
# 更新记录但不删除
await self.add_or_update_conflict(
conflict_content=original_question,
create_time=existing_conflict.create_time,
update_time=time.time(),
answer="",
chat_id=tracker.chat_id,
)
logger.info(f"记录冲突内容(未解答): {len(original_question)} 字符")
else:
# 如果没有现有记录，创建新记录
await self.add_or_update_conflict(
conflict_content=original_question,
create_time=time.time(),
update_time=time.time(),
answer="",
chat_id=tracker.chat_id,
)
logger.info(f"记录冲突内容(未解答): {len(original_question)} 字符")
logger.info(f"问题跟踪结束：{original_question}")
except Exception as e:
logger.error(f"后台问题跟踪任务异常: {e}")
finally:
# 无论任务成功还是失败，都要从追踪列表中移除
tracker.stop()
self.remove_tracker(tracker)
def remove_tracker(self, tracker: QuestionTracker) -> None:
"""
从追踪列表中移除指定的追踪器
Args:
tracker: 要移除的追踪器对象
"""
try:
if tracker in self.question_tracker_list:
self.question_tracker_list.remove(tracker)
logger.info(f"已从追踪列表中移除追踪器: {tracker.question}")
else:
logger.warning(f"尝试移除不存在的追踪器: {tracker.question}")
except Exception as e:
logger.error(f"移除追踪器时出错: {e}")
async def add_or_update_conflict(
self,
conflict_content: str,
@ -429,7 +143,6 @@ class ConflictTracker:
await self.record_conflict(
conflict_content=question["question"],
context=reasoning_content,
start_following=False,
chat_id=chat_id,
)
return True