feat:将关系改为及时构建

Author: SengokuCola

src/chat/focus_chat/info_processors/relationship_processor.py

@@ -16,6 +16,8 @@
 from json_repair import repair_json
 from src.person_info.person_info import person_info_manager
 import json
+import asyncio
+from src.chat.utils.chat_message_builder import get_raw_msg_by_timestamp_with_chat
 
 logger = get_logger("processor")
 
@@ -37,6 +39,7 @@ def init_prompt():
 1. 根据聊天记录的需求，如果需要你和某个人的信息，请输出你和这个人之间精简的信息
 2. 如果没有特别需要提及的信息，就不用输出这个人的信息
 3. 如果有人问你对他的看法或者关系，请输出你和这个人之间的信息
+4. 你可以完全不输出任何信息，或者不输出某个人
 
 请从这些信息中提取出你对某人的了解信息，信息提取成一串文本：
 
@@ -58,6 +61,11 @@ def __init__(self, subheartflow_id: str):
         super().__init__()
 
         self.subheartflow_id = subheartflow_id
+        self.person_cache: Dict[str, Dict[str, any]] = {}  # {person_id: {"info": str, "ttl": int, "start_time": float}}
+        self.pending_updates: Dict[str, Dict[str, any]] = (
+            {}
+        )  # {person_id: {"start_time": float, "end_time": float, "grace_period_ttl": int, "chat_id": str}}
+        self.grace_period_rounds = 5
 
         self.llm_model = LLMRequest(
             model=global_config.model.relation,
@@ -91,54 +99,109 @@ async def process_info(
         return [relation_info]
 
     async def relation_identify(
-        self, observations: Optional[List[Observation]] = None,
+        self,
+        observations: Optional[List[Observation]] = None,
     ):
         """
         在回复前进行思考，生成内心想法并收集工具调用结果
-
-        参数:
-            observations: 观察信息
-
-        返回:
-            如果return_prompt为False:
-                tuple: (current_mind, past_mind) 当前想法和过去的想法列表
-            如果return_prompt为True:
-                tuple: (current_mind, past_mind, prompt) 当前想法、过去的想法列表和使用的prompt
         """
-
-        if observations is None:
-            observations = []
-        for observation in observations:
-            if isinstance(observation, ChattingObservation):
-                # 获取聊天元信息
-                is_group_chat = observation.is_group_chat
-                chat_target_info = observation.chat_target_info
-                chat_target_name = "对方"  # 私聊默认名称
-                if not is_group_chat and chat_target_info:
-                    # 优先使用person_name，其次user_nickname，最后回退到默认值
-                    chat_target_name = (
-                        chat_target_info.get("person_name") or chat_target_info.get("user_nickname") or chat_target_name
+        # 0. 从观察信息中提取所需数据
+        person_list = []
+        chat_observe_info = ""
+        is_group_chat = False
+        if observations:
+            for observation in observations:
+                if isinstance(observation, ChattingObservation):
+                    is_group_chat = observation.is_group_chat
+                    chat_observe_info = observation.get_observe_info()
+                    person_list = observation.person_list
+                    break
+
+        # 1. 处理等待更新的条目（仅检查TTL，不检查是否被重提）
+        persons_to_update_now = []  # 等待期结束，需要立即更新的用户
+        for person_id, data in list(self.pending_updates.items()):
+            data["grace_period_ttl"] -= 1
+            if data["grace_period_ttl"] <= 0:
+                persons_to_update_now.append(person_id)
+
+        # 触发等待期结束的更新任务
+        for person_id in persons_to_update_now:
+            if person_id in self.pending_updates:
+                update_data = self.pending_updates.pop(person_id)
+                logger.info(f"{self.log_prefix} 用户 {person_id} 等待期结束，开始印象更新。")
+                asyncio.create_task(
+                    self.update_impression_on_cache_expiry(
+                        person_id, update_data["chat_id"], update_data["start_time"], update_data["end_time"]
                     )
-                # 获取聊天内容
-                chat_observe_info = observation.get_observe_info()
-                person_list = observation.person_list
-
-        nickname_str = ""
-        for nicknames in global_config.bot.alias_names:
-            nickname_str += f"{nicknames},"
+                )
+
+        # 2. 维护活动缓存，并将过期条目移至等待区或立即更新
+        persons_moved_to_pending = []
+        for person_id, cache_data in self.person_cache.items():
+            cache_data["ttl"] -= 1
+            if cache_data["ttl"] <= 0:
+                persons_moved_to_pending.append(person_id)
+
+        for person_id in persons_moved_to_pending:
+            if person_id in self.person_cache:
+                cache_item = self.person_cache.pop(person_id)
+                start_time = cache_item.get("start_time")
+                end_time = time.time()
+                time_elapsed = end_time - start_time
+
+                impression_messages = get_raw_msg_by_timestamp_with_chat(self.subheartflow_id, start_time, end_time)
+                message_count = len(impression_messages)
+
+                if message_count > 50 or (time_elapsed > 600 and message_count > 20):
+                    logger.info(
+                        f"{self.log_prefix} 用户 {person_id} 缓存过期，满足立即更新条件 (消息数: {message_count}, 持续时间: {time_elapsed:.0f}s)，立即更新。"
+                    )
+                    asyncio.create_task(
+                        self.update_impression_on_cache_expiry(person_id, self.subheartflow_id, start_time, end_time)
+                    )
+                else:
+                    logger.info(f"{self.log_prefix} 用户 {person_id} 缓存过期，进入更新等待区。")
+                    self.pending_updates[person_id] = {
+                        "start_time": start_time,
+                        "end_time": end_time,
+                        "grace_period_ttl": self.grace_period_rounds,
+                        "chat_id": self.subheartflow_id,
+                    }
+
+        # 3. 准备LLM输入和直接使用缓存
+        if not person_list:
+            return ""
+
+        cached_person_info_str = ""
+        persons_to_process = []
+        person_name_list_for_llm = []
+
+        for person_id in person_list:
+            if person_id in self.person_cache:
+                logger.info(f"{self.log_prefix} 关系识别 (缓存): {person_id}")
+                person_name = await person_info_manager.get_value(person_id, "person_name")
+                info = self.person_cache[person_id]["info"]
+                cached_person_info_str += f"你对 {person_name} 的了解：{info}\n"
+            else:
+                # 所有不在活动缓存中的用户（包括等待区的）都将由LLM处理
+                persons_to_process.append(person_id)
+                person_name_list_for_llm.append(await person_info_manager.get_value(person_id, "person_name"))
+
+        # 4. 如果没有需要LLM处理的人员，直接返回缓存信息
+        if not persons_to_process:
+            final_result = cached_person_info_str.strip()
+            if final_result:
+                logger.info(f"{self.log_prefix} 关系识别 (全部缓存): {final_result}")
+            return final_result
+
+        # 5. 为需要处理的人员准备LLM prompt
+        nickname_str = ",".join(global_config.bot.alias_names)
         name_block = f"你的名字是{global_config.bot.nickname},你的昵称有{nickname_str}，有人也会用这些昵称称呼你。"
-
-        if is_group_chat:
-            relation_prompt_init = "你对群聊里的人的印象是：\n"
-        else:
-            relation_prompt_init = "你对对方的印象是：\n"
-        
+        relation_prompt_init = "你对群聊里的人的印象是：\n" if is_group_chat else "你对对方的印象是：\n"
         relation_prompt = ""
-        person_name_list = []
-        for person in person_list:
-            relation_prompt += f"{await relationship_manager.build_relationship_info(person, is_id=True)}\n\n"
-            person_name_list.append(await person_info_manager.get_value(person, "person_name"))
-            
+        for person_id in persons_to_process:
+            relation_prompt += f"{await relationship_manager.build_relationship_info(person_id, is_id=True)}\n\n"
+
         if relation_prompt:
             relation_prompt = relation_prompt_init + relation_prompt
         else:
@@ -151,45 +214,76 @@ async def relation_identify(
             chat_observe_info=chat_observe_info,
         )
 
-        # print(prompt)
-
-        content = ""
+        # 6. 调用LLM并处理结果
+        newly_processed_info_str = ""
         try:
             logger.info(f"{self.log_prefix} 关系识别prompt: \n{prompt}\n")
             content, _ = await self.llm_model.generate_response_async(prompt=prompt)
-            if not content:
+            if content:
+                print(f"content: {content}")
+                content_json = json.loads(repair_json(content))
+
+                for person_name, person_info in content_json.items():
+                    if person_name in person_name_list_for_llm:
+                        try:
+                            idx = person_name_list_for_llm.index(person_name)
+                            person_id = persons_to_process[idx]
+
+                            # 关键：检查此人是否在等待区，如果是，则为"唤醒"
+                            start_time = time.time()  # 新用户的默认start_time
+                            if person_id in self.pending_updates:
+                                logger.info(f"{self.log_prefix} 用户 {person_id} 在等待期被LLM重提，重新激活缓存。")
+                                revived_item = self.pending_updates.pop(person_id)
+                                start_time = revived_item["start_time"]
+
+                            self.person_cache[person_id] = {
+                                "info": person_info,
+                                "ttl": 5,
+                                "start_time": start_time,
+                            }
+                            newly_processed_info_str += f"你对 {person_name} 的了解：{person_info}\n"
+                        except (ValueError, IndexError):
+                            continue
+            else:
                 logger.warning(f"{self.log_prefix} LLM返回空结果，关系识别失败。")
-            
-            print(f"content: {content}")
-            
-            content = repair_json(content)
-            content = json.loads(content)
-            
-            person_info_str = ""
-            
-            for person_name, person_info in content.items():
-                # print(f"person_name: {person_name}, person_info: {person_info}")
-                # print(f"person_list: {person_name_list}")
-                if person_name not in person_name_list:
-                    continue
-                person_str = f"你对 {person_name} 的了解：{person_info}\n"
-                person_info_str += person_str
-                
-                
+
         except Exception as e:
-            # 处理总体异常
             logger.error(f"{self.log_prefix} 执行LLM请求或处理响应时出错: {e}")
             logger.error(traceback.format_exc())
-            person_info_str = "关系识别过程中出现错误"
+            newly_processed_info_str = "关系识别过程中出现错误"
+
+        # 7. 合并缓存和新处理的信息
+        person_info_str = (cached_person_info_str + newly_processed_info_str).strip()
 
         if person_info_str == "None":
             person_info_str = ""
-            
-        # 记录初步思考结果
 
         logger.info(f"{self.log_prefix} 关系识别: {person_info_str}")
 
         return person_info_str
 
+    async def update_impression_on_cache_expiry(
+        self, person_id: str, chat_id: str, start_time: float, end_time: float
+    ):
+        """
+        在缓存过期时，获取聊天记录并更新用户印象
+        """
+        logger.info(f"缓存过期，开始为 {person_id} 更新印象。时间范围：{start_time} -> {end_time}")
+        try:
+            
+
+            impression_messages = get_raw_msg_by_timestamp_with_chat(chat_id, start_time, end_time)
+            if impression_messages:
+                logger.info(f"为 {person_id} 获取到 {len(impression_messages)} 条消息用于印象更新。")
+                await relationship_manager.update_person_impression(
+                    person_id=person_id, timestamp=end_time, bot_engaged_messages=impression_messages
+                )
+            else:
+                logger.info(f"在指定时间范围内没有找到 {person_id} 的消息，不更新印象。")
+
+        except Exception as e:
+            logger.error(f"为 {person_id} 更新印象时发生错误: {e}")
+            logger.error(traceback.format_exc())
+
 
 init_prompt()

src/chat/focus_chat/info_processors/self_processor.py

@@ -20,22 +20,27 @@
 
 def init_prompt():
     indentify_prompt = """
+<聊天记录>
+{chat_observe_info}
+</聊天记录>
+
+<人格>
 {name_block}
-你是一个AI，但是你伪装成了一个人类，不要让别人发现这一点
-请参考以下人格，不要被当前聊天内容中的内容误导：
+你是一个AI，但是你伪装成了一个人类，不要让别人发现这一点，你的性格是：
 {prompt_personality}
 {indentify_block}
+</人格>
 
-以下是正在进行的聊天内容：
-现在是{time_now}，你正在参与聊天
-{chat_observe_info}
+请区分聊天记录的内容和你稳定的人格，聊天记录是现在发生的事情，人格是你稳定的独特的特质。
 
-现在请你输出对自己的描述：请严格遵守以下规则
+{name_block}
+现在请你提取你人格的关键信息，提取成一串文本：
 1. 根据聊天记录，输出与聊天记录相关的自我描述，包括人格，形象等等，对人格形象进行精简
 2. 思考有没有内容与你的描述相关
 3. 如果没有明显相关内容，请输出十几个字的简短自我描述
 
-现在请输出你的自我描述,请注意不要输出多余内容(包括前后缀，括号()，表情包，at或 @等 )：
+现在请输出你的自我描述,格式是：“你是.....，你.................（描述）”
+请注意不要输出多余内容(包括前后缀，括号()，表情包，at或 @等 )：
 
 """
     Prompt(indentify_prompt, "indentify_prompt")

src/chat/focus_chat/replyer/default_replyer.py

@@ -37,11 +37,13 @@ def init_prompt():
         
 {extra_info_block}
 
+{relation_info_block}
+
 {time_block}
 你现在正在群里聊天，以下是群里正在进行的聊天内容：
 {chat_info}
 
-{relation_info_block}
+
 
 以上是聊天内容，你需要了解聊天记录中的内容
 
@@ -605,6 +607,8 @@ async def _build_single_sending_message(
             platform=self.chat_stream.platform,
         )
 
+        # await anchor_message.process()
+
         bot_message = MessageSending(
             message_id=message_id,  # 使用片段的唯一ID
             chat_stream=self.chat_stream,

src/chat/heart_flow/observation/chatting_observation.py

@@ -132,13 +132,13 @@ def search_message_by_text(self, text: str) -> Optional[MessageRecv]:
                 # logger.debug(f"找到的锚定消息：find_msg: {find_msg}")
                 break
             else:
-                similarity = difflib.SequenceMatcher(None, text, message["processed_plain_text"]).ratio()
+                similarity = difflib.SequenceMatcher(None, text, message["raw_message"]).ratio()
                 msg_list.append({"message": message, "similarity": similarity})
             # logger.debug(f"对锚定消息检查：message: {message['processed_plain_text']},similarity: {similarity}")
         if not find_msg:
             if msg_list:
                 msg_list.sort(key=lambda x: x["similarity"], reverse=True)
-                if msg_list[0]["similarity"] >= 0.5:  # 只返回相似度大于等于0.5的消息
+                if msg_list[0]["similarity"] >= 0.9:  # 只返回相似度大于等于0.5的消息
                     find_msg = msg_list[0]["message"]
                 else:
                     logger.debug("没有找到锚定消息,相似度低")
@@ -191,6 +191,7 @@ def search_message_by_text(self, text: str) -> Optional[MessageRecv]:
             "detailed_plain_text": find_msg.get("processed_plain_text"),
             "processed_plain_text": find_msg.get("processed_plain_text"),
         }
+        # print(f"message_dict: {message_dict}")
         find_rec_msg = MessageRecv(message_dict)
         # logger.debug(f"锚定消息处理后：find_rec_msg: {find_rec_msg}")
         return find_rec_msg

src/main.py

@@ -20,7 +20,6 @@
 from rich.traceback import install
 from .chat.focus_chat.expressors.exprssion_learner import expression_learner
 from .api.main import start_api_server
-from .person_info.impression_update_task import impression_update_task
 
 install(extra_lines=3)
 
@@ -60,9 +59,6 @@ async def _init_components(self):
         # 添加遥测心跳任务
         await async_task_manager.add_task(TelemetryHeartBeatTask())
 
-        # 添加印象更新任务
-        await async_task_manager.add_task(impression_update_task)
-
         # 启动API服务器
         start_api_server()
         logger.success("API服务器启动成功")