Merge branch 'dev' into main

changelogs/changelog_dev.md

@@ -1,4 +1,9 @@
 这里放置了测试版本的细节更新
+
+## [test-0.6.1-snapshot-1] - 2025-4-5
+- 修复pfc回复出错bug
+- 修复表情包打字时间，不会卡表情包
+
 ## [test-0.6.0-snapshot-9] - 2025-4-4
 - 可以识别gif表情包
 

flake.nix

@@ -18,10 +18,11 @@
       devShells.default = pkgs.mkShell {
         name = "python-venv";
         venvDir = "./.venv";
-        buildInputs = [
-          pythonPackages.python
-          pythonPackages.venvShellHook
-          pythonPackages.numpy
+        buildInputs = with pythonPackages; [
+          python
+          venvShellHook
+          scipy
+          numpy
         ];
 
         postVenvCreation = ''

scripts/run.sh

@@ -4,7 +4,7 @@
 # 适用于Arch/Ubuntu 24.10/Debian 12/CentOS 9
 # 请小心使用任何一键脚本！
 
-INSTALLER_VERSION="0.0.1-refactor"
+INSTALLER_VERSION="0.0.2-refactor"
 LANG=C.UTF-8
 
 # 如无法访问GitHub请修改此处镜像地址
@@ -62,7 +62,7 @@ show_menu() {
             "4" "启动Nonebot adapter" \
             "5" "停止Nonebot adapter" \
             "6" "重启Nonebot adapter" \
-            "7" "更新MaiCore及其依赖" \
+            "7" "拉取最新MaiCore仓库" \
             "8" "切换分支" \
             "9" "退出" 3>&1 1>&2 2>&3)
 
@@ -111,6 +111,8 @@ show_menu() {
 
 # 更新依赖
 update_dependencies() {
+    whiptail --title "⚠" --msgbox "更新后请阅读教程" 10 60
+    systemctl stop ${SERVICE_NAME}
     cd "${INSTALL_DIR}/MaiBot" || {
         whiptail --msgbox "🚫 无法进入安装目录！" 10 60
         return 1
@@ -126,8 +128,7 @@ update_dependencies() {
         return 1
     fi
     deactivate
-    systemctl restart ${SERVICE_NAME}
-    whiptail --msgbox "✅ 依赖已更新并重启服务！" 10 60
+    whiptail --msgbox "✅ 已停止服务并拉取最新仓库提交" 10 60
 }
 
 # 切换分支
@@ -157,7 +158,7 @@ switch_branch() {
         whiptail --msgbox "🚫 代码拉取失败！" 10 60
         return 1
     fi
-
+    systemctl stop ${SERVICE_NAME}
     source "${INSTALL_DIR}/venv/bin/activate"
     pip install -r requirements.txt
     deactivate
@@ -165,8 +166,7 @@ switch_branch() {
     sed -i "s/^BRANCH=.*/BRANCH=${new_branch}/" /etc/maicore_install.conf
     BRANCH="${new_branch}"
     check_eula
-    systemctl restart ${SERVICE_NAME}
-    whiptail --msgbox "✅ 已切换到分支 ${new_branch} 并重启服务！" 10 60
+    whiptail --msgbox "✅ 已停止服务并切换到分支 ${new_branch} ！" 10 60
 }
 
 check_eula() {
@@ -228,6 +228,8 @@ run_installation() {
         fi
     fi
 
+    whiptail --title "ℹ️ 提示" --msgbox "如果您没有特殊需求，请优先使用docker方式部署。" 10 60
+
     # 协议确认
     if ! (whiptail --title "ℹ️ [1/6] 使用协议" --yes-button "我同意" --no-button "我拒绝" --yesno "使用MaiCore及此脚本前请先阅读EULA协议及隐私协议\nhttps://github.com/MaiM-with-u/MaiBot/blob/refactor/EULA.md\nhttps://github.com/MaiM-with-u/MaiBot/blob/refactor/PRIVACY.md\n\n您是否同意上述协议？" 12 70); then
         exit 1
@@ -370,12 +372,13 @@ run_installation() {
     # 选择分支
     choose_branch() {
     BRANCH=$(whiptail --title "🔀 选择分支" --radiolist "请选择要安装的分支：" 15 60 4 \
-        "main" "稳定最新版（推荐）" ON \
-        "classical" "经典版" OFF \
+        "main" "稳定版本（推荐）" ON \
+        "dev" "开发版（不知道什么意思就别选）" OFF \
+        "classical" "经典版（0.6.0以前的版本）" OFF \
         "custom" "自定义分支" OFF 3>&1 1>&2 2>&3)
     RETVAL=$?
     if [ $RETVAL -ne 0 ]; then
-        whiptail --msgbox "操作取消！" 10 60
+        whiptail --msgbox "🚫 操作取消！" 10 60
         exit 1
     fi
 
@@ -383,7 +386,7 @@ run_installation() {
         BRANCH=$(whiptail --title "🔀 自定义分支" --inputbox "请输入自定义分支名称：" 10 60 "refactor" 3>&1 1>&2 2>&3)
         RETVAL=$?
         if [ $RETVAL -ne 0 ]; then
-            whiptail --msgbox "输入取消！" 10 60
+            whiptail --msgbox "🚫 输入取消！" 10 60
             exit 1
         fi
         if [[ -z "$BRANCH" ]]; then

src/common/database.py

@@ -15,9 +15,16 @@ def __create_database_instance():
     password = os.getenv("MONGODB_PASSWORD")
     auth_source = os.getenv("MONGODB_AUTH_SOURCE")
 
-    if uri and uri.startswith("mongodb://"):
-        # 优先使用URI连接
-        return MongoClient(uri)
+    if uri:
+        # 支持标准mongodb://和mongodb+srv://连接字符串
+        if uri.startswith(("mongodb://", "mongodb+srv://")):
+            return MongoClient(uri)
+        else:
+            raise ValueError(
+                "Invalid MongoDB URI format. URI must start with 'mongodb://' or 'mongodb+srv://'. "
+                "For MongoDB Atlas, use 'mongodb+srv://' format. "
+                "See: https://www.mongodb.com/docs/manual/reference/connection-string/"
+            )
 
     if username and password:
         # 如果有用户名和密码，使用认证连接

src/plugins/PFC/chat_observer.py

@@ -1,6 +1,6 @@
 import time
 import asyncio
-from typing import Optional, Dict, Any, List
+from typing import Optional, Dict, Any, List, Tuple 
 from src.common.logger import get_module_logger
 from src.common.database import db
 from ..message.message_base import UserInfo
@@ -57,6 +57,35 @@ class ChatObserver:
         self._update_event = asyncio.Event()  # 触发更新的事件
         self._update_complete = asyncio.Event()  # 更新完成的事件
     
+    def check(self) -> bool:
+        """检查距离上一次观察之后是否有了新消息
+        
+        Returns:
+            bool: 是否有新消息
+        """
+        logger.debug(f"检查距离上一次观察之后是否有了新消息: {self.last_check_time}")
+        
+        query = {
+            "chat_id": self.stream_id,
+            "time": {"$gt": self.last_check_time}
+        }
+        
+        # 只需要查询是否存在，不需要获取具体消息
+        new_message_exists = db.messages.find_one(query) is not None
+        
+        if new_message_exists:
+            logger.debug("发现新消息")
+            self.last_check_time = time.time()
+        
+        return new_message_exists
+    
+    def get_new_message(self) -> Tuple[List[Dict[str, Any]], List[Dict[str, Any]]]:
+        """获取上一次观察的时间点后的新消息，插入到历史记录中，并返回新消息和历史记录两个对象"""
+        messages = self.get_message_history(self.last_check_time)
+        for message in messages:
+            self._add_message_to_history(message)
+        return messages, self.message_history
+    
     def new_message_after(self, time_point: float) -> bool:
         """判断是否在指定时间点后有新消息
         
@@ -66,6 +95,7 @@ class ChatObserver:
         Returns:
             bool: 是否有新消息
         """
+        logger.debug(f"判断是否在指定时间点后有新消息: {self.last_message_time} > {time_point}")
         return self.last_message_time is None or self.last_message_time > time_point
      
     def _add_message_to_history(self, message: Dict[str, Any]):

src/plugins/PFC/pfc.py

@@ -17,6 +17,7 @@ from ..storage.storage import MessageStorage
 from .chat_observer import ChatObserver
 from .pfc_KnowledgeFetcher import KnowledgeFetcher
 from .reply_checker import ReplyChecker
+from .pfc_utils import get_items_from_json
 import json
 import time
 
@@ -128,43 +129,18 @@ judge_conversation: 判断对话是否结束，当发现对话目标已经达到
             content, _ = await self.llm.generate_response_async(prompt)
             logger.debug(f"LLM原始返回内容: {content}")
             
-            # 清理内容，尝试提取JSON部分
-            content = content.strip()
-            try:
-                # 尝试直接解析
-                result = json.loads(content)
-            except json.JSONDecodeError:
-                # 如果直接解析失败，尝试查找和提取JSON部分
-                import re
-                json_pattern = r'\{[^{}]*\}'
-                json_match = re.search(json_pattern, content)
-                if json_match:
-                    try:
-                        result = json.loads(json_match.group())
-                    except json.JSONDecodeError:
-                        logger.error("提取的JSON内容解析失败，返回默认行动")
-                        return "direct_reply", "JSON解析失败，选择直接回复"
-                else:
-                    # 如果找不到JSON，尝试从文本中提取行动和原因
-                    if "direct_reply" in content.lower():
-                        return "direct_reply", "从文本中提取的行动"
-                    elif "fetch_knowledge" in content.lower():
-                        return "fetch_knowledge", "从文本中提取的行动"
-                    elif "wait" in content.lower():
-                        return "wait", "从文本中提取的行动"
-                    elif "listening" in content.lower():
-                        return "listening", "从文本中提取的行动"
-                    elif "rethink_goal" in content.lower():
-                        return "rethink_goal", "从文本中提取的行动"
-                    elif "judge_conversation" in content.lower():
-                        return "judge_conversation", "从文本中提取的行动"
-                    else:
-                        logger.error("无法从返回内容中提取行动类型")
-                        return "direct_reply", "无法解析响应，选择直接回复"
+            # 使用简化函数提取JSON内容
+            success, result = get_items_from_json(
+                content,
+                "action", "reason",
+                default_values={"action": "direct_reply", "reason": "默认原因"}
+            )
             
-            # 验证JSON字段
-            action = result.get("action", "direct_reply")
-            reason = result.get("reason", "默认原因")
+            if not success:
+                return "direct_reply", "JSON解析失败，选择直接回复"
+            
+            action = result["action"]
+            reason = result["reason"]
             
             # 验证action类型
             if action not in ["direct_reply", "fetch_knowledge", "wait", "listening", "rethink_goal", "judge_conversation"]:
@@ -196,6 +172,8 @@ class GoalAnalyzer:
         self.nick_name = global_config.BOT_ALIAS_NAMES
         self.chat_observer = ChatObserver.get_instance(stream_id)
         
+        self.current_goal_and_reason = None
+
     async def analyze_goal(self) -> Tuple[str, str, str]:
         """分析对话历史并设定目标
         
@@ -239,48 +217,20 @@ class GoalAnalyzer:
                 content, _ = await self.llm.generate_response_async(prompt)
                 logger.debug(f"LLM原始返回内容: {content}")
                 
-                # 清理和验证返回内容
-                if not content or not isinstance(content, str):
-                    logger.error("LLM返回内容为空或格式不正确")
-                    continue
+                # 使用简化函数提取JSON内容
+                success, result = get_items_from_json(
+                    content,
+                    "goal", "reasoning",
+                    required_types={"goal": str, "reasoning": str}
+                )
                 
-                # 尝试提取JSON部分
-                content = content.strip()
-                try:
-                    # 尝试直接解析
-                    result = json.loads(content)
-                except json.JSONDecodeError:
-                    # 如果直接解析失败，尝试查找和提取JSON部分
-                    import re
-                    json_pattern = r'\{[^{}]*\}'
-                    json_match = re.search(json_pattern, content)
-                    if json_match:
-                        try:
-                            result = json.loads(json_match.group())
-                        except json.JSONDecodeError:
-                            logger.error(f"提取的JSON内容解析失败，重试第{retry + 1}次")
-                            continue
-                    else:
-                        logger.error(f"无法在返回内容中找到有效的JSON，重试第{retry + 1}次")
-                        continue
-                
-                # 验证JSON字段
-                if not all(key in result for key in ["goal", "reasoning"]):
-                    logger.error(f"JSON缺少必要字段，实际内容: {result}，重试第{retry + 1}次")
+                if not success:
+                    logger.error(f"无法解析JSON，重试第{retry + 1}次")
                     continue
                     
                 goal = result["goal"]
                 reasoning = result["reasoning"]
                 
-                # 验证字段内容
-                if not isinstance(goal, str) or not isinstance(reasoning, str):
-                    logger.error(f"JSON字段类型错误，goal和reasoning必须是字符串，重试第{retry + 1}次")
-                    continue
-                    
-                if not goal.strip() or not reasoning.strip():
-                    logger.error(f"JSON字段内容为空，重试第{retry + 1}次")
-                    continue
-                    
                 # 使用默认的方法
                 method = "以友好的态度回应"
                 return goal, method, reasoning
@@ -330,58 +280,21 @@ class GoalAnalyzer:
             content, _ = await self.llm.generate_response_async(prompt)
             logger.debug(f"LLM原始返回内容: {content}")
             
-            # 清理和验证返回内容
-            if not content or not isinstance(content, str):
-                logger.error("LLM返回内容为空或格式不正确")
+            # 使用简化函数提取JSON内容
+            success, result = get_items_from_json(
+                content,
+                "goal_achieved", "stop_conversation", "reason",
+                required_types={
+                    "goal_achieved": bool,
+                    "stop_conversation": bool,
+                    "reason": str
+                }
+            )
+            
+            if not success:
                 return False, False, "确保对话顺利进行"
             
-            # 尝试提取JSON部分
-            content = content.strip()
-            try:
-                # 尝试直接解析
-                result = json.loads(content)
-            except json.JSONDecodeError:
-                # 如果直接解析失败，尝试查找和提取JSON部分
-                import re
-                json_pattern = r'\{[^{}]*\}'
-                json_match = re.search(json_pattern, content)
-                if json_match:
-                    try:
-                        result = json.loads(json_match.group())
-                    except json.JSONDecodeError as e:
-                        logger.error(f"提取的JSON内容解析失败: {e}")
-                        return False, False, "确保对话顺利进行"
-                else:
-                    logger.error("无法在返回内容中找到有效的JSON")
-                    return False, False, "确保对话顺利进行"
-            
-            # 验证JSON字段
-            if not all(key in result for key in ["goal_achieved", "stop_conversation", "reason"]):
-                logger.error(f"JSON缺少必要字段，实际内容: {result}")
-                return False, False, "确保对话顺利进行"
-                
-            goal_achieved = result["goal_achieved"]
-            stop_conversation = result["stop_conversation"]
-            reason = result["reason"]
-            
-            # 验证字段类型
-            if not isinstance(goal_achieved, bool):
-                logger.error("goal_achieved 必须是布尔值")
-                return False, False, "确保对话顺利进行"
-                
-            if not isinstance(stop_conversation, bool):
-                logger.error("stop_conversation 必须是布尔值")
-                return False, False, "确保对话顺利进行"
-                
-            if not isinstance(reason, str):
-                logger.error("reason 必须是字符串")
-                return False, False, "确保对话顺利进行"
-                
-            if not reason.strip():
-                logger.error("reason 不能为空")
-                return False, False, "确保对话顺利进行"
-            
-            return goal_achieved, stop_conversation, reason
+            return result["goal_achieved"], result["stop_conversation"], result["reason"]
             
         except Exception as e:
             logger.error(f"分析对话目标时出错: {str(e)}")
@@ -435,19 +348,18 @@ class ReplyGenerator:
         knowledge_cache: Dict[str, str],
         previous_reply: Optional[str] = None,
         retry_count: int = 0
-    ) -> Tuple[str, bool]:
+    ) -> str:
         """生成回复
         
         Args:
             goal: 对话目标
-            method: 实现方式
             chat_history: 聊天历史
             knowledge_cache: 知识缓存
             previous_reply: 上一次生成的回复（如果有）
             retry_count: 当前重试次数
             
         Returns:
-            Tuple[str, bool]: (生成的回复, 是否需要重新规划)
+            str: 生成的回复
         """
         # 构建提示词
         logger.debug(f"开始生成回复：当前目标: {goal}")
@@ -508,53 +420,105 @@ class ReplyGenerator:
         try:
             content, _ = await self.llm.generate_response_async(prompt)
             logger.info(f"生成的回复: {content}")
+            is_new = self.chat_observer.check()
+            logger.debug(f"再看一眼聊天记录，{'有' if is_new else '没有'}新消息")
             
-            # 检查生成的回复是否合适
-            is_suitable, reason, need_replan = await self.reply_checker.check(
-                content, goal, retry_count
-            )
+            # 如果有新消息,重新生成回复
+            if is_new:
+                logger.info("检测到新消息,重新生成回复")
+                return await self.generate(
+                    goal, chat_history, knowledge_cache,
+                    None, retry_count
+                )
                 
-            if not is_suitable:
-                logger.warning(f"生成的回复不合适，原因: {reason}")
-                if need_replan:
-                    logger.info("需要重新规划对话目标")
-                    return "让我重新思考一下...", True
-                else:
-                    # 递归调用，将当前回复作为previous_reply传入
-                    return await self.generate(
-                        goal, chat_history, knowledge_cache, 
-                        content, retry_count + 1
-                    )
-                
-            return content, False
+            return content
             
         except Exception as e:
             logger.error(f"生成回复时出错: {e}")
-            return "抱歉，我现在有点混乱，让我重新思考一下...", True
+            return "抱歉，我现在有点混乱，让我重新思考一下..."
+
+    async def check_reply(
+        self,
+        reply: str,
+        goal: str,
+        retry_count: int = 0
+    ) -> Tuple[bool, str, bool]:
+        """检查回复是否合适
+        
+        Args:
+            reply: 生成的回复
+            goal: 对话目标
+            retry_count: 当前重试次数
+            
+        Returns:
+            Tuple[bool, str, bool]: (是否合适, 原因, 是否需要重新规划)
+        """
+        return await self.reply_checker.check(reply, goal, retry_count)
 
 
 class Conversation:
     # 类级别的实例管理
     _instances: Dict[str, 'Conversation'] = {}
+    _instance_lock = asyncio.Lock()  # 类级别的全局锁
+    _init_events: Dict[str, asyncio.Event] = {}  # 初始化完成事件
+    _initializing: Dict[str, bool] = {}  # 标记是否正在初始化
     
     @classmethod
-    def get_instance(cls, stream_id: str) -> 'Conversation':
-        """获取或创建对话实例"""
-        if stream_id not in cls._instances:
-            cls._instances[stream_id] = cls(stream_id)
-            logger.info(f"创建新的对话实例: {stream_id}")
-        return cls._instances[stream_id]
+    async def get_instance(cls, stream_id: str) -> Optional['Conversation']:
+        """获取或创建对话实例
+        
+        Args:
+            stream_id: 聊天流ID
+            
+        Returns:
+            Optional[Conversation]: 对话实例，如果创建或等待失败则返回None
+        """
+        try:
+            # 使用全局锁来确保线程安全
+            async with cls._instance_lock:
+                # 如果已经在初始化中，等待初始化完成
+                if stream_id in cls._initializing and cls._initializing[stream_id]:
+                    # 释放锁等待初始化
+                    cls._instance_lock.release()
+                    try:
+                        await asyncio.wait_for(cls._init_events[stream_id].wait(), timeout=5.0)
+                    except asyncio.TimeoutError:
+                        logger.error(f"等待实例 {stream_id} 初始化超时")
+                        return None
+                    finally:
+                        await cls._instance_lock.acquire()
+                
+                # 如果实例不存在，创建新实例
+                if stream_id not in cls._instances:
+                    cls._instances[stream_id] = cls(stream_id)
+                    cls._init_events[stream_id] = asyncio.Event()
+                    cls._initializing[stream_id] = True
+                    logger.info(f"创建新的对话实例: {stream_id}")
+                
+                return cls._instances[stream_id]
+        except Exception as e:
+            logger.error(f"获取对话实例失败: {e}")
+            return None
     
     @classmethod
-    def remove_instance(cls, stream_id: str):
-        """删除对话实例"""
-        if stream_id in cls._instances:
-            # 停止相关组件
-            instance = cls._instances[stream_id]
-            instance.chat_observer.stop()
-            # 删除实例
-            del cls._instances[stream_id]
-            logger.info(f"已删除对话实例 {stream_id}")
+    async def remove_instance(cls, stream_id: str):
+        """删除对话实例
+        
+        Args:
+            stream_id: 聊天流ID
+        """
+        async with cls._instance_lock:
+            if stream_id in cls._instances:
+                # 停止相关组件
+                instance = cls._instances[stream_id]
+                instance.chat_observer.stop()
+                # 删除实例
+                del cls._instances[stream_id]
+                if stream_id in cls._init_events:
+                    del cls._init_events[stream_id]
+                if stream_id in cls._initializing:
+                    del cls._initializing[stream_id]
+                logger.info(f"已删除对话实例 {stream_id}")
 
     def __init__(self, stream_id: str):
         """初始化对话系统"""
@@ -592,12 +556,20 @@ class Conversation:
         
     async def start(self):
         """开始对话流程"""
-        logger.info("对话系统启动")
-        self.should_continue = True
-        self.chat_observer.start()  # 启动观察器
-        await asyncio.sleep(1)
-        # 启动对话循环
-        await self._conversation_loop()
+        try:
+            logger.info("对话系统启动")
+            self.should_continue = True
+            self.chat_observer.start()  # 启动观察器
+            await asyncio.sleep(1)
+            # 启动对话循环
+            await self._conversation_loop()
+        except Exception as e:
+            logger.error(f"启动对话系统失败: {e}")
+            raise
+        finally:
+            # 标记初始化完成
+            self._init_events[self.stream_id].set()
+            self._initializing[self.stream_id] = False
 
     async def _conversation_loop(self):
         """对话循环"""
@@ -658,17 +630,53 @@ class Conversation:
         if action == "direct_reply":
             self.state = ConversationState.GENERATING
             messages = self.chat_observer.get_message_history(limit=30)
-            self.generated_reply, need_replan = await self.reply_generator.generate(
+            self.generated_reply = await self.reply_generator.generate(
                 self.current_goal,
                 self.current_method,
                 [self._convert_to_message(msg) for msg in messages],
                 self.knowledge_cache
             )
-            if need_replan:
-                self.state = ConversationState.RETHINKING
-                self.current_goal, self.current_method, self.goal_reasoning = await self.goal_analyzer.analyze_goal()
-            else:
-                await self._send_reply()
+            
+            # 检查回复是否合适
+            is_suitable, reason, need_replan = await self.reply_generator.check_reply(
+                self.generated_reply,
+                self.current_goal
+            )
+            
+            if not is_suitable:
+                    logger.warning(f"生成的回复不合适，原因: {reason}")
+                    if need_replan:
+                        self.state = ConversationState.RETHINKING
+                        self.current_goal, self.current_method, self.goal_reasoning = await self.goal_analyzer.analyze_goal()
+                        return
+                    else:
+                        # 重新生成回复
+                        self.generated_reply = await self.reply_generator.generate(
+                            self.current_goal,
+                            self.current_method,
+                            [self._convert_to_message(msg) for msg in messages],
+                            self.knowledge_cache,
+                            self.generated_reply  # 将不合适的回复作为previous_reply传入
+                        )
+            
+            while self.chat_observer.check():
+                if not is_suitable:
+                    logger.warning(f"生成的回复不合适，原因: {reason}")
+                    if need_replan:
+                        self.state = ConversationState.RETHINKING
+                        self.current_goal, self.current_method, self.goal_reasoning = await self.goal_analyzer.analyze_goal()
+                        return
+                    else:
+                        # 重新生成回复
+                        self.generated_reply = await self.reply_generator.generate(
+                            self.current_goal,
+                            self.current_method,
+                            [self._convert_to_message(msg) for msg in messages],
+                            self.knowledge_cache,
+                            self.generated_reply  # 将不合适的回复作为previous_reply传入
+                        )
+            
+            await self._send_reply()
             
         elif action == "fetch_knowledge":
             self.state = ConversationState.GENERATING
@@ -682,17 +690,36 @@ class Conversation:
             if knowledge != "未找到相关知识":
                 self.knowledge_cache[sources] = knowledge
             
-            self.generated_reply, need_replan = await self.reply_generator.generate(
+            self.generated_reply = await self.reply_generator.generate(
                 self.current_goal,
                 self.current_method,
                 [self._convert_to_message(msg) for msg in messages],
                 self.knowledge_cache
             )
-            if need_replan:
-                self.state = ConversationState.RETHINKING
-                self.current_goal, self.current_method, self.goal_reasoning = await self.goal_analyzer.analyze_goal()
-            else:
-                await self._send_reply()
+            
+            # 检查回复是否合适
+            is_suitable, reason, need_replan = await self.reply_generator.check_reply(
+                self.generated_reply,
+                self.current_goal
+            )
+            
+            if not is_suitable:
+                logger.warning(f"生成的回复不合适，原因: {reason}")
+                if need_replan:
+                    self.state = ConversationState.RETHINKING
+                    self.current_goal, self.current_method, self.goal_reasoning = await self.goal_analyzer.analyze_goal()
+                    return
+                else:
+                    # 重新生成回复
+                    self.generated_reply = await self.reply_generator.generate(
+                        self.current_goal,
+                        self.current_method,
+                        [self._convert_to_message(msg) for msg in messages],
+                        self.knowledge_cache,
+                        self.generated_reply  # 将不合适的回复作为previous_reply传入
+                    )
+            
+            await self._send_reply()
         
         elif action == "rethink_goal":
             self.state = ConversationState.RETHINKING
@@ -724,7 +751,7 @@ class Conversation:
         self.should_continue = False
         self.state = ConversationState.ENDED
         # 删除实例（这会同时停止chat_observer）
-        self.remove_instance(self.stream_id)
+        await self.remove_instance(self.stream_id)
 
     async def _send_timeout_message(self):
         """发送超时结束消息"""
@@ -821,7 +848,7 @@ class DirectMessageSender:
             if not end_point:
                 raise ValueError(f"未找到平台：{chat_stream.platform} 的url配置")
                 
-            await global_api.send_message(end_point, message_json)
+            await global_api.send_message_REST(end_point, message_json)
             
             # 存储消息
             await self.storage.store_message(message, message.chat_stream)

src/plugins/PFC/pfc_utils.py

@@ -0,0 +1,72 @@
+import json
+import re
+from typing import Dict, Any, Optional, List, Tuple, Union
+from src.common.logger import get_module_logger
+
+logger = get_module_logger("pfc_utils")
+
+def get_items_from_json(
+    content: str,
+    *items: str,
+    default_values: Optional[Dict[str, Any]] = None,
+    required_types: Optional[Dict[str, type]] = None
+) -> Tuple[bool, Dict[str, Any]]:
+    """从文本中提取JSON内容并获取指定字段
+    
+    Args:
+        content: 包含JSON的文本
+        *items: 要提取的字段名
+        default_values: 字段的默认值，格式为 {字段名: 默认值}
+        required_types: 字段的必需类型，格式为 {字段名: 类型}
+        
+    Returns:
+        Tuple[bool, Dict[str, Any]]: (是否成功, 提取的字段字典)
+    """
+    content = content.strip()
+    result = {}
+    
+    # 设置默认值
+    if default_values:
+        result.update(default_values)
+    
+    # 尝试解析JSON
+    try:
+        json_data = json.loads(content)
+    except json.JSONDecodeError:
+        # 如果直接解析失败，尝试查找和提取JSON部分
+        json_pattern = r'\{[^{}]*\}'
+        json_match = re.search(json_pattern, content)
+        if json_match:
+            try:
+                json_data = json.loads(json_match.group())
+            except json.JSONDecodeError:
+                logger.error("提取的JSON内容解析失败")
+                return False, result
+        else:
+            logger.error("无法在返回内容中找到有效的JSON")
+            return False, result
+    
+    # 提取字段
+    for item in items:
+        if item in json_data:
+            result[item] = json_data[item]
+    
+    # 验证必需字段
+    if not all(item in result for item in items):
+        logger.error(f"JSON缺少必要字段，实际内容: {json_data}")
+        return False, result
+    
+    # 验证字段类型
+    if required_types:
+        for field, expected_type in required_types.items():
+            if field in result and not isinstance(result[field], expected_type):
+                logger.error(f"{field} 必须是 {expected_type.__name__} 类型")
+                return False, result
+    
+    # 验证字符串字段不为空
+    for field in items:
+        if isinstance(result[field], str) and not result[field].strip():
+            logger.error(f"{field} 不能为空")
+            return False, result
+    
+    return True, result 

src/plugins/chat/bot.py

@@ -42,11 +42,24 @@ class ChatBot:
             
             if global_config.enable_pfc_chatting:
                 # 获取或创建对话实例
-                conversation = Conversation.get_instance(chat_id)
+                conversation = await Conversation.get_instance(chat_id)
+                if conversation is None:
+                    logger.error(f"创建或获取对话实例失败: {chat_id}")
+                    return
+                    
                 # 如果是新创建的实例，启动对话系统
                 if conversation.state == ConversationState.INIT:
                     asyncio.create_task(conversation.start())
                     logger.info(f"为聊天 {chat_id} 创建新的对话实例")
+                elif conversation.state == ConversationState.ENDED:
+                    # 如果实例已经结束，重新创建
+                    await Conversation.remove_instance(chat_id)
+                    conversation = await Conversation.get_instance(chat_id)
+                    if conversation is None:
+                        logger.error(f"重新创建对话实例失败: {chat_id}")
+                        return
+                    asyncio.create_task(conversation.start())
+                    logger.info(f"为聊天 {chat_id} 重新创建对话实例")
         except Exception as e:
             logger.error(f"创建PFC聊天流失败: {e}")
 
@@ -78,7 +91,12 @@ class ChatBot:
         try:
             message = MessageRecv(message_data)
             groupinfo = message.message_info.group_info
-            logger.debug(f"处理消息:{str(message_data)[:50]}...")
+            userinfo = message.message_info.user_info
+            logger.debug(f"处理消息:{str(message_data)[:80]}...")
+
+            if userinfo.user_id in global_config.ban_user_id:
+                logger.debug(f"用户{userinfo.user_id}被禁止回复")
+                return
             
             if global_config.enable_pfc_chatting:
                 try:
@@ -96,11 +114,11 @@ class ChatBot:
                         await self._create_PFC_chat(message)
                     else:
                         if groupinfo.group_id in global_config.talk_allowed_groups:
-                            logger.debug(f"开始群聊模式{message_data}")
+                            logger.debug(f"开始群聊模式{str(message_data)[:50]}...")
                             if global_config.response_mode == "heart_flow":
                                 await self.think_flow_chat.process_message(message_data)
                             elif global_config.response_mode == "reasoning":
-                                logger.debug(f"开始推理模式{message_data}")
+                                logger.debug(f"开始推理模式{str(message_data)[:50]}...")
                                 await self.reasoning_chat.process_message(message_data)
                             else:
                                 logger.error(f"未知的回复模式，请检查配置文件！！: {global_config.response_mode}")

src/plugins/chat/chat_stream.py

@@ -28,7 +28,7 @@ class ChatStream:
         self.platform = platform
         self.user_info = user_info
         self.group_info = group_info
-        self.create_time = data.get("create_time", int(time.time())) if data else int(time.time())
+        self.create_time = data.get("create_time", time.time()) if data else time.time()
         self.last_active_time = data.get("last_active_time", self.create_time) if data else self.create_time
         self.saved = False
 
@@ -60,7 +60,7 @@ class ChatStream:
 
     def update_active_time(self):
         """更新最后活跃时间"""
-        self.last_active_time = int(time.time())
+        self.last_active_time = time.time()
         self.saved = False
 
 

src/plugins/chat/message.py

@@ -168,7 +168,7 @@ class MessageProcessBase(Message):
         # 调用父类初始化
         super().__init__(
             message_id=message_id,
-            time=int(time.time()),
+            time=round(time.time(), 3),  # 保留3位小数
             chat_stream=chat_stream,
             user_info=bot_user_info,
             message_segment=message_segment,

src/plugins/chat/message_sender.py

@@ -43,6 +43,12 @@ class Message_Sender:
         # 按thinking_start_time排序，时间早的在前面
         return recalled_messages
 
+    async def send_via_ws(self, message: MessageSending) -> None:
+        try:
+            await global_api.send_message(message)
+        except Exception as e:
+            raise ValueError(f"未找到平台：{message.message_info.platform} 的url配置，请检查配置文件") from e
+
     async def send_message(
         self,
         message: MessageSending,
@@ -58,8 +64,14 @@ class Message_Sender:
                     logger.warning(f"消息“{message.processed_plain_text}”已被撤回，不发送")
                     break
             if not is_recalled:
-                typing_time = calculate_typing_time(message.processed_plain_text)
+                # print(message.processed_plain_text + str(message.is_emoji))
+                typing_time = calculate_typing_time(
+                    input_string=message.processed_plain_text,
+                    thinking_start_time=message.thinking_start_time,
+                    is_emoji=message.is_emoji)
+                logger.debug(f"{message.processed_plain_text},{typing_time},计算输入时间结束")
                 await asyncio.sleep(typing_time)
+                logger.debug(f"{message.processed_plain_text},{typing_time},等待输入时间结束")
 
                 message_json = message.to_dict()
 
@@ -69,14 +81,14 @@ class Message_Sender:
                     if end_point:
                         # logger.info(f"发送消息到{end_point}")
                         # logger.info(message_json)
-                        await global_api.send_message_REST(end_point, message_json)
-                    else:
                         try:
-                            await global_api.send_message(message)
+                            await global_api.send_message_REST(end_point, message_json)
                         except Exception as e:
-                            raise ValueError(
-                                f"未找到平台：{message.message_info.platform} 的url配置，请检查配置文件"
-                            ) from e
+                            logger.error(f"REST方式发送失败，出现错误: {str(e)}")
+                            logger.info("尝试使用ws发送")
+                            await self.send_via_ws(message)
+                    else:
+                        await self.send_via_ws(message)
                     logger.success(f"发送消息“{message_preview}”成功")
                 except Exception as e:
                     logger.error(f"发送消息“{message_preview}”失败: {str(e)}")
@@ -214,6 +226,8 @@ class MessageManager:
 
                 await message_earliest.process()
 
+                # print(f"message_earliest.thinking_start_tim22222e:{message_earliest.thinking_start_time}")
+                
                 await message_sender.send_message(message_earliest)
 
                 await self.storage.store_message(message_earliest, message_earliest.chat_stream)

src/plugins/chat/utils.py

@@ -334,26 +334,19 @@ def process_llm_response(text: str) -> List[str]:
     return sentences
 
 
-def calculate_typing_time(input_string: str, chinese_time: float = 0.2, english_time: float = 0.1) -> float:
+def calculate_typing_time(input_string: str, thinking_start_time: float, chinese_time: float = 0.2, english_time: float = 0.1, is_emoji: bool = False) -> float:
     """
     计算输入字符串所需的时间，中文和英文字符有不同的输入时间
         input_string (str): 输入的字符串
         chinese_time (float): 中文字符的输入时间，默认为0.2秒
         english_time (float): 英文字符的输入时间，默认为0.1秒
+        is_emoji (bool): 是否为emoji，默认为False
 
     特殊情况：
     - 如果只有一个中文字符，将使用3倍的中文输入时间
     - 在所有输入结束后，额外加上回车时间0.3秒
+    - 如果is_emoji为True，将使用固定1秒的输入时间
     """
-
-    # 如果输入是列表，将其连接成字符串
-    if isinstance(input_string, list):
-        input_string = ''.join(input_string)
-    
-    # 确保现在是字符串类型
-    if not isinstance(input_string, str):
-        input_string = str(input_string)
-
     mood_manager = MoodManager.get_instance()
     # 将0-1的唤醒度映射到-1到1
     mood_arousal = mood_manager.current_mood.arousal
@@ -376,7 +369,19 @@ def calculate_typing_time(input_string: str, chinese_time: float = 0.2, english_
         else:  # 其他字符（如英文）
             total_time += english_time
     
-    return total_time + 0.3  # 加上回车时间
+    
+    if is_emoji:
+        total_time = 1
+    
+    if time.time() - thinking_start_time > 10:
+        total_time = 1
+    
+    # print(f"thinking_start_time:{thinking_start_time}")
+    # print(f"nowtime:{time.time()}")
+    # print(f"nowtime - thinking_start_time:{time.time() - thinking_start_time}")
+    # print(f"{total_time}")
+    
+    return total_time  # 加上回车时间
 
 
 def cosine_similarity(v1, v2):

src/plugins/chat_module/reasoning_chat/reasoning_prompt_builder.py

@@ -145,12 +145,13 @@ class PromptBuilder:
         logger.info("开始构建prompt")
         
         prompt = f"""
+{relation_prompt_all}
 {memory_prompt}
 {prompt_info}
 {schedule_prompt}
 {chat_target}
 {chat_talking_prompt}
-现在"{sender_name}"说的:{message_txt}。引起了你的注意，你想要在群里发言发言或者回复这条消息。{relation_prompt_all}\n
+现在"{sender_name}"说的:{message_txt}。引起了你的注意，你想要在群里发言发言或者回复这条消息。\n
 你的网名叫{global_config.BOT_NICKNAME}，有人也叫你{"/".join(global_config.BOT_ALIAS_NAMES)}，{prompt_personality}。
 你正在{chat_target_2},现在请你读读之前的聊天记录，{mood_prompt}，然后给出日常且口语化的回复，平淡一些，
 尽量简短一些。{keywords_reaction_prompt}请注意把握聊天内容，不要回复的太有条理，可以有个性。{prompt_ger}

src/plugins/chat_module/think_flow_chat/think_flow_chat.py

@@ -95,6 +95,8 @@ class ThinkFlowChat:
             )
             if not mark_head:
                 mark_head = True
+                
+            # print(f"thinking_start_time:{bot_message.thinking_start_time}")
             message_set.add_message(bot_message)
         message_manager.add_message(message_set)
 

src/plugins/chat_module/think_flow_chat/think_flow_prompt_builder.py

@@ -123,7 +123,7 @@ class PromptBuilder:
 {chat_talking_prompt}
 你刚刚脑子里在想：
 {current_mind_info}
-现在"{sender_name}"说的:{message_txt}。引起了你的注意，你想要在群里发言发言或者回复这条消息。{relation_prompt_all}\n
+现在"{sender_name}"说的:{message_txt}。引起了你的注意，你想要在群里发言发言或者回复这条消息。\n
 你的网名叫{global_config.BOT_NICKNAME}，有人也叫你{"/".join(global_config.BOT_ALIAS_NAMES)}，{prompt_personality}。
 你正在{chat_target_2},现在请你读读之前的聊天记录，然后给出日常且口语化的回复，平淡一些，
 尽量简短一些。{keywords_reaction_prompt}请注意把握聊天内容，不要回复的太有条理，可以有个性。{prompt_ger}

src/plugins/config/config.py

@@ -25,8 +25,8 @@ config_config = LogConfig(
 logger = get_module_logger("config", config=config_config)
 
 #考虑到，实际上配置文件中的mai_version是不会自动更新的,所以采用硬编码
-mai_version_main = "0.6.0"
-mai_version_fix = ""
+mai_version_main = "0.6.1"
+mai_version_fix = "snapshot-1"
 if mai_version_fix:
     mai_version = f"{mai_version_main}-{mai_version_fix}"
 else:

src/plugins/message/api.py

@@ -29,7 +29,10 @@ class BaseMessageHandler:
             try:
                 tasks.append(handler(message))
             except Exception as e:
-                raise RuntimeError(str(e)) from e
+                logger.error(f"消息处理出错: {str(e)}")
+                logger.error(traceback.format_exc())
+                # 不抛出异常，而是记录错误并继续处理其他消息
+                continue
         if tasks:
             await asyncio.gather(*tasks, return_exceptions=True)
 
@@ -212,9 +215,8 @@ class MessageServer(BaseMessageHandler):
             try:
                 async with session.post(url, json=data, headers={"Content-Type": "application/json"}) as response:
                     return await response.json()
-            except Exception:
-                # logger.error(f"发送消息失败: {str(e)}")
-                pass
+            except Exception as e:
+                raise e
 
 
 class BaseMessageAPI:

template/bot_config_template.toml

@@ -162,7 +162,7 @@ pfc_chatting = false # 是否启用PFC聊天
 # stream = <true|false> : 用于指定模型是否是使用流式输出
 # 如果不指定，则该项是 False
 
-[model.llm_reasoning] #暂时未使用
+[model.llm_reasoning] #只在回复模式为reasoning时启用
 name = "Pro/deepseek-ai/DeepSeek-R1"
 # name = "Qwen/QwQ-32B"
 provider = "SILICONFLOW"