From e1c26518151b36b838a70a85ee0995d051217c4a Mon Sep 17 00:00:00 2001 From: SXP-Simon Date: Thu, 9 Oct 2025 16:29:27 +0800 Subject: [PATCH] =?UTF-8?q?[refactor]=20(llm=5Fanalyzer)=20=E4=B8=BA=20llm?= =?UTF-8?q?=5Fanalyzer=20=E8=A7=A3=E8=80=A6=E5=8C=96=20[fix]=20(LLM=20?= =?UTF-8?q?=E8=BE=93=E5=87=BA=E6=8F=90=E5=8F=96=E5=92=8C=E5=A2=9E=E5=BC=BA?= =?UTF-8?q?)=20=E8=A1=A5=E5=85=85=E5=8E=9F=E6=9C=AC=E5=9C=A8=E8=AF=9D?= =?UTF-8?q?=E9=A2=98=E5=88=86=E6=9E=90=E5=AD=98=E5=9C=A8=E4=BD=86=E6=98=AF?= =?UTF-8?q?=E4=B8=8D=E5=AD=98=E5=9C=A8=E4=BA=8E=20=E7=94=A8=E6=88=B7?= =?UTF-8?q?=E5=A4=B4=E8=A1=94=E5=92=8C=E7=BE=A4=E5=9C=A3=E7=BB=8F=E7=9A=84?= =?UTF-8?q?=20json=20=E6=8F=90=E5=8F=96=E9=80=BB=E8=BE=91?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .gitignore | 5 + src/analysis/analyzers/__init__.py | 16 + src/analysis/analyzers/base_analyzer.py | 177 +++++ .../analyzers/golden_quote_analyzer.py | 195 +++++ src/analysis/analyzers/topic_analyzer.py | 207 ++++++ src/analysis/analyzers/user_title_analyzer.py | 222 ++++++ src/analysis/llm_analyzer.py | 673 ++++-------------- src/analysis/utils/__init__.py | 32 + src/analysis/utils/json_utils.py | 263 +++++++ src/analysis/utils/llm_utils.py | 172 +++++ 10 files changed, 1408 insertions(+), 554 deletions(-) create mode 100644 .gitignore create mode 100644 src/analysis/analyzers/__init__.py create mode 100644 src/analysis/analyzers/base_analyzer.py create mode 100644 src/analysis/analyzers/golden_quote_analyzer.py create mode 100644 src/analysis/analyzers/topic_analyzer.py create mode 100644 src/analysis/analyzers/user_title_analyzer.py create mode 100644 src/analysis/utils/__init__.py create mode 100644 src/analysis/utils/json_utils.py create mode 100644 src/analysis/utils/llm_utils.py diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..af57d0f --- /dev/null +++ b/.gitignore @@ -0,0 +1,5 @@ + +.kilocode/ +.kiro/ +.vscode/ +src/analysis/ARCHITECTURE.md diff --git a/src/analysis/analyzers/__init__.py b/src/analysis/analyzers/__init__.py new file mode 100644 index 0000000..e4d49df --- /dev/null +++ b/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' +] \ No newline at end of file diff --git a/src/analysis/analyzers/base_analyzer.py b/src/analysis/analyzers/base_analyzer.py new file mode 100644 index 0000000..e42551b --- /dev/null +++ b/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 + + \ No newline at end of file diff --git a/src/analysis/analyzers/golden_quote_analyzer.py b/src/analysis/analyzers/golden_quote_analyzer.py new file mode 100644 index 0000000..eb9f22e --- /dev/null +++ b/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() \ No newline at end of file diff --git a/src/analysis/analyzers/topic_analyzer.py b/src/analysis/analyzers/topic_analyzer.py new file mode 100644 index 0000000..a867491 --- /dev/null +++ b/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() \ No newline at end of file diff --git a/src/analysis/analyzers/user_title_analyzer.py b/src/analysis/analyzers/user_title_analyzer.py new file mode 100644 index 0000000..bd47fdd --- /dev/null +++ b/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() \ No newline at end of file diff --git a/src/analysis/llm_analyzer.py b/src/analysis/llm_analyzer.py index a2eb2af..1f9a776 100644 --- a/src/analysis/llm_analyzer.py +++ b/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() \ No newline at end of file + 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] \ No newline at end of file diff --git a/src/analysis/utils/__init__.py b/src/analysis/utils/__init__.py new file mode 100644 index 0000000..440c513 --- /dev/null +++ b/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' +] \ No newline at end of file diff --git a/src/analysis/utils/json_utils.py b/src/analysis/utils/json_utils.py new file mode 100644 index 0000000..357357d --- /dev/null +++ b/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 [] diff --git a/src/analysis/utils/llm_utils.py b/src/analysis/utils/llm_utils.py new file mode 100644 index 0000000..4239208 --- /dev/null +++ b/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 "" \ No newline at end of file