feat: 添加系统基本信息接口,包含操作系统、Python版本、CPU和内存使用情况

This commit is contained in:
墨梓柒
2025-05-11 17:12:39 +08:00
parent 5833b78001
commit 47b3e1af28
4 changed files with 322 additions and 35 deletions

152
src/api/basic_info_api.py Normal file
View File

@@ -0,0 +1,152 @@
import platform
import psutil
import sys
import os
def get_system_info():
"""获取操作系统信息"""
return {
"system": platform.system(),
"release": platform.release(),
"version": platform.version(),
"machine": platform.machine(),
"processor": platform.processor(),
}
def get_python_version():
"""获取 Python 版本信息"""
return sys.version
def get_cpu_usage():
"""获取系统总CPU使用率"""
return psutil.cpu_percent(interval=1)
def get_process_cpu_usage():
"""获取当前进程CPU使用率"""
process = psutil.Process(os.getpid())
return process.cpu_percent(interval=1)
def get_memory_usage():
"""获取系统内存使用情况 (单位 MB)"""
mem = psutil.virtual_memory()
bytes_to_mb = lambda x: round(x / (1024 * 1024), 2) #noqa
return {
"total_mb": bytes_to_mb(mem.total),
"available_mb": bytes_to_mb(mem.available),
"percent": mem.percent,
"used_mb": bytes_to_mb(mem.used),
"free_mb": bytes_to_mb(mem.free),
}
def get_process_memory_usage():
"""获取当前进程内存使用情况 (单位 MB)"""
process = psutil.Process(os.getpid())
mem_info = process.memory_info()
bytes_to_mb = lambda x: round(x / (1024 * 1024), 2) #noqa
return {
"rss_mb": bytes_to_mb(mem_info.rss), # Resident Set Size: 实际使用物理内存
"vms_mb": bytes_to_mb(mem_info.vms), # Virtual Memory Size: 虚拟内存大小
"percent": process.memory_percent() # 进程内存使用百分比
}
def get_disk_usage(path="/"):
"""获取指定路径磁盘使用情况 (单位 GB)"""
disk = psutil.disk_usage(path)
bytes_to_gb = lambda x: round(x / (1024 * 1024 * 1024), 2) #noqa
return {
"total_gb": bytes_to_gb(disk.total),
"used_gb": bytes_to_gb(disk.used),
"free_gb": bytes_to_gb(disk.free),
"percent": disk.percent,
}
def get_all_basic_info():
"""获取所有基本信息并封装返回"""
# 对于进程CPU使用率需要先初始化
process = psutil.Process(os.getpid())
process.cpu_percent(interval=None) # 初始化调用
process_cpu = process.cpu_percent(interval=0.1) # 短暂间隔获取
return {
"system_info": get_system_info(),
"python_version": get_python_version(),
"cpu_usage_percent": get_cpu_usage(),
"process_cpu_usage_percent": process_cpu,
"memory_usage": get_memory_usage(),
"process_memory_usage": get_process_memory_usage(),
"disk_usage_root": get_disk_usage("/"),
}
def get_all_basic_info_string() -> str:
"""获取所有基本信息并以带解释的字符串形式返回"""
info = get_all_basic_info()
sys_info = info["system_info"]
mem_usage = info["memory_usage"]
proc_mem_usage = info["process_memory_usage"]
disk_usage = info["disk_usage_root"]
# 对进程内存使用百分比进行格式化,保留两位小数
proc_mem_percent = round(proc_mem_usage['percent'], 2)
output_string = f"""[系统信息]
- 操作系统: {sys_info['system']} (例如: Windows, Linux)
- 发行版本: {sys_info['release']} (例如: 11, Ubuntu 20.04)
- 详细版本: {sys_info['version']}
- 硬件架构: {sys_info['machine']} (例如: AMD64)
- 处理器信息: {sys_info['processor']}
[Python 环境]
- Python 版本: {info['python_version']}
[CPU 状态]
- 系统总 CPU 使用率: {info['cpu_usage_percent']}%
- 当前进程 CPU 使用率: {info['process_cpu_usage_percent']}%
[系统内存使用情况]
- 总物理内存: {mem_usage['total_mb']} MB
- 可用物理内存: {mem_usage['available_mb']} MB
- 物理内存使用率: {mem_usage['percent']}%
- 已用物理内存: {mem_usage['used_mb']} MB
- 空闲物理内存: {mem_usage['free_mb']} MB
[当前进程内存使用情况]
- 实际使用物理内存 (RSS): {proc_mem_usage['rss_mb']} MB
- 占用虚拟内存 (VMS): {proc_mem_usage['vms_mb']} MB
- 进程内存使用率: {proc_mem_percent}%
[磁盘使用情况 (根目录)]
- 总空间: {disk_usage['total_gb']} GB
- 已用空间: {disk_usage['used_gb']} GB
- 可用空间: {disk_usage['free_gb']} GB
- 磁盘使用率: {disk_usage['percent']}%
"""
return output_string
if __name__ == '__main__':
print(f"System Info: {get_system_info()}")
print(f"Python Version: {get_python_version()}")
print(f"CPU Usage: {get_cpu_usage()}%")
# 第一次调用 process.cpu_percent() 会返回0.0或一个无意义的值,需要间隔一段时间再调用
# 或者在初始化Process对象后先调用一次cpu_percent(interval=None)然后再调用cpu_percent(interval=1)
current_process = psutil.Process(os.getpid())
current_process.cpu_percent(interval=None) # 初始化
print(f"Process CPU Usage: {current_process.cpu_percent(interval=1)}%") # 实际获取
memory_usage_info = get_memory_usage()
print(f"Memory Usage: Total={memory_usage_info['total_mb']}MB, Used={memory_usage_info['used_mb']}MB, Percent={memory_usage_info['percent']}%")
process_memory_info = get_process_memory_usage()
print(f"Process Memory Usage: RSS={process_memory_info['rss_mb']}MB, VMS={process_memory_info['vms_mb']}MB, Percent={process_memory_info['percent']}%")
disk_usage_info = get_disk_usage('/')
print(f"Disk Usage (Root): Total={disk_usage_info['total_gb']}GB, Used={disk_usage_info['used_gb']}GB, Percent={disk_usage_info['percent']}%")
print("\n--- All Basic Info (JSON) ---")
all_info = get_all_basic_info()
import json
print(json.dumps(all_info, indent=4, ensure_ascii=False))
print("\n--- All Basic Info (String with Explanations) ---")
info_string = get_all_basic_info_string()
print(info_string)

View File

@@ -10,7 +10,7 @@ ROOT_PATH = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".."))
class APIBotConfig:
"""机器人配置类"""
INNER_VERSION: Version # 配置文件内部版本号
INNER_VERSION: str # 配置文件内部版本号toml为字符串
MAI_VERSION: str # 硬编码的版本信息
# bot
@@ -34,28 +34,28 @@ class APIBotConfig:
appearance: str # 外貌特征描述
# schedule
ENABLE_SCHEDULE_GEN: bool # 是否启用日程生成
ENABLE_SCHEDULE_INTERACTION: bool # 是否启用日程交互
PROMPT_SCHEDULE_GEN: str # 日程生成提示词
SCHEDULE_DOING_UPDATE_INTERVAL: int # 日程进行中更新间隔
SCHEDULE_TEMPERATURE: float # 日程生成温度
TIME_ZONE: str # 时区
enable_schedule_gen: bool # 是否启用日程
enable_schedule_interaction: bool # 日程表是否影响回复模式
prompt_schedule_gen: str # 日程生成提示词
schedule_doing_update_interval: int # 日程更新间隔(秒)
schedule_temperature: float # 日程温度
time_zone: str # 时区
# platforms
platforms: Dict[str, str] # 平台信息
# chat
allow_focus_mode: bool # 是否允许专注模式
base_normal_chat_num: int # 基础普通聊天次数
base_focused_chat_num: int # 基础专注聊天次数
observation_context_size: int # 观察上下文大小
allow_focus_mode: bool # 是否允许专注聊天状态
base_normal_chat_num: int # 最多允许多少个群进行普通聊天
base_focused_chat_num: int # 最多允许多少个群进行专注聊天
observation_context_size: int # 观察到的最长上下文大小
message_buffer: bool # 是否启用消息缓冲
ban_words: List[str] # 禁止词列表
ban_msgs_regex: List[str] # 禁止消息的正则表达式列表
# normal_chat
MODEL_R1_PROBABILITY: float # 模型推理概率
MODEL_V3_PROBABILITY: float # 模型普通概率
model_reasoning_probability: float # 推理模型概率
model_normal_probability: float # 普通模型概率
emoji_chance: float # 表情符号出现概率
thinking_timeout: int # 思考超时时间
willing_mode: str # 意愿模式
@@ -63,8 +63,8 @@ class APIBotConfig:
response_interested_rate_amplifier: float # 回复兴趣率放大器
down_frequency_rate: float # 降低频率率
emoji_response_penalty: float # 表情回复惩罚
mentioned_bot_inevitable_reply: bool # 提到机器人时是否必定回复
at_bot_inevitable_reply: bool # @机器人时是否必定回复
mentioned_bot_inevitable_reply: bool # 提及 bot 必然回复
at_bot_inevitable_reply: bool # @bot 必然回复
# focus_chat
reply_trigger_threshold: float # 回复触发阈值
@@ -78,24 +78,25 @@ class APIBotConfig:
# emoji
max_emoji_num: int # 最大表情符号数量
max_reach_deletion: bool # 达到最大数量时是否删除
EMOJI_CHECK_INTERVAL: int # 表情检查间隔
EMOJI_REGISTER_INTERVAL: Optional[int] # 表情注册间隔(兼容性保留)
EMOJI_SAVE: bool # 是否保存表情
EMOJI_CHECK: bool # 是否检查表情
EMOJI_CHECK_PROMPT: str # 表情检查提示词
check_interval: int # 检查表情包的时间间隔(分钟)
save_pic: bool # 是否保存图片
save_emoji: bool # 是否保存表情
steal_emoji: bool # 是否偷取表情
enable_check: bool # 是否启用表情包过滤
check_prompt: str # 表情包过滤要求
# memory
build_memory_interval: int # 构建记忆间隔
memory_build_distribution: List[float] # 记忆构建分布
build_memory_sample_num: int # 构建记忆样本数量
build_memory_sample_length: int # 构建记忆样本长度
build_memory_interval: int # 记忆构建间隔
build_memory_distribution: List[float] # 记忆构建分布
build_memory_sample_num: int # 采样数量
build_memory_sample_length: int # 采样长度
memory_compress_rate: float # 记忆压缩率
forget_memory_interval: int # 忘记记忆间隔
memory_forget_time: int # 记忆忘记时间
memory_forget_percentage: float # 记忆忘记百分比
consolidate_memory_interval: int # 巩固记忆间隔
consolidation_similarity_threshold: float # 巩固相似度阈值
consolidation_check_percentage: float # 巩固检查百分比
forget_memory_interval: int # 记忆遗忘间隔
memory_forget_time: int # 记忆遗忘时间(小时)
memory_forget_percentage: float # 记忆遗忘比例
consolidate_memory_interval: int # 记忆整合间隔
consolidation_similarity_threshold: float # 相似度阈值
consolidation_check_percentage: float # 检查节点比例
memory_ban_words: List[str] # 记忆禁止词列表
# mood
@@ -128,21 +129,19 @@ class APIBotConfig:
# experimental
enable_friend_chat: bool # 是否启用好友聊天
talk_allowed_private: List[int] # 允许私聊的QQ号列表
enable_pfc_chatting: bool # 是否启用PFC聊天
pfc_chatting: bool # 是否启用PFC聊天
# 模型配置
llm_reasoning: Dict[str, Any] # 推理模型配置
llm_normal: Dict[str, Any] # 普通模型配置
llm_topic_judge: Dict[str, Any] # 主题判断模型配置
llm_summary: Dict[str, Any] # 总结模型配置
llm_emotion_judge: Optional[Dict[str, Any]] # 情绪判断模型配置(兼容性保留)
embedding: Dict[str, Any] # 嵌入模型配置
vlm: Dict[str, Any] # VLM模型配置
moderation: Optional[Dict[str, Any]] # 审核模型配置(兼容性保留)
llm_heartflow: Dict[str, Any] # 心流模型配置
llm_observation: Dict[str, Any] # 观察模型配置
llm_sub_heartflow: Dict[str, Any] # 子心流模型配置
llm_heartflow: Dict[str, Any] # 心流模型配置
llm_plan: Optional[Dict[str, Any]] # 计划模型配置
embedding: Dict[str, Any] # 嵌入模型配置
llm_PFC_action_planner: Optional[Dict[str, Any]] # PFC行动计划模型配置
llm_PFC_chat: Optional[Dict[str, Any]] # PFC聊天模型配置
llm_PFC_reply_checker: Optional[Dict[str, Any]] # PFC回复检查模型配置
@@ -150,6 +149,63 @@ class APIBotConfig:
api_urls: Optional[Dict[str, str]] # API地址配置
@staticmethod
def validate_config(config: dict):
"""
校验传入的 toml 配置字典是否合法。
:param config: toml库load后的配置字典
:raises: ValueError, KeyError, TypeError
"""
# 检查主层级
required_sections = [
"inner", "bot", "groups", "personality", "identity", "schedule",
"platforms", "chat", "normal_chat", "focus_chat", "emoji", "memory",
"mood", "keywords_reaction", "chinese_typo", "response_splitter",
"remote", "experimental", "model"
]
for section in required_sections:
if section not in config:
raise KeyError(f"缺少配置段: [{section}]")
# 检查部分关键字段
if "version" not in config["inner"]:
raise KeyError("缺少 inner.version 字段")
if not isinstance(config["inner"]["version"], str):
raise TypeError("inner.version 必须为字符串")
if "qq" not in config["bot"]:
raise KeyError("缺少 bot.qq 字段")
if not isinstance(config["bot"]["qq"], int):
raise TypeError("bot.qq 必须为整数")
if "personality_core" not in config["personality"]:
raise KeyError("缺少 personality.personality_core 字段")
if not isinstance(config["personality"]["personality_core"], str):
raise TypeError("personality.personality_core 必须为字符串")
if "identity_detail" not in config["identity"]:
raise KeyError("缺少 identity.identity_detail 字段")
if not isinstance(config["identity"]["identity_detail"], list):
raise TypeError("identity.identity_detail 必须为列表")
# 可继续添加更多字段的类型和值检查
# ...
# 检查模型配置
model_keys = [
"llm_reasoning", "llm_normal", "llm_topic_judge", "llm_summary",
"vlm", "llm_heartflow", "llm_observation", "llm_sub_heartflow",
"embedding"
]
if "model" not in config:
raise KeyError("缺少 [model] 配置段")
for key in model_keys:
if key not in config["model"]:
raise KeyError(f"缺少 model.{key} 配置")
# 检查通过
return True
@strawberry.type
class APIEnvConfig:
@@ -182,6 +238,56 @@ class APIEnvConfig:
def get_env(self) -> str:
return "env"
@staticmethod
def validate_config(config: dict):
"""
校验环境变量配置字典是否合法。
:param config: 环境变量配置字典
:raises: KeyError, TypeError
"""
required_fields = [
"HOST", "PORT", "PLUGINS", "MONGODB_HOST", "MONGODB_PORT", "DATABASE_NAME",
"CHAT_ANY_WHERE_BASE_URL", "SILICONFLOW_BASE_URL", "DEEP_SEEK_BASE_URL"
]
for field in required_fields:
if field not in config:
raise KeyError(f"缺少环境变量配置字段: {field}")
if not isinstance(config["HOST"], str):
raise TypeError("HOST 必须为字符串")
if not isinstance(config["PORT"], int):
raise TypeError("PORT 必须为整数")
if not isinstance(config["PLUGINS"], list):
raise TypeError("PLUGINS 必须为列表")
if not isinstance(config["MONGODB_HOST"], str):
raise TypeError("MONGODB_HOST 必须为字符串")
if not isinstance(config["MONGODB_PORT"], int):
raise TypeError("MONGODB_PORT 必须为整数")
if not isinstance(config["DATABASE_NAME"], str):
raise TypeError("DATABASE_NAME 必须为字符串")
if not isinstance(config["CHAT_ANY_WHERE_BASE_URL"], str):
raise TypeError("CHAT_ANY_WHERE_BASE_URL 必须为字符串")
if not isinstance(config["SILICONFLOW_BASE_URL"], str):
raise TypeError("SILICONFLOW_BASE_URL 必须为字符串")
if not isinstance(config["DEEP_SEEK_BASE_URL"], str):
raise TypeError("DEEP_SEEK_BASE_URL 必须为字符串")
# 可选字段类型检查
optional_str_fields = [
"DEEP_SEEK_KEY", "CHAT_ANY_WHERE_KEY", "SILICONFLOW_KEY",
"CONSOLE_LOG_LEVEL", "FILE_LOG_LEVEL",
"DEFAULT_CONSOLE_LOG_LEVEL", "DEFAULT_FILE_LOG_LEVEL"
]
for field in optional_str_fields:
if field in config and config[field] is not None and not isinstance(config[field], str):
raise TypeError(f"{field} 必须为字符串或None")
if "SIMPLE_OUTPUT" in config and config["SIMPLE_OUTPUT"] is not None and not isinstance(config["SIMPLE_OUTPUT"], bool):
raise TypeError("SIMPLE_OUTPUT 必须为布尔值或None")
# 检查通过
return True
print("当前路径:")
print(ROOT_PATH)

View File

@@ -16,6 +16,7 @@ from src.api.apiforgui import (
get_all_states,
)
from src.heart_flow.sub_heartflow import ChatState
from src.api.basic_info_api import get_all_basic_info # 新增导入
# import uvicorn
# import os
@@ -97,6 +98,18 @@ async def get_all_states_api():
return {"status": "failed", "reason": "failed to get all states"}
@router.get("/info")
async def get_system_basic_info():
"""获取系统基本信息"""
logger.info("请求系统基本信息")
try:
info = get_all_basic_info()
return {"status": "success", "data": info}
except Exception as e:
logger.error(f"获取系统基本信息失败: {e}")
return {"status": "failed", "reason": str(e)}
def start_api_server():
"""启动API服务器"""
global_server.register_router(router, prefix="/api/v1")