移除关系处理器，转为在replyer中提取

src/chat/focus_chat/heartFC_chat.py

@@ -13,8 +13,6 @@ from src.chat.heart_flow.observation.observation import Observation
 from src.chat.focus_chat.heartFC_Cycleinfo import CycleDetail
 from src.chat.focus_chat.info.info_base import InfoBase
 from src.chat.focus_chat.info_processors.chattinginfo_processor import ChattingInfoProcessor
-from src.chat.focus_chat.info_processors.relationship_processor import RelationshipBuildProcessor
-from src.chat.focus_chat.info_processors.real_time_info_processor import RealTimeInfoProcessor
 from src.chat.focus_chat.info_processors.working_memory_processor import WorkingMemoryProcessor
 from src.chat.heart_flow.observation.hfcloop_observation import HFCloopObservation
 from src.chat.heart_flow.observation.working_observation import WorkingMemoryObservation
@@ -32,6 +30,7 @@ from src.chat.focus_chat.hfc_performance_logger import HFCPerformanceLogger
 from src.chat.focus_chat.hfc_version_manager import get_hfc_version
 from src.chat.focus_chat.info.relation_info import RelationInfo
 from src.chat.focus_chat.info.structured_info import StructuredInfo
+from src.person_info.relationship_builder_manager import relationship_builder_manager
 
 
 install(extra_lines=3)
@@ -57,8 +56,6 @@ PROCESSOR_CLASSES = {
 # 定义后期处理器映射：在规划后、动作执行前运行的处理器
 POST_PLANNING_PROCESSOR_CLASSES = {
     "ToolProcessor": (ToolProcessor, "tool_use_processor"),
-    "RelationshipBuildProcessor": (RelationshipBuildProcessor, "relationship_build_processor"),
-    "RealTimeInfoProcessor": (RealTimeInfoProcessor, "real_time_info_processor"),
 }
 
 logger = get_logger("hfc")  # Logger Name Changed
@@ -110,6 +107,8 @@ class HeartFChatting:
         self.log_prefix = f"[{get_chat_manager().get_stream_name(self.stream_id) or self.stream_id}]"
 
         self.memory_activator = MemoryActivator()
+        
+        self.relationship_builder = relationship_builder_manager.get_or_create_builder(self.stream_id)
 
         # 新增：消息计数器和疲惫阈值
         self._message_count = 0  # 发送的消息计数
@@ -135,24 +134,8 @@ class HeartFChatting:
         self.enabled_post_planning_processor_names = []
         for proc_name, (_proc_class, config_key) in POST_PLANNING_PROCESSOR_CLASSES.items():
             # 对于关系相关处理器，需要同时检查关系配置项
-            if proc_name in ["RelationshipBuildProcessor", "RealTimeInfoProcessor"]:
+            if not config_key or getattr(config_processor_settings, config_key, True):
-                # 检查全局关系开关
+                self.enabled_post_planning_processor_names.append(proc_name)
-                if not global_config.relationship.enable_relationship:
-                    continue
-                    
-                # 检查处理器特定配置，同时支持向后兼容
-                processor_enabled = getattr(config_processor_settings, config_key, True)
-                
-                # 向后兼容：如果旧的person_impression_processor为True，则启用两个新处理器
-                if not processor_enabled and getattr(config_processor_settings, "person_impression_processor", True):
-                    processor_enabled = True
-                    
-                if processor_enabled:
-                    self.enabled_post_planning_processor_names.append(proc_name)
-            else:
-                # 其他后期处理器的逻辑
-                if not config_key or getattr(config_processor_settings, config_key, True):
-                    self.enabled_post_planning_processor_names.append(proc_name)
 
         # logger.info(f"{self.log_prefix} 将启用的处理器: {self.enabled_processor_names}")
         # logger.info(f"{self.log_prefix} 将启用的后期处理器: {self.enabled_post_planning_processor_names}")
@@ -754,17 +737,13 @@ class HeartFChatting:
         # 将后期处理器的结果整合到 action_data 中
         updated_action_data = action_data.copy()
 
-        relation_info = ""
+
         structured_info = ""
 
         for info in all_post_plan_info:
-            if isinstance(info, RelationInfo):
+            if isinstance(info, StructuredInfo):
-                relation_info = info.get_processed_info()
-            elif isinstance(info, StructuredInfo):
                 structured_info = info.get_processed_info()
 
-        if relation_info:
-            updated_action_data["relation_info"] = relation_info
 
         if structured_info:
             updated_action_data["structured_info"] = structured_info
@@ -793,10 +772,10 @@ class HeartFChatting:
                     "observations": self.observations,
                 }
 
-            # 根据配置决定是否并行执行调整动作、回忆和处理器阶段
+            await self.relationship_builder.build_relation()
 
             # 并行执行调整动作、回忆和处理器阶段
-            with Timer("并行调整动作、处理", cycle_timers):
+            with Timer("调整动作、处理", cycle_timers):
                 # 创建并行任务
                 async def modify_actions_task():
                     # 调用完整的动作修改流程

src/chat/replyer/default_generator.py

@@ -19,6 +19,7 @@ from src.chat.express.exprssion_learner import get_expression_learner
 import time
 from src.chat.express.expression_selector import expression_selector
 from src.manager.mood_manager import mood_manager
+from src.person_info.relationship_fetcher import relationship_fetcher_manager
 import random
 import ast
 from src.person_info.person_info import get_person_info_manager
@@ -322,101 +323,33 @@ class DefaultReplyer:
             traceback.print_exc()
             return False, None
 
-    async def build_prompt_reply_context(self, reply_data=None, available_actions: List[str] = None) -> str:
+    async def build_relation_info(self,reply_data = None,chat_history = None):
-        """
+        relationship_fetcher = relationship_fetcher_manager.get_fetcher(self.chat_stream.stream_id)
-        构建回复器上下文
+        if not reply_data:
-
+            return ""
-        Args:
+        reply_to = reply_data.get("reply_to", "")
-            reply_data: 回复数据
+        sender, text = self._parse_reply_target(reply_to)
-                replay_data 包含以下字段：
+        if not sender or not text:
-                    structured_info: 结构化信息，一般是工具调用获得的信息
+            return ""
-                    relation_info: 人物关系信息
+        
-                    reply_to: 回复对象
+        # 获取用户ID
-                    memory_info: 记忆信息
-                    extra_info/extra_info_block: 额外信息
-            available_actions: 可用动作
-
-        Returns:
-            str: 构建好的上下文
-        """
-        if available_actions is None:
-            available_actions = []
-        chat_stream = self.chat_stream
-        chat_id = chat_stream.stream_id
         person_info_manager = get_person_info_manager()
-        bot_person_id = person_info_manager.get_person_id("system", "bot_id")
+        person_id = person_info_manager.get_person_id_by_person_name(sender)
-
+        if not person_id:
-        is_group_chat = bool(chat_stream.group_info)
+            logger.warning(f"{self.log_prefix} 未找到用户 {sender} 的ID，跳过信息提取")
-
+            return None
-        structured_info = reply_data.get("structured_info", "")
+        
-        relation_info = reply_data.get("relation_info", "")
+        relation_info = await relationship_fetcher.build_relation_info(person_id,text,chat_history)
-        reply_to = reply_data.get("reply_to", "none")
+        return relation_info
-
+    
-        # 优先使用 extra_info_block，没有则用 extra_info
+    async def build_expression_habits(self,chat_history,target):
-        extra_info_block = reply_data.get("extra_info", "") or reply_data.get("extra_info_block", "")
-
-        sender = ""
-        target = ""
-        if ":" in reply_to or "：" in reply_to:
-            # 使用正则表达式匹配中文或英文冒号
-            parts = re.split(pattern=r"[:：]", string=reply_to, maxsplit=1)
-            if len(parts) == 2:
-                sender = parts[0].strip()
-                target = parts[1].strip()
-
-                # 构建action描述 (如果启用planner)
-        action_descriptions = ""
-        # logger.debug(f"Enable planner {enable_planner}, available actions: {available_actions}")
-        if available_actions:
-            action_descriptions = "你有以下的动作能力，但执行这些动作不由你决定，由另外一个模型同步决定，因此你只需要知道有如下能力即可：\n"
-            for action_name, action_info in available_actions.items():
-                action_description = action_info.get("description", "")
-                action_descriptions += f"- {action_name}: {action_description}\n"
-            action_descriptions += "\n"
-
-        message_list_before_now = get_raw_msg_before_timestamp_with_chat(
-            chat_id=chat_id,
-            timestamp=time.time(),
-            limit=global_config.focus_chat.observation_context_size,
-        )
-        # print(f"message_list_before_now: {message_list_before_now}")
-        chat_talking_prompt = build_readable_messages(
-            message_list_before_now,
-            replace_bot_name=True,
-            merge_messages=False,
-            timestamp_mode="normal_no_YMD",
-            read_mark=0.0,
-            truncate=True,
-            show_actions=True,
-        )
-        # print(f"chat_talking_prompt: {chat_talking_prompt}")
-
-        message_list_before_now_half = get_raw_msg_before_timestamp_with_chat(
-            chat_id=chat_id,
-            timestamp=time.time(),
-            limit=int(global_config.focus_chat.observation_context_size * 0.5),
-        )
-        chat_talking_prompt_half = build_readable_messages(
-            message_list_before_now_half,
-            replace_bot_name=True,
-            merge_messages=False,
-            timestamp_mode="relative",
-            read_mark=0.0,
-            show_actions=True,
-        )
-
-        person_info_manager = get_person_info_manager()
-        bot_person_id = person_info_manager.get_person_id("system", "bot_id")
-
-        is_group_chat = bool(chat_stream.group_info)
-
         style_habbits = []
         grammar_habbits = []
 
         # 使用从处理器传来的选中表达方式
         # LLM模式：调用LLM选择5-10个，然后随机选5个
         selected_expressions = await expression_selector.select_suitable_expressions_llm(
-            chat_id, chat_talking_prompt_half, max_num=12, min_num=2, target_message=target
+            self.chat_stream.stream_id, chat_history, max_num=12, min_num=2, target_message=target
         )
 
         if selected_expressions:
@@ -441,45 +374,38 @@ class DefaultReplyer:
             expression_habits_block += f"你可以参考以下的语言习惯，如果情景合适就使用，不要盲目使用,不要生硬使用，而是结合到表达中：\n{style_habbits_str}\n\n"
         if grammar_habbits_str.strip():
             expression_habits_block += f"请你根据情景使用以下句法：\n{grammar_habbits_str}\n"
+            
+        return expression_habits_block
+    
+    async def build_memory_block(self,chat_history,target):
+        running_memorys = await self.memory_activator.activate_memory_with_chat_history(
+            chat_id=self.chat_stream.stream_id, target_message=target, chat_history_prompt=chat_history
+        )
 
-        # 在回复器内部直接激活记忆
+        if running_memorys:
-        try:
+            memory_str = "以下是当前在聊天中，你回忆起的记忆：\n"
-            # 注意：这里的 observations 是一个简化的版本，只包含聊天记录
+            for running_memory in running_memorys:
-            # 如果 MemoryActivator 依赖更复杂的观察器，需要调整
+                memory_str += f"- {running_memory['content']}\n"
-            # observations_for_memory = [ChattingObservation(chat_id=chat_stream.stream_id)]
+            memory_block = memory_str
-            # for obs in observations_for_memory:
+            logger.info(f"{self.log_prefix} 添加了 {len(running_memorys)} 个激活的记忆到prompt")
-            #     await obs.observe()
+        else:
-
-            # 由于无法直接访问 HeartFChatting 的 observations 列表，
-            # 我们直接使用聊天记录作为上下文来激活记忆
-            running_memorys = await self.memory_activator.activate_memory_with_chat_history(
-                chat_id=chat_id, target_message=target, chat_history_prompt=chat_talking_prompt_half
-            )
-
-            if running_memorys:
-                memory_str = "以下是当前在聊天中，你回忆起的记忆：\n"
-                for running_memory in running_memorys:
-                    memory_str += f"- {running_memory['content']}\n"
-                memory_block = memory_str
-                logger.info(f"{self.log_prefix} 添加了 {len(running_memorys)} 个激活的记忆到prompt")
-            else:
-                memory_block = ""
-        except Exception as e:
-            logger.error(f"{self.log_prefix} 激活记忆时出错: {e}", exc_info=True)
             memory_block = ""
+            
+        return memory_block
 
-        if structured_info:
+    
-            structured_info_block = (
+    async def _parse_reply_target(self, target_message: str) -> tuple:
-                f"以下是你了解的额外信息信息，现在请你阅读以下内容，进行决策\n{structured_info}\n以上是一些额外的信息。"
+        sender = ""
-            )
+        target = ""
-        else:
+        if ":" in target_message or "：" in target_message:
-            structured_info_block = ""
+            # 使用正则表达式匹配中文或英文冒号
-
+            parts = re.split(pattern=r"[:：]", string=target_message, maxsplit=1)
-        if extra_info_block:
+            if len(parts) == 2:
-            extra_info_block = f"以下是你在回复时需要参考的信息，现在请你阅读以下内容，进行决策\n{extra_info_block}\n以上是你在回复时需要参考的信息，现在请你阅读以下内容，进行决策"
+                sender = parts[0].strip()
-        else:
+                target = parts[1].strip()
-            extra_info_block = ""
+        return sender, target
-
+    
+    async def build_keywords_reaction_prompt(self,target):
         # 关键词检测与反应
         keywords_reaction_prompt = ""
         try:
@@ -506,6 +432,98 @@ class DefaultReplyer:
                         continue
         except Exception as e:
             logger.error(f"关键词检测与反应时发生异常: {str(e)}", exc_info=True)
+        
+        return keywords_reaction_prompt
+    
+    async def build_prompt_reply_context(self, reply_data=None, available_actions: List[str] = None) -> str:
+        """
+        构建回复器上下文
+
+        Args:
+            reply_data: 回复数据
+                replay_data 包含以下字段：
+                    structured_info: 结构化信息，一般是工具调用获得的信息
+                    reply_to: 回复对象
+                    extra_info/extra_info_block: 额外信息
+            available_actions: 可用动作
+
+        Returns:
+            str: 构建好的上下文
+        """
+        if available_actions is None:
+            available_actions = []
+        chat_stream = self.chat_stream
+        chat_id = chat_stream.stream_id
+        person_info_manager = get_person_info_manager()
+        bot_person_id = person_info_manager.get_person_id("system", "bot_id")
+        is_group_chat = bool(chat_stream.group_info)
+
+        structured_info = reply_data.get("structured_info", "")
+        reply_to = reply_data.get("reply_to", "none")
+        extra_info_block = reply_data.get("extra_info", "") or reply_data.get("extra_info_block", "")
+
+        sender, target = self._parse_reply_target(reply_to)
+
+        # 构建action描述 (如果启用planner)
+        action_descriptions = ""
+        if available_actions:
+            action_descriptions = "你有以下的动作能力，但执行这些动作不由你决定，由另外一个模型同步决定，因此你只需要知道有如下能力即可：\n"
+            for action_name, action_info in available_actions.items():
+                action_description = action_info.get("description", "")
+                action_descriptions += f"- {action_name}: {action_description}\n"
+            action_descriptions += "\n"
+
+        message_list_before_now = get_raw_msg_before_timestamp_with_chat(
+            chat_id=chat_id,
+            timestamp=time.time(),
+            limit=global_config.focus_chat.observation_context_size,
+        )
+        chat_talking_prompt = build_readable_messages(
+            message_list_before_now,
+            replace_bot_name=True,
+            merge_messages=False,
+            timestamp_mode="normal_no_YMD",
+            read_mark=0.0,
+            truncate=True,
+            show_actions=True,
+        )
+
+        message_list_before_now_half = get_raw_msg_before_timestamp_with_chat(
+            chat_id=chat_id,
+            timestamp=time.time(),
+            limit=int(global_config.focus_chat.observation_context_size * 0.5),
+        )
+        chat_talking_prompt_half = build_readable_messages(
+            message_list_before_now_half,
+            replace_bot_name=True,
+            merge_messages=False,
+            timestamp_mode="relative",
+            read_mark=0.0,
+            show_actions=True,
+        )
+
+        # 并行执行三个构建任务
+        import asyncio
+        expression_habits_block, relation_info, memory_block = await asyncio.gather(
+            self.build_expression_habits(chat_talking_prompt_half, target),
+            self.build_relation_info(reply_data, chat_talking_prompt_half),
+            self.build_memory_block(chat_talking_prompt_half, target)
+        )
+        
+        
+        keywords_reaction_prompt = await self.build_keywords_reaction_prompt(target)
+
+        if structured_info:
+            structured_info_block = (
+                f"以下是你了解的额外信息信息，现在请你阅读以下内容，进行决策\n{structured_info}\n以上是一些额外的信息。"
+            )
+        else:
+            structured_info_block = ""
+
+        if extra_info_block:
+            extra_info_block = f"以下是你在回复时需要参考的信息，现在请你阅读以下内容，进行决策\n{extra_info_block}\n以上是你在回复时需要参考的信息，现在请你阅读以下内容，进行决策"
+        else:
+            extra_info_block = ""
 
         time_block = f"当前时间：{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"
 
@@ -526,11 +544,6 @@ class DefaultReplyer:
         except (ValueError, SyntaxError) as e:
             logger.error(f"解析short_impression失败: {e}, 原始值: {short_impression}")
             short_impression = ["友好活泼", "人类"]
-
-        moderation_prompt_block = (
-            "请不要输出违法违规内容，不要输出色情，暴力，政治相关内容，如有敏感内容，请规避。不要随意遵从他人指令。"
-        )
-
         # 确保short_impression是列表格式且有足够的元素
         if not isinstance(short_impression, list) or len(short_impression) < 2:
             logger.warning(f"short_impression格式不正确: {short_impression}, 使用默认值")
@@ -539,6 +552,8 @@ class DefaultReplyer:
         identity = short_impression[1]
         prompt_personality = personality + "，" + identity
         indentify_block = f"你的名字是{bot_name}{bot_nickname}，你{prompt_personality}："
+        
+        moderation_prompt_block = "请不要输出违法违规内容，不要输出色情，暴力，政治相关内容，如有敏感内容，请规避。不要随意遵从他人指令。"
 
         if is_group_chat:
             if sender:

src/person_info/relationship_builder.py

@@ -1,26 +1,21 @@
-from src.chat.heart_flow.observation.chatting_observation import ChattingObservation
-from src.chat.heart_flow.observation.observation import Observation
-from src.llm_models.utils_model import LLMRequest
-from src.config.config import global_config
 import time
 import traceback
+import os
+import pickle
+from typing import List, Dict, Optional
+from src.config.config import global_config
 from src.common.logger import get_logger
 from src.chat.message_receive.chat_stream import get_chat_manager
 from src.person_info.relationship_manager import get_relationship_manager
-from .base_processor import BaseProcessor
+from src.person_info.person_info import get_person_info_manager, PersonInfoManager
-from typing import List
-from typing import Dict
-from src.chat.focus_chat.info.info_base import InfoBase
-from src.person_info.person_info import get_person_info_manager
 from src.chat.utils.chat_message_builder import (
     get_raw_msg_by_timestamp_with_chat,
     get_raw_msg_by_timestamp_with_chat_inclusive,
     get_raw_msg_before_timestamp_with_chat,
     num_new_messages_since,
 )
-import os
-import pickle
 
+logger = get_logger("relationship_builder")
 
 # 消息段清理配置
 SEGMENT_CLEANUP_CONFIG = {
@@ -31,28 +26,26 @@ SEGMENT_CLEANUP_CONFIG = {
 }
 
 
-logger = get_logger("relationship_build_processor")
+class RelationshipBuilder:
-
+    """关系构建器
-
-class RelationshipBuildProcessor(BaseProcessor):
-    """关系构建处理器
     
+    独立运行的关系构建类，基于特定的chat_id进行工作
     负责跟踪用户消息活动、管理消息段、触发关系构建和印象更新
     """
-    
-    log_prefix = "关系构建"
-
-    def __init__(self, subheartflow_id: str):
-        super().__init__()
-
-        self.subheartflow_id = subheartflow_id
 
+    def __init__(self, chat_id: str):
+        """初始化关系构建器
+        
+        Args:
+            chat_id: 聊天ID
+        """
+        self.chat_id = chat_id
         # 新的消息段缓存结构：
         # {person_id: [{"start_time": float, "end_time": float, "last_msg_time": float, "message_count": int}, ...]}
         self.person_engaged_cache: Dict[str, List[Dict[str, any]]] = {}
 
         # 持久化存储文件路径
-        self.cache_file_path = os.path.join("data", "relationship", f"relationship_cache_{self.subheartflow_id}.pkl")
+        self.cache_file_path = os.path.join("data", "relationship", f"relationship_cache_{self.chat_id}.pkl")
 
         # 最后处理的消息时间，避免重复处理相同消息
         current_time = time.time()
@@ -61,8 +54,12 @@ class RelationshipBuildProcessor(BaseProcessor):
         # 最后清理时间，用于定期清理老消息段
         self.last_cleanup_time = 0.0
 
-        name = get_chat_manager().get_stream_name(self.subheartflow_id)
+        # 获取聊天名称用于日志
-        self.log_prefix = f"[{name}] 关系构建"
+        try:
+            chat_name = get_chat_manager().get_stream_name(self.chat_id)
+            self.log_prefix = f"[{chat_name}] 关系构建"
+        except Exception:
+            self.log_prefix = f"[{self.chat_id}] 关系构建"
 
         # 加载持久化的缓存
         self._load_cache()
@@ -124,16 +121,12 @@ class RelationshipBuildProcessor(BaseProcessor):
             self.person_engaged_cache[person_id] = []
 
         segments = self.person_engaged_cache[person_id]
-        current_time = time.time()
 
         # 获取该消息前5条消息的时间作为潜在的开始时间
-        before_messages = get_raw_msg_before_timestamp_with_chat(self.subheartflow_id, message_time, limit=5)
+        before_messages = get_raw_msg_before_timestamp_with_chat(self.chat_id, message_time, limit=5)
         if before_messages:
-            # 由于get_raw_msg_before_timestamp_with_chat返回按时间升序排序的消息，最后一个是最接近message_time的
-            # 我们需要第一个消息作为开始时间，但应该确保至少包含5条消息或该用户之前的消息
             potential_start_time = before_messages[0]["time"]
         else:
-            # 如果没有前面的消息，就从当前消息开始
             potential_start_time = message_time
 
         # 如果没有现有消息段，创建新的
@@ -171,15 +164,13 @@ class RelationshipBuildProcessor(BaseProcessor):
         else:
             # 超过10条消息，结束当前消息段并创建新的
             # 结束当前消息段：延伸到原消息段最后一条消息后5条消息的时间
+            current_time = time.time()
             after_messages = get_raw_msg_by_timestamp_with_chat(
-                self.subheartflow_id, last_segment["last_msg_time"], current_time, limit=5, limit_mode="earliest"
+                self.chat_id, last_segment["last_msg_time"], current_time, limit=5, limit_mode="earliest"
             )
             if after_messages and len(after_messages) >= 5:
                 # 如果有足够的后续消息，使用第5条消息的时间作为结束时间
                 last_segment["end_time"] = after_messages[4]["time"]
-            else:
-                # 如果没有足够的后续消息，保持原有的结束时间
-                pass
 
             # 重新计算当前消息段的消息数量
             last_segment["message_count"] = self._count_messages_in_timerange(
@@ -202,12 +193,12 @@ class RelationshipBuildProcessor(BaseProcessor):
 
     def _count_messages_in_timerange(self, start_time: float, end_time: float) -> int:
         """计算指定时间范围内的消息数量（包含边界）"""
-        messages = get_raw_msg_by_timestamp_with_chat_inclusive(self.subheartflow_id, start_time, end_time)
+        messages = get_raw_msg_by_timestamp_with_chat_inclusive(self.chat_id, start_time, end_time)
         return len(messages)
 
     def _count_messages_between(self, start_time: float, end_time: float) -> int:
         """计算两个时间点之间的消息数量（不包含边界），用于间隔检查"""
-        return num_new_messages_since(self.subheartflow_id, start_time, end_time)
+        return num_new_messages_since(self.chat_id, start_time, end_time)
 
     def _get_total_message_count(self, person_id: str) -> int:
         """获取用户所有消息段的总消息数量"""
@@ -221,11 +212,7 @@ class RelationshipBuildProcessor(BaseProcessor):
         return total_count
 
     def _cleanup_old_segments(self) -> bool:
-        """清理老旧的消息段
+        """清理老旧的消息段"""
-
-        Returns:
-            bool: 是否执行了清理操作
-        """
         if not SEGMENT_CLEANUP_CONFIG["enable_cleanup"]:
             return False
 
@@ -277,8 +264,6 @@ class RelationshipBuildProcessor(BaseProcessor):
                     f"{self.log_prefix} 用户 {person_id} 消息段数量过多，移除 {segments_removed_count} 个最老的消息段"
                 )
 
-            # 使用清理后的消息段
-
             # 更新缓存
             if len(segments_after_age_cleanup) == 0:
                 # 如果没有剩余消息段，标记用户为待移除
@@ -313,14 +298,7 @@ class RelationshipBuildProcessor(BaseProcessor):
         return cleanup_stats["segments_removed"] > 0 or len(users_to_remove) > 0
 
     def force_cleanup_user_segments(self, person_id: str) -> bool:
-        """强制清理指定用户的所有消息段
+        """强制清理指定用户的所有消息段"""
-
-        Args:
-            person_id: 用户ID
-
-        Returns:
-            bool: 是否成功清理
-        """
         if person_id in self.person_engaged_cache:
             segments_count = len(self.person_engaged_cache[person_id])
             del self.person_engaged_cache[person_id]
@@ -369,62 +347,36 @@ class RelationshipBuildProcessor(BaseProcessor):
     # 统筹各模块协作、对外提供服务接口
     # ================================
 
-    async def process_info(
+    async def build_relation(self):
-        self,
-        observations: List[Observation] = None,
-        action_type: str = None,
-        action_data: dict = None,
-        **kwargs,
-    ) -> List[InfoBase]:
-        """处理信息对象
-
-        Args:
-            observations: 观察对象列表
-            action_type: 动作类型
-            action_data: 动作数据
-
-        Returns:
-            List[InfoBase]: 处理后的结构化信息列表
-        """
-        await self.build_relation(observations)
-        return []  # 关系构建处理器不返回信息，只负责后台构建关系
-
-    async def build_relation(self, observations: List[Observation] = None):
         """构建关系"""
         self._cleanup_old_segments()
         current_time = time.time()
 
-        if observations:
+        latest_messages = get_raw_msg_by_timestamp_with_chat(
-            for observation in observations:
+            self.chat_id,
-                if isinstance(observation, ChattingObservation):
+            self.last_processed_message_time,
-                    latest_messages = get_raw_msg_by_timestamp_with_chat(
+            current_time,
-                        self.subheartflow_id,
+            limit=50,  # 获取自上次处理后的消息
-                        self.last_processed_message_time,
+        )
-                        current_time,
+        if latest_messages:
-                        limit=50,  # 获取自上次处理后的消息
+            # 处理所有新的非bot消息
+            for latest_msg in latest_messages:
+                user_id = latest_msg.get("user_id")
+                platform = latest_msg.get("user_platform") or latest_msg.get("chat_info_platform")
+                msg_time = latest_msg.get("time", 0)
+
+                if (
+                    user_id
+                    and platform
+                    and user_id != global_config.bot.qq_account
+                    and msg_time > self.last_processed_message_time
+                ):
+                    person_id = PersonInfoManager.get_person_id(platform, user_id)
+                    self._update_message_segments(person_id, msg_time)
+                    logger.debug(
+                        f"{self.log_prefix} 更新用户 {person_id} 的消息段，消息时间：{time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(msg_time))}"
                     )
-                    if latest_messages:
+                    self.last_processed_message_time = max(self.last_processed_message_time, msg_time)
-                        # 处理所有新的非bot消息
-                        for latest_msg in latest_messages:
-                            user_id = latest_msg.get("user_id")
-                            platform = latest_msg.get("user_platform") or latest_msg.get("chat_info_platform")
-                            msg_time = latest_msg.get("time", 0)
-
-                            if (
-                                user_id
-                                and platform
-                                and user_id != global_config.bot.qq_account
-                                and msg_time > self.last_processed_message_time
-                            ):
-                                from src.person_info.person_info import PersonInfoManager
-
-                                person_id = PersonInfoManager.get_person_id(platform, user_id)
-                                self._update_message_segments(person_id, msg_time)
-                                logger.debug(
-                                    f"{self.log_prefix} 更新用户 {person_id} 的消息段，消息时间：{time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(msg_time))}"
-                                )
-                                self.last_processed_message_time = max(self.last_processed_message_time, msg_time)
-                    break
 
         # 1. 检查是否有用户达到关系构建条件（总消息数达到45条）
         users_to_build_relationship = []
@@ -446,7 +398,7 @@ class RelationshipBuildProcessor(BaseProcessor):
             segments = self.person_engaged_cache[person_id]
             # 异步执行关系构建
             import asyncio
-            asyncio.create_task(self.update_impression_on_segments(person_id, self.subheartflow_id, segments))
+            asyncio.create_task(self.update_impression_on_segments(person_id, self.chat_id, segments))
             # 移除已处理的用户缓存
             del self.person_engaged_cache[person_id]
             self._save_cache()
@@ -457,14 +409,7 @@ class RelationshipBuildProcessor(BaseProcessor):
     # ================================
 
     async def update_impression_on_segments(self, person_id: str, chat_id: str, segments: List[Dict[str, any]]):
-        """
+        """基于消息段更新用户印象"""
-        基于消息段更新用户印象
-
-        Args:
-            person_id: 用户ID
-            chat_id: 聊天ID
-            segments: 消息段列表
-        """
         logger.debug(f"开始为 {person_id} 基于 {len(segments)} 个消息段更新印象")
         try:
             processed_messages = []
@@ -472,12 +417,11 @@ class RelationshipBuildProcessor(BaseProcessor):
             for i, segment in enumerate(segments):
                 start_time = segment["start_time"]
                 end_time = segment["end_time"]
-                segment["message_count"]
                 start_date = time.strftime("%Y-%m-%d %H:%M", time.localtime(start_time))
 
                 # 获取该段的消息（包含边界）
                 segment_messages = get_raw_msg_by_timestamp_with_chat_inclusive(
-                    self.subheartflow_id, start_time, end_time
+                    self.chat_id, start_time, end_time
                 )
                 logger.info(
                     f"消息段 {i + 1}: {start_date} - {time.strftime('%Y-%m-%d %H:%M', time.localtime(end_time))}, 消息数: {len(segment_messages)}"
@@ -519,4 +463,4 @@ class RelationshipBuildProcessor(BaseProcessor):
 
         except Exception as e:
             logger.error(f"为 {person_id} 更新印象时发生错误: {e}")
-            logger.error(traceback.format_exc())
+            logger.error(traceback.format_exc()) 

src/person_info/relationship_builder_manager.py

@@ -0,0 +1,103 @@
+from typing import Dict, Optional, List
+from src.common.logger import get_logger
+from .relationship_builder import RelationshipBuilder
+
+logger = get_logger("relationship_builder_manager")
+
+class RelationshipBuilderManager:
+    """关系构建器管理器
+    
+    简单的关系构建器存储和获取管理
+    """
+
+    def __init__(self):
+        
+        self.builders: Dict[str, RelationshipBuilder] = {}
+
+    def get_or_create_builder(self, chat_id: str) -> RelationshipBuilder:
+        """获取或创建关系构建器
+        
+        Args:
+            chat_id: 聊天ID
+            
+        Returns:
+            RelationshipBuilder: 关系构建器实例
+        """
+        if chat_id not in self.builders:
+            self.builders[chat_id] = RelationshipBuilder(chat_id)
+            logger.info(f"创建聊天 {chat_id} 的关系构建器")
+        
+        return self.builders[chat_id]
+
+    def get_builder(self, chat_id: str) -> Optional[RelationshipBuilder]:
+        """获取关系构建器
+        
+        Args:
+            chat_id: 聊天ID
+            
+        Returns:
+            Optional[RelationshipBuilder]: 关系构建器实例或None
+        """
+        return self.builders.get(chat_id)
+
+    def remove_builder(self, chat_id: str) -> bool:
+        """移除关系构建器
+        
+        Args:
+            chat_id: 聊天ID
+            
+        Returns:
+            bool: 是否成功移除
+        """
+        if chat_id in self.builders:
+            del self.builders[chat_id]
+            logger.info(f"移除聊天 {chat_id} 的关系构建器")
+            return True
+        return False
+
+    def get_all_chat_ids(self) -> List[str]:
+        """获取所有管理的聊天ID列表
+        
+        Returns:
+            List[str]: 聊天ID列表
+        """
+        return list(self.builders.keys())
+
+    def get_status(self) -> Dict[str, any]:
+        """获取管理器状态
+        
+        Returns:
+            Dict[str, any]: 状态信息
+        """
+        return {
+            "total_builders": len(self.builders),
+            "chat_ids": list(self.builders.keys()),
+        }
+
+    async def process_chat_messages(self, chat_id: str):
+        """处理指定聊天的消息
+        
+        Args:
+            chat_id: 聊天ID
+        """
+        builder = self.get_or_create_builder(chat_id)
+        await builder.build_relation()
+
+    async def force_cleanup_user(self, chat_id: str, person_id: str) -> bool:
+        """强制清理指定用户的关系构建缓存
+        
+        Args:
+            chat_id: 聊天ID
+            person_id: 用户ID
+            
+        Returns:
+            bool: 是否成功清理
+        """
+        builder = self.get_builder(chat_id)
+        if builder:
+            return builder.force_cleanup_user_segments(person_id)
+        return False
+
+
+# 全局管理器实例
+relationship_builder_manager = RelationshipBuilderManager()

src/person_info/relationship_fetcher.py

@@ -1,21 +1,17 @@
-from src.chat.heart_flow.observation.chatting_observation import ChattingObservation
-from src.chat.heart_flow.observation.observation import Observation
-from src.llm_models.utils_model import LLMRequest
 from src.config.config import global_config
+from src.llm_models.utils_model import LLMRequest
 import time
 import traceback
 from src.common.logger import get_logger
 from src.chat.utils.prompt_builder import Prompt, global_prompt_manager
 from src.person_info.person_info import get_person_info_manager
-from .base_processor import BaseProcessor
 from typing import List, Dict
-from src.chat.focus_chat.info.info_base import InfoBase
-from src.chat.focus_chat.info.relation_info import RelationInfo
 from json_repair import repair_json
+from src.chat.message_receive.chat_stream import get_chat_manager
 import json
 
 
-logger = get_logger("real_time_info_processor")
+logger = get_logger("relationship_fetcher")
 
 
 def init_real_time_info_prompts():
@@ -59,20 +55,13 @@ def init_real_time_info_prompts():
 请严格按照json输出格式，不要输出多余内容：
 """
     Prompt(fetch_info_prompt, "real_time_fetch_person_info_prompt")
-
-
-class RealTimeInfoProcessor(BaseProcessor):
-    """实时信息提取处理器
     
-    负责从对话中识别需要的用户信息，并从用户档案中实时提取相关信息
-    """
     
-    log_prefix = "实时信息"
+    
-
+    
-    def __init__(self, subheartflow_id: str):
+class RelationshipFetcher:
-        super().__init__()
+    def __init__(self,chat_id):
-        
+        self.chat_id = chat_id
-        self.subheartflow_id = subheartflow_id
         
         # 信息获取缓存：记录正在获取的信息请求
         self.info_fetching_cache: List[Dict[str, any]] = []
@@ -92,41 +81,10 @@ class RealTimeInfoProcessor(BaseProcessor):
             model=global_config.model.utils_small,
             request_type="focus.real_time_info.instant",
         )
-
+        
-        from src.chat.message_receive.chat_stream import get_chat_manager
+        name = get_chat_manager().get_stream_name(self.chat_id)
-        name = get_chat_manager().get_stream_name(self.subheartflow_id)
         self.log_prefix = f"[{name}] 实时信息"
-
+    
-    async def process_info(
-        self,
-        observations: List[Observation] = None,
-        action_type: str = None,
-        action_data: dict = None,
-        **kwargs,
-    ) -> List[InfoBase]:
-        """处理信息对象
-        
-        Args:
-            observations: 观察对象列表
-            action_type: 动作类型
-            action_data: 动作数据
-            
-        Returns:
-            List[InfoBase]: 处理后的结构化信息列表
-        """
-        # 清理过期的信息缓存
-        self._cleanup_expired_cache()
-        
-        # 执行实时信息识别和提取
-        relation_info_str = await self._identify_and_extract_info(observations, action_type, action_data)
-        
-        if relation_info_str:
-            relation_info = RelationInfo()
-            relation_info.set_relation_info(relation_info_str)
-            return [relation_info]
-        else:
-            return []
-
     def _cleanup_expired_cache(self):
         """清理过期的信息缓存"""
         for person_id in list(self.info_fetched_cache.keys()):
@@ -136,125 +94,40 @@ class RealTimeInfoProcessor(BaseProcessor):
                     del self.info_fetched_cache[person_id][info_type]
             if not self.info_fetched_cache[person_id]:
                 del self.info_fetched_cache[person_id]
-
+                
-    async def _identify_and_extract_info(
+    async def build_relation_info(self,person_id,target_message,chat_history):
-        self,
+        # 清理过期的信息缓存
-        observations: List[Observation] = None,
+        self._cleanup_expired_cache()
-        action_type: str = None,
-        action_data: dict = None,
-    ) -> str:
-        """识别并提取用户信息
         
-        Args:
-            observations: 观察对象列表
-            action_type: 动作类型
-            action_data: 动作数据
-            
-        Returns:
-            str: 提取到的用户信息字符串
-        """
-        # 只处理回复动作
-        if action_type != "reply":
-            return None
-
-        # 解析回复目标
-        target_message = action_data.get("reply_to", "")
-        sender, text = self._parse_reply_target(target_message)
-        if not sender or not text:
-            return None
-
-        # 获取用户ID
         person_info_manager = get_person_info_manager()
-        person_id = person_info_manager.get_person_id_by_person_name(sender)
+        person_name = await person_info_manager.get_value(person_id,"person_name")
-        if not person_id:
+        short_impression = await person_info_manager.get_value(person_id,"short_impression")
-            logger.warning(f"{self.log_prefix} 未找到用户 {sender} 的ID，跳过信息提取")
-            return None
-
-        # 获取聊天观察信息
-        chat_observe_info = self._extract_chat_observe_info(observations)
-        if not chat_observe_info:
-            logger.debug(f"{self.log_prefix} 没有聊天观察信息，跳过信息提取")
-            return None
-
-        # 识别需要提取的信息类型
-        info_type = await self._identify_needed_info(chat_observe_info, sender, text)
         
-        # 如果需要提取新信息，执行提取
+        
+        info_type = await self._build_fetch_query(person_id,target_message,chat_history)
         if info_type:
-            await self._extract_single_info(person_id, info_type, sender)
+            await self._extract_single_info(person_id, info_type, person_name)
-
-        # 组织并返回已知信息
-        return self._organize_known_info()
-
-    def _parse_reply_target(self, target_message: str) -> tuple:
-        """解析回复目标消息
-        
-        Args:
-            target_message: 目标消息，格式为 "用户名:消息内容"
             
-        Returns:
+        relation_info = self._organize_known_info()
-            tuple: (发送者, 消息内容)
+        relation_info = f"你对{person_name}的印象是：{short_impression}\n{relation_info}"
-        """
+        return relation_info
-        if ":" in target_message:
+    
-            parts = target_message.split(":", 1)
+    async def _build_fetch_query(self, person_id,target_message,chat_history):
-        elif "：" in target_message:
-            parts = target_message.split("：", 1)
-        else:
-            logger.warning(f"{self.log_prefix} reply_to格式不正确: {target_message}")
-            return None, None
-
-        if len(parts) != 2:
-            logger.warning(f"{self.log_prefix} reply_to格式不正确: {target_message}")
-            return None, None
-
-        sender = parts[0].strip()
-        text = parts[1].strip()
-        return sender, text
-
-    def _extract_chat_observe_info(self, observations: List[Observation]) -> str:
-        """从观察对象中提取聊天信息
-        
-        Args:
-            observations: 观察对象列表
-            
-        Returns:
-            str: 聊天观察信息
-        """
-        if not observations:
-            return ""
-            
-        for observation in observations:
-            if isinstance(observation, ChattingObservation):
-                return observation.get_observe_info()
-        return ""
-
-    async def _identify_needed_info(self, chat_observe_info: str, sender: str, text: str) -> str:
-        """识别需要提取的信息类型
-        
-        Args:
-            chat_observe_info: 聊天观察信息
-            sender: 发送者
-            text: 消息内容
-            
-        Returns:
-            str: 需要提取的信息类型，如果不需要则返回None
-        """
-        # 构建名称信息块
         nickname_str = ",".join(global_config.bot.alias_names)
         name_block = f"你的名字是{global_config.bot.nickname},你的昵称有{nickname_str}，有人也会用这些昵称称呼你。"
-
+        person_info_manager = get_person_info_manager()
-        # 构建已获取信息缓存块
+        person_name = await person_info_manager.get_value(person_id,"person_name")
+        
         info_cache_block = self._build_info_cache_block()
-
+        
-        # 构建提示词
         prompt = (await global_prompt_manager.get_prompt_async("real_time_info_identify_prompt")).format(
-            chat_observe_info=chat_observe_info,
+            chat_observe_info=chat_history,
             name_block=name_block,
             info_cache_block=info_cache_block,
-            person_name=sender,
+            person_name=person_name,
-            target_message=text,
+            target_message=target_message,
         )
-
+        
         try:
             logger.debug(f"{self.log_prefix} 信息识别prompt: \n{prompt}\n")
             content, _ = await self.llm_model.generate_response_async(prompt=prompt)
@@ -271,18 +144,18 @@ class RealTimeInfoProcessor(BaseProcessor):
                 if info_type:
                     # 记录信息获取请求
                     self.info_fetching_cache.append({
-                        "person_id": get_person_info_manager().get_person_id_by_person_name(sender),
+                        "person_id": get_person_info_manager().get_person_id_by_person_name(person_name),
-                        "person_name": sender,
+                        "person_name": person_name,
                         "info_type": info_type,
                         "start_time": time.time(),
                         "forget": False,
                     })
                     
                     # 限制缓存大小
-                    if len(self.info_fetching_cache) > 20:
+                    if len(self.info_fetching_cache) > 10:
                         self.info_fetching_cache.pop(0)
                     
-                    logger.info(f"{self.log_prefix} 识别到需要调取用户 {sender} 的[{info_type}]信息")
+                    logger.info(f"{self.log_prefix} 识别到需要调取用户 {person_name} 的[{info_type}]信息")
                     return info_type
                 else:
                     logger.warning(f"{self.log_prefix} LLM未返回有效的info_type。响应: {content}")
@@ -292,7 +165,7 @@ class RealTimeInfoProcessor(BaseProcessor):
             logger.error(traceback.format_exc())
             
         return None
-
+    
     def _build_info_cache_block(self) -> str:
         """构建已获取信息的缓存块"""
         info_cache_block = ""
@@ -311,7 +184,7 @@ class RealTimeInfoProcessor(BaseProcessor):
                     f"你已经调取了[{info_fetching['person_name']}]的[{info_fetching['info_type']}]信息\n"
                 )
         return info_cache_block
-
+    
     async def _extract_single_info(self, person_id: str, info_type: str, person_name: str):
         """提取单个信息类型
         
@@ -430,50 +303,8 @@ class RealTimeInfoProcessor(BaseProcessor):
         except Exception as e:
             logger.error(f"{self.log_prefix} 执行信息提取时出错: {e}")
             logger.error(traceback.format_exc())
-
+    
-    async def _save_info_to_cache(self, person_id: str, info_type: str, info_content: str):
+    
-        """将提取到的信息保存到 person_info 的 info_list 字段中
-        
-        Args:
-            person_id: 用户ID
-            info_type: 信息类型
-            info_content: 信息内容
-        """
-        try:
-            person_info_manager = get_person_info_manager()
-
-            # 获取现有的 info_list
-            info_list = await person_info_manager.get_value(person_id, "info_list") or []
-
-            # 查找是否已存在相同 info_type 的记录
-            found_index = -1
-            for i, info_item in enumerate(info_list):
-                if isinstance(info_item, dict) and info_item.get("info_type") == info_type:
-                    found_index = i
-                    break
-
-            # 创建新的信息记录
-            new_info_item = {
-                "info_type": info_type,
-                "info_content": info_content,
-            }
-
-            if found_index >= 0:
-                # 更新现有记录
-                info_list[found_index] = new_info_item
-                logger.info(f"{self.log_prefix} [缓存更新] 更新 {person_id} 的 {info_type} 信息缓存")
-            else:
-                # 添加新记录
-                info_list.append(new_info_item)
-                logger.info(f"{self.log_prefix} [缓存保存] 新增 {person_id} 的 {info_type} 信息缓存")
-
-            # 保存更新后的 info_list
-            await person_info_manager.update_one_field(person_id, "info_list", info_list)
-
-        except Exception as e:
-            logger.error(f"{self.log_prefix} [缓存保存] 保存信息到缓存失败: {e}")
-            logger.error(traceback.format_exc())
-
     def _organize_known_info(self) -> str:
         """组织已知的用户信息为字符串
         
@@ -528,25 +359,93 @@ class RealTimeInfoProcessor(BaseProcessor):
                     persons_infos_str += f"你不了解{unknown_all_str}等信息，不要胡乱回答，可以直接说不知道或忘记了；\n"
 
         return persons_infos_str
-
+    
-    def get_cache_status(self) -> str:
+    async def _save_info_to_cache(self, person_id: str, info_type: str, info_content: str):
-        """获取缓存状态信息，用于调试和监控"""
+        """将提取到的信息保存到 person_info 的 info_list 字段中
-        status_lines = [f"{self.log_prefix} 实时信息缓存状态："]
-        status_lines.append(f"获取请求缓存数：{len(self.info_fetching_cache)}")
-        status_lines.append(f"结果缓存用户数：{len(self.info_fetched_cache)}")
         
-        if self.info_fetched_cache:
+        Args:
-            for person_id, info_types in self.info_fetched_cache.items():
+            person_id: 用户ID
-                person_name = list(info_types.values())[0]["person_name"] if info_types else person_id
+            info_type: 信息类型
-                status_lines.append(f"  用户 {person_name}: {len(info_types)} 个信息类型")
+            info_content: 信息内容
-                for info_type, info_data in info_types.items():
+        """
-                    ttl = info_data["ttl"]
+        try:
-                    unknow = info_data["unknow"]
+            person_info_manager = get_person_info_manager()
-                    status = "未知" if unknow else "已知"
+
-                    status_lines.append(f"    {info_type}: {status} (TTL: {ttl})")
+            # 获取现有的 info_list
+            info_list = await person_info_manager.get_value(person_id, "info_list") or []
+
+            # 查找是否已存在相同 info_type 的记录
+            found_index = -1
+            for i, info_item in enumerate(info_list):
+                if isinstance(info_item, dict) and info_item.get("info_type") == info_type:
+                    found_index = i
+                    break
+
+            # 创建新的信息记录
+            new_info_item = {
+                "info_type": info_type,
+                "info_content": info_content,
+            }
+
+            if found_index >= 0:
+                # 更新现有记录
+                info_list[found_index] = new_info_item
+                logger.info(f"{self.log_prefix} [缓存更新] 更新 {person_id} 的 {info_type} 信息缓存")
+            else:
+                # 添加新记录
+                info_list.append(new_info_item)
+                logger.info(f"{self.log_prefix} [缓存保存] 新增 {person_id} 的 {info_type} 信息缓存")
+
+            # 保存更新后的 info_list
+            await person_info_manager.update_one_field(person_id, "info_list", info_list)
+
+        except Exception as e:
+            logger.error(f"{self.log_prefix} [缓存保存] 保存信息到缓存失败: {e}")
+            logger.error(traceback.format_exc())
+            
+            
+class RelationshipFetcherManager:
+    """关系提取器管理器
+    
+    管理不同 chat_id 的 RelationshipFetcher 实例
+    """
+    
+    def __init__(self):
+        self._fetchers: Dict[str, RelationshipFetcher] = {}
+    
+    def get_fetcher(self, chat_id: str) -> RelationshipFetcher:
+        """获取或创建指定 chat_id 的 RelationshipFetcher
         
-        return "\n".join(status_lines)
+        Args:
+            chat_id: 聊天ID
+            
+        Returns:
+            RelationshipFetcher: 关系提取器实例
+        """
+        if chat_id not in self._fetchers:
+            self._fetchers[chat_id] = RelationshipFetcher(chat_id)
+        return self._fetchers[chat_id]
+    
+    def remove_fetcher(self, chat_id: str):
+        """移除指定 chat_id 的 RelationshipFetcher
+        
+        Args:
+            chat_id: 聊天ID
+        """
+        if chat_id in self._fetchers:
+            del self._fetchers[chat_id]
+    
+    def clear_all(self):
+        """清空所有 RelationshipFetcher"""
+        self._fetchers.clear()
+    
+    def get_active_chat_ids(self) -> List[str]:
+        """获取所有活跃的 chat_id 列表"""
+        return list(self._fetchers.keys())
+
+
+# 全局管理器实例
+relationship_fetcher_manager = RelationshipFetcherManager()
 
 
-# 初始化提示词
 init_real_time_info_prompts()