fix：修复炸索引

src/chat/focus_chat/expressors/exprssion_learner.py

@@ -283,13 +283,31 @@ class ExpressionLearner:
             if len(old_data) > MAX_EXPRESSION_COUNT:
                 # 计算每个表达方式的权重（count的倒数，这样count越小的越容易被选中）
                 weights = [1 / (expr.get("count", 1) + 0.1) for expr in old_data]
-                # 归一化权重
-                total_weight = sum(weights)
-                weights = [w / total_weight for w in weights]
                 
-                # 随机选择要移除的表达方式
+                # 随机选择要移除的表达方式，避免重复索引
                 remove_count = len(old_data) - MAX_EXPRESSION_COUNT
-                remove_indices = random.choices(range(len(old_data)), weights=weights, k=remove_count)
+                
+                # 使用一种不会选到重复索引的方法
+                indices = list(range(len(old_data)))
+                
+                # 方法1：使用numpy.random.choice
+                # 把列表转成一个映射字典，保证不会有重复
+                remove_set = set()
+                total_attempts = 0
+                
+                # 尝试按权重随机选择，直到选够数量
+                while len(remove_set) < remove_count and total_attempts < len(old_data) * 2:
+                    idx = random.choices(indices, weights=weights, k=1)[0]
+                    remove_set.add(idx)
+                    total_attempts += 1
+                
+                # 如果没选够，随机补充
+                if len(remove_set) < remove_count:
+                    remaining = set(indices) - remove_set
+                    remove_set.update(random.sample(remaining, remove_count - len(remove_set)))
+                
+                remove_indices = list(remove_set)
+                
                 # 从后往前删除，避免索引变化
                 for idx in sorted(remove_indices, reverse=True):
                     old_data.pop(idx)

src/chat/focus_chat/info_processors/relationship_processor.py

@@ -146,9 +146,9 @@ class RelationshipProcessor(BaseProcessor):
             time_elapsed = current_time - record["start_time"]
             message_count = len(get_raw_msg_by_timestamp_with_chat(self.subheartflow_id, record["start_time"], current_time))
             
-            if (record["rounds"] > 20 or 
+            if (record["rounds"] > 50 or 
                 time_elapsed > 1800 or  # 30分钟
-                message_count > 50):
+                message_count > 75):
                 logger.info(f"{self.log_prefix} 用户 {record['person_id']} 满足关系构建条件，开始构建关系。")
                 asyncio.create_task(
                     self.update_impression_on_cache_expiry(

src/chat/focus_chat/planners/actions/no_reply_complex_action.py

@@ -1,134 +0,0 @@
-import asyncio
-import traceback
-from src.common.logger_manager import get_logger
-from src.chat.utils.timer_calculator import Timer
-from src.chat.focus_chat.planners.actions.base_action import BaseAction, register_action
-from typing import Tuple, List
-from src.chat.heart_flow.observation.observation import Observation
-from src.chat.heart_flow.observation.chatting_observation import ChattingObservation
-from src.chat.focus_chat.hfc_utils import parse_thinking_id_to_timestamp
-
-logger = get_logger("action_taken")
-
-# 常量定义
-WAITING_TIME_THRESHOLD = 1200  # 等待新消息时间阈值，单位秒
-
-
-@register_action
-class NoReplyAction(BaseAction):
-    """不回复动作处理类
-
-    处理决定不回复的动作。
-    """
-
-    action_name = "no_reply"
-    action_description = "不回复"
-    action_parameters = {}
-    action_require = [
-        "话题无关/无聊/不感兴趣/不懂",
-        "聊天记录中最新一条消息是你自己发的且无人回应你",
-        "你连续发送了太多消息，且无人回复",
-    ]
-    default = True
-
-    def __init__(
-        self,
-        action_data: dict,
-        reasoning: str,
-        cycle_timers: dict,
-        thinking_id: str,
-        observations: List[Observation],
-        log_prefix: str,
-        shutting_down: bool = False,
-        **kwargs,
-    ):
-        """初始化不回复动作处理器
-
-        Args:
-            action_name: 动作名称
-            action_data: 动作数据
-            reasoning: 执行该动作的理由
-            cycle_timers: 计时器字典
-            thinking_id: 思考ID
-            observations: 观察列表
-            log_prefix: 日志前缀
-            shutting_down: 是否正在关闭
-        """
-        super().__init__(action_data, reasoning, cycle_timers, thinking_id)
-        self.observations = observations
-        self.log_prefix = log_prefix
-        self._shutting_down = shutting_down
-
-    async def handle_action(self) -> Tuple[bool, str]:
-        """
-        处理不回复的情况
-
-        工作流程：
-        1. 等待新消息、超时或关闭信号
-        2. 根据等待结果更新连续不回复计数
-        3. 如果达到阈值，触发回调
-
-        Returns:
-            Tuple[bool, str]: (是否执行成功, 空字符串)
-        """
-        logger.info(f"{self.log_prefix} 决定不回复: {self.reasoning}")
-
-        observation = self.observations[0] if self.observations else None
-
-        try:
-            with Timer("等待新消息", self.cycle_timers):
-                # 等待新消息、超时或关闭信号，并获取结果
-                await self._wait_for_new_message(observation, self.thinking_id, self.log_prefix)
-
-            return True, ""  # 不回复动作没有回复文本
-
-        except asyncio.CancelledError:
-            logger.info(f"{self.log_prefix} 处理 'no_reply' 时等待被中断 (CancelledError)")
-            raise
-        except Exception as e:  # 捕获调用管理器或其他地方可能发生的错误
-            logger.error(f"{self.log_prefix} 处理 'no_reply' 时发生错误: {e}")
-            logger.error(traceback.format_exc())
-            return False, ""
-
-    async def _wait_for_new_message(self, observation: ChattingObservation, thinking_id: str, log_prefix: str) -> bool:
-        """
-        等待新消息 或 检测到关闭信号
-
-        参数:
-            observation: 观察实例
-            thinking_id: 思考ID
-            log_prefix: 日志前缀
-
-        返回:
-            bool: 是否检测到新消息 (如果因关闭信号退出则返回 False)
-        """
-        wait_start_time = asyncio.get_event_loop().time()
-        while True:
-            # --- 在每次循环开始时检查关闭标志 ---
-            if self._shutting_down:
-                logger.info(f"{log_prefix} 等待新消息时检测到关闭信号，中断等待。")
-                return False  # 表示因为关闭而退出
-            # -----------------------------------
-
-            thinking_id_timestamp = parse_thinking_id_to_timestamp(thinking_id)
-
-            # 检查新消息
-            if await observation.has_new_messages_since(thinking_id_timestamp):
-                logger.info(f"{log_prefix} 检测到新消息")
-                return True
-
-            # 检查超时 (放在检查新消息和关闭之后)
-            if asyncio.get_event_loop().time() - wait_start_time > WAITING_TIME_THRESHOLD:
-                logger.warning(f"{log_prefix} 等待新消息超时({WAITING_TIME_THRESHOLD}秒)")
-                return False
-
-            try:
-                # 短暂休眠，让其他任务有机会运行，并能更快响应取消或关闭
-                await asyncio.sleep(0.5)  # 缩短休眠时间
-            except asyncio.CancelledError:
-                # 如果在休眠时被取消，再次检查关闭标志
-                # 如果是正常关闭，则不需要警告
-                if not self._shutting_down:
-                    logger.warning(f"{log_prefix} _wait_for_new_message 的休眠被意外取消")
-                # 无论如何，重新抛出异常，让上层处理
-                raise

src/chat/focus_chat/planners/actions/plugin_action.py

@@ -1,5 +1,5 @@
 import traceback
-from typing import Tuple, Dict, List, Any, Optional
+from typing import Tuple, Dict, List, Any, Optional, Union, Type
 from src.chat.focus_chat.planners.actions.base_action import BaseAction, register_action  # noqa F401
 from src.chat.heart_flow.observation.chatting_observation import ChattingObservation
 from src.chat.focus_chat.hfc_utils import create_empty_anchor_message
@@ -12,6 +12,9 @@ import os
 import inspect
 import toml  # 导入 toml 库
 from src.common.database.database_model import ActionRecords
+from src.common.database.database import db
+from peewee import Model, DoesNotExist
+import json
 import time
 
 # 以下为类型注解需要
@@ -434,3 +437,332 @@ class PluginAction(BaseAction):
         except Exception as e:
             logger.error(f"{self.log_prefix} 存储action信息时出错: {e}")
             traceback.print_exc()
+            
+    async def db_query(
+        self,
+        model_class: Type[Model],
+        query_type: str = "get",
+        filters: Dict[str, Any] = None,
+        data: Dict[str, Any] = None,
+        limit: int = None,
+        order_by: List[str] = None,
+        single_result: bool = False
+    ) -> Union[List[Dict[str, Any]], Dict[str, Any], None]:
+        """执行数据库查询操作
+        
+        这个方法提供了一个通用接口来执行数据库操作，包括查询、创建、更新和删除记录。
+        
+        Args:
+            model_class: Peewee 模型类，例如 ActionRecords, Messages 等
+            query_type: 查询类型，可选值: "get", "create", "update", "delete", "count"
+            filters: 过滤条件字典，键为字段名，值为要匹配的值
+            data: 用于创建或更新的数据字典
+            limit: 限制结果数量
+            order_by: 排序字段列表，使用字段名，前缀'-'表示降序
+            single_result: 是否只返回单个结果
+            
+        Returns:
+            根据查询类型返回不同的结果:
+            - "get": 返回查询结果列表或单个结果（如果 single_result=True）
+            - "create": 返回创建的记录
+            - "update": 返回受影响的行数
+            - "delete": 返回受影响的行数
+            - "count": 返回记录数量
+            
+        示例:
+            # 查询最近10条消息
+            messages = await self.db_query(
+                Messages, 
+                query_type="get",
+                filters={"chat_id": chat_stream.stream_id},
+                limit=10,
+                order_by=["-time"]
+            )
+            
+            # 创建一条记录
+            new_record = await self.db_query(
+                ActionRecords,
+                query_type="create",
+                data={"action_id": "123", "time": time.time(), "action_name": "TestAction"}
+            )
+            
+            # 更新记录
+            updated_count = await self.db_query(
+                ActionRecords,
+                query_type="update",
+                filters={"action_id": "123"},
+                data={"action_done": True}
+            )
+            
+            # 删除记录
+            deleted_count = await self.db_query(
+                ActionRecords,
+                query_type="delete",
+                filters={"action_id": "123"}
+            )
+            
+            # 计数
+            count = await self.db_query(
+                Messages,
+                query_type="count",
+                filters={"chat_id": chat_stream.stream_id}
+            )
+        """
+        try:
+            # 构建基本查询
+            if query_type in ["get", "update", "delete", "count"]:
+                query = model_class.select()
+                
+                # 应用过滤条件
+                if filters:
+                    for field, value in filters.items():
+                        query = query.where(getattr(model_class, field) == value)
+            
+            # 执行查询
+            if query_type == "get":
+                # 应用排序
+                if order_by:
+                    for field in order_by:
+                        if field.startswith("-"):
+                            query = query.order_by(getattr(model_class, field[1:]).desc())
+                        else:
+                            query = query.order_by(getattr(model_class, field))
+                
+                # 应用限制
+                if limit:
+                    query = query.limit(limit)
+                
+                # 执行查询
+                results = list(query.dicts())
+                
+                # 返回结果
+                if single_result:
+                    return results[0] if results else None
+                return results
+                
+            elif query_type == "create":
+                if not data:
+                    raise ValueError("创建记录需要提供data参数")
+                
+                # 创建记录
+                record = model_class.create(**data)
+                # 返回创建的记录
+                return model_class.select().where(model_class.id == record.id).dicts().get()
+                
+            elif query_type == "update":
+                if not data:
+                    raise ValueError("更新记录需要提供data参数")
+                
+                # 更新记录
+                return query.update(**data).execute()
+                
+            elif query_type == "delete":
+                # 删除记录
+                return query.delete().execute()
+                
+            elif query_type == "count":
+                # 计数
+                return query.count()
+                
+            else:
+                raise ValueError(f"不支持的查询类型: {query_type}")
+                
+        except DoesNotExist:
+            # 记录不存在
+            if query_type == "get" and single_result:
+                return None
+            return []
+            
+        except Exception as e:
+            logger.error(f"{self.log_prefix} 数据库操作出错: {e}")
+            traceback.print_exc()
+            
+            # 根据查询类型返回合适的默认值
+            if query_type == "get":
+                return None if single_result else []
+            elif query_type in ["create", "update", "delete", "count"]:
+                return None
+                
+    async def db_raw_query(
+        self, 
+        sql: str, 
+        params: List[Any] = None,
+        fetch_results: bool = True
+    ) -> Union[List[Dict[str, Any]], int, None]:
+        """执行原始SQL查询
+        
+        警告: 使用此方法需要小心，确保SQL语句已正确构造以避免SQL注入风险。
+        
+        Args:
+            sql: 原始SQL查询字符串
+            params: 查询参数列表，用于替换SQL中的占位符
+            fetch_results: 是否获取查询结果，对于SELECT查询设为True，对于
+                          UPDATE/INSERT/DELETE等操作设为False
+                          
+        Returns:
+            如果fetch_results为True，返回查询结果列表；
+            如果fetch_results为False，返回受影响的行数；
+            如果出错，返回None
+        """
+        try:
+            cursor = db.execute_sql(sql, params or [])
+            
+            if fetch_results:
+                # 获取列名
+                columns = [col[0] for col in cursor.description]
+                
+                # 构建结果字典列表
+                results = []
+                for row in cursor.fetchall():
+                    results.append(dict(zip(columns, row)))
+                
+                return results
+            else:
+                # 返回受影响的行数
+                return cursor.rowcount
+                
+        except Exception as e:
+            logger.error(f"{self.log_prefix} 执行原始SQL查询出错: {e}")
+            traceback.print_exc()
+            return None
+            
+    async def db_save(
+        self, 
+        model_class: Type[Model], 
+        data: Dict[str, Any], 
+        key_field: str = None, 
+        key_value: Any = None
+    ) -> Union[Dict[str, Any], None]:
+        """保存数据到数据库（创建或更新）
+        
+        如果提供了key_field和key_value，会先尝试查找匹配的记录进行更新；
+        如果没有找到匹配记录，或未提供key_field和key_value，则创建新记录。
+        
+        Args:
+            model_class: Peewee模型类，如ActionRecords, Messages等
+            data: 要保存的数据字典
+            key_field: 用于查找现有记录的字段名，例如"action_id"
+            key_value: 用于查找现有记录的字段值
+            
+        Returns:
+            Dict[str, Any]: 保存后的记录数据
+            None: 如果操作失败
+            
+        示例:
+            # 创建或更新一条记录
+            record = await self.db_save(
+                ActionRecords,
+                {
+                    "action_id": "123", 
+                    "time": time.time(), 
+                    "action_name": "TestAction",
+                    "action_done": True
+                },
+                key_field="action_id",
+                key_value="123"
+            )
+        """
+        try:
+            # 如果提供了key_field和key_value，尝试更新现有记录
+            if key_field and key_value is not None:
+                # 查找现有记录
+                existing_records = list(model_class.select().where(
+                    getattr(model_class, key_field) == key_value
+                ).limit(1))
+                
+                if existing_records:
+                    # 更新现有记录
+                    existing_record = existing_records[0]
+                    for field, value in data.items():
+                        setattr(existing_record, field, value)
+                    existing_record.save()
+                    
+                    # 返回更新后的记录
+                    updated_record = model_class.select().where(
+                        model_class.id == existing_record.id
+                    ).dicts().get()
+                    return updated_record
+            
+            # 如果没有找到现有记录或未提供key_field和key_value，创建新记录
+            new_record = model_class.create(**data)
+            
+            # 返回创建的记录
+            created_record = model_class.select().where(
+                model_class.id == new_record.id
+            ).dicts().get()
+            return created_record
+            
+        except Exception as e:
+            logger.error(f"{self.log_prefix} 保存数据库记录出错: {e}")
+            traceback.print_exc()
+            return None
+            
+    async def db_get(
+        self,
+        model_class: Type[Model],
+        filters: Dict[str, Any] = None,
+        order_by: str = None,
+        limit: int = None
+    ) -> Union[List[Dict[str, Any]], Dict[str, Any], None]:
+        """从数据库获取记录
+        
+        这是db_query方法的简化版本，专注于数据检索操作。
+        
+        Args:
+            model_class: Peewee模型类
+            filters: 过滤条件，字段名和值的字典
+            order_by: 排序字段，前缀'-'表示降序，例如'-time'表示按时间降序
+            limit: 结果数量限制，如果为1则返回单个记录而不是列表
+            
+        Returns:
+            如果limit=1，返回单个记录字典或None；
+            否则返回记录字典列表或空列表。
+            
+        示例:
+            # 获取单个记录
+            record = await self.db_get(
+                ActionRecords,
+                filters={"action_id": "123"},
+                limit=1
+            )
+            
+            # 获取最近10条记录
+            records = await self.db_get(
+                Messages,
+                filters={"chat_id": chat_stream.stream_id},
+                order_by="-time",
+                limit=10
+            )
+        """
+        try:
+            # 构建查询
+            query = model_class.select()
+            
+            # 应用过滤条件
+            if filters:
+                for field, value in filters.items():
+                    query = query.where(getattr(model_class, field) == value)
+            
+            # 应用排序
+            if order_by:
+                if order_by.startswith("-"):
+                    query = query.order_by(getattr(model_class, order_by[1:]).desc())
+                else:
+                    query = query.order_by(getattr(model_class, order_by))
+            
+            # 应用限制
+            if limit:
+                query = query.limit(limit)
+            
+            # 执行查询
+            results = list(query.dicts())
+            
+            # 返回结果
+            if limit == 1:
+                return results[0] if results else None
+            return results
+            
+        except Exception as e:
+            logger.error(f"{self.log_prefix} 获取数据库记录出错: {e}")
+            traceback.print_exc()
+            return None if limit == 1 else []

src/chat/focus_chat/planners/actions/reply_action.py

@@ -32,6 +32,7 @@ class ReplyAction(BaseAction):
     action_require: list[str] = [
         "你想要闲聊或者随便附和",
         "有人提到你",
+        "如果你刚刚回复，不要对同一个话题重复回应"
     ]
 
     associated_types: list[str] = ["text", "emoji"]

src/person_info/relationship_manager.py

@@ -241,7 +241,8 @@ class RelationshipManager:
             readable_messages = readable_messages.replace(f"{original_name}", f"{mapped_name}")
         
         prompt = f"""
-你的名字是{global_config.bot.nickname}，别名是{alias_str}。
+你的名字是{global_config.bot.nickname}，{global_config.bot.nickname}的别名是{alias_str}。
+请不要混淆你自己和{global_config.bot.nickname}和{person_name}。
 请你基于用户 {person_name}(昵称:{nickname}) 的最近发言，总结出其中是否有有关{person_name}的内容引起了你的兴趣，或者有什么需要你记忆的点。
 如果没有，就输出none
 
@@ -432,8 +433,10 @@ class RelationshipManager:
                 impression = await person_info_manager.get_value(person_id, "impression") or ""
                 
                 compress_prompt = f"""
-你的名字是{global_config.bot.nickname}，别名是{alias_str}。
+你的名字是{global_config.bot.nickname}，{global_config.bot.nickname}的别名是{alias_str}。
-请根据以下历史记录，添加，修改，整合，原有的印象和关系，总结出对{person_name}(昵称:{nickname})的信息。
+请不要混淆你自己和{global_config.bot.nickname}和{person_name}。
+
+请根据以下历史记录，添加，修改，整合，原有的印象和关系，总结出对用户 {person_name}(昵称:{nickname})的信息。
 
 你之前对他的印象和关系是：
 印象impression：{impression}