gardel/Mofox-Core
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

🤖 自动格式化代码 [skip ci]

Browse Source
This commit is contained in:
github-actions[bot]
2025-06-22 09:19:09 +00:00
parent 396cb9822b
commit 5757fef0f5
11 changed files with 264 additions and 298 deletions
Download Patch File Download Diff File Expand all files Collapse all files

101
scripts/test_hfc_performance.py
View File

@@ -16,149 +16,126 @@ from src.chat.focus_chat.hfc_version_manager import set_hfc_version, get_hfc_ver
def test_performance_logger(): def test_performance_logger():
"""测试性能记录器功能""" """测试性能记录器功能"""
# 设置测试版本号 # 设置测试版本号
test_version = "v1.2.3_test" test_version = "v1.2.3_test"
set_hfc_version(test_version) set_hfc_version(test_version)
print(f"设置测试版本号: {test_version}") print(f"设置测试版本号: {test_version}")
print(f"当前版本号: {get_hfc_version()}") print(f"当前版本号: {get_hfc_version()}")
# 创建测试用的性能记录器 # 创建测试用的性能记录器
test_chat_id = "test_chat_123" test_chat_id = "test_chat_123"
logger = HFCPerformanceLogger(test_chat_id, test_version) logger = HFCPerformanceLogger(test_chat_id, test_version)
print(f"测试 HFC 性能记录器 - Chat ID: {test_chat_id}, Version: {logger.version}") print(f"测试 HFC 性能记录器 - Chat ID: {test_chat_id}, Version: {logger.version}")
# 模拟记录几个循环的数据 # 模拟记录几个循环的数据
test_cycles = [ test_cycles = [
{ {
"cycle_id": 1, "cycle_id": 1,
"action_type": "reply", "action_type": "reply",
"total_time": 2.5, "total_time": 2.5,
"step_times": { "step_times": {"观察": 0.1, "并行调整动作、处理": 1.2, "规划器": 0.8, "执行动作": 0.4},
"观察": 0.1,
"并行调整动作、处理": 1.2,
"规划器": 0.8,
"执行动作": 0.4
},
"reasoning": "用户询问天气，需要回复", "reasoning": "用户询问天气，需要回复",
"success": True "success": True,
}, },
{ {
"cycle_id": 2, "cycle_id": 2,
"action_type": "no_reply", "action_type": "no_reply",
"total_time": 1.8, "total_time": 1.8,
"step_times": { "step_times": {"观察": 0.08, "并行调整动作、处理": 0.9, "规划器": 0.6, "执行动作": 0.22},
"观察": 0.08,
"并行调整动作、处理": 0.9,
"规划器": 0.6,
"执行动作": 0.22
},
"reasoning": "无需回复的日常对话", "reasoning": "无需回复的日常对话",
"success": True "success": True,
}, },
{ {
"cycle_id": 3, "cycle_id": 3,
"action_type": "reply", "action_type": "reply",
"total_time": 3.2, "total_time": 3.2,
"step_times": { "step_times": {"观察": 0.12, "并行调整动作、处理": 1.5, "规划器": 1.1, "执行动作": 0.48},
"观察": 0.12,
"并行调整动作、处理": 1.5,
"规划器": 1.1,
"执行动作": 0.48
},
"reasoning": "用户提出复杂问题，需要详细回复", "reasoning": "用户提出复杂问题，需要详细回复",
"success": True "success": True,
}, },
{ {
"cycle_id": 4, "cycle_id": 4,
"action_type": "no_reply", "action_type": "no_reply",
"total_time": 1.5, "total_time": 1.5,
"step_times": { "step_times": {"观察": 0.07, "并行调整动作、处理": 0.8, "规划器": 0.5, "执行动作": 0.13},
"观察": 0.07,
"并行调整动作、处理": 0.8,
"规划器": 0.5,
"执行动作": 0.13
},
"reasoning": "群聊中的无关对话", "reasoning": "群聊中的无关对话",
"success": True "success": True,
}, },
{ {
"cycle_id": 5, "cycle_id": 5,
"action_type": "error", "action_type": "error",
"total_time": 0.5, "total_time": 0.5,
"step_times": { "step_times": {"观察": 0.05, "并行调整动作、处理": 0.2, "规划器": 0.15, "执行动作": 0.1},
"观察": 0.05,
"并行调整动作、处理": 0.2,
"规划器": 0.15,
"执行动作": 0.1
},
"reasoning": "处理过程中出现错误", "reasoning": "处理过程中出现错误",
"success": False "success": False,
} },
] ]
# 记录测试数据 # 记录测试数据
for cycle_data in test_cycles: for cycle_data in test_cycles:
logger.record_cycle(cycle_data) logger.record_cycle(cycle_data)
print(f"已记录循环 {cycle_data['cycle_id']}: {cycle_data['action_type']} ({cycle_data['total_time']:.1f}s)") print(f"已记录循环 {cycle_data['cycle_id']}: {cycle_data['action_type']} ({cycle_data['total_time']:.1f}s)")
# 获取当前会话统计 # 获取当前会话统计
current_stats = logger.get_current_session_stats() current_stats = logger.get_current_session_stats()
print("\n=== 当前会话统计 ===") print("\n=== 当前会话统计 ===")
print(json.dumps(current_stats, ensure_ascii=False, indent=2)) print(json.dumps(current_stats, ensure_ascii=False, indent=2))
# 完成会话 # 完成会话
logger.finalize_session() logger.finalize_session()
print(f"\n=== 会话已完成 ===") print("\n=== 会话已完成 ===")
print(f"日志文件: {logger.session_file}") print(f"日志文件: {logger.session_file}")
print(f"统计文件: {logger.stats_file}") print(f"统计文件: {logger.stats_file}")
# 检查生成的文件 # 检查生成的文件
if logger.session_file.exists(): if logger.session_file.exists():
print(f"\n会话文件大小: {logger.session_file.stat().st_size} 字节") print(f"\n会话文件大小: {logger.session_file.stat().st_size} 字节")
if logger.stats_file.exists(): if logger.stats_file.exists():
print(f"统计文件大小: {logger.stats_file.stat().st_size} 字节") print(f"统计文件大小: {logger.stats_file.stat().st_size} 字节")
# 读取并显示统计数据 # 读取并显示统计数据
with open(logger.stats_file, 'r', encoding='utf-8') as f: with open(logger.stats_file, "r", encoding="utf-8") as f:
stats_data = json.load(f) stats_data = json.load(f)
print(f"\n=== 最终统计数据 ===") print("\n=== 最终统计数据 ===")
if test_chat_id in stats_data: if test_chat_id in stats_data:
chat_stats = stats_data[test_chat_id] chat_stats = stats_data[test_chat_id]
print(f"Chat ID: {test_chat_id}") print(f"Chat ID: {test_chat_id}")
print(f"最后更新: {chat_stats['last_updated']}") print(f"最后更新: {chat_stats['last_updated']}")
print(f"总记录数: {chat_stats['overall']['total_records']}") print(f"总记录数: {chat_stats['overall']['total_records']}")
print(f"平均总时间: {chat_stats['overall']['avg_total_time']:.2f}秒") print(f"平均总时间: {chat_stats['overall']['avg_total_time']:.2f}秒")
print(f"\n各步骤平均时间:") print("\n各步骤平均时间:")
for step, avg_time in chat_stats['overall']['avg_step_times'].items(): for step, avg_time in chat_stats["overall"]["avg_step_times"].items():
print(f" {step}: {avg_time:.3f}秒") print(f" {step}: {avg_time:.3f}秒")
print(f"\n按动作类型统计:") print("\n按动作类型统计:")
for action, action_stats in chat_stats['by_action'].items(): for action, action_stats in chat_stats["by_action"].items():
print(f" {action}: {action_stats['count']}次 ({action_stats['percentage']:.1f}%), 平均{action_stats['avg_total_time']:.2f}秒") print(
f" {action}: {action_stats['count']}次 ({action_stats['percentage']:.1f}%), 平均{action_stats['avg_total_time']:.2f}秒"
)
def test_version_manager(): def test_version_manager():
"""测试版本号管理功能""" """测试版本号管理功能"""
print("\n=== 测试版本号管理器 ===") print("\n=== 测试版本号管理器 ===")
# 测试默认版本 # 测试默认版本
print(f"默认版本: {get_hfc_version()}") print(f"默认版本: {get_hfc_version()}")
# 测试设置版本 # 测试设置版本
test_versions = ["v2.0.0", "1.5.0", "v1.0.0.beta", "v1.0.build123"] test_versions = ["v2.0.0", "1.5.0", "v1.0.0.beta", "v1.0.build123"]
for version in test_versions: for version in test_versions:
success = set_hfc_version(version) success = set_hfc_version(version)
print(f"设置版本 '{version}': {'成功' if success else '失败'} -> {get_hfc_version()}") print(f"设置版本 '{version}': {'成功' if success else '失败'} -> {get_hfc_version()}")
# 测试自动生成版本 # 测试自动生成版本
auto_version = auto_generate_hfc_version() auto_version = auto_generate_hfc_version()
print(f"自动生成版本: {auto_version}") print(f"自动生成版本: {auto_version}")
# 测试基于现有版本的自动生成 # 测试基于现有版本的自动生成
auto_version2 = auto_generate_hfc_version("v2.1.0") auto_version2 = auto_generate_hfc_version("v2.1.0")
print(f"基于v2.1.0自动生成: {auto_version2}") print(f"基于v2.1.0自动生成: {auto_version2}")
@@ -166,4 +143,4 @@ def test_version_manager():
if __name__ == "__main__": if __name__ == "__main__":
test_version_manager() test_version_manager()
test_performance_logger() test_performance_logger()

109
scripts/view_hfc_stats.py
View File

@@ -7,7 +7,6 @@ import sys
import json import json
import argparse import argparse
from pathlib import Path from pathlib import Path
from datetime import datetime
from typing import Dict, Any from typing import Dict, Any
# 添加项目根目录到Python路径 # 添加项目根目录到Python路径
@@ -27,30 +26,30 @@ def display_chat_stats(chat_id: str, stats: Dict[str, Any]):
print(f"\n=== Chat ID: {chat_id} ===") print(f"\n=== Chat ID: {chat_id} ===")
print(f"版本: {stats.get('version', 'unknown')}") print(f"版本: {stats.get('version', 'unknown')}")
print(f"最后更新: {stats['last_updated']}") print(f"最后更新: {stats['last_updated']}")
overall = stats['overall'] overall = stats["overall"]
print(f"\n📊 总体统计:") print("\n📊 总体统计:")
print(f" 总记录数: {overall['total_records']}") print(f" 总记录数: {overall['total_records']}")
print(f" 平均总时间: {format_time(overall['avg_total_time'])}") print(f" 平均总时间: {format_time(overall['avg_total_time'])}")
print(f"\n⏱️ 各步骤平均时间:") print("\n⏱️ 各步骤平均时间:")
for step, avg_time in overall['avg_step_times'].items(): for step, avg_time in overall["avg_step_times"].items():
print(f" {step}: {format_time(avg_time)}") print(f" {step}: {format_time(avg_time)}")
print(f"\n🎯 按动作类型统计:") print("\n🎯 按动作类型统计:")
by_action = stats['by_action'] by_action = stats["by_action"]
# 按比例排序 # 按比例排序
sorted_actions = sorted(by_action.items(), key=lambda x: x[1]['percentage'], reverse=True) sorted_actions = sorted(by_action.items(), key=lambda x: x[1]["percentage"], reverse=True)
for action, action_stats in sorted_actions: for action, action_stats in sorted_actions:
print(f" 📌 {action}:") print(f" 📌 {action}:")
print(f" 次数: {action_stats['count']} ({action_stats['percentage']:.1f}%)") print(f" 次数: {action_stats['count']} ({action_stats['percentage']:.1f}%)")
print(f" 平均总时间: {format_time(action_stats['avg_total_time'])}") print(f" 平均总时间: {format_time(action_stats['avg_total_time'])}")
if action_stats['avg_step_times']: if action_stats["avg_step_times"]:
print(f" 步骤时间:") print(" 步骤时间:")
for step, step_time in action_stats['avg_step_times'].items(): for step, step_time in action_stats["avg_step_times"].items():
print(f" {step}: {format_time(step_time)}") print(f" {step}: {format_time(step_time)}")
@@ -58,27 +57,29 @@ def display_comparison(stats_data: Dict[str, Dict[str, Any]]):
"""显示多个聊天的对比数据""" """显示多个聊天的对比数据"""
if len(stats_data) < 2: if len(stats_data) < 2:
return return
print(f"\n=== 多聊天对比 ===") print("\n=== 多聊天对比 ===")
# 创建对比表格 # 创建对比表格
chat_ids = list(stats_data.keys()) chat_ids = list(stats_data.keys())
print(f"\n📊 总体对比:") print("\n📊 总体对比:")
print(f"{'Chat ID':<20} {'版本':<12} {'记录数':<8} {'平均时间':<12} {'最常见动作':<15}") print(f"{'Chat ID':<20} {'版本':<12} {'记录数':<8} {'平均时间':<12} {'最常见动作':<15}")
print("-" * 70) print("-" * 70)
for chat_id in chat_ids: for chat_id in chat_ids:
stats = stats_data[chat_id] stats = stats_data[chat_id]
overall = stats['overall'] overall = stats["overall"]
# 找到最常见的动作 # 找到最常见的动作
most_common_action = max(stats['by_action'].items(), key=lambda x: x[1]['count']) most_common_action = max(stats["by_action"].items(), key=lambda x: x[1]["count"])
most_common_name = most_common_action[0] most_common_name = most_common_action[0]
most_common_pct = most_common_action[1]['percentage'] most_common_pct = most_common_action[1]["percentage"]
version = stats.get('version', 'unknown') version = stats.get("version", "unknown")
print(f"{chat_id:<20} {version:<12} {overall['total_records']:<8} {format_time(overall['avg_total_time']):<12} {most_common_name}({most_common_pct:.0f}%)") print(
f"{chat_id:<20} {version:<12} {overall['total_records']:<8} {format_time(overall['avg_total_time']):<12} {most_common_name}({most_common_pct:.0f}%)"
)
def view_session_logs(chat_id: str = None, latest: bool = False): def view_session_logs(chat_id: str = None, latest: bool = False):
@@ -87,50 +88,50 @@ def view_session_logs(chat_id: str = None, latest: bool = False):
if not log_dir.exists(): if not log_dir.exists():
print("❌ 日志目录不存在") print("❌ 日志目录不存在")
return return
if chat_id: if chat_id:
pattern = f"{chat_id}_*.json" pattern = f"{chat_id}_*.json"
else: else:
pattern = "*.json" pattern = "*.json"
log_files = list(log_dir.glob(pattern)) log_files = list(log_dir.glob(pattern))
if not log_files: if not log_files:
print(f"❌ 没有找到匹配的日志文件: {pattern}") print(f"❌ 没有找到匹配的日志文件: {pattern}")
return return
if latest: if latest:
# 按文件修改时间排序，取最新的 # 按文件修改时间排序，取最新的
log_files.sort(key=lambda f: f.stat().st_mtime, reverse=True) log_files.sort(key=lambda f: f.stat().st_mtime, reverse=True)
log_files = log_files[:1] log_files = log_files[:1]
for log_file in log_files: for log_file in log_files:
print(f"\n=== 会话日志: {log_file.name} ===") print(f"\n=== 会话日志: {log_file.name} ===")
try: try:
with open(log_file, 'r', encoding='utf-8') as f: with open(log_file, "r", encoding="utf-8") as f:
records = json.load(f) records = json.load(f)
if not records: if not records:
print(" 空文件") print(" 空文件")
continue continue
print(f" 记录数: {len(records)}") print(f" 记录数: {len(records)}")
print(f" 时间范围: {records[0]['timestamp']} ~ {records[-1]['timestamp']}") print(f" 时间范围: {records[0]['timestamp']} ~ {records[-1]['timestamp']}")
# 统计动作分布 # 统计动作分布
action_counts = {} action_counts = {}
total_time = 0 total_time = 0
for record in records: for record in records:
action = record['action_type'] action = record["action_type"]
action_counts[action] = action_counts.get(action, 0) + 1 action_counts[action] = action_counts.get(action, 0) + 1
total_time += record['total_time'] total_time += record["total_time"]
print(f" 总耗时: {format_time(total_time)}") print(f" 总耗时: {format_time(total_time)}")
print(f" 平均耗时: {format_time(total_time / len(records))}") print(f" 平均耗时: {format_time(total_time / len(records))}")
print(f" 动作分布: {dict(action_counts)}") print(f" 动作分布: {dict(action_counts)}")
except Exception as e: except Exception as e:
print(f" ❌ 读取文件失败: {e}") print(f" ❌ 读取文件失败: {e}")
@@ -141,30 +142,30 @@ def main():
parser.add_argument("--logs", action="store_true", help="查看会话日志文件") parser.add_argument("--logs", action="store_true", help="查看会话日志文件")
parser.add_argument("--latest", action="store_true", help="只显示最新的日志文件") parser.add_argument("--latest", action="store_true", help="只显示最新的日志文件")
parser.add_argument("--compare", action="store_true", help="显示多聊天对比") parser.add_argument("--compare", action="store_true", help="显示多聊天对比")
args = parser.parse_args() args = parser.parse_args()
if args.logs: if args.logs:
view_session_logs(args.chat_id, args.latest) view_session_logs(args.chat_id, args.latest)
return return
# 读取统计数据 # 读取统计数据
stats_file = Path("data/hfc/time.json") stats_file = Path("data/hfc/time.json")
if not stats_file.exists(): if not stats_file.exists():
print("❌ 统计数据文件不存在，请先运行一些HFC循环以生成数据") print("❌ 统计数据文件不存在，请先运行一些HFC循环以生成数据")
return return
try: try:
with open(stats_file, 'r', encoding='utf-8') as f: with open(stats_file, "r", encoding="utf-8") as f:
stats_data = json.load(f) stats_data = json.load(f)
except Exception as e: except Exception as e:
print(f"❌ 读取统计数据失败: {e}") print(f"❌ 读取统计数据失败: {e}")
return return
if not stats_data: if not stats_data:
print("❌ 统计数据为空") print("❌ 统计数据为空")
return return
if args.chat_id: if args.chat_id:
if args.chat_id in stats_data: if args.chat_id in stats_data:
display_chat_stats(args.chat_id, stats_data[args.chat_id]) display_chat_stats(args.chat_id, stats_data[args.chat_id])
@@ -175,10 +176,10 @@ def main():
# 显示所有聊天的统计数据 # 显示所有聊天的统计数据
for chat_id, stats in stats_data.items(): for chat_id, stats in stats_data.items():
display_chat_stats(chat_id, stats) display_chat_stats(chat_id, stats)
if args.compare: if args.compare:
display_comparison(stats_data) display_comparison(stats_data)
if __name__ == "__main__": if __name__ == "__main__":
main() main()

12
src/chat/focus_chat/heartFC_chat.py
View File

@@ -151,7 +151,7 @@ class HeartFChatting:
# 存储回调函数 # 存储回调函数
self.on_stop_focus_chat = on_stop_focus_chat self.on_stop_focus_chat = on_stop_focus_chat
# 初始化性能记录器 # 初始化性能记录器
# 如果没有指定版本号，则使用全局版本管理器的版本号 # 如果没有指定版本号，则使用全局版本管理器的版本号
actual_version = performance_version or get_hfc_version() actual_version = performance_version or get_hfc_version()
@@ -407,17 +407,17 @@ class HeartFChatting:
+ (f"\n详情: {'; '.join(timer_strings)}" if timer_strings else "") + (f"\n详情: {'; '.join(timer_strings)}" if timer_strings else "")
+ processor_time_log + processor_time_log
) )
# 记录性能数据 # 记录性能数据
try: try:
action_result = self._current_cycle_detail.loop_plan_info.get('action_result', {}) action_result = self._current_cycle_detail.loop_plan_info.get("action_result", {})
cycle_performance_data = { cycle_performance_data = {
"cycle_id": self._current_cycle_detail.cycle_id, "cycle_id": self._current_cycle_detail.cycle_id,
"action_type": action_result.get('action_type', 'unknown'), "action_type": action_result.get("action_type", "unknown"),
"total_time": self._current_cycle_detail.end_time - self._current_cycle_detail.start_time, "total_time": self._current_cycle_detail.end_time - self._current_cycle_detail.start_time,
"step_times": cycle_timers.copy(), "step_times": cycle_timers.copy(),
"reasoning": action_result.get('reasoning', ''), "reasoning": action_result.get("reasoning", ""),
"success": self._current_cycle_detail.loop_action_info.get('action_taken', False), "success": self._current_cycle_detail.loop_action_info.get("action_taken", False),
} }
self.performance_logger.record_cycle(cycle_performance_data) self.performance_logger.record_cycle(cycle_performance_data)
except Exception as perf_e: except Exception as perf_e:

136
src/chat/focus_chat/hfc_performance_logger.py
View File

@@ -1,8 +1,6 @@
import json import json
import os
import time
from datetime import datetime from datetime import datetime
from typing import Dict, List, Any, Optional from typing import Dict, List, Any
from pathlib import Path from pathlib import Path
from src.common.logger import get_logger from src.common.logger import get_logger
@@ -11,32 +9,34 @@ logger = get_logger("hfc_performance")
class HFCPerformanceLogger: class HFCPerformanceLogger:
"""HFC性能记录管理器""" """HFC性能记录管理器"""
# 版本号常量，可在启动时修改 # 版本号常量，可在启动时修改
INTERNAL_VERSION = "v1.0.0" INTERNAL_VERSION = "v1.0.0"
def __init__(self, chat_id: str, version: str = None): def __init__(self, chat_id: str, version: str = None):
self.chat_id = chat_id self.chat_id = chat_id
self.version = version or self.INTERNAL_VERSION self.version = version or self.INTERNAL_VERSION
self.log_dir = Path("log/hfc_loop") self.log_dir = Path("log/hfc_loop")
self.data_dir = Path("data/hfc") self.data_dir = Path("data/hfc")
self.session_start_time = datetime.now() self.session_start_time = datetime.now()
# 确保目录存在 # 确保目录存在
self.log_dir.mkdir(parents=True, exist_ok=True) self.log_dir.mkdir(parents=True, exist_ok=True)
self.data_dir.mkdir(parents=True, exist_ok=True) self.data_dir.mkdir(parents=True, exist_ok=True)
# 当前会话的日志文件，包含版本号 # 当前会话的日志文件，包含版本号
version_suffix = self.version.replace(".", "_") version_suffix = self.version.replace(".", "_")
self.session_file = self.log_dir / f"{chat_id}_{version_suffix}_{self.session_start_time.strftime('%Y%m%d_%H%M%S')}.json" self.session_file = (
self.log_dir / f"{chat_id}_{version_suffix}_{self.session_start_time.strftime('%Y%m%d_%H%M%S')}.json"
)
self.current_session_data = [] self.current_session_data = []
# 统计数据文件 # 统计数据文件
self.stats_file = self.data_dir / "time.json" self.stats_file = self.data_dir / "time.json"
# 初始化时计算历史统计数据 # 初始化时计算历史统计数据
self._update_historical_stats() self._update_historical_stats()
def record_cycle(self, cycle_data: Dict[str, Any]): def record_cycle(self, cycle_data: Dict[str, Any]):
"""记录单次循环数据""" """记录单次循环数据"""
try: try:
@@ -50,112 +50,110 @@ class HFCPerformanceLogger:
"total_time": cycle_data.get("total_time", 0), "total_time": cycle_data.get("total_time", 0),
"step_times": cycle_data.get("step_times", {}), "step_times": cycle_data.get("step_times", {}),
"reasoning": cycle_data.get("reasoning", ""), "reasoning": cycle_data.get("reasoning", ""),
"success": cycle_data.get("success", False) "success": cycle_data.get("success", False),
} }
# 添加到当前会话数据 # 添加到当前会话数据
self.current_session_data.append(record) self.current_session_data.append(record)
# 立即写入文件（防止数据丢失） # 立即写入文件（防止数据丢失）
self._write_session_data() self._write_session_data()
logger.debug(f"记录HFC循环数据: cycle_id={record['cycle_id']}, action={record['action_type']}, time={record['total_time']:.2f}s") logger.debug(
f"记录HFC循环数据: cycle_id={record['cycle_id']}, action={record['action_type']}, time={record['total_time']:.2f}s"
)
except Exception as e: except Exception as e:
logger.error(f"记录HFC循环数据失败: {e}") logger.error(f"记录HFC循环数据失败: {e}")
def _write_session_data(self): def _write_session_data(self):
"""写入当前会话数据到文件""" """写入当前会话数据到文件"""
try: try:
with open(self.session_file, 'w', encoding='utf-8') as f: with open(self.session_file, "w", encoding="utf-8") as f:
json.dump(self.current_session_data, f, ensure_ascii=False, indent=2) json.dump(self.current_session_data, f, ensure_ascii=False, indent=2)
except Exception as e: except Exception as e:
logger.error(f"写入会话数据失败: {e}") logger.error(f"写入会话数据失败: {e}")
def _update_historical_stats(self): def _update_historical_stats(self):
"""更新历史统计数据""" """更新历史统计数据"""
try: try:
# 读取所有历史会话文件 # 读取所有历史会话文件
all_records = [] all_records = []
# 读取当前chat_id的所有历史文件（包括不同版本） # 读取当前chat_id的所有历史文件（包括不同版本）
for file_path in self.log_dir.glob(f"{self.chat_id}_*.json"): for file_path in self.log_dir.glob(f"{self.chat_id}_*.json"):
if file_path == self.session_file: if file_path == self.session_file:
continue # 跳过当前会话文件 continue # 跳过当前会话文件
try: try:
with open(file_path, 'r', encoding='utf-8') as f: with open(file_path, "r", encoding="utf-8") as f:
records = json.load(f) records = json.load(f)
if isinstance(records, list): if isinstance(records, list):
all_records.extend(records) all_records.extend(records)
except Exception as e: except Exception as e:
logger.warning(f"读取历史文件 {file_path} 失败: {e}") logger.warning(f"读取历史文件 {file_path} 失败: {e}")
if not all_records: if not all_records:
logger.info(f"没有找到 chat_id={self.chat_id} 的历史数据") logger.info(f"没有找到 chat_id={self.chat_id} 的历史数据")
return return
# 计算统计数据 # 计算统计数据
stats = self._calculate_stats(all_records) stats = self._calculate_stats(all_records)
# 更新统计文件 # 更新统计文件
self._update_stats_file(stats) self._update_stats_file(stats)
logger.info(f"更新了 chat_id={self.chat_id} 的历史统计数据，共 {len(all_records)} 条记录") logger.info(f"更新了 chat_id={self.chat_id} 的历史统计数据，共 {len(all_records)} 条记录")
except Exception as e: except Exception as e:
logger.error(f"更新历史统计数据失败: {e}") logger.error(f"更新历史统计数据失败: {e}")
def _calculate_stats(self, records: List[Dict[str, Any]]) -> Dict[str, Any]: def _calculate_stats(self, records: List[Dict[str, Any]]) -> Dict[str, Any]:
"""计算统计数据""" """计算统计数据"""
if not records: if not records:
return {} return {}
# 按动作类型分组 # 按动作类型分组
action_groups = {} action_groups = {}
total_times = [] total_times = []
step_time_totals = {} step_time_totals = {}
for record in records: for record in records:
action_type = record.get("action_type", "unknown") action_type = record.get("action_type", "unknown")
total_time = record.get("total_time", 0) total_time = record.get("total_time", 0)
step_times = record.get("step_times", {}) step_times = record.get("step_times", {})
if action_type not in action_groups: if action_type not in action_groups:
action_groups[action_type] = { action_groups[action_type] = {"count": 0, "total_times": [], "step_times": {}}
"count": 0,
"total_times": [],
"step_times": {}
}
action_groups[action_type]["count"] += 1 action_groups[action_type]["count"] += 1
action_groups[action_type]["total_times"].append(total_time) action_groups[action_type]["total_times"].append(total_time)
total_times.append(total_time) total_times.append(total_time)
# 记录步骤时间 # 记录步骤时间
for step_name, step_time in step_times.items(): for step_name, step_time in step_times.items():
if step_name not in action_groups[action_type]["step_times"]: if step_name not in action_groups[action_type]["step_times"]:
action_groups[action_type]["step_times"][step_name] = [] action_groups[action_type]["step_times"][step_name] = []
action_groups[action_type]["step_times"][step_name].append(step_time) action_groups[action_type]["step_times"][step_name].append(step_time)
if step_name not in step_time_totals: if step_name not in step_time_totals:
step_time_totals[step_name] = [] step_time_totals[step_name] = []
step_time_totals[step_name].append(step_time) step_time_totals[step_name].append(step_time)
# 计算各种平均值和比例 # 计算各种平均值和比例
total_records = len(records) total_records = len(records)
# 整体统计 # 整体统计
overall_stats = { overall_stats = {
"total_records": total_records, "total_records": total_records,
"avg_total_time": sum(total_times) / len(total_times) if total_times else 0, "avg_total_time": sum(total_times) / len(total_times) if total_times else 0,
"avg_step_times": {} "avg_step_times": {},
} }
# 各步骤平均时间 # 各步骤平均时间
for step_name, times in step_time_totals.items(): for step_name, times in step_time_totals.items():
overall_stats["avg_step_times"][step_name] = sum(times) / len(times) if times else 0 overall_stats["avg_step_times"][step_name] = sum(times) / len(times) if times else 0
# 按动作类型统计 # 按动作类型统计
action_stats = {} action_stats = {}
for action_type, data in action_groups.items(): for action_type, data in action_groups.items():
@@ -163,76 +161,78 @@ class HFCPerformanceLogger:
"count": data["count"], "count": data["count"],
"percentage": (data["count"] / total_records) * 100, "percentage": (data["count"] / total_records) * 100,
"avg_total_time": sum(data["total_times"]) / len(data["total_times"]) if data["total_times"] else 0, "avg_total_time": sum(data["total_times"]) / len(data["total_times"]) if data["total_times"] else 0,
"avg_step_times": {} "avg_step_times": {},
} }
# 该动作各步骤平均时间 # 该动作各步骤平均时间
for step_name, times in data["step_times"].items(): for step_name, times in data["step_times"].items():
action_stats[action_type]["avg_step_times"][step_name] = sum(times) / len(times) if times else 0 action_stats[action_type]["avg_step_times"][step_name] = sum(times) / len(times) if times else 0
return { return {
"chat_id": self.chat_id, "chat_id": self.chat_id,
"version": self.version, "version": self.version,
"last_updated": datetime.now().isoformat(), "last_updated": datetime.now().isoformat(),
"overall": overall_stats, "overall": overall_stats,
"by_action": action_stats "by_action": action_stats,
} }
def _update_stats_file(self, new_stats: Dict[str, Any]): def _update_stats_file(self, new_stats: Dict[str, Any]):
"""更新统计文件""" """更新统计文件"""
try: try:
# 读取现有统计数据 # 读取现有统计数据
existing_stats = {} existing_stats = {}
if self.stats_file.exists(): if self.stats_file.exists():
with open(self.stats_file, 'r', encoding='utf-8') as f: with open(self.stats_file, "r", encoding="utf-8") as f:
existing_stats = json.load(f) existing_stats = json.load(f)
# 更新当前chat_id和版本的统计数据 # 更新当前chat_id和版本的统计数据
stats_key = f"{self.chat_id}_{self.version}" stats_key = f"{self.chat_id}_{self.version}"
existing_stats[stats_key] = new_stats existing_stats[stats_key] = new_stats
# 写回文件 # 写回文件
with open(self.stats_file, 'w', encoding='utf-8') as f: with open(self.stats_file, "w", encoding="utf-8") as f:
json.dump(existing_stats, f, ensure_ascii=False, indent=2) json.dump(existing_stats, f, ensure_ascii=False, indent=2)
except Exception as e: except Exception as e:
logger.error(f"更新统计文件失败: {e}") logger.error(f"更新统计文件失败: {e}")
def get_current_session_stats(self) -> Dict[str, Any]: def get_current_session_stats(self) -> Dict[str, Any]:
"""获取当前会话的统计数据""" """获取当前会话的统计数据"""
if not self.current_session_data: if not self.current_session_data:
return {} return {}
return self._calculate_stats(self.current_session_data) return self._calculate_stats(self.current_session_data)
def finalize_session(self): def finalize_session(self):
"""结束会话，进行最终统计""" """结束会话，进行最终统计"""
try: try:
if self.current_session_data: if self.current_session_data:
# 计算当前会话统计数据 # 计算当前会话统计数据
current_stats = self._calculate_stats(self.current_session_data) self._calculate_stats(self.current_session_data)
# 合并历史数据重新计算总体统计 # 合并历史数据重新计算总体统计
all_records = self.current_session_data[:] all_records = self.current_session_data[:]
# 读取历史数据 # 读取历史数据
for file_path in self.log_dir.glob(f"{self.chat_id}_*.json"): for file_path in self.log_dir.glob(f"{self.chat_id}_*.json"):
if file_path == self.session_file: if file_path == self.session_file:
continue continue
try: try:
with open(file_path, 'r', encoding='utf-8') as f: with open(file_path, "r", encoding="utf-8") as f:
records = json.load(f) records = json.load(f)
if isinstance(records, list): if isinstance(records, list):
all_records.extend(records) all_records.extend(records)
except Exception as e: except Exception as e:
logger.warning(f"读取历史文件 {file_path} 失败: {e}") logger.warning(f"读取历史文件 {file_path} 失败: {e}")
# 重新计算总体统计 # 重新计算总体统计
total_stats = self._calculate_stats(all_records) total_stats = self._calculate_stats(all_records)
self._update_stats_file(total_stats) self._update_stats_file(total_stats)
logger.info(f"完成会话统计，当前会话 {len(self.current_session_data)} 条记录，总共 {len(all_records)} 条记录") logger.info(
f"完成会话统计，当前会话 {len(self.current_session_data)} 条记录，总共 {len(all_records)} 条记录"
)
except Exception as e: except Exception as e:
logger.error(f"结束会话统计失败: {e}") logger.error(f"结束会话统计失败: {e}")

70
src/chat/focus_chat/hfc_version_manager.py
View File

@@ -18,21 +18,21 @@ logger = get_logger("hfc_version")
class HFCVersionManager: class HFCVersionManager:
"""HFC版本号管理器""" """HFC版本号管理器"""
# 默认版本号 # 默认版本号
DEFAULT_VERSION = "v1.0.0" DEFAULT_VERSION = "v1.0.0"
# 当前运行时版本号 # 当前运行时版本号
_current_version: Optional[str] = None _current_version: Optional[str] = None
@classmethod @classmethod
def set_version(cls, version: str) -> bool: def set_version(cls, version: str) -> bool:
""" """
设置当前运行时版本号 设置当前运行时版本号
参数: 参数:
version: 版本号字符串，格式如 v1.0.0 或 1.0.0 version: 版本号字符串，格式如 v1.0.0 或 1.0.0
返回: 返回:
bool: 设置是否成功 bool: 设置是否成功
""" """
@@ -48,103 +48,103 @@ class HFCVersionManager:
except Exception as e: except Exception as e:
logger.error(f"设置版本号失败: {e}") logger.error(f"设置版本号失败: {e}")
return False return False
@classmethod @classmethod
def get_version(cls) -> str: def get_version(cls) -> str:
""" """
获取当前版本号 获取当前版本号
返回: 返回:
str: 当前版本号 str: 当前版本号
""" """
if cls._current_version: if cls._current_version:
return cls._current_version return cls._current_version
# 尝试从环境变量获取 # 尝试从环境变量获取
env_version = os.getenv("HFC_PERFORMANCE_VERSION") env_version = os.getenv("HFC_PERFORMANCE_VERSION")
if env_version: if env_version:
if cls.set_version(env_version): if cls.set_version(env_version):
return cls._current_version return cls._current_version
# 返回默认版本号 # 返回默认版本号
return cls.DEFAULT_VERSION return cls.DEFAULT_VERSION
@classmethod @classmethod
def auto_generate_version(cls, base_version: str = None) -> str: def auto_generate_version(cls, base_version: str = None) -> str:
""" """
自动生成版本号（基于时间戳） 自动生成版本号（基于时间戳）
参数: 参数:
base_version: 基础版本号，如果不提供则使用默认版本 base_version: 基础版本号，如果不提供则使用默认版本
返回: 返回:
str: 生成的版本号 str: 生成的版本号
""" """
if not base_version: if not base_version:
base_version = cls.DEFAULT_VERSION base_version = cls.DEFAULT_VERSION
# 提取基础版本号的主要部分 # 提取基础版本号的主要部分
base_match = re.match(r'v?(\d+\.\d+)', base_version) base_match = re.match(r"v?(\d+\.\d+)", base_version)
if base_match: if base_match:
base_part = base_match.group(1) base_part = base_match.group(1)
else: else:
base_part = "1.0" base_part = "1.0"
# 添加时间戳 # 添加时间戳
timestamp = datetime.now().strftime("%Y%m%d_%H%M") timestamp = datetime.now().strftime("%Y%m%d_%H%M")
generated_version = f"v{base_part}.{timestamp}" generated_version = f"v{base_part}.{timestamp}"
cls.set_version(generated_version) cls.set_version(generated_version)
logger.info(f"自动生成版本号: {generated_version}") logger.info(f"自动生成版本号: {generated_version}")
return generated_version return generated_version
@classmethod @classmethod
def _validate_version(cls, version: str) -> Optional[str]: def _validate_version(cls, version: str) -> Optional[str]:
""" """
验证版本号格式 验证版本号格式
参数: 参数:
version: 待验证的版本号 version: 待验证的版本号
返回: 返回:
Optional[str]: 验证后的版本号，失败返回None Optional[str]: 验证后的版本号，失败返回None
""" """
if not version or not isinstance(version, str): if not version or not isinstance(version, str):
return None return None
version = version.strip() version = version.strip()
# 支持的格式： # 支持的格式：
# v1.0.0, 1.0.0, v1.0, 1.0, v1.0.0.20241222_1530 等 # v1.0.0, 1.0.0, v1.0, 1.0, v1.0.0.20241222_1530 等
patterns = [ patterns = [
r'^v?(\d+\.\d+\.\d+)$', # v1.0.0 或 1.0.0 r"^v?(\d+\.\d+\.\d+)$", # v1.0.0 或 1.0.0
r'^v?(\d+\.\d+)$', # v1.0 或 1.0 r"^v?(\d+\.\d+)$", # v1.0 或 1.0
r'^v?(\d+\.\d+\.\d+\.\w+)$', # v1.0.0.build 或 1.0.0.build r"^v?(\d+\.\d+\.\d+\.\w+)$", # v1.0.0.build 或 1.0.0.build
r'^v?(\d+\.\d+\.\w+)$', # v1.0.build 或 1.0.build r"^v?(\d+\.\d+\.\w+)$", # v1.0.build 或 1.0.build
] ]
for pattern in patterns: for pattern in patterns:
match = re.match(pattern, version) match = re.match(pattern, version)
if match: if match:
# 确保版本号以v开头 # 确保版本号以v开头
if not version.startswith('v'): if not version.startswith("v"):
version = 'v' + version version = "v" + version
return version return version
return None return None
@classmethod @classmethod
def reset_version(cls): def reset_version(cls):
"""重置版本号为默认值""" """重置版本号为默认值"""
cls._current_version = None cls._current_version = None
logger.info("HFC版本号已重置为默认值") logger.info("HFC版本号已重置为默认值")
@classmethod @classmethod
def get_version_info(cls) -> dict: def get_version_info(cls) -> dict:
""" """
获取版本信息 获取版本信息
返回: 返回:
dict: 版本相关信息 dict: 版本相关信息
""" """
@@ -154,7 +154,7 @@ class HFCVersionManager:
"default_version": cls.DEFAULT_VERSION, "default_version": cls.DEFAULT_VERSION,
"is_custom": current != cls.DEFAULT_VERSION, "is_custom": current != cls.DEFAULT_VERSION,
"env_version": os.getenv("HFC_PERFORMANCE_VERSION"), "env_version": os.getenv("HFC_PERFORMANCE_VERSION"),
"timestamp": datetime.now().isoformat() "timestamp": datetime.now().isoformat(),
} }
@@ -182,4 +182,4 @@ def reset_hfc_version():
# 在模块加载时显示当前版本信息 # 在模块加载时显示当前版本信息
if __name__ != "__main__": if __name__ != "__main__":
current_version = HFCVersionManager.get_version() current_version = HFCVersionManager.get_version()
logger.debug(f"HFC性能记录模块已加载，当前版本: {current_version}") logger.debug(f"HFC性能记录模块已加载，当前版本: {current_version}")

1
src/chat/message_receive/chat_stream.py
View File

@@ -345,6 +345,7 @@ class ChatManager:
async def load_all_streams(self): async def load_all_streams(self):
"""从数据库加载所有聊天流""" """从数据库加载所有聊天流"""
logger.info("正在从数据库加载所有聊天流") logger.info("正在从数据库加载所有聊天流")
def _db_load_all_streams_sync(): def _db_load_all_streams_sync():
loaded_streams_data = [] loaded_streams_data = []
for model_instance in ChatStreams.select(): for model_instance in ChatStreams.select():

94
src/chat/utils/statistic.py
View File

@@ -187,36 +187,30 @@ class StatisticOutputTask(AsyncTask):
logger.info("\n" + "\n".join(output)) logger.info("\n" + "\n".join(output))
async def run(self): async def run(self):
try: try:
now = datetime.now() now = datetime.now()
# 使用线程池并行执行耗时操作 # 使用线程池并行执行耗时操作
loop = asyncio.get_event_loop() loop = asyncio.get_event_loop()
# 在线程池中并行执行数据收集和之前的HTML生成（如果存在） # 在线程池中并行执行数据收集和之前的HTML生成（如果存在）
with concurrent.futures.ThreadPoolExecutor() as executor: with concurrent.futures.ThreadPoolExecutor() as executor:
logger.info("正在收集统计数据...") logger.info("正在收集统计数据...")
# 数据收集任务 # 数据收集任务
collect_task = loop.run_in_executor( collect_task = loop.run_in_executor(executor, self._collect_all_statistics, now)
executor, self._collect_all_statistics, now
)
# 等待数据收集完成 # 等待数据收集完成
stats = await collect_task stats = await collect_task
logger.info("统计数据收集完成") logger.info("统计数据收集完成")
# 并行执行控制台输出和HTML报告生成 # 并行执行控制台输出和HTML报告生成
console_task = loop.run_in_executor( console_task = loop.run_in_executor(executor, self._statistic_console_output, stats, now)
executor, self._statistic_console_output, stats, now html_task = loop.run_in_executor(executor, self._generate_html_report, stats, now)
)
html_task = loop.run_in_executor(
executor, self._generate_html_report, stats, now
)
# 等待两个输出任务完成 # 等待两个输出任务完成
await asyncio.gather(console_task, html_task) await asyncio.gather(console_task, html_task)
logger.info("统计数据输出完成") logger.info("统计数据输出完成")
except Exception as e: except Exception as e:
logger.exception(f"输出统计数据过程中发生异常，错误信息：{e}") logger.exception(f"输出统计数据过程中发生异常，错误信息：{e}")
@@ -226,41 +220,38 @@ class StatisticOutputTask(AsyncTask):
备选方案：完全异步后台运行统计输出 备选方案：完全异步后台运行统计输出
使用此方法可以让统计任务完全非阻塞 使用此方法可以让统计任务完全非阻塞
""" """
async def _async_collect_and_output(): async def _async_collect_and_output():
try: try:
import concurrent.futures import concurrent.futures
now = datetime.now() now = datetime.now()
loop = asyncio.get_event_loop() loop = asyncio.get_event_loop()
with concurrent.futures.ThreadPoolExecutor() as executor: with concurrent.futures.ThreadPoolExecutor() as executor:
logger.info("正在后台收集统计数据...") logger.info("正在后台收集统计数据...")
# 创建后台任务，不等待完成 # 创建后台任务，不等待完成
collect_task = asyncio.create_task( collect_task = asyncio.create_task(
loop.run_in_executor(executor, self._collect_all_statistics, now) loop.run_in_executor(executor, self._collect_all_statistics, now)
) )
stats = await collect_task stats = await collect_task
logger.info("统计数据收集完成") logger.info("统计数据收集完成")
# 创建并发的输出任务 # 创建并发的输出任务
output_tasks = [ output_tasks = [
asyncio.create_task( asyncio.create_task(loop.run_in_executor(executor, self._statistic_console_output, stats, now)),
loop.run_in_executor(executor, self._statistic_console_output, stats, now) asyncio.create_task(loop.run_in_executor(executor, self._generate_html_report, stats, now)),
),
asyncio.create_task(
loop.run_in_executor(executor, self._generate_html_report, stats, now)
)
] ]
# 等待所有输出任务完成 # 等待所有输出任务完成
await asyncio.gather(*output_tasks) await asyncio.gather(*output_tasks)
logger.info("统计数据后台输出完成") logger.info("统计数据后台输出完成")
except Exception as e: except Exception as e:
logger.exception(f"后台统计数据输出过程中发生异常：{e}") logger.exception(f"后台统计数据输出过程中发生异常：{e}")
# 创建后台任务，立即返回 # 创建后台任务，立即返回
asyncio.create_task(_async_collect_and_output()) asyncio.create_task(_async_collect_and_output())
@@ -1223,7 +1214,7 @@ class AsyncStatisticOutputTask(AsyncTask):
def __init__(self, record_file_path: str = "maibot_statistics.html"): def __init__(self, record_file_path: str = "maibot_statistics.html"):
# 延迟0秒启动，运行间隔300秒 # 延迟0秒启动，运行间隔300秒
super().__init__(task_name="Async Statistics Data Output Task", wait_before_start=0, run_interval=300) super().__init__(task_name="Async Statistics Data Output Task", wait_before_start=0, run_interval=300)
# 直接复用 StatisticOutputTask 的初始化逻辑 # 直接复用 StatisticOutputTask 的初始化逻辑
temp_stat_task = StatisticOutputTask(record_file_path) temp_stat_task = StatisticOutputTask(record_file_path)
self.name_mapping = temp_stat_task.name_mapping self.name_mapping = temp_stat_task.name_mapping
@@ -1232,49 +1223,46 @@ class AsyncStatisticOutputTask(AsyncTask):
async def run(self): async def run(self):
"""完全异步执行统计任务""" """完全异步执行统计任务"""
async def _async_collect_and_output(): async def _async_collect_and_output():
try: try:
now = datetime.now() now = datetime.now()
loop = asyncio.get_event_loop() loop = asyncio.get_event_loop()
with concurrent.futures.ThreadPoolExecutor() as executor: with concurrent.futures.ThreadPoolExecutor() as executor:
logger.info("正在后台收集统计数据...") logger.info("正在后台收集统计数据...")
# 数据收集任务 # 数据收集任务
collect_task = asyncio.create_task( collect_task = asyncio.create_task(
loop.run_in_executor(executor, self._collect_all_statistics, now) loop.run_in_executor(executor, self._collect_all_statistics, now)
) )
stats = await collect_task stats = await collect_task
logger.info("统计数据收集完成") logger.info("统计数据收集完成")
# 创建并发的输出任务 # 创建并发的输出任务
output_tasks = [ output_tasks = [
asyncio.create_task( asyncio.create_task(loop.run_in_executor(executor, self._statistic_console_output, stats, now)),
loop.run_in_executor(executor, self._statistic_console_output, stats, now) asyncio.create_task(loop.run_in_executor(executor, self._generate_html_report, stats, now)),
),
asyncio.create_task(
loop.run_in_executor(executor, self._generate_html_report, stats, now)
)
] ]
# 等待所有输出任务完成 # 等待所有输出任务完成
await asyncio.gather(*output_tasks) await asyncio.gather(*output_tasks)
logger.info("统计数据后台输出完成") logger.info("统计数据后台输出完成")
except Exception as e: except Exception as e:
logger.exception(f"后台统计数据输出过程中发生异常：{e}") logger.exception(f"后台统计数据输出过程中发生异常：{e}")
# 创建后台任务，立即返回 # 创建后台任务，立即返回
asyncio.create_task(_async_collect_and_output()) asyncio.create_task(_async_collect_and_output())
# 复用 StatisticOutputTask 的所有方法 # 复用 StatisticOutputTask 的所有方法
def _collect_all_statistics(self, now: datetime): def _collect_all_statistics(self, now: datetime):
return StatisticOutputTask._collect_all_statistics(self, now) return StatisticOutputTask._collect_all_statistics(self, now)
def _statistic_console_output(self, stats: Dict[str, Any], now: datetime): def _statistic_console_output(self, stats: Dict[str, Any], now: datetime):
return StatisticOutputTask._statistic_console_output(self, stats, now) return StatisticOutputTask._statistic_console_output(self, stats, now)
def _generate_html_report(self, stats: dict[str, Any], now: datetime): def _generate_html_report(self, stats: dict[str, Any], now: datetime):
return StatisticOutputTask._generate_html_report(self, stats, now) return StatisticOutputTask._generate_html_report(self, stats, now)
@@ -1282,11 +1270,11 @@ class AsyncStatisticOutputTask(AsyncTask):
@staticmethod @staticmethod
def _collect_model_request_for_period(collect_period: List[Tuple[str, datetime]]) -> Dict[str, Any]: def _collect_model_request_for_period(collect_period: List[Tuple[str, datetime]]) -> Dict[str, Any]:
return StatisticOutputTask._collect_model_request_for_period(collect_period) return StatisticOutputTask._collect_model_request_for_period(collect_period)
@staticmethod @staticmethod
def _collect_online_time_for_period(collect_period: List[Tuple[str, datetime]], now: datetime) -> Dict[str, Any]: def _collect_online_time_for_period(collect_period: List[Tuple[str, datetime]], now: datetime) -> Dict[str, Any]:
return StatisticOutputTask._collect_online_time_for_period(collect_period, now) return StatisticOutputTask._collect_online_time_for_period(collect_period, now)
def _collect_message_count_for_period(self, collect_period: List[Tuple[str, datetime]]) -> Dict[str, Any]: def _collect_message_count_for_period(self, collect_period: List[Tuple[str, datetime]]) -> Dict[str, Any]:
return StatisticOutputTask._collect_message_count_for_period(self, collect_period) return StatisticOutputTask._collect_message_count_for_period(self, collect_period)
@@ -1300,12 +1288,12 @@ class AsyncStatisticOutputTask(AsyncTask):
def _format_chat_stat(self, stats: Dict[str, Any]) -> str: def _format_chat_stat(self, stats: Dict[str, Any]) -> str:
return StatisticOutputTask._format_chat_stat(self, stats) return StatisticOutputTask._format_chat_stat(self, stats)
def _generate_chart_data(self, stat: dict[str, Any]) -> dict: def _generate_chart_data(self, stat: dict[str, Any]) -> dict:
return StatisticOutputTask._generate_chart_data(self, stat) return StatisticOutputTask._generate_chart_data(self, stat)
def _collect_interval_data(self, now: datetime, hours: int, interval_minutes: int) -> dict: def _collect_interval_data(self, now: datetime, hours: int, interval_minutes: int) -> dict:
return StatisticOutputTask._collect_interval_data(self, now, hours, interval_minutes) return StatisticOutputTask._collect_interval_data(self, now, hours, interval_minutes)
def _generate_chart_tab(self, chart_data: dict) -> str: def _generate_chart_tab(self, chart_data: dict) -> str:
return StatisticOutputTask._generate_chart_tab(self, chart_data) return StatisticOutputTask._generate_chart_tab(self, chart_data)

14
src/common/database/database.py
View File

@@ -72,11 +72,11 @@ os.makedirs(_DB_DIR, exist_ok=True)
db = SqliteDatabase( db = SqliteDatabase(
_DB_FILE, _DB_FILE,
pragmas={ pragmas={
'journal_mode': 'wal', # WAL模式提高并发性能 "journal_mode": "wal", # WAL模式提高并发性能
'cache_size': -64 * 1000, # 64MB缓存 "cache_size": -64 * 1000, # 64MB缓存
'foreign_keys': 1, "foreign_keys": 1,
'ignore_check_constraints': 0, "ignore_check_constraints": 0,
'synchronous': 0, # 异步写入提高性能 "synchronous": 0, # 异步写入提高性能
'busy_timeout': 1000, # 1秒超时而不是3秒 "busy_timeout": 1000, # 1秒超时而不是3秒
} },
) )

2
src/individuality/individuality.py
View File

@@ -48,7 +48,6 @@ class Individuality:
person_info_manager = get_person_info_manager() person_info_manager = get_person_info_manager()
self.bot_person_id = person_info_manager.get_person_id("system", "bot_id") self.bot_person_id = person_info_manager.get_person_id("system", "bot_id")
self.name = bot_nickname self.name = bot_nickname
# 检查配置变化，如果变化则清空 # 检查配置变化，如果变化则清空
await self._check_config_and_clear_if_changed( await self._check_config_and_clear_if_changed(
@@ -63,7 +62,6 @@ class Individuality:
# 初始化身份 # 初始化身份
self.identity = Identity(identity_detail=identity_detail) self.identity = Identity(identity_detail=identity_detail)
logger.info("正在将所有人设写入impression") logger.info("正在将所有人设写入impression")
# 将所有人设写入impression # 将所有人设写入impression
impression_parts = [] impression_parts = []

2
src/main.py
View File

@@ -101,7 +101,7 @@ class MainSystem:
logger.info("willing管理器初始化成功") logger.info("willing管理器初始化成功")
# 初始化聊天管理器 # 初始化聊天管理器
await get_chat_manager()._initialize() await get_chat_manager()._initialize()
asyncio.create_task(get_chat_manager()._auto_save_task()) asyncio.create_task(get_chat_manager()._auto_save_task())

21
src/person_info/person_info.py
View File

@@ -62,11 +62,11 @@ class PersonInfoManager:
try: try:
db.connect(reuse_if_open=True) db.connect(reuse_if_open=True)
# 设置连接池参数 # 设置连接池参数
if hasattr(db, 'execute_sql'): if hasattr(db, "execute_sql"):
# 设置SQLite优化参数 # 设置SQLite优化参数
db.execute_sql('PRAGMA cache_size = -64000') # 64MB缓存 db.execute_sql("PRAGMA cache_size = -64000") # 64MB缓存
db.execute_sql('PRAGMA temp_store = memory') # 临时存储在内存中 db.execute_sql("PRAGMA temp_store = memory") # 临时存储在内存中
db.execute_sql('PRAGMA mmap_size = 268435456') # 256MB内存映射 db.execute_sql("PRAGMA mmap_size = 268435456") # 256MB内存映射
db.create_tables([PersonInfo], safe=True) db.create_tables([PersonInfo], safe=True)
except Exception as e: except Exception as e:
logger.error(f"数据库连接或 PersonInfo 表创建失败: {e}") logger.error(f"数据库连接或 PersonInfo 表创建失败: {e}")
@@ -165,7 +165,6 @@ class PersonInfoManager:
async def update_one_field(self, person_id: str, field_name: str, value, data: dict = None): async def update_one_field(self, person_id: str, field_name: str, value, data: dict = None):
"""更新某一个字段，会补全""" """更新某一个字段，会补全"""
if field_name not in PersonInfo._meta.fields: if field_name not in PersonInfo._meta.fields:
logger.debug(f"更新'{field_name}'失败，未在 PersonInfo Peewee 模型中定义的字段。") logger.debug(f"更新'{field_name}'失败，未在 PersonInfo Peewee 模型中定义的字段。")
return return
@@ -178,20 +177,23 @@ class PersonInfoManager:
def _db_update_sync(p_id: str, f_name: str, val_to_set): def _db_update_sync(p_id: str, f_name: str, val_to_set):
import time import time
start_time = time.time() start_time = time.time()
try: try:
record = PersonInfo.get_or_none(PersonInfo.person_id == p_id) record = PersonInfo.get_or_none(PersonInfo.person_id == p_id)
query_time = time.time() query_time = time.time()
if record: if record:
setattr(record, f_name, val_to_set) setattr(record, f_name, val_to_set)
record.save() record.save()
save_time = time.time() save_time = time.time()
total_time = save_time - start_time total_time = save_time - start_time
if total_time > 0.5: # 如果超过500ms就记录日志 if total_time > 0.5: # 如果超过500ms就记录日志
logger.warning(f"数据库更新操作耗时 {total_time:.3f}秒 (查询: {query_time-start_time:.3f}s, 保存: {save_time-query_time:.3f}s) person_id={p_id}, field={f_name}") logger.warning(
f"数据库更新操作耗时 {total_time:.3f}秒 (查询: {query_time - start_time:.3f}s, 保存: {save_time - query_time:.3f}s) person_id={p_id}, field={f_name}"
)
return True, False # Found and updated, no creation needed return True, False # Found and updated, no creation needed
else: else:
total_time = time.time() - start_time total_time = time.time() - start_time
@@ -202,7 +204,6 @@ class PersonInfoManager:
total_time = time.time() - start_time total_time = time.time() - start_time
logger.error(f"数据库操作异常，耗时 {total_time:.3f}秒: {e}") logger.error(f"数据库操作异常，耗时 {total_time:.3f}秒: {e}")
raise raise
found, needs_creation = await asyncio.to_thread(_db_update_sync, person_id, field_name, processed_value) found, needs_creation = await asyncio.to_thread(_db_update_sync, person_id, field_name, processed_value)