[refactor] (llm_analyzer) 为 llm_analyzer 解耦化

[fix] (LLM 输出提取和增强) 补充原本在话题分析存在但是不存在于 用户头衔和群圣经的 json 提取逻辑

.gitignore

@@ -0,0 +1,5 @@
+
+.kilocode/
+.kiro/
+.vscode/
+src/analysis/ARCHITECTURE.md

src/analysis/analyzers/__init__.py

@@ -0,0 +1,16 @@
+"""
+分析器模块
+包含各种LLM分析功能的实现
+"""
+
+from .base_analyzer import BaseAnalyzer
+from .topic_analyzer import TopicAnalyzer
+from .user_title_analyzer import UserTitleAnalyzer
+from .golden_quote_analyzer import GoldenQuoteAnalyzer
+
+__all__ = [
+    'BaseAnalyzer',
+    'TopicAnalyzer',
+    'UserTitleAnalyzer',
+    'GoldenQuoteAnalyzer'
+]

src/analysis/analyzers/base_analyzer.py

@@ -0,0 +1,177 @@
+"""
+基础分析器抽象类
+定义通用分析流程和接口
+"""
+
+from abc import ABC, abstractmethod
+from typing import List, Dict, Tuple, Any, Optional
+from datetime import datetime
+from astrbot.api import logger
+from ...models.data_models import TokenUsage
+from ..utils.json_utils import parse_json_response
+from ..utils.llm_utils import call_provider_with_retry, extract_token_usage, extract_response_text
+import re
+
+class BaseAnalyzer(ABC):
+    """
+    基础分析器抽象类
+    定义所有分析器的通用接口和流程
+    """
+    
+    def __init__(self, context, config_manager):
+        """
+        初始化基础分析器
+        
+        Args:
+            context: AstrBot上下文对象
+            config_manager: 配置管理器
+        """
+        self.context = context
+        self.config_manager = config_manager
+    
+    @abstractmethod
+    def get_data_type(self) -> str:
+        """
+        获取数据类型标识
+        
+        Returns:
+            数据类型字符串
+        """
+        pass
+    
+    @abstractmethod
+    def get_max_count(self) -> int:
+        """
+        获取最大提取数量
+        
+        Returns:
+            最大数量
+        """
+        pass
+    
+    @abstractmethod
+    def build_prompt(self, data: Any) -> str:
+        """
+        构建LLM提示词
+        
+        Args:
+            data: 输入数据
+            
+        Returns:
+            提示词字符串
+        """
+        pass
+    
+    @abstractmethod
+    def extract_with_regex(self, result_text: str, max_count: int) -> List[Dict]:
+        """
+        使用正则表达式提取数据
+        
+        Args:
+            result_text: LLM响应文本
+            max_count: 最大提取数量
+            
+        Returns:
+            提取到的数据列表
+        """
+        pass
+    
+    @abstractmethod
+    def create_data_objects(self, data_list: List[Dict]) -> List[Any]:
+        """
+        创建数据对象列表
+        
+        Args:
+            data_list: 原始数据列表
+            
+        Returns:
+            数据对象列表
+        """
+        pass
+    
+    async def analyze(self, data: Any, umo: str = None) -> Tuple[List[Any], TokenUsage]:
+        """
+        统一的分析流程
+        
+        Args:
+            data: 输入数据
+            umo: 模型唯一标识符
+            
+        Returns:
+            (分析结果列表, Token使用统计)
+        """
+        try:
+            # 1. 构建提示词
+            prompt = self.build_prompt(data)
+            logger.info(f"开始{self.get_data_type()}分析，构建提示词完成")
+            
+            # 2. 调用LLM
+            max_tokens = self.get_max_tokens()
+            temperature = self.get_temperature()
+            
+            response = await call_provider_with_retry(
+                self.context, self.config_manager, prompt, 
+                max_tokens, temperature, umo
+            )
+            
+            if response is None:
+                logger.error(f"{self.get_data_type()}分析调用LLM失败: provider返回None（重试失败）")
+                return [], TokenUsage()
+            
+            # 3. 提取token使用统计
+            token_usage_dict = extract_token_usage(response)
+            token_usage = TokenUsage(
+                prompt_tokens=token_usage_dict["prompt_tokens"],
+                completion_tokens=token_usage_dict["completion_tokens"],
+                total_tokens=token_usage_dict["total_tokens"]
+            )
+            
+            # 4. 提取响应文本
+            result_text = extract_response_text(response)
+            logger.debug(f"{self.get_data_type()}分析原始响应: {result_text[:500]}...")
+            
+            # 5. 尝试JSON解析
+            success, parsed_data, error_msg = parse_json_response(result_text, self.get_data_type())
+            
+            if success and parsed_data:
+                # JSON解析成功，创建数据对象
+                data_objects = self.create_data_objects(parsed_data)
+                logger.info(f"{self.get_data_type()}分析成功，解析到 {len(data_objects)} 条数据")
+                return data_objects, token_usage
+            
+            # 6. JSON解析失败，使用正则表达式降级
+            logger.warning(f"{self.get_data_type()}JSON解析失败，尝试正则表达式提取: {error_msg}")
+            regex_data = self.extract_with_regex(result_text, self.get_max_count())
+            
+            if regex_data:
+                logger.info(f"{self.get_data_type()}正则表达式提取成功，获得 {len(regex_data)} 条数据")
+                data_objects = self.create_data_objects(regex_data)
+                return data_objects, token_usage
+            else:
+                # 最后的降级方案
+                logger.warning(f"{self.get_data_type()}正则表达式提取失败，返回空列表")
+                return [], token_usage
+                
+        except Exception as e:
+            logger.error(f"{self.get_data_type()}分析失败: {e}")
+            return [], TokenUsage()
+    
+    def get_max_tokens(self) -> int:
+        """
+        获取最大token数，子类可重写
+        
+        Returns:
+            最大token数
+        """
+        return 10000
+    
+    def get_temperature(self) -> float:
+        """
+        获取温度参数，子类可重写
+        
+        Returns:
+            温度参数
+        """
+        return 0.6
+    
+    

src/analysis/analyzers/golden_quote_analyzer.py

@@ -0,0 +1,195 @@
+"""
+金句分析模块
+专门处理群聊金句提取和分析
+"""
+
+from typing import List, Dict, Tuple
+from datetime import datetime
+from astrbot.api import logger
+from ...models.data_models import GoldenQuote, TokenUsage
+from .base_analyzer import BaseAnalyzer
+from ..utils.json_utils import extract_golden_quotes_with_regex
+
+
+
+class GoldenQuoteAnalyzer(BaseAnalyzer):
+    """
+    金句分析器
+    专门处理群聊金句的提取和分析
+    """
+    
+    def get_data_type(self) -> str:
+        """获取数据类型标识"""
+        return "金句"
+    
+    def get_max_count(self) -> int:
+        """获取最大金句数量"""
+        return self.config_manager.get_max_golden_quotes()
+    
+    def get_max_tokens(self) -> int:
+        """获取最大token数"""
+        return 1500
+    
+    def get_temperature(self) -> float:
+        """获取温度参数"""
+        return 0.7
+    
+    def build_prompt(self, messages: List[Dict]) -> str:
+        """
+        构建金句分析提示词
+        
+        Args:
+            messages: 圣经的文本消息列表
+            
+        Returns:
+            提示词字符串
+        """
+        if not messages:
+            return ""
+        
+        # 构建消息文本
+        messages_text = "\n".join([
+            f"[{msg['time']}] {msg['sender']}: {msg['content']}"
+            for msg in messages
+        ])
+        
+        max_golden_quotes = self.get_max_count()
+        
+        prompt = f"""
+请从以下群聊记录中挑选出{max_golden_quotes}句最具冲击力、最令人惊叹的"金句"。这些金句需满足：
+- 核心标准：**逆天的神人发言**，即具备颠覆常识的脑洞、逻辑跳脱的表达或强烈反差感的原创内容
+- 典型特征：包含某些争议话题元素、夸张类比、反常规结论、一本正经的"胡说八道"或突破语境的清奇思路，并且具备一定的冲击力，让人印象深刻。
+
+对于每个金句，请提供：
+1. 原文内容（完整保留发言细节）
+2. 发言人昵称
+3. 选择理由（具体说明其"逆天"之处，如逻辑颠覆点/脑洞角度/反差感/争议话题元素）
+
+此外，我将对你进行严格约束：
+- 优先筛选 **逆天指数最高** 的内容：发情、性压抑话题 > 争议话题 > 元素级 > 颠覆认知级 > 逻辑跳脱级 > 趣味调侃级，剔除单纯玩梗或网络热词堆砌的普通发言
+- 重点标记包含极端类比、反常识论证或无厘头结论的内容，并且包含一定的争议话题元素。
+
+群聊记录：
+{messages_text}
+
+请以JSON格式返回，格式如下：
+[
+  {{
+    "content": "金句原文",
+    "sender": "发言人昵称",
+    "reason": "选择这句话的理由（需明确说明逆天特质）"
+  }}
+]
+"""
+        return prompt
+    
+    def extract_with_regex(self, result_text: str, max_count: int) -> List[Dict]:
+        """
+        使用正则表达式提取金句信息
+        
+        Args:
+            result_text: LLM响应文本
+            max_count: 最大提取数量
+            
+        Returns:
+            金句数据列表
+        """
+        return extract_golden_quotes_with_regex(result_text, max_count)
+    
+    def create_data_objects(self, quotes_data: List[Dict]) -> List[GoldenQuote]:
+        """
+        创建金句对象列表
+        
+        Args:
+            quotes_data: 原始金句数据列表
+            
+        Returns:
+            GoldenQuote对象列表
+        """
+        try:
+            quotes = []
+            max_quotes = self.get_max_count()
+            
+            for quote_data in quotes_data[:max_quotes]:
+                # 确保数据格式正确
+                content = quote_data.get("content", "").strip()
+                sender = quote_data.get("sender", "").strip()
+                reason = quote_data.get("reason", "").strip()
+                
+                # 验证必要字段
+                if not content or not sender or not reason:
+                    logger.warning(f"金句数据格式不完整，跳过: {quote_data}")
+                    continue
+                
+                quotes.append(GoldenQuote(
+                    content=content,
+                    sender=sender,
+                    reason=reason
+                ))
+            
+            return quotes
+            
+        except Exception as e:
+            logger.error(f"创建金句对象失败: {e}")
+            return []
+    
+    def extract_interesting_messages(self, messages: List[Dict]) -> List[Dict]:
+        """
+        提取圣经的文本消息
+        
+        Args:
+            messages: 群聊消息列表
+            
+        Returns:
+            圣经的文本消息列表
+        """
+        try:
+            interesting_messages = []
+            
+            for msg in messages:
+                sender = msg.get("sender", {})
+                nickname = sender.get("nickname", "") or sender.get("card", "")
+                msg_time = datetime.fromtimestamp(msg.get("time", 0)).strftime("%H:%M")
+                
+                for content in msg.get("message", []):
+                    if content.get("type") == "text":
+                        text = content.get("data", {}).get("text", "").strip()
+                        # 过滤长度适中、可能圣经的消息
+                        if 5 <= len(text) <= 100 and not text.startswith(("http", "www", "/")):
+                            interesting_messages.append({
+                                "sender": nickname,
+                                "time": msg_time,
+                                "content": text
+                            })
+            
+            return interesting_messages
+            
+        except Exception as e:
+            logger.error(f"提取圣经消息失败: {e}")
+            return []
+    
+    async def analyze_golden_quotes(self, messages: List[Dict], umo: str = None) -> Tuple[List[GoldenQuote], TokenUsage]:
+        """
+        分析群聊金句
+        
+        Args:
+            messages: 群聊消息列表
+            umo: 模型唯一标识符
+            
+        Returns:
+            (金句列表, Token使用统计)
+        """
+        try:
+            # 提取圣经的文本消息
+            interesting_messages = self.extract_interesting_messages(messages)
+            
+            if not interesting_messages:
+                logger.info("没有符合条件的圣经消息，返回空结果")
+                return [], TokenUsage()
+            
+            logger.info(f"开始从 {len(interesting_messages)} 条圣经消息中提取金句")
+            return await self.analyze(interesting_messages, umo)
+            
+        except Exception as e:
+            logger.error(f"金句分析失败: {e}")
+            return [], TokenUsage()

src/analysis/analyzers/topic_analyzer.py

@@ -0,0 +1,207 @@
+"""
+话题分析模块
+专门处理群聊话题分析
+"""
+
+from typing import List, Dict, Tuple
+from datetime import datetime
+import re
+from astrbot.api import logger
+from ...models.data_models import SummaryTopic, TokenUsage
+from .base_analyzer import BaseAnalyzer
+from ..utils.json_utils import extract_topics_with_regex
+
+
+class TopicAnalyzer(BaseAnalyzer):
+    """
+    话题分析器
+    专门处理群聊话题的提取和分析
+    """
+    
+    def get_data_type(self) -> str:
+        """获取数据类型标识"""
+        return "话题"
+    
+    def get_max_count(self) -> int:
+        """获取最大话题数量"""
+        return self.config_manager.get_max_topics()
+    
+    def get_max_tokens(self) -> int:
+        """获取最大token数"""
+        return 10000
+    
+    def get_temperature(self) -> float:
+        """获取温度参数"""
+        return 0.6
+    
+    def build_prompt(self, messages: List[Dict]) -> str:
+        """
+        构建话题分析提示词
+        
+        Args:
+            messages: 群聊消息列表
+            
+        Returns:
+            提示词字符串
+        """
+        # 提取文本消息
+        text_messages = []
+        for msg in messages:
+            sender = msg.get("sender", {})
+            nickname = sender.get("nickname", "") or sender.get("card", "")
+            msg_time = datetime.fromtimestamp(msg.get("time", 0)).strftime("%H:%M")
+            
+            for content in msg.get("message", []):
+                if content.get("type") == "text":
+                    text = content.get("data", {}).get("text", "").strip()
+                    if text and len(text) > 2 and not text.startswith("/"):
+                        # 清理消息内容
+                        text = text.replace('"', '"').replace('"', '"')
+                        text = text.replace(''', "'").replace(''', "'")
+                        text = text.replace('\n', ' ').replace('\r', ' ')
+                        text = text.replace('\t', ' ')
+                        text = re.sub(r'[\x00-\x1f\x7f-\x9f]', '', text)
+                        text_messages.append({
+                            "sender": nickname,
+                            "time": msg_time,
+                            "content": text.strip()
+                        })
+        
+        if not text_messages:
+            return ""
+        
+        # 构建消息文本
+        messages_text = "\n".join([
+            f"[{msg['time']}] {msg['sender']}: {msg['content']}"
+            for msg in text_messages
+        ])
+        
+        max_topics = self.get_max_count()
+        
+        prompt = f"""
+你是一个帮我进行群聊信息总结的助手，生成总结内容时，你需要严格遵守下面的几个准则：
+请分析接下来提供的群聊记录，提取出最多{max_topics}个主要话题。
+
+对于每个话题，请提供：
+1. 话题名称（突出主题内容，尽量简明扼要）
+2. 主要参与者（最多5人）
+3. 话题详细描述（包含关键信息和结论）
+
+注意：
+- 对于比较有价值的点，稍微用一两句话详细讲讲，比如不要生成 "Nolan 和 SOV 讨论了 galgame 中关于性符号的衍生情况" 这种宽泛的内容，而是生成更加具体的讨论内容，让其他人只看这个消息就能知道讨论中有价值的，有营养的信息。
+- 对于其中的部分信息，你需要特意提到主题施加的主体是谁，是哪个群友做了什么事情，而不要直接生成和群友没有关系的语句。
+- 对于每一条总结，尽量讲清楚前因后果，以及话题的结论，是什么，为什么，怎么做，如果用户没有讲到细节，则可以不用这么做。
+
+群聊记录：
+{messages_text}
+
+重要：必须返回标准JSON格式，严格遵守以下规则：
+1. 只使用英文双引号 " 不要使用中文引号 " "
+2. 字符串内容中的引号必须转义为 \"
+3. 多个对象之间用逗号分隔
+4. 数组元素之间用逗号分隔
+5. 不要在JSON外添加任何文字说明
+6. 描述内容避免使用特殊符号，用普通文字表达
+
+请严格按照以下JSON格式返回，确保可以被标准JSON解析器解析：
+[
+  {{
+    "topic": "话题名称",
+    "contributors": ["用户1", "用户2"],
+    "detail": "话题描述内容"
+  }},
+  {{
+    "topic": "另一个话题",
+    "contributors": ["用户3", "用户4"],
+    "detail": "另一个话题的描述"
+  }}
+]
+
+注意：返回的内容必须是纯JSON，不要包含markdown代码块标记或其他格式
+"""
+        return prompt
+    
+    def extract_with_regex(self, result_text: str, max_topics: int) -> List[Dict]:
+        """
+        使用正则表达式提取话题信息
+        
+        Args:
+            result_text: LLM响应文本
+            max_topics: 最大话题数量
+            
+        Returns:
+            话题数据列表
+        """
+        return extract_topics_with_regex(result_text, max_topics)
+    
+    def create_data_objects(self, topics_data: List[Dict]) -> List[SummaryTopic]:
+        """
+        创建话题对象列表
+        
+        Args:
+            topics_data: 原始话题数据列表
+            
+        Returns:
+            SummaryTopic对象列表
+        """
+        try:
+            topics = []
+            max_topics = self.get_max_count()
+            
+            for topic_data in topics_data[:max_topics]:
+                # 确保数据格式正确
+                topic_name = topic_data.get("topic", "").strip()
+                contributors = topic_data.get("contributors", [])
+                detail = topic_data.get("detail", "").strip()
+                
+                # 验证必要字段
+                if not topic_name or not detail:
+                    logger.warning(f"话题数据格式不完整，跳过: {topic_data}")
+                    continue
+                
+                # 确保参与者列表有效
+                if not contributors or not isinstance(contributors, list):
+                    contributors = ["群友"]
+                else:
+                    # 清理参与者名称
+                    contributors = [str(c).strip() for c in contributors if c and str(c).strip()]
+                    if not contributors:
+                        contributors = ["群友"]
+                
+                topics.append(SummaryTopic(
+                    topic=topic_name,
+                    contributors=contributors[:5],  # 最多5个参与者
+                    detail=detail
+                ))
+            
+            return topics
+            
+        except Exception as e:
+            logger.error(f"创建话题对象失败: {e}")
+            return []
+    
+    async def analyze_topics(self, messages: List[Dict], umo: str = None) -> Tuple[List[SummaryTopic], TokenUsage]:
+        """
+        分析群聊话题
+        
+        Args:
+            messages: 群聊消息列表
+            umo: 模型唯一标识符
+            
+        Returns:
+            (话题列表, Token使用统计)
+        """
+        try:
+            # 提取文本消息
+            text_messages = self.extract_text_messages(messages)
+            
+            if not text_messages:
+                logger.info("没有有效的文本消息，返回空结果")
+                return [], TokenUsage()
+            
+            logger.info(f"开始分析 {len(text_messages)} 条文本消息中的话题")
+            return await self.analyze(text_messages, umo)
+            
+        except Exception as e:
+            logger.error(f"话题分析失败: {e}")
+            return [], TokenUsage()

src/analysis/analyzers/user_title_analyzer.py

@@ -0,0 +1,222 @@
+"""
+用户称号分析模块
+专门处理用户称号和MBTI类型分析
+"""
+
+from typing import List, Dict, Tuple
+from astrbot.api import logger
+from ...models.data_models import UserTitle, TokenUsage
+from .base_analyzer import BaseAnalyzer
+from ..utils.json_utils import extract_user_titles_with_regex
+
+
+class UserTitleAnalyzer(BaseAnalyzer):
+    """
+    用户称号分析器
+    专门处理用户称号分配和MBTI类型分析
+    """
+    
+    def get_data_type(self) -> str:
+        """获取数据类型标识"""
+        return "用户称号"
+    
+    def get_max_count(self) -> int:
+        """获取最大用户称号数量"""
+        return self.config_manager.get_max_user_titles()
+    
+    def get_max_tokens(self) -> int:
+        """获取最大token数"""
+        return 1500
+    
+    def get_temperature(self) -> float:
+        """获取温度参数"""
+        return 0.5
+    
+    def build_prompt(self, user_data: Dict) -> str:
+        """
+        构建用户称号分析提示词
+        
+        Args:
+            user_data: 用户数据字典，包含用户统计信息
+            
+        Returns:
+            提示词字符串
+        """
+        user_summaries = user_data.get("user_summaries", [])
+        
+        if not user_summaries:
+            return ""
+        
+        # 构建用户数据文本
+        users_text = "\n".join([
+            f"- {user['name']} (QQ:{user['qq']}): "
+            f"发言{user['message_count']}条, 平均{user['avg_chars']}字, "
+            f"表情比例{user['emoji_ratio']}, 夜间发言比例{user['night_ratio']}, "
+            f"回复比例{user['reply_ratio']}"
+            for user in user_summaries
+        ])
+        
+        prompt = f"""
+请为以下群友分配合适的称号和MBTI类型。每个人只能有一个称号，每个称号只能给一个人。
+
+可选称号：
+- 龙王: 发言频繁但内容轻松的人
+- 技术专家: 经常讨论技术话题的人
+- 夜猫子: 经常在深夜发言的人
+- 表情包军火库: 经常发表情的人
+- 沉默终结者: 经常开启话题的人
+- 评论家: 平均发言长度很长的人
+- 阳角: 在群里很有影响力的人
+- 互动达人: 经常回复别人的人
+- ... (你可以自行进行拓展添加)
+
+用户数据：
+{users_text}
+
+请以JSON格式返回，格式如下：
+[
+  {{
+    "name": "用户名",
+    "qq": 123456789,
+    "title": "称号",
+    "mbti": "MBTI类型",
+    "reason": "获得此称号的原因"
+  }}
+]
+"""
+        return prompt
+    
+    def extract_with_regex(self, result_text: str, max_count: int) -> List[Dict]:
+        """
+        使用正则表达式提取用户称号信息
+        
+        Args:
+            result_text: LLM响应文本
+            max_count: 最大提取数量
+            
+        Returns:
+            用户称号数据列表
+        """
+        return extract_user_titles_with_regex(result_text, max_count)
+    
+    def create_data_objects(self, titles_data: List[Dict]) -> List[UserTitle]:
+        """
+        创建用户称号对象列表
+        
+        Args:
+            titles_data: 原始用户称号数据列表
+            
+        Returns:
+            UserTitle对象列表
+        """
+        try:
+            titles = []
+            max_titles = self.get_max_count()
+            
+            for title_data in titles_data[:max_titles]:
+                # 确保数据格式正确
+                name = title_data.get("name", "").strip()
+                qq = title_data.get("qq")
+                title = title_data.get("title", "").strip()
+                mbti = title_data.get("mbti", "").strip()
+                reason = title_data.get("reason", "").strip()
+                
+                # 验证必要字段
+                if not name or not title or not mbti or not reason:
+                    logger.warning(f"用户称号数据格式不完整，跳过: {title_data}")
+                    continue
+                
+                # 验证QQ号格式
+                try:
+                    qq = int(qq)
+                except (ValueError, TypeError):
+                    logger.warning(f"QQ号格式无效，跳过: {qq}")
+                    continue
+                
+                titles.append(UserTitle(
+                    name=name,
+                    qq=qq,
+                    title=title,
+                    mbti=mbti,
+                    reason=reason
+                ))
+            
+            return titles
+            
+        except Exception as e:
+            logger.error(f"创建用户称号对象失败: {e}")
+            return []
+    
+    def prepare_user_data(self, messages: List[Dict], user_analysis: Dict) -> Dict:
+        """
+        准备用户数据
+        
+        Args:
+            messages: 群聊消息列表
+            user_analysis: 用户分析统计
+            
+        Returns:
+            准备好的用户数据字典
+        """
+        try:
+            user_summaries = []
+            
+            for user_id, stats in user_analysis.items():
+                if stats["message_count"] < 5:  # 过滤活跃度太低的用户
+                    continue
+                
+                # 分析用户特征
+                night_messages = sum(stats["hours"][h] for h in range(0, 6))
+                day_messages = stats["message_count"] - night_messages
+                avg_chars = stats["char_count"] / stats["message_count"] if stats["message_count"] > 0 else 0
+                
+                user_summaries.append({
+                    "name": stats["nickname"],
+                    "qq": int(user_id),
+                    "message_count": stats["message_count"],
+                    "avg_chars": round(avg_chars, 1),
+                    "emoji_ratio": round(stats["emoji_count"] / stats["message_count"], 2),
+                    "night_ratio": round(night_messages / stats["message_count"], 2),
+                    "reply_ratio": round(stats["reply_count"] / stats["message_count"], 2)
+                })
+            
+            if not user_summaries:
+                return {"user_summaries": []}
+            
+            # 按消息数量排序，取前N名
+            max_user_titles = self.get_max_count()
+            user_summaries.sort(key=lambda x: x["message_count"], reverse=True)
+            user_summaries = user_summaries[:max_user_titles]
+            
+            return {"user_summaries": user_summaries}
+            
+        except Exception as e:
+            logger.error(f"准备用户数据失败: {e}")
+            return {"user_summaries": []}
+    
+    async def analyze_user_titles(self, messages: List[Dict], user_analysis: Dict, umo: str = None) -> Tuple[List[UserTitle], TokenUsage]:
+        """
+        分析用户称号
+        
+        Args:
+            messages: 群聊消息列表
+            user_analysis: 用户分析统计
+            umo: 模型唯一标识符
+            
+        Returns:
+            (用户称号列表, Token使用统计)
+        """
+        try:
+            # 准备用户数据
+            user_data = self.prepare_user_data(messages, user_analysis)
+            
+            if not user_data["user_summaries"]:
+                logger.info("没有符合条件的用户，返回空结果")
+                return [], TokenUsage()
+            
+            logger.info(f"开始分析 {len(user_data['user_summaries'])} 个用户的称号")
+            return await self.analyze(user_data, umo)
+            
+        except Exception as e:
+            logger.error(f"用户称号分析失败: {e}")
+            return [], TokenUsage()

src/analysis/llm_analyzer.py

@@ -1,580 +1,145 @@
 """
 LLM分析器模块
-负责使用LLM进行话题分析、用户称号分析和金句分析
+负责协调各个分析器进行话题分析、用户称号分析和金句分析
 """
 
-import json
-import re
-from datetime import datetime
-import asyncio
 from typing import List, Dict, Tuple
 from astrbot.api import logger
-from ...src.models.data_models import SummaryTopic, UserTitle, GoldenQuote, TokenUsage
+from ..models.data_models import SummaryTopic, UserTitle, GoldenQuote, TokenUsage
+from .analyzers.topic_analyzer import TopicAnalyzer
+from .analyzers.user_title_analyzer import UserTitleAnalyzer
+from .analyzers.golden_quote_analyzer import GoldenQuoteAnalyzer
+from .utils.llm_utils import call_provider_with_retry
+from .utils.json_utils import fix_json
+from .utils.json_utils import extract_topics_with_regex, extract_user_titles_with_regex, extract_golden_quotes_with_regex
 
 
 class LLMAnalyzer:
-    """LLM分析器"""
-
+    """
+    LLM分析器
+    作为统一入口，协调各个专门的分析器进行不同类型的分析
+    保持向后兼容性，提供原有的接口
+    """
+    
     def __init__(self, context, config_manager):
+        """
+        初始化LLM分析器
+        
+        Args:
+            context: AstrBot上下文对象
+            config_manager: 配置管理器
+        """
         self.context = context
         self.config_manager = config_manager
-
-    async def _call_provider_with_retry(self, provider, prompt: str, max_tokens: int, temperature: float, umo: str = None):
-        """
-        调用LLM提供者，带超时、重试与退避。支持自定义服务商。
-
-        Args:
-            provider: LLM服务商实例或None。
-            prompt (str): 输入的提示语。
-            max_tokens (int): 最大生成token数。
-            temperature (float): 采样温度。
-            umo (str, optional): 指定使用的模型唯一标识符（Unique Model Object），
-                用于选择特定的LLM服务商或模型。格式通常为字符串，例如 "gpt-3.5-turbo"。
-                如果为None，则使用默认模型。
-
-        Returns:
-            LLM生成的结果。
-        """
-        timeout = self.config_manager.get_llm_timeout()
-        retries = self.config_manager.get_llm_retries()
-        backoff = self.config_manager.get_llm_backoff()
-
-        # 获取自定义服务商参数
-        custom_api_key = self.config_manager.get_custom_api_key()
-        custom_api_base = self.config_manager.get_custom_api_base_url()
-        custom_model = self.config_manager.get_custom_model_name()
-
-        last_exc = None
-        for attempt in range(1, retries + 1):
-            try:
-                if custom_api_key and custom_api_base and custom_model:
-                    logger.info(f"使用自定义LLM提供商: {custom_api_base} model={custom_model}")
-                    import aiohttp
-                    async with aiohttp.ClientSession() as session:
-                        headers = {
-                            "Authorization": f"Bearer {custom_api_key}",
-                            "Content-Type": "application/json"
-                        }
-                        payload = {
-                            "model": custom_model,
-                            "messages": [{"role": "user", "content": prompt}],
-                            "max_tokens": max_tokens,
-                            "temperature": temperature
-                        }
-                        aio_timeout = aiohttp.ClientTimeout(total=timeout)
-                        async with session.post(custom_api_base, json=payload, headers=headers, timeout=aio_timeout) as resp:
-                            if resp.status != 200:
-                                error_text = await resp.text()
-                                logger.error(f"自定义LLM服务商请求失败: HTTP {resp.status}, 内容: {error_text}")
-                            try:
-                                response_json = await resp.json()
-                            except Exception as json_err:
-                                error_text = await resp.text()
-                                logger.error(f"自定义LLM服务商响应JSON解析失败: {json_err}, 内容: {error_text}")
-                                return None
-                            # 兼容 OpenAI 格式，安全访问嵌套字段
-                            content = None
-                            try:
-                                choices = response_json.get("choices")
-                                if choices and isinstance(choices, list) and len(choices) > 0:
-                                    message = choices[0].get("message")
-                                    if message and isinstance(message, dict):
-                                        content = message.get("content")
-                                if content is None:
-                                    logger.error(f"自定义LLM响应格式异常: {response_json}")
-                                    return None
-                            except Exception as key_err:
-                                logger.error(f"自定义LLM响应结构解析失败: {key_err}, 响应内容: {response_json}")
-                                return None
-                            # 构造一个兼容原有逻辑的对象
-                            class CustomResponse:
-                                completion_text = content
-                                raw_completion = response_json
-                            return CustomResponse()
-                else:
-                    # 确保使用当前指定的模型
-                    if provider is None:
-                        provider = self.context.get_using_provider(umo=umo)
-                        provider_id = 'unknown'
-                        if provider:
-                            try:
-                                meta = provider.meta()
-                                provider_id = meta.id
-                            except Exception as e:
-                                logger.debug(f"获取提供商ID失败: {e}")
-                        logger.info(f"获取到的 provider ID: {provider_id}")
-                        if not provider or provider_id == 'unknown':
-                            logger.warning(f"获取的提供商不正确 (Provider ID: {provider_id})")
-
-                    
-                    logger.info(f"使用LLM provider: {provider}")
-                    if not provider:
-                        logger.error("provider 为空，无法调用 text_chat，直接返回 None")
-                        return None
-                    coro = provider.text_chat(prompt=prompt, max_tokens=max_tokens, temperature=temperature)
-                    return await asyncio.wait_for(coro, timeout=timeout)
-            except asyncio.TimeoutError as e:
-                last_exc = e
-                logger.warning(f"LLM请求超时: 第{attempt}次, timeout={timeout}s")
-            except Exception as e:
-                last_exc = e
-                logger.warning(f"LLM请求失败: 第{attempt}次, 错误: {last_exc}")
-            # 若非最后一次，等待退避后重试
-            if attempt < retries:
-                await asyncio.sleep(backoff * attempt)
-
-        # 最终仍失败，记录错误并返回 None 由调用方处理降级，避免抛出异常
-        logger.error(f"LLM请求全部重试失败: {last_exc}")
-        return None
-
+        
+        # 初始化各个专门的分析器
+        self.topic_analyzer = TopicAnalyzer(context, config_manager)
+        self.user_title_analyzer = UserTitleAnalyzer(context, config_manager)
+        self.golden_quote_analyzer = GoldenQuoteAnalyzer(context, config_manager)
+    
     async def analyze_topics(self, messages: List[Dict], umo: str = None) -> Tuple[List[SummaryTopic], TokenUsage]:
-        """使用LLM分析话题"""
+        """
+        使用LLM分析话题
+        保持原有接口，委托给专门的TopicAnalyzer处理
+        
+        Args:
+            messages: 群聊消息列表
+            umo: 模型唯一标识符
+            
+        Returns:
+            (话题列表, Token使用统计)
+        """
         try:
-            # 提取文本消息
-            text_messages = []
-            for msg in messages:
-                sender = msg.get("sender", {})
-                nickname = sender.get("nickname", "") or sender.get("card", "")
-                msg_time = datetime.fromtimestamp(msg.get("time", 0)).strftime("%H:%M")
-
-                for content in msg.get("message", []):
-                    if content.get("type") == "text":
-                        text = content.get("data", {}).get("text", "").strip()
-                        if text and len(text) > 2 and not text.startswith(("/")):
-                            text_messages.append({
-                                "sender": nickname,
-                                "time": msg_time,
-                                "content": text
-                            })
-
-            if not text_messages:
-                return [], TokenUsage()
-
-            # 构建LLM提示词，清理消息内容
-            def clean_message_content(content):
-                """清理消息内容，移除可能影响JSON解析的字符"""
-                # 替换中文引号
-                content = content.replace('"', '"').replace('"', '"')
-                content = content.replace(''', "'").replace(''', "'")
-                # 移除或替换其他特殊字符
-                content = content.replace('\n', ' ').replace('\r', ' ')
-                content = content.replace('\t', ' ')
-                # 移除可能的控制字符
-                content = re.sub(r'[\x00-\x1f\x7f-\x9f]', '', content)
-                return content.strip()
-
-            messages_text = "\n".join([
-                f"[{msg['time']}] {msg['sender']}: {clean_message_content(msg['content'])}"
-                for msg in text_messages
-            ])
-
-            max_topics = self.config_manager.get_max_topics()
-            prompt = f"""
-你是一个帮我进行群聊信息总结的助手，生成总结内容时，你需要严格遵守下面的几个准则：
-请分析接下来提供的群聊记录，提取出最多{max_topics}个主要话题。
-
-对于每个话题，请提供：
-1. 话题名称（突出主题内容，尽量简明扼要）
-2. 主要参与者（最多5人）
-3. 话题详细描述（包含关键信息和结论）
-
-注意：
-- 对于比较有价值的点，稍微用一两句话详细讲讲，比如不要生成 "Nolan 和 SOV 讨论了 galgame 中关于性符号的衍生情况" 这种宽泛的内容，而是生成更加具体的讨论内容，让其他人只看这个消息就能知道讨论中有价值的，有营养的信息。
-- 对于其中的部分信息，你需要特意提到主题施加的主体是谁，是哪个群友做了什么事情，而不要直接生成和群友没有关系的语句。
-- 对于每一条总结，尽量讲清楚前因后果，以及话题的结论，是什么，为什么，怎么做，如果用户没有讲到细节，则可以不用这么做。
-
-群聊记录：
-{messages_text}
-
-重要：必须返回标准JSON格式，严格遵守以下规则：
-1. 只使用英文双引号 " 不要使用中文引号 " "
-2. 字符串内容中的引号必须转义为 \"
-3. 多个对象之间用逗号分隔
-4. 数组元素之间用逗号分隔
-5. 不要在JSON外添加任何文字说明
-6. 描述内容避免使用特殊符号，用普通文字表达
-
-请严格按照以下JSON格式返回，确保可以被标准JSON解析器解析：
-[
-  {{
-    "topic": "话题名称",
-    "contributors": ["用户1", "用户2"],
-    "detail": "话题描述内容"
-  }},
-  {{
-    "topic": "另一个话题",
-    "contributors": ["用户3", "用户4"],
-    "detail": "另一个话题的描述"
-  }}
-]
-
-注意：返回的内容必须是纯JSON，不要包含markdown代码块标记或其他格式
-"""
-
-            # 调用LLM
-            response = await self._call_provider_with_retry(None, prompt, max_tokens=10000, temperature=0.6, umo=umo)
-            if response is None:
-                logger.error("话题分析调用LLM失败: provider返回None（重试失败）")
-                return [], TokenUsage()
-
-            # 提取token使用统计
-            token_usage = TokenUsage()
-            # 安全地提取 usage，避免 response.raw_completion.usage 为 None 导致的 AttributeError
-            usage = None
-            if getattr(response, 'raw_completion', None) is not None:
-                usage = getattr(response.raw_completion, 'usage', None)
-                if usage:
-                    token_usage.prompt_tokens = getattr(usage, 'prompt_tokens', 0) or 0
-                    token_usage.completion_tokens = getattr(usage, 'completion_tokens', 0) or 0
-                    token_usage.total_tokens = getattr(usage, 'total_tokens', 0) or 0
-
-            # 解析响应
-            if hasattr(response, 'completion_text'):
-                result_text = response.completion_text
-            else:
-                result_text = str(response)
-
-            # 尝试解析JSON
-            try:
-                # 提取JSON部分
-                json_match = re.search(r'\[.*?\]', result_text, re.DOTALL)
-                if json_match:
-                    json_text = json_match.group()
-                    logger.debug(f"话题分析JSON原文: {json_text[:500]}...")
-
-                    # 强化JSON清理和修复
-                    json_text = self._fix_json(json_text)
-                    logger.debug(f"修复后的JSON: {json_text[:300]}...")
-
-                    topics_data = json.loads(json_text)
-                    topics = [SummaryTopic(**topic) for topic in topics_data[:max_topics]]
-                    logger.info(f"话题分析成功，解析到 {len(topics)} 个话题")
-                    return topics, token_usage
-                else:
-                    logger.warning(f"话题分析响应中未找到JSON格式，响应内容: {result_text[:200]}...")
-            except json.JSONDecodeError as e:
-                logger.warning(f"话题分析JSON解析失败: {e}")
-                logger.debug(f"修复后的JSON: {json_text if 'json_text' in locals() else 'N/A'}")
-                logger.debug(f"原始响应: {result_text}")
-
-                # 如果JSON解析失败，尝试用正则表达式提取话题信息
-                topics = self._extract_topics_with_regex(result_text, max_topics)
-                if topics:
-                    logger.info(f"正则表达式提取成功，获得 {len(topics)} 个话题，话题分析 warning 可忽略")
-                    return topics, token_usage
-                else:
-                    # 最后的降级方案
-                    logger.info("正则表达式提取失败，使用默认话题...")
-                    return [SummaryTopic(
-                        topic="群聊讨论",
-                        contributors=["群友"],
-                        detail="今日群聊内容丰富，涵盖多个话题"
-                    )], token_usage
-
-            return [], token_usage
-
+            logger.info("开始话题分析")
+            return await self.topic_analyzer.analyze_topics(messages, umo)
         except Exception as e:
             logger.error(f"话题分析失败: {e}")
             return [], TokenUsage()
-
-    def _fix_json(self, text: str) -> str:
-        """修复JSON格式问题"""
-        # 移除markdown代码块标记
-        text = re.sub(r'```json\s*', '', text)
-        text = re.sub(r'```\s*$', '', text)
-
-        # 基础清理
-        text = text.replace('\n', ' ').replace('\r', ' ')
-        text = re.sub(r'\s+', ' ', text)
-
-        # 替换中文引号为英文引号
-        text = text.replace('"', '"').replace('"', '"')
-        text = text.replace(''', "'").replace(''', "'")
-
-        # 处理字符串内容中的特殊字符
-        # 转义字符串内的双引号
-        def escape_quotes_in_strings(match):
-            content = match.group(1)
-            # 转义内部的双引号
-            content = content.replace('"', '\\"')
-            return f'"{content}"'
-
-        # 先处理字段值中的引号
-        text = re.sub(r'"([^"]*(?:"[^"]*)*)"', escape_quotes_in_strings, text)
-
-        # 修复截断的JSON
-        if not text.endswith(']'):
-            last_complete = text.rfind('}')
-            if last_complete > 0:
-                text = text[:last_complete + 1] + ']'
-
-        # 修复常见的JSON格式问题
-        # 1. 修复缺失的逗号
-        text = re.sub(r'}\s*{', '}, {', text)
-
-        # 2. 确保字段名有引号
-        text = re.sub(r'([{,]\s*)([a-zA-Z_][a-zA-Z0-9_]*)\s*:', r'\1"\2":', text)
-
-        # 3. 移除多余的逗号
-        text = re.sub(r',\s*}', '}', text)
-        text = re.sub(r',\s*]', ']', text)
-
-        return text
-
-    def _extract_topics_with_regex(self, result_text: str, max_topics: int) -> List[SummaryTopic]:
-        """使用正则表达式提取话题信息"""
-        try:
-            topics = []
-
-            # 更强的正则表达式提取话题信息，处理转义字符
-            # 匹配每个完整的话题对象
-            topic_pattern = r'\{\s*"topic":\s*"([^"]+)"\s*,\s*"contributors":\s*\[([^\]]+)\]\s*,\s*"detail":\s*"([^"]*(?:\\.[^"]*)*)"\s*\}'
-            matches = re.findall(topic_pattern, result_text, re.DOTALL)
-
-            if not matches:
-                # 尝试更宽松的匹配
-                topic_pattern = r'"topic":\s*"([^"]+)"[^}]*"contributors":\s*\[([^\]]+)\][^}]*"detail":\s*"([^"]*(?:\\.[^"]*)*)"'
-                matches = re.findall(topic_pattern, result_text, re.DOTALL)
-
-            for match in matches[:max_topics]:
-                topic_name = match[0].strip()
-                contributors_str = match[1].strip()
-                detail = match[2].strip()
-
-                # 清理detail中的转义字符
-                detail = detail.replace('\\"', '"').replace('\\n', ' ').replace('\\t', ' ')
-
-                # 解析参与者列表
-                contributors = []
-                for contrib in re.findall(r'"([^"]+)"', contributors_str):
-                    contributors.append(contrib.strip())
-
-                if not contributors:
-                    contributors = ["群友"]
-
-                topics.append(SummaryTopic(
-                    topic=topic_name,
-                    contributors=contributors[:5],  # 最多5个参与者
-                    detail=detail
-                ))
-
-            return topics
-        except Exception as e:
-            logger.error(f"正则表达式提取失败: {e}")
-            return []
-
+    
     async def analyze_user_titles(self, messages: List[Dict], user_analysis: Dict, umo: str = None) -> Tuple[List[UserTitle], TokenUsage]:
-        """使用LLM分析用户称号"""
+        """
+        使用LLM分析用户称号
+        保持原有接口，委托给专门的UserTitleAnalyzer处理
+        
+        Args:
+            messages: 群聊消息列表
+            user_analysis: 用户分析统计
+            umo: 模型唯一标识符
+            
+        Returns:
+            (用户称号列表, Token使用统计)
+        """
         try:
-            # 准备用户数据
-            user_summaries = []
-            for user_id, stats in user_analysis.items():
-                if stats["message_count"] < 5:  # 过滤活跃度太低的用户
-                    continue
-
-                # 分析用户特征
-                night_messages = sum(stats["hours"][h] for h in range(0, 6))
-                day_messages = stats["message_count"] - night_messages
-                avg_chars = stats["char_count"] / stats["message_count"] if stats["message_count"] > 0 else 0
-
-                user_summaries.append({
-                    "name": stats["nickname"],
-                    "qq": int(user_id),
-                    "message_count": stats["message_count"],
-                    "avg_chars": round(avg_chars, 1),
-                    "emoji_ratio": round(stats["emoji_count"] / stats["message_count"], 2),
-                    "night_ratio": round(night_messages / stats["message_count"], 2),
-                    "reply_ratio": round(stats["reply_count"] / stats["message_count"], 2)
-                })
-
-            if not user_summaries:
-                return [], TokenUsage()
-
-            # 按消息数量排序，取前N名
-            max_user_titles = self.config_manager.get_max_user_titles()
-            user_summaries.sort(key=lambda x: x["message_count"], reverse=True)
-            user_summaries = user_summaries[:max_user_titles]
-
-            # 构建LLM提示词
-            users_text = "\n".join([
-                f"- {user['name']} (QQ:{user['qq']}): "
-                f"发言{user['message_count']}条, 平均{user['avg_chars']}字, "
-                f"表情比例{user['emoji_ratio']}, 夜间发言比例{user['night_ratio']}, "
-                f"回复比例{user['reply_ratio']}"
-                for user in user_summaries
-            ])
-
-            prompt = f"""
-请为以下群友分配合适的称号和MBTI类型。每个人只能有一个称号，每个称号只能给一个人。
-
-可选称号：
-- 龙王: 发言频繁但内容轻松的人
-- 技术专家: 经常讨论技术话题的人
-- 夜猫子: 经常在深夜发言的人
-- 表情包军火库: 经常发表情的人
-- 沉默终结者: 经常开启话题的人
-- 评论家: 平均发言长度很长的人
-- 阳角: 在群里很有影响力的人
-- 互动达人: 经常回复别人的人
-- ... (你可以自行进行拓展添加)
-
-用户数据：
-{users_text}
-
-请以JSON格式返回，格式如下：
-[
-  {{
-    "name": "用户名",
-    "qq": 123456789,
-    "title": "称号",
-    "mbti": "MBTI类型",
-    "reason": "获得此称号的原因"
-  }}
-]
-"""
-
-            # 调用LLM
-            response = await self._call_provider_with_retry(None, prompt, max_tokens=1500, temperature=0.5, umo=umo)
-            if response is None:
-                logger.error("用户称号分析调用LLM失败: provider返回None（重试失败）")
-                return [], TokenUsage()
-
-            # 提取token使用统计
-            token_usage = TokenUsage()
-            # 安全地提取 usage，避免 response.raw_completion.usage 为 None 导致的 AttributeError
-            usage = None
-            if getattr(response, 'raw_completion', None) is not None:
-                usage = getattr(response.raw_completion, 'usage', None)
-                if usage:
-                    token_usage.prompt_tokens = getattr(usage, 'prompt_tokens', 0) or 0
-                    token_usage.completion_tokens = getattr(usage, 'completion_tokens', 0) or 0
-                    token_usage.total_tokens = getattr(usage, 'total_tokens', 0) or 0
-
-            # 解析响应
-            if hasattr(response, 'completion_text'):
-                result_text = response.completion_text
-            else:
-                result_text = str(response)
-
-            # debug日志：打印原始响应
-            logger.debug(f"用户称号分析原始响应: {result_text[:500]}...")
-
-            # 尝试解析JSON
-            try:
-                json_match = re.search(r'\[.*\]', result_text, re.DOTALL)
-                if json_match:
-                    logger.debug(f"用户称号分析JSON原文: {json_match.group()[:500]}...")
-                    titles_data = json.loads(json_match.group())
-                    return [UserTitle(**title) for title in titles_data], token_usage
-            except Exception as e:
-                logger.warning(f"用户称号分析JSON解析失败: {e}")
-                logger.warning(f"原始响应: {result_text}")
-
-            return [], token_usage
-
+            logger.info("开始用户称号分析")
+            return await self.user_title_analyzer.analyze_user_titles(messages, user_analysis, umo)
         except Exception as e:
             logger.error(f"用户称号分析失败: {e}")
             return [], TokenUsage()
-
+    
     async def analyze_golden_quotes(self, messages: List[Dict], umo: str = None) -> Tuple[List[GoldenQuote], TokenUsage]:
-        """使用LLM分析群聊金句"""
+        """
+        使用LLM分析群聊金句
+        保持原有接口，委托给专门的GoldenQuoteAnalyzer处理
+        
+        Args:
+            messages: 群聊消息列表
+            umo: 模型唯一标识符
+            
+        Returns:
+            (金句列表, Token使用统计)
+        """
         try:
-            # 提取有趣的文本消息
-            interesting_messages = []
-            for msg in messages:
-                sender = msg.get("sender", {})
-                nickname = sender.get("nickname", "") or sender.get("card", "")
-                msg_time = datetime.fromtimestamp(msg.get("time", 0)).strftime("%H:%M")
-
-                for content in msg.get("message", []):
-                    if content.get("type") == "text":
-                        text = content.get("data", {}).get("text", "").strip()
-                        # 过滤长度适中、可能有趣的消息
-                        if 5 <= len(text) <= 100 and not text.startswith(("http", "www", "/")):
-                            interesting_messages.append({
-                                "sender": nickname,
-                                "time": msg_time,
-                                "content": text
-                            })
-
-            if not interesting_messages:
-                return [], TokenUsage()
-
-            # 构建LLM提示词
-            messages_text = "\n".join([
-                f"[{msg['time']}] {msg['sender']}: {msg['content']}"
-                for msg in interesting_messages
-            ])
-
-            # 计算金句数量
-            max_golden_quotes = self.config_manager.get_max_golden_quotes()
-
-            prompt = f"""
-请从以下群聊记录中挑选出{max_golden_quotes}句最具冲击力、最令人惊叹的"金句"。这些金句需满足：
-- 核心标准：**逆天的神人发言**，即具备颠覆常识的脑洞、逻辑跳脱的表达或强烈反差感的原创内容
-- 典型特征：包含某些争议话题元素、夸张类比、反常规结论、一本正经的"胡说八道"或突破语境的清奇思路，并且具备一定的冲击力，让人印象深刻。
-
-对于每个金句，请提供：
-1. 原文内容（完整保留发言细节）
-2. 发言人昵称
-3. 选择理由（具体说明其"逆天"之处，如逻辑颠覆点/脑洞角度/反差感/争议话题元素）
-
-此外，我将对你进行严格约束：
-- 优先筛选 **逆天指数最高** 的内容：发情、性压抑话题 > 争议话题 > 元素级 > 颠覆认知级 > 逻辑跳脱级 > 趣味调侃级，剔除单纯玩梗或网络热词堆砌的普通发言
-- 重点标记包含极端类比、反常识论证或无厘头结论的内容，并且包含一定的争议话题元素。
-
-群聊记录：
-{messages_text}
-
-请以JSON格式返回，格式如下：
-[
-  {{
-    "content": "金句原文",
-    "sender": "发言人昵称",
-    "reason": "选择这句话的理由（需明确说明逆天特质）"
-  }}
-]
-"""
-
-            # 调用LLM
-            response = await self._call_provider_with_retry(None, prompt, max_tokens=1500, temperature=0.7, umo=umo)
-            if response is None:
-                logger.error("金句分析调用LLM失败: provider返回None（重试失败）")
-                return [], TokenUsage()
-
-            # 提取token使用统计
-            token_usage = TokenUsage()
-            # 安全地提取 usage，避免 response.raw_completion.usage 为 None 导致的 AttributeError
-            usage = None
-            if getattr(response, 'raw_completion', None) is not None:
-                usage = getattr(response.raw_completion, 'usage', None)
-                if usage:
-                    token_usage.prompt_tokens = getattr(usage, 'prompt_tokens', 0) or 0
-                    token_usage.completion_tokens = getattr(usage, 'completion_tokens', 0) or 0
-                    token_usage.total_tokens = getattr(usage, 'total_tokens', 0) or 0
-
-            # 解析响应
-            if hasattr(response, 'completion_text'):
-                result_text = response.completion_text
-            else:
-                result_text = str(response)
-
-            # debug日志：打印原始响应
-            logger.debug(f"金句分析原始响应: {result_text[:500]}...")
-
-            # 尝试解析JSON
-            try:
-                json_match = re.search(r'\[.*\]', result_text, re.DOTALL)
-                if json_match:
-                    logger.debug(f"金句分析JSON原文: {json_match.group()[:500]}...")
-                    quotes_data = json.loads(json_match.group())
-                    return [GoldenQuote(**quote) for quote in quotes_data[:max_golden_quotes]], token_usage
-            except Exception as e:
-                logger.warning(f"金句分析JSON解析失败: {e}")
-                logger.warning(f"原始响应: {result_text}")
-
-            return [], token_usage
-
+            logger.info("开始金句分析")
+            return await self.golden_quote_analyzer.analyze_golden_quotes(messages, umo)
         except Exception as e:
             logger.error(f"金句分析失败: {e}")
-            return [], TokenUsage()
+            return [], TokenUsage()
+    
+    # 向后兼容的方法，保持原有调用方式
+    async def _call_provider_with_retry(self, provider, prompt: str, max_tokens: int, 
+                                      temperature: float, umo: str = None):
+        """
+        向后兼容的LLM调用方法
+        现在委托给llm_utils模块处理
+        
+        Args:
+            provider: LLM服务商实例或None
+            prompt: 输入的提示语
+            max_tokens: 最大生成token数
+            temperature: 采样温度
+            umo: 指定使用的模型唯一标识符
+            
+        Returns:
+            LLM生成的结果
+        """
+        return await call_provider_with_retry(self.context, self.config_manager, 
+                                            prompt, max_tokens, temperature, umo)
+    
+    def _fix_json(self, text: str) -> str:
+        """
+        向后兼容的JSON修复方法
+        现在委托给json_utils模块处理
+        
+        Args:
+            text: 需要修复的JSON文本
+            
+        Returns:
+            修复后的JSON文本
+        """
+        return fix_json(text)
+    
+    def _extract_topics_with_regex(self, result_text: str, max_topics: int) -> List[SummaryTopic]:
+        """
+        向后兼容的话题正则提取方法
+        现在委托给json_utils模块处理
+        
+        Args:
+            result_text: 需要提取的文本
+            max_topics: 最大话题数量
+            
+        Returns:
+            话题对象列表
+        """
+        
+        topics_data = extract_topics_with_regex(result_text, max_topics)
+        return [SummaryTopic(**topic) for topic in topics_data]

src/analysis/utils/__init__.py

@@ -0,0 +1,32 @@
+"""
+分析工具模块
+包含JSON处理和LLM API请求处理工具
+"""
+
+from .json_utils import (
+    fix_json,
+    parse_json_response,
+    extract_topics_with_regex,
+    extract_user_titles_with_regex,
+    extract_golden_quotes_with_regex
+)
+
+from .llm_utils import (
+    call_provider_with_retry,
+    extract_token_usage,
+    extract_response_text
+)
+
+__all__ = [
+    # JSON处理工具
+    'fix_json',
+    'parse_json_response',
+    'extract_topics_with_regex',
+    'extract_user_titles_with_regex',
+    'extract_golden_quotes_with_regex',
+    
+    # LLM工具
+    'call_provider_with_retry',
+    'extract_token_usage',
+    'extract_response_text'
+]

src/analysis/utils/json_utils.py

@@ -0,0 +1,263 @@
+
+"""
+JSON处理工具模块
+提供JSON解析、修复和正则提取功能
+"""
+
+import json
+import re
+from typing import List, Dict, Tuple, Any, Optional
+from astrbot.api import logger
+
+
+def fix_json(text: str) -> str:
+    """
+    修复JSON格式问题，包括中文符号替换
+    
+    Args:
+        text: 需要修复的JSON文本
+        
+    Returns:
+        修复后的JSON文本
+    """
+    try:
+        # 1. 移除markdown代码块标记
+        text = re.sub(r'```json\s*', '', text)
+        text = re.sub(r'```\s*$', '', text)
+        
+        # 2. 基础清理
+        text = text.replace('\n', ' ').replace('\r', ' ')
+        text = re.sub(r'\s+', ' ', text)
+        
+        # 3. 替换中文符号为英文符号（修复）
+        # 中文引号 -> 英文引号
+        text = text.replace('“', '"').replace('”', '"')
+        text = text.replace('‘', "'").replace('’', "'")
+        # 中文逗号 -> 英文逗号
+        text = text.replace('，', ',')
+        # 中文冒号 -> 英文冒号
+        text = text.replace('：', ':')
+        # 中文括号 -> 英文括号
+        text = text.replace('（', '(').replace('）', ')')
+        text = text.replace('【', '[').replace('】', ']')
+        
+        # 4. 处理字符串内容中的特殊字符
+        # 转义字符串内的双引号
+        def escape_quotes_in_strings(match):
+            content = match.group(1)
+            # 转义内部的双引号
+            content = content.replace('"', '\\"')
+            return f'"{content}"'
+        
+        # 先处理字段值中的引号
+        text = re.sub(r'"([^"]*(?:"[^"]*)*)"', escape_quotes_in_strings, text)
+        
+        # 5. 修复截断的JSON
+        if not text.endswith(']'):
+            last_complete = text.rfind('}')
+            if last_complete > 0:
+                text = text[:last_complete + 1] + ']'
+        
+        # 6. 修复常见的JSON格式问题
+        # 1. 修复缺失的逗号
+        text = re.sub(r'}\s*{', '}, {', text)
+        
+        # 2. 确保字段名有引号
+        text = re.sub(r'([{,]\s*)([a-zA-Z_][a-zA-Z0-9_]*)\s*:', r'\1"\2":', text)
+        
+        # 3. 移除多余的逗号
+        text = re.sub(r',\s*}', '}', text)
+        text = re.sub(r',\s*]', ']', text)
+        
+        return text.strip()
+        
+    except Exception as e:
+        logger.error(f"JSON修复失败: {e}")
+        return text
+
+
+def parse_json_response(result_text: str, data_type: str) -> Tuple[bool, Optional[List[Dict]], Optional[str]]:
+    """
+    统一的JSON解析方法
+    
+    Args:
+        result_text: LLM返回的原始文本
+        data_type: 数据类型 ('topics' | 'user_titles' | 'golden_quotes')
+    
+    Returns:
+        (成功标志, 解析后的数据列表, 错误消息)
+    """
+    try:
+        # 1. 提取JSON部分
+        json_match = re.search(r'\[.*?\]', result_text, re.DOTALL)
+        if not json_match:
+            error_msg = f"{data_type}响应中未找到JSON格式"
+            logger.warning(error_msg)
+            return False, None, error_msg
+        
+        json_text = json_match.group()
+        logger.debug(f"{data_type}分析JSON原文: {json_text[:500]}...")
+        
+        # 2. 修复JSON
+        json_text = fix_json(json_text)
+        logger.debug(f"{data_type}修复后的JSON: {json_text[:300]}...")
+        
+        # 3. 解析JSON
+        data = json.loads(json_text)
+        logger.info(f"{data_type}分析成功，解析到 {len(data)} 条数据")
+        return True, data, None
+        
+    except json.JSONDecodeError as e:
+        error_msg = f"{data_type}JSON解析失败: {e}"
+        logger.warning(error_msg)
+        logger.debug(f"修复后的JSON: {json_text if 'json_text' in locals() else 'N/A'}")
+        return False, None, error_msg
+    except Exception as e:
+        error_msg = f"{data_type}解析异常: {e}"
+        logger.error(error_msg)
+        return False, None, error_msg
+
+
+def extract_topics_with_regex(result_text: str, max_topics: int) -> List[Dict]:
+    """
+    使用正则表达式提取话题信息
+    
+    Args:
+        result_text: 需要提取的文本
+        max_topics: 最大话题数量
+        
+    Returns:
+        话题数据列表
+    """
+    try:
+        # 更强的正则表达式提取话题信息，处理转义字符
+        # 匹配每个完整的话题对象
+        topic_pattern = r'\{\s*"topic":\s*"([^"]+)"\s*,\s*"contributors":\s*\[([^\]]+)\]\s*,\s*"detail":\s*"([^"]*(?:\\.[^"]*)*)"\s*\}'
+        matches = re.findall(topic_pattern, result_text, re.DOTALL)
+        
+        if not matches:
+            # 尝试更宽松的匹配
+            topic_pattern = r'"topic":\s*"([^"]+)"[^}]*"contributors":\s*\[([^\]]+)\][^}]*"detail":\s*"([^"]*(?:\\.[^"]*)*)"'
+            matches = re.findall(topic_pattern, result_text, re.DOTALL)
+        
+        topics = []
+        for match in matches[:max_topics]:
+            topic_name = match[0].strip()
+            contributors_str = match[1].strip()
+            detail = match[2].strip()
+            
+            # 清理detail中的转义字符
+            detail = detail.replace('\\"', '"').replace('\\n', ' ').replace('\\t', ' ')
+            
+            # 解析参与者列表
+            contributors = []
+            for contrib in re.findall(r'"([^"]+)"', contributors_str):
+                contributors.append(contrib.strip())
+            
+            if not contributors:
+                contributors = ["群友"]
+            
+            topics.append({
+                "topic": topic_name,
+                "contributors": contributors[:5],  # 最多5个参与者
+                "detail": detail
+            })
+        
+        return topics
+        
+    except Exception as e:
+        logger.error(f"话题正则表达式提取失败: {e}")
+        return []
+
+
+def extract_user_titles_with_regex(result_text: str, max_count: int) -> List[Dict]:
+    """
+    使用正则表达式提取用户称号信息
+    
+    Args:
+        result_text: 需要提取的文本
+        max_count: 最大提取数量
+        
+    Returns:
+        用户称号数据列表
+    """
+    try:
+        titles = []
+        
+        # 正则模式：匹配完整的用户称号对象
+        pattern = r'\{\s*"name":\s*"([^"]+)"\s*,\s*"qq":\s*(\d+)\s*,\s*"title":\s*"([^"]+)"\s*,\s*"mbti":\s*"([^"]+)"\s*,\s*"reason":\s*"([^"]*(?:\\.[^"]*)*)"\s*\}'
+        matches = re.findall(pattern, result_text, re.DOTALL)
+        
+        if not matches:
+            # 尝试更宽松的匹配（字段顺序可变）
+            pattern = r'"name":\s*"([^"]+)"[^}]*"qq":\s*(\d+)[^}]*"title":\s*"([^"]+)"[^}]*"mbti":\s*"([^"]+)"[^}]*"reason":\s*"([^"]*(?:\\.[^"]*)*)"'
+            matches = re.findall(pattern, result_text, re.DOTALL)
+        
+        for match in matches[:max_count]:
+            name = match[0].strip()
+            qq = int(match[1])
+            title = match[2].strip()
+            mbti = match[3].strip()
+            reason = match[4].strip()
+            
+            # 清理转义字符
+            reason = reason.replace('\\"', '"').replace('\\n', ' ').replace('\\t', ' ')
+            
+            titles.append({
+                "name": name,
+                "qq": qq,
+                "title": title,
+                "mbti": mbti,
+                "reason": reason
+            })
+        
+        return titles
+        
+    except Exception as e:
+        logger.error(f"用户称号正则表达式提取失败: {e}")
+        return []
+
+
+def extract_golden_quotes_with_regex(result_text: str, max_count: int) -> List[Dict]:
+    """
+    使用正则表达式提取金句信息
+    
+    Args:
+        result_text: 需要提取的文本
+        max_count: 最大提取数量
+        
+    Returns:
+        金句数据列表
+    """
+    try:
+        quotes = []
+        
+        # 正则模式：匹配完整的金句对象
+        pattern = r'\{\s*"content":\s*"([^"]*(?:\\.[^"]*)*)"\s*,\s*"sender":\s*"([^"]+)"\s*,\s*"reason":\s*"([^"]*(?:\\.[^"]*)*)"\s*\}'
+        matches = re.findall(pattern, result_text, re.DOTALL)
+        
+        if not matches:
+            # 尝试更宽松的匹配（字段顺序可变）
+            pattern = r'"content":\s*"([^"]*(?:\\.[^"]*)*)"[^}]*"sender":\s*"([^"]+)"[^}]*"reason":\s*"([^"]*(?:\\.[^"]*)*)"'
+            matches = re.findall(pattern, result_text, re.DOTALL)
+        
+        for match in matches[:max_count]:
+            content = match[0].strip()
+            sender = match[1].strip()
+            reason = match[2].strip()
+            
+            # 清理转义字符
+            content = content.replace('\\"', '"').replace('\\n', ' ').replace('\\t', ' ')
+            reason = reason.replace('\\"', '"').replace('\\n', ' ').replace('\\t', ' ')
+            
+            quotes.append({
+                "content": content,
+                "sender": sender,
+                "reason": reason
+            })
+        
+        return quotes
+        
+    except Exception as e:
+        logger.error(f"金句正则表达式提取失败: {e}")
+        return []

src/analysis/utils/llm_utils.py

@@ -0,0 +1,172 @@
+"""
+LLM API请求处理工具模块
+提供LLM调用和token统计功能
+"""
+
+import asyncio
+from typing import Optional, Any
+from astrbot.api import logger
+import aiohttp
+
+
+async def call_provider_with_retry(context, config_manager, prompt: str, max_tokens: int, 
+                                 temperature: float, umo: str = None) -> Optional[Any]:
+    """
+    调用LLM提供者，带超时、重试与退避。支持自定义服务商。
+    
+    Args:
+        context: AstrBot上下文对象
+        config_manager: 配置管理器
+        prompt: 输入的提示语
+        max_tokens: 最大生成token数
+        temperature: 采样温度
+        umo: 指定使用的模型唯一标识符
+        
+    Returns:
+        LLM生成的结果，失败时返回None
+    """
+    timeout = config_manager.get_llm_timeout()
+    retries = config_manager.get_llm_retries()
+    backoff = config_manager.get_llm_backoff()
+    
+    # 获取自定义服务商参数
+    custom_api_key = config_manager.get_custom_api_key()
+    custom_api_base = config_manager.get_custom_api_base_url()
+    custom_model = config_manager.get_custom_model_name()
+    
+    last_exc = None
+    for attempt in range(1, retries + 1):
+        try:
+            if custom_api_key and custom_api_base and custom_model:
+                logger.info(f"使用自定义LLM提供商: {custom_api_base} model={custom_model}")
+                async with aiohttp.ClientSession() as session:
+                    headers = {
+                        "Authorization": f"Bearer {custom_api_key}",
+                        "Content-Type": "application/json"
+                    }
+                    payload = {
+                        "model": custom_model,
+                        "messages": [{"role": "user", "content": prompt}],
+                        "max_tokens": max_tokens,
+                        "temperature": temperature
+                    }
+                    aio_timeout = aiohttp.ClientTimeout(total=timeout)
+                    async with session.post(custom_api_base, json=payload, headers=headers, timeout=aio_timeout) as resp:
+                        if resp.status != 200:
+                            error_text = await resp.text()
+                            logger.error(f"自定义LLM服务商请求失败: HTTP {resp.status}, 内容: {error_text}")
+                        try:
+                            response_json = await resp.json()
+                        except Exception as json_err:
+                            error_text = await resp.text()
+                            logger.error(f"自定义LLM服务商响应JSON解析失败: {json_err}, 内容: {error_text}")
+                            return None
+                        # 兼容 OpenAI 格式，安全访问嵌套字段
+                        content = None
+                        try:
+                            choices = response_json.get("choices")
+                            if choices and isinstance(choices, list) and len(choices) > 0:
+                                message = choices[0].get("message")
+                                if message and isinstance(message, dict):
+                                    content = message.get("content")
+                            if content is None:
+                                logger.error(f"自定义LLM响应格式异常: {response_json}")
+                                return None
+                        except Exception as key_err:
+                            logger.error(f"自定义LLM响应结构解析失败: {key_err}, 响应内容: {response_json}")
+                            return None
+                        # 构造一个兼容原有逻辑的对象
+                        class CustomResponse:
+                            completion_text = content
+                            raw_completion = response_json
+                        return CustomResponse()
+            else:
+                # 确保使用当前指定的模型
+                provider = context.get_using_provider(umo=umo)
+                provider_id = 'unknown'
+                if provider:
+                    try:
+                        meta = provider.meta()
+                        provider_id = meta.id
+                    except Exception as e:
+                        logger.debug(f"获取提供商ID失败: {e}")
+                logger.info(f"获取到的 provider ID: {provider_id}")
+                if not provider or provider_id == 'unknown':
+                    logger.warning(f"获取的提供商不正确 (Provider ID: {provider_id})")
+                
+                logger.info(f"使用LLM provider: {provider}")
+                if not provider:
+                    logger.error("provider 为空，无法调用 text_chat，直接返回 None")
+                    return None
+                coro = provider.text_chat(prompt=prompt, max_tokens=max_tokens, temperature=temperature)
+                return await asyncio.wait_for(coro, timeout=timeout)
+        except asyncio.TimeoutError as e:
+            last_exc = e
+            logger.warning(f"LLM请求超时: 第{attempt}次, timeout={timeout}s")
+        except Exception as e:
+            last_exc = e
+            logger.warning(f"LLM请求失败: 第{attempt}次, 错误: {last_exc}")
+        # 若非最后一次，等待退避后重试
+        if attempt < retries:
+            await asyncio.sleep(backoff * attempt)
+    
+    # 最终仍失败，记录错误并返回 None 由调用方处理降级，避免抛出异常
+    logger.error(f"LLM请求全部重试失败: {last_exc}")
+    return None
+
+
+def extract_token_usage(response) -> Optional[dict]:
+    """
+    从LLM响应中提取token使用统计
+    
+    Args:
+        response: LLM响应对象
+        
+    Returns:
+        Token使用统计字典，包含prompt_tokens, completion_tokens, total_tokens
+    """
+    try:
+        token_usage = {
+            "prompt_tokens": 0,
+            "completion_tokens": 0,
+            "total_tokens": 0
+        }
+        
+        # 安全地提取 usage，避免 response.raw_completion.usage 为 None 导致的 AttributeError
+        usage = None
+        if getattr(response, 'raw_completion', None) is not None:
+            usage = getattr(response.raw_completion, 'usage', None)
+            if usage:
+                token_usage["prompt_tokens"] = getattr(usage, 'prompt_tokens', 0) or 0
+                token_usage["completion_tokens"] = getattr(usage, 'completion_tokens', 0) or 0
+                token_usage["total_tokens"] = getattr(usage, 'total_tokens', 0) or 0
+        
+        return token_usage
+        
+    except Exception as e:
+        logger.error(f"提取token使用统计失败: {e}")
+        return {
+            "prompt_tokens": 0,
+            "completion_tokens": 0,
+            "total_tokens": 0
+        }
+
+
+def extract_response_text(response) -> str:
+    """
+    从LLM响应中提取文本内容
+    
+    Args:
+        response: LLM响应对象
+        
+    Returns:
+        响应文本内容
+    """
+    try:
+        if hasattr(response, 'completion_text'):
+            return response.completion_text
+        else:
+            return str(response)
+    except Exception as e:
+        logger.error(f"提取响应文本失败: {e}")
+        return ""