nezumi/astrbot_plugin_qq_group_daily_analysis
1
0
Fork 0
mirror of https://github.com/Nezumi-2711/astrbot_plugin_qq_group_daily_analysis.git synced 2026-09-22 13:38:43 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(debug): 调试模式，启用后，会在 plugin_data/debug_data 目录下保存每次分析的平台 API 原始历史消息和 Prompt 原文，用于调试和优化提示词

Browse Source
This commit is contained in:
SXP-Simon
2026-02-08 21:25:20 +08:00
parent bd096679b5
commit 185c5edcfb
7 changed files with 150 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files
_conf_schema.json
+6
View File
@@ -21,6 +21,12 @@
"default": 3,
"hint": "同时进行的群聊分析任务数量，建议根据机器性能调整，过高可能导致LLM API RPM 超出限制，卡顿或被风控"
},
"debug_mode": {
"type": "bool",
"description": "调试模式（建议关闭）",
"default": false,
"hint": "启用后，会在 plugin_data/debug_data 目录下保存每次分析的平台 API 原始历史消息和 Prompt 原文，用于调试和优化提示词。"
},
"max_messages": {
"type": "int",
"description": "单次分析的获取最大消息条数基准",
src/analysis/analyzers/base_analyzer.py
+47 -1
View File
@@ -104,13 +104,48 @@ class BaseAnalyzer(ABC):
"""
pass
async def analyze(self, data: Any, umo: str = None) -> tuple[list[Any], TokenUsage]:
def _save_debug_data(self, prompt: str, session_id: str):
"""
保存调试数据到文件
Args:
prompt: 提示词内容
session_id: 会话ID
"""
try:
from astrbot.core.utils.astrbot_path import get_astrbot_plugin_data_path
from pathlib import Path
plugin_name = "astrbot_plugin_qq_group_daily_analysis"
base_data_path = get_astrbot_plugin_data_path()
if isinstance(base_data_path, str):
base_data_path = Path(base_data_path)
data_path = base_data_path / plugin_name / "debug_data"
data_path.mkdir(parents=True, exist_ok=True)
file_name = f"{session_id}_{self.get_data_type()}.txt"
file_path = data_path / file_name
logger.info(f"正在保存调试数据到: {file_path}")
with open(file_path, "w", encoding="utf-8") as f:
f.write(prompt)
logger.info(f"已保存 {self.get_data_type()} 分析 Prompt 到 {file_path}")
except Exception as e:
logger.error(f"保存调试数据失败: {e}", exc_info=True)
async def analyze(
self, data: Any, umo: str = None, session_id: str = None
) -> tuple[list[Any], TokenUsage]:
"""
统一的分析流程
Args:
data: 输入数据
umo: 模型唯一标识符
session_id: 会话ID (用于调试模式)
Returns:
(分析结果列表, Token使用统计)
@@ -133,6 +168,17 @@ class BaseAnalyzer(ABC):
f"{self.get_data_type()}分析prompt前100字符: {prompt[:100] if prompt else 'None'}..."
)
# 保存调试数据
debug_mode = self.config_manager.get_debug_mode()
logger.info(
f"[Debug] debug_mode={debug_mode}, session_id={session_id}, prompt_len={len(prompt) if prompt else 0}"
) # Added log
if debug_mode and session_id and prompt:
self._save_debug_data(prompt, session_id)
elif debug_mode and not session_id:
logger.warning("[Debug] Debug mode enabled but no session_id provided")
# 检查 prompt 是否为空
if not prompt or not prompt.strip():
logger.warning(
src/analysis/analyzers/golden_quote_analyzer.py
+3 -2
View File
@@ -127,7 +127,7 @@ class GoldenQuoteAnalyzer(BaseAnalyzer):
return []
async def analyze_golden_quotes(
self, messages: list[dict], umo: str = None
self, messages: list[dict], umo: str = None, session_id: str = None
) -> tuple[list[GoldenQuote], TokenUsage]:
"""
分析群聊金句
@@ -135,6 +135,7 @@ class GoldenQuoteAnalyzer(BaseAnalyzer):
Args:
messages: 群聊消息列表
umo: 模型唯一标识符
session_id: 会话ID (用于调试模式)
Returns:
(金句列表, Token使用统计)
@@ -148,7 +149,7 @@ class GoldenQuoteAnalyzer(BaseAnalyzer):
return [], TokenUsage()
logger.info(f"开始从 {len(interesting_messages)} 条圣经消息中提取金句")
quotes, usage = await self.analyze(interesting_messages, umo)
quotes, usage = await self.analyze(interesting_messages, umo, session_id)
# 回填 User ID
for quote in quotes:
src/analysis/analyzers/topic_analyzer.py
+3 -2
View File
@@ -338,7 +338,7 @@ class TopicAnalyzer(BaseAnalyzer):
return text_messages
async def analyze_topics(
self, messages: list[dict], umo: str = None
self, messages: list[dict], umo: str = None, session_id: str = None
) -> tuple[list[SummaryTopic], TokenUsage]:
"""
分析群聊话题
@@ -346,6 +346,7 @@ class TopicAnalyzer(BaseAnalyzer):
Args:
messages: 群聊消息列表
umo: 模型唯一标识符
session_id: 会话ID (用于调试模式)
Returns:
(话题列表, Token使用统计)
@@ -376,7 +377,7 @@ class TopicAnalyzer(BaseAnalyzer):
logger.debug(f"第一条文本消息内容: {text_messages[0]}")
# 直接传入原始消息，让 build_prompt 方法处理
return await self.analyze(messages, umo)
return await self.analyze(messages, umo, session_id)
except Exception as e:
logger.error(f"话题分析失败: {e}", exc_info=True)
src/analysis/analyzers/user_title_analyzer.py
+3 -1
View File
@@ -237,6 +237,7 @@ class UserTitleAnalyzer(BaseAnalyzer):
user_analysis: dict,
umo: str = None,
top_users: list[dict] = None,
session_id: str = None,
) -> tuple[list[UserTitle], TokenUsage]:
"""
分析用户称号
@@ -246,6 +247,7 @@ class UserTitleAnalyzer(BaseAnalyzer):
user_analysis: 用户分析统计
umo: 模型唯一标识符
top_users: 活跃用户列表(从get_top_users获取,可选)
session_id: 会话ID (用于调试模式)
Returns:
(用户称号列表, Token使用统计)
@@ -259,7 +261,7 @@ class UserTitleAnalyzer(BaseAnalyzer):
return [], TokenUsage()
logger.info(f"开始分析 {len(user_data['user_summaries'])} 个活跃用户的称号")
return await self.analyze(user_data, umo)
return await self.analyze(user_data, umo, session_id)
except Exception as e:
logger.error(f"用户称号分析失败: {e}")
src/analysis/llm_analyzer.py
+84 -12
View File
@@ -39,7 +39,7 @@ class LLMAnalyzer:
self.golden_quote_analyzer = GoldenQuoteAnalyzer(context, config_manager)
async def analyze_topics(
self, messages: list[dict], umo: str = None
self, messages: list[dict], umo: str = None, session_id: str = None
) -> tuple[list[SummaryTopic], TokenUsage]:
"""
使用LLM分析话题
@@ -48,13 +48,25 @@ class LLMAnalyzer:
Args:
messages: 群聊消息列表
umo: 模型唯一标识符
session_id: 会话ID (用于调试模式)
Returns:
(话题列表, Token使用统计)
"""
try:
logger.info("开始话题分析")
return await self.topic_analyzer.analyze_topics(messages, umo)
if not session_id:
from datetime import datetime
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
if umo:
# Sanitize umo for filename (replace : with _)
safe_umo = umo.replace(":", "_")
session_id = f"{timestamp}_{safe_umo}"
else:
session_id = timestamp
logger.info(f"开始话题分析, session_id: {session_id}")
return await self.topic_analyzer.analyze_topics(messages, umo, session_id)
except Exception as e:
logger.error(f"话题分析失败: {e}")
return [], TokenUsage()
@@ -65,6 +77,7 @@ class LLMAnalyzer:
user_analysis: dict,
umo: str = None,
top_users: list[dict] = None,
session_id: str = None,
) -> tuple[list[UserTitle], TokenUsage]:
"""
使用LLM分析用户称号
@@ -75,21 +88,32 @@ class LLMAnalyzer:
user_analysis: 用户分析统计
umo: 模型唯一标识符
top_users: 活跃用户列表(可选)
session_id: 会话ID (用于调试模式)
Returns:
(用户称号列表, Token使用统计)
"""
try:
logger.info("开始用户称号分析")
if not session_id:
from datetime import datetime
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
if umo:
safe_umo = umo.replace(":", "_")
session_id = f"{timestamp}_{safe_umo}"
else:
session_id = timestamp
logger.info(f"开始用户称号分析, session_id: {session_id}")
return await self.user_title_analyzer.analyze_user_titles(
messages, user_analysis, umo, top_users
messages, user_analysis, umo, top_users, session_id
)
except Exception as e:
logger.error(f"用户称号分析失败: {e}")
return [], TokenUsage()
async def analyze_golden_quotes(
self, messages: list[dict], umo: str = None
self, messages: list[dict], umo: str = None, session_id: str = None
) -> tuple[list[GoldenQuote], TokenUsage]:
"""
使用LLM分析群聊金句
@@ -98,13 +122,26 @@ class LLMAnalyzer:
Args:
messages: 群聊消息列表
umo: 模型唯一标识符
session_id: 会话ID (用于调试模式)
Returns:
(金句列表, Token使用统计)
"""
try:
logger.info("开始金句分析")
return await self.golden_quote_analyzer.analyze_golden_quotes(messages, umo)
if not session_id:
from datetime import datetime
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
if umo:
safe_umo = umo.replace(":", "_")
session_id = f"{timestamp}_{safe_umo}"
else:
session_id = timestamp
logger.info(f"开始金句分析, session_id: {session_id}")
return await self.golden_quote_analyzer.analyze_golden_quotes(
messages, umo, session_id
)
except Exception as e:
logger.error(f"金句分析失败: {e}")
return [], TokenUsage()
@@ -129,15 +166,50 @@ class LLMAnalyzer:
(话题列表, 用户称号列表, 金句列表, 总Token使用统计)
"""
try:
logger.info("开始并发执行所有分析任务")
from datetime import datetime
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
if umo:
safe_umo = umo.replace(":", "_")
session_id = f"{timestamp}_{safe_umo}"
else:
session_id = timestamp
logger.info(f"开始并发执行所有分析任务，会话ID: {session_id}")
# 保存原始消息数据 (Debug Mode)
if self.config_manager.get_debug_mode():
try:
import json
from astrbot.core.utils.astrbot_path import (
get_astrbot_plugin_data_path,
)
from pathlib import Path
plugin_name = "astrbot_plugin_qq_group_daily_analysis"
base_data_path = get_astrbot_plugin_data_path()
if isinstance(base_data_path, str):
base_data_path = Path(base_data_path)
debug_dir = base_data_path / plugin_name / "debug_data"
debug_dir.mkdir(parents=True, exist_ok=True)
msg_file_path = debug_dir / f"{session_id}_messages.json"
with open(msg_file_path, "w", encoding="utf-8") as f:
json.dump(messages, f, ensure_ascii=False, indent=2)
logger.info(f"已保存原始消息数据到: {msg_file_path}")
except Exception as e:
logger.error(f"保存原始消息数据失败: {e}", exc_info=True)
# 并发执行三个分析任务
results = await asyncio.gather(
self.topic_analyzer.analyze_topics(messages, umo),
self.topic_analyzer.analyze_topics(messages, umo, session_id),
self.user_title_analyzer.analyze_user_titles(
messages, user_analysis, umo, top_users
messages, user_analysis, umo, top_users, session_id
),
self.golden_quote_analyzer.analyze_golden_quotes(
messages, umo, session_id
),
self.golden_quote_analyzer.analyze_golden_quotes(messages, umo),
return_exceptions=True,
)
src/core/config.py
+4
View File
@@ -144,6 +144,10 @@ class ConfigManager:
"""获取用户称号分析最大token数"""
return self.config.get("user_title_max_tokens", 4096)
def get_debug_mode(self) -> bool:
"""获取是否启用调试模式"""
return self.config.get("debug_mode", False)
def get_llm_provider_id(self) -> str:
"""获取主 LLM Provider ID"""
return self.config.get("llm_provider_id", "")