fix：使用即时提取relation值

2025-06-09 19:31:59 +08:00
parent 95cb24c11d
commit 03a3be18aa
2 changed files with 164 additions and 30 deletions
--- a/src/chat/focus_chat/info_processors/relationship_processor.py
+++ b/src/chat/focus_chat/info_processors/relationship_processor.py
@@ -19,6 +19,11 @@ import json
 import asyncio
 from src.chat.utils.chat_message_builder import get_raw_msg_by_timestamp_with_chat

+# 配置常量：是否启用小模型即时信息提取
+# 开启时：使用小模型并行即时提取，速度更快，但精度可能略低
+# 关闭时：使用原来的异步模式，精度更高但速度较慢
+ENABLE_INSTANT_INFO_EXTRACTION = True  
+
 logger = get_logger("processor")


@@ -97,6 +102,13 @@ class RelationshipProcessor(BaseProcessor):
            request_type="focus.relationship",
        )
        
+        # 小模型用于即时信息提取
+        if ENABLE_INSTANT_INFO_EXTRACTION:
+            self.instant_llm_model = LLMRequest(
+                model=global_config.model.utils_small,
+                request_type="focus.relationship.instant",
+            )
+
        name = chat_manager.get_stream_name(self.subheartflow_id)
        self.log_prefix = f"[{name}] "

@@ -151,9 +163,9 @@ class RelationshipProcessor(BaseProcessor):
            # 根据消息数量和时间设置不同的触发条件
            should_trigger = (
                message_count >= 50 or  # 50条消息必定满足
-                (message_count >= 35 and time_elapsed >= 600) or  # 35条且10分钟
-                (message_count >= 25 and time_elapsed >= 1800) or  # 25条且30分钟
-                (message_count >= 10 and time_elapsed >= 3600)  # 10条且1小时
+                (message_count >= 35 and time_elapsed >= 300) or  # 35条且10分钟
+                (message_count >= 25 and time_elapsed >= 900) or  # 25条且30分钟
+                (message_count >= 10 and time_elapsed >= 2000)  # 10条且1小时
            )
            
            if should_trigger:
@@ -219,6 +231,10 @@ class RelationshipProcessor(BaseProcessor):
                print(f"content: {content}")
                content_json = json.loads(repair_json(content))

+                # 收集即时提取任务
+                instant_tasks = []
+                async_tasks = []
+                
                for person_name, info_type in content_json.items():
                    person_id = person_info_manager.get_person_id_by_person_name(person_name)
                    if person_id:
@@ -233,6 +249,7 @@ class RelationshipProcessor(BaseProcessor):
                            self.info_fetching_cache.pop(0)
                    else:
                        logger.warning(f"{self.log_prefix} 未找到用户 {person_name} 的ID，跳过调取信息。")
+                        continue
                    
                    logger.info(f"{self.log_prefix} 调取用户 {person_name} 的 {info_type} 信息。")
                    
@@ -244,7 +261,22 @@ class RelationshipProcessor(BaseProcessor):
                            "start_time": time.time(),
                            "rounds": 0
                        })
-                    asyncio.create_task(self.fetch_person_info(person_id, [info_type], start_time=time.time()))
+                    
+                    if ENABLE_INSTANT_INFO_EXTRACTION:
+                        # 收集即时提取任务
+                        instant_tasks.append((person_id, info_type, time.time()))
+                    else:
+                        # 使用原来的异步模式
+                        async_tasks.append(asyncio.create_task(self.fetch_person_info(person_id, [info_type], start_time=time.time())))
+
+                # 执行即时提取任务
+                if ENABLE_INSTANT_INFO_EXTRACTION and instant_tasks:
+                    await self._execute_instant_extraction_batch(instant_tasks)
+                
+                # 启动异步任务（如果不是即时模式）
+                if async_tasks:
+                    # 异步任务不需要等待完成
+                    pass

            else:
                logger.warning(f"{self.log_prefix} LLM返回空结果，关系识别失败。")
@@ -265,11 +297,12 @@ class RelationshipProcessor(BaseProcessor):
                        info_content = self.info_fetched_cache[person_id][info_type]["info"]
                        person_infos_str += f"[{info_type}]：{info_content}；"
                    else:
-                        person_infos_str += f"你不了解{person_name}有关[{info_type}]的信息，不要胡乱回答；"
+                        person_infos_str += f"你不了解{person_name}有关[{info_type}]的信息，不要胡乱回答，你可以直接说你不知道，或者你忘记了；"
                if person_infos_str:
                    persons_infos_str += f"你对 {person_name} 的了解：{person_infos_str}\n"
        
-        # 处理正在调取但还没有结果的项目
+        # 处理正在调取但还没有结果的项目（只在非即时提取模式下显示）
+        if not ENABLE_INSTANT_INFO_EXTRACTION:
            pending_info_dict = {}
            for record in self.info_fetching_cache:
                if not record["forget"]:
@@ -297,6 +330,104 @@ class RelationshipProcessor(BaseProcessor):

        return persons_infos_str
    
+    async def _execute_instant_extraction_batch(self, instant_tasks: list):
+        """
+        批量执行即时提取任务
+        """
+        if not instant_tasks:
+            return
+            
+        logger.info(f"{self.log_prefix} [即时提取] 开始批量提取 {len(instant_tasks)} 个信息")
+        
+        # 创建所有提取任务
+        extraction_tasks = []
+        for person_id, info_type, start_time in instant_tasks:
+            # 检查缓存中是否已存在且未过期的信息
+            if (person_id in self.info_fetched_cache and 
+                info_type in self.info_fetched_cache[person_id]):
+                logger.info(f"{self.log_prefix} 用户 {person_id} 的 {info_type} 信息已存在且未过期，跳过调取。")
+                continue
+                
+            task = asyncio.create_task(self._fetch_single_info_instant(person_id, info_type, start_time))
+            extraction_tasks.append(task)
+        
+        # 并行执行所有提取任务并等待完成
+        if extraction_tasks:
+            await asyncio.gather(*extraction_tasks, return_exceptions=True)
+            logger.info(f"{self.log_prefix} [即时提取] 批量提取完成")
+    
+
+    async def _fetch_single_info_instant(self, person_id: str, info_type: str, start_time: float):
+        """
+        使用小模型提取单个信息类型
+        """
+        nickname_str = ",".join(global_config.bot.alias_names)
+        name_block = f"你的名字是{global_config.bot.nickname},你的昵称有{nickname_str}，有人也会用这些昵称称呼你。"
+        
+        person_name = await person_info_manager.get_value(person_id, "person_name")
+        
+        person_impression = await person_info_manager.get_value(person_id, "impression")
+        if not person_impression:
+            impression_block = "你对ta没有什么深刻的印象"
+        else:
+            impression_block = f"{person_impression}"
+        
+        points = await person_info_manager.get_value(person_id, "points")
+        if points:
+            points_text = "\n".join([
+                f"{point[2]}:{point[0]}"
+                for point in points
+            ])
+        else:
+            points_text = "你不记得ta最近发生了什么"
+        
+        prompt = (await global_prompt_manager.get_prompt_async("fetch_info_prompt")).format(
+            name_block=name_block,
+            info_type=info_type,
+            person_impression=impression_block,
+            person_name=person_name,
+            info_json_str=f'"{info_type}": "信息内容"',
+            points_text=points_text,
+        )
+
+        try:
+            # 使用小模型进行即时提取
+            content, _ = await self.instant_llm_model.generate_response_async(prompt=prompt)
+            
+            logger.info(f"{self.log_prefix} [即时提取] {person_name} 的 {info_type} 结果: {content}")
+            
+            if content:
+                content_json = json.loads(repair_json(content))
+                if info_type in content_json:
+                    info_content = content_json[info_type]
+                    if info_content != "none" and info_content:
+                        if person_id not in self.info_fetched_cache:
+                            self.info_fetched_cache[person_id] = {}
+                        self.info_fetched_cache[person_id][info_type] = {
+                            "info": info_content,
+                            "ttl": 8,  # 小模型提取的信息TTL稍短
+                            "start_time": start_time,
+                            "person_name": person_name,
+                            "unknow": False,
+                        }
+                        logger.info(f"{self.log_prefix} [即时提取] 成功获取 {person_name} 的 {info_type}: {info_content}")
+                    else:
+                        if person_id not in self.info_fetched_cache:
+                            self.info_fetched_cache[person_id] = {}
+                        self.info_fetched_cache[person_id][info_type] = {
+                            "info": "unknow",
+                            "ttl": 8,
+                            "start_time": start_time,
+                            "person_name": person_name,
+                            "unknow": True,
+                        }
+                        logger.info(f"{self.log_prefix} [即时提取] {person_name} 的 {info_type} 信息不明确")
+            else:
+                logger.warning(f"{self.log_prefix} [即时提取] 小模型返回空结果，获取 {person_name} 的 {info_type} 信息失败。")
+        except Exception as e:
+            logger.error(f"{self.log_prefix} [即时提取] 执行小模型请求获取用户信息时出错: {e}")
+            logger.error(traceback.format_exc())
+    
    async def fetch_person_info(self, person_id: str, info_types: list[str], start_time: float):
        """
        获取某个人的信息
--- a/src/chat/focus_chat/info_processors/self_processor.py
+++ b/src/chat/focus_chat/info_processors/self_processor.py
@@ -133,6 +133,9 @@ class SelfProcessor(BaseProcessor):
        name_block = f"你的名字是{global_config.bot.nickname},你的昵称有{nickname_str}，有人也会用这些昵称称呼你。"

        personality_block = individuality.get_personality_prompt(x_person=2, level=2)
+        
+    
+        
        identity_block = individuality.get_identity_prompt(x_person=2, level=2)

        prompt = (await global_prompt_manager.get_prompt_async("indentify_prompt")).format(