510 lines
20 KiB
Python
510 lines
20 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
OpenClaw 问题诊断工具
|
|
仅识别问题，不自动修复
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
import json
|
|
import psutil
|
|
import socket
|
|
import requests
|
|
import subprocess
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
|
|
class OpenClawDiagnoser:
|
|
def __init__(self):
|
|
self.problems = []
|
|
self.config_path = Path.home() / ".openclaw" / "openclaw.json"
|
|
self.log_dir = Path.home() / ".openclaw"
|
|
self.tmp_dir = Path("/tmp")
|
|
|
|
def log_problem(self, severity, category, message, evidence=None):
|
|
"""记录发现的问题"""
|
|
problem = {
|
|
"severity": severity, # CRITICAL, HIGH, MEDIUM, LOW, INFO
|
|
"category": category,
|
|
"message": message,
|
|
"evidence": evidence,
|
|
"timestamp": datetime.now().isoformat()
|
|
}
|
|
self.problems.append(problem)
|
|
print(f"[{severity}] {category}: {message}")
|
|
if evidence:
|
|
print(f" 证据: {evidence}")
|
|
|
|
def check_gateway_process(self):
|
|
"""检查网关进程状态"""
|
|
print("\n🔍 检查网关进程...")
|
|
gateway_processes = []
|
|
|
|
for proc in psutil.process_iter(['pid', 'name', 'cmdline', 'status']):
|
|
try:
|
|
if proc.info['cmdline']:
|
|
cmdline = ' '.join(proc.info['cmdline'])
|
|
if 'openclaw' in cmdline.lower() and 'gateway' in cmdline.lower():
|
|
gateway_processes.append({
|
|
'pid': proc.info['pid'],
|
|
'cmdline': cmdline,
|
|
'status': proc.info['status']
|
|
})
|
|
except (psutil.NoSuchProcess, psutil.AccessDenied):
|
|
continue
|
|
|
|
if not gateway_processes:
|
|
self.log_problem("CRITICAL", "进程", "未找到OpenClaw网关进程")
|
|
return False
|
|
|
|
print(f"找到 {len(gateway_processes)} 个网关进程:")
|
|
for proc in gateway_processes:
|
|
print(f" PID: {proc['pid']}, 状态: {proc['status']}")
|
|
|
|
# 检查进程状态
|
|
if proc['status'] not in ['running', 'sleeping']:
|
|
self.log_problem("HIGH", "进程", f"进程状态异常: {proc['status']}",
|
|
f"PID: {proc['pid']}")
|
|
|
|
return True
|
|
|
|
def check_port_listening(self):
|
|
"""检查端口监听状态"""
|
|
print("\n🔍 检查端口监听...")
|
|
port = 18789
|
|
|
|
try:
|
|
# 检查端口是否被占用
|
|
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
|
sock.settimeout(1)
|
|
result = sock.connect_ex(('127.0.0.1', port))
|
|
sock.close()
|
|
|
|
if result == 0:
|
|
print(f"✅ 端口 {port} 正在监听")
|
|
|
|
# 查看哪个进程在监听
|
|
try:
|
|
output = subprocess.check_output(
|
|
f"ss -tulpn | grep :{port}",
|
|
shell=True,
|
|
text=True
|
|
).strip()
|
|
if output:
|
|
print(f" 监听进程: {output}")
|
|
except:
|
|
pass
|
|
else:
|
|
self.log_problem("HIGH", "网络", f"端口 {port} 未监听",
|
|
"网关可能未启动或绑定失败")
|
|
return False
|
|
|
|
except Exception as e:
|
|
self.log_problem("MEDIUM", "网络", f"检查端口时出错: {str(e)}")
|
|
|
|
return True
|
|
|
|
def check_web_access(self):
|
|
"""检查Web服务访问"""
|
|
print("\n🔍 检查Web服务...")
|
|
urls = [
|
|
("http://localhost:18789/", "本地访问"),
|
|
("http://127.0.0.1:18789/", "本地回环"),
|
|
]
|
|
|
|
accessible = False
|
|
for url, desc in urls:
|
|
try:
|
|
response = requests.get(url, timeout=5)
|
|
if response.status_code == 200:
|
|
print(f"✅ {desc}: 成功 (HTTP {response.status_code})")
|
|
accessible = True
|
|
|
|
# 检查返回内容
|
|
if "openclaw" in response.text.lower():
|
|
print(" 页面包含 'openclaw'")
|
|
else:
|
|
self.log_problem("MEDIUM", "服务", "返回页面可能不是OpenClaw控制台")
|
|
else:
|
|
print(f"⚠️ {desc}: HTTP {response.status_code}")
|
|
self.log_problem("MEDIUM", "服务", f"{desc}返回异常状态码",
|
|
f"URL: {url}, 状态码: {response.status_code}")
|
|
except requests.ConnectionError:
|
|
print(f"❌ {desc}: 连接失败")
|
|
if desc == "本地访问":
|
|
self.log_problem("HIGH", "服务", "无法连接到本地网关", url)
|
|
except requests.Timeout:
|
|
print(f"❌ {desc}: 连接超时")
|
|
self.log_problem("MEDIUM", "服务", "连接网关超时", url)
|
|
except Exception as e:
|
|
print(f"❌ {desc}: 错误 {str(e)}")
|
|
|
|
return accessible
|
|
|
|
def check_lock_files(self):
|
|
"""检查锁文件"""
|
|
print("\n🔍 检查锁文件...")
|
|
lock_files = []
|
|
|
|
# 检查配置目录
|
|
if self.log_dir.exists():
|
|
lock_patterns = ['*.lock', '*lock*']
|
|
for pattern in lock_patterns:
|
|
for lock_file in self.log_dir.rglob(pattern):
|
|
if lock_file.is_file():
|
|
lock_files.append(str(lock_file))
|
|
|
|
# 检查/tmp目录
|
|
for pattern in ['openclaw*lock*', '*openclaw*lock*']:
|
|
for lock_file in self.tmp_dir.glob(pattern):
|
|
if lock_file.is_file():
|
|
lock_files.append(str(lock_file))
|
|
|
|
if lock_files:
|
|
print(f"⚠️ 发现 {len(lock_files)} 个锁文件:")
|
|
for lock in sorted(lock_files)[:10]: # 只显示前10个
|
|
print(f" {lock}")
|
|
|
|
if len(lock_files) > 10:
|
|
print(f" ... 还有 {len(lock_files)-10} 个未显示")
|
|
|
|
if len(lock_files) > 5:
|
|
self.log_problem("HIGH", "文件锁", f"发现 {len(lock_files)} 个锁文件",
|
|
"过多的锁文件可能导致死锁")
|
|
elif len(lock_files) > 0:
|
|
self.log_problem("MEDIUM", "文件锁", f"发现 {len(lock_files)} 个锁文件")
|
|
else:
|
|
print("✅ 未发现锁文件")
|
|
|
|
return len(lock_files)
|
|
|
|
def check_config_file(self):
|
|
"""检查配置文件"""
|
|
print("\n🔍 检查配置文件...")
|
|
|
|
if not self.config_path.exists():
|
|
self.log_problem("CRITICAL", "配置", "配置文件不存在", str(self.config_path))
|
|
return False
|
|
|
|
try:
|
|
with open(self.config_path, 'r') as f:
|
|
config = json.load(f)
|
|
|
|
print(f"✅ 配置文件存在: {self.config_path}")
|
|
print(f" 文件大小: {os.path.getsize(self.config_path)} 字节")
|
|
|
|
# 检查关键配置项
|
|
required_sections = ['models', 'agents']
|
|
for section in required_sections:
|
|
if section in config:
|
|
print(f" ✅ 包含 {section} 配置")
|
|
else:
|
|
self.log_problem("HIGH", "配置", f"缺少关键配置项: {section}")
|
|
|
|
# 检查模型配置
|
|
if 'models' in config and 'providers' in config['models']:
|
|
providers = config['models']['providers']
|
|
print(f" 配置了 {len(providers)} 个模型提供商")
|
|
|
|
for provider, details in providers.items():
|
|
if provider == 'deepseek':
|
|
if 'apiKey' in details:
|
|
key_len = len(details['apiKey'])
|
|
masked = details['apiKey'][:4] + '...' + details['apiKey'][-4:] if key_len > 8 else '***'
|
|
print(f" DeepSeek API密钥: {masked}")
|
|
else:
|
|
self.log_problem("HIGH", "配置", "DeepSeek缺少API密钥")
|
|
|
|
return True
|
|
|
|
except json.JSONDecodeError as e:
|
|
self.log_problem("CRITICAL", "配置", "配置文件JSON格式错误", str(e))
|
|
return False
|
|
except Exception as e:
|
|
self.log_problem("HIGH", "配置", "读取配置文件时出错", str(e))
|
|
return False
|
|
|
|
def check_model_connectivity(self):
|
|
"""检查模型连接性"""
|
|
print("\n🔍 检查模型连接性...")
|
|
|
|
# 从配置文件读取API密钥
|
|
api_key = None
|
|
try:
|
|
if self.config_path.exists():
|
|
with open(self.config_path, 'r') as f:
|
|
config = json.load(f)
|
|
|
|
if 'models' in config and 'providers' in config['models']:
|
|
providers = config['models']['providers']
|
|
if 'deepseek' in providers and 'apiKey' in providers['deepseek']:
|
|
api_key = providers['deepseek']['apiKey']
|
|
except:
|
|
pass
|
|
|
|
if not api_key:
|
|
print("⚠️ 未找到DeepSeek API密钥，跳过连接测试")
|
|
return
|
|
|
|
# 测试DeepSeek API
|
|
print("测试DeepSeek API连接...")
|
|
try:
|
|
headers = {
|
|
'Authorization': f'Bearer {api_key}',
|
|
'Content-Type': 'application/json'
|
|
}
|
|
data = {
|
|
'model': 'deepseek-chat',
|
|
'messages': [{'role': 'user', 'content': 'ping'}],
|
|
'max_tokens': 5
|
|
}
|
|
|
|
response = requests.post(
|
|
'https://api.deepseek.com/v1/chat/completions',
|
|
headers=headers,
|
|
json=data,
|
|
timeout=10
|
|
)
|
|
|
|
if response.status_code == 200:
|
|
print(f"✅ DeepSeek API: 连接成功 (HTTP 200)")
|
|
try:
|
|
result = response.json()
|
|
if 'choices' in result and len(result['choices']) > 0:
|
|
print(f" 模型响应正常")
|
|
else:
|
|
self.log_problem("MEDIUM", "模型", "API返回异常格式")
|
|
except:
|
|
print(f" API返回: {response.text[:100]}...")
|
|
elif response.status_code == 401:
|
|
self.log_problem("CRITICAL", "模型", "DeepSeek API密钥无效或过期", "HTTP 401")
|
|
elif response.status_code == 429:
|
|
self.log_problem("HIGH", "模型", "DeepSeek API调用频率限制", "HTTP 429")
|
|
elif response.status_code == 500:
|
|
self.log_problem("MEDIUM", "模型", "DeepSeek服务器内部错误", "HTTP 500")
|
|
else:
|
|
self.log_problem("MEDIUM", "模型", f"DeepSeek API返回异常状态码",
|
|
f"HTTP {response.status_code}")
|
|
|
|
except requests.Timeout:
|
|
self.log_problem("HIGH", "模型", "DeepSeek API连接超时", "10秒超时")
|
|
except requests.ConnectionError:
|
|
self.log_problem("HIGH", "模型", "无法连接到DeepSeek API", "网络连接失败")
|
|
except Exception as e:
|
|
self.log_problem("MEDIUM", "模型", f"测试DeepSeek API时出错: {str(e)}")
|
|
|
|
def check_system_resources(self):
|
|
"""检查系统资源"""
|
|
print("\n🔍 检查系统资源...")
|
|
|
|
# 内存使用
|
|
memory = psutil.virtual_memory()
|
|
print(f"内存: {memory.percent}% 已使用 ({memory.used//1024//1024}MB/{memory.total//1024//1024}MB)")
|
|
|
|
if memory.percent > 90:
|
|
self.log_problem("HIGH", "资源", "系统内存使用过高", f"{memory.percent}% 已使用")
|
|
elif memory.percent > 80:
|
|
self.log_problem("MEDIUM", "资源", "系统内存使用较高", f"{memory.percent}% 已使用")
|
|
|
|
# 查找OpenClaw进程内存使用
|
|
openclaw_memory = 0
|
|
for proc in psutil.process_iter(['pid', 'name', 'memory_info']):
|
|
try:
|
|
if 'openclaw' in proc.info['name'].lower():
|
|
mem = proc.info['memory_info']
|
|
if mem:
|
|
openclaw_memory += mem.rss
|
|
except:
|
|
continue
|
|
|
|
if openclaw_memory > 0:
|
|
print(f"OpenClaw进程内存: {openclaw_memory//1024//1024}MB")
|
|
if openclaw_memory > 1024*1024*1024: # 超过1GB
|
|
self.log_problem("HIGH", "资源", "OpenClaw内存使用过高",
|
|
f"{openclaw_memory//1024//1024}MB")
|
|
|
|
def check_log_files(self):
|
|
"""检查日志文件"""
|
|
print("\n🔍 检查日志文件...")
|
|
|
|
log_files = []
|
|
error_patterns = ['error', 'fail', 'timeout', 'locked', 'exception', 'crash']
|
|
|
|
# 查找日志文件
|
|
log_patterns = ['*.log', 'gateway*.log', 'openclaw*.log']
|
|
for pattern in log_patterns:
|
|
# 检查 ~/.openclaw
|
|
for log_file in self.log_dir.rglob(pattern):
|
|
if log_file.is_file():
|
|
log_files.append(log_file)
|
|
|
|
# 检查 /tmp/openclaw
|
|
tmp_log_dir = self.tmp_dir / "openclaw"
|
|
if tmp_log_dir.exists():
|
|
for log_file in tmp_log_dir.rglob(pattern):
|
|
if log_file.is_file():
|
|
log_files.append(log_file)
|
|
|
|
if log_files:
|
|
print(f"找到 {len(log_files)} 个日志文件")
|
|
recent_errors = []
|
|
|
|
for log_file in sorted(log_files, key=lambda x: x.stat().st_mtime, reverse=True)[:3]:
|
|
print(f" {log_file} (修改时间: {datetime.fromtimestamp(log_file.stat().st_mtime)})")
|
|
|
|
# 检查最近错误
|
|
try:
|
|
with open(log_file, 'r', encoding='utf-8', errors='ignore') as f:
|
|
lines = f.readlines()[-100:] # 最后100行
|
|
|
|
errors = []
|
|
for i, line in enumerate(lines):
|
|
line_lower = line.lower()
|
|
if any(pattern in line_lower for pattern in error_patterns):
|
|
# 获取上下文
|
|
context_start = max(0, i-1)
|
|
context_end = min(len(lines), i+2)
|
|
context = ''.join(lines[context_start:context_end])
|
|
errors.append(context.strip())
|
|
|
|
if errors:
|
|
recent_errors.extend(errors[-3:]) # 最多记录3个错误
|
|
|
|
except Exception as e:
|
|
print(f" 读取日志失败: {str(e)}")
|
|
|
|
if recent_errors:
|
|
print(f" 发现 {len(recent_errors)} 个最近错误")
|
|
for err in recent_errors[:3]:
|
|
print(f" - {err[:200]}...")
|
|
self.log_problem("MEDIUM", "日志", "日志中发现错误信息",
|
|
f"共 {len(recent_errors)} 个错误")
|
|
else:
|
|
print("未找到日志文件")
|
|
|
|
def run_diagnostics(self):
|
|
"""运行所有诊断"""
|
|
print("="*60)
|
|
print("OpenClaw 问题诊断工具")
|
|
print(f"运行时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
|
|
print("="*60)
|
|
|
|
# 运行所有检查
|
|
checks = [
|
|
("系统资源", self.check_system_resources),
|
|
("配置文件", self.check_config_file),
|
|
("网关进程", self.check_gateway_process),
|
|
("端口监听", self.check_port_listening),
|
|
("Web服务", self.check_web_access),
|
|
("文件锁", self.check_lock_files),
|
|
("模型连接", self.check_model_connectivity),
|
|
("日志文件", self.check_log_files),
|
|
]
|
|
|
|
for name, check_func in checks:
|
|
try:
|
|
check_func()
|
|
except Exception as e:
|
|
print(f"❌ 检查 {name} 时出错: {str(e)}")
|
|
|
|
# 生成报告
|
|
self.generate_report()
|
|
|
|
def generate_report(self):
|
|
"""生成诊断报告"""
|
|
print("\n" + "="*60)
|
|
print("诊断报告摘要")
|
|
print("="*60)
|
|
|
|
if not self.problems:
|
|
print("✅ 未发现严重问题")
|
|
return
|
|
|
|
# 按严重程度分组
|
|
severity_order = ['CRITICAL', 'HIGH', 'MEDIUM', 'LOW', 'INFO']
|
|
problems_by_severity = {severity: [] for severity in severity_order}
|
|
|
|
for problem in self.problems:
|
|
problems_by_severity[problem['severity']].append(problem)
|
|
|
|
# 输出摘要
|
|
for severity in severity_order:
|
|
problems = problems_by_severity[severity]
|
|
if problems:
|
|
print(f"\n{severity}级别问题 ({len(problems)}个):")
|
|
for i, problem in enumerate(problems, 1):
|
|
print(f" {i}. {problem['category']}: {problem['message']}")
|
|
|
|
# 保存详细报告
|
|
report_path = Path.home() / ".openclaw" / f"diagnosis_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"
|
|
try:
|
|
report = {
|
|
"timestamp": datetime.now().isoformat(),
|
|
"summary": {
|
|
"total_problems": len(self.problems),
|
|
"critical": len(problems_by_severity['CRITICAL']),
|
|
"high": len(problems_by_severity['HIGH']),
|
|
"medium": len(problems_by_severity['MEDIUM']),
|
|
"low": len(problems_by_severity['LOW']),
|
|
},
|
|
"problems": self.problems
|
|
}
|
|
|
|
with open(report_path, 'w', encoding='utf-8') as f:
|
|
json.dump(report, f, indent=2, ensure_ascii=False)
|
|
|
|
print(f"\n📄 详细报告已保存到: {report_path}")
|
|
|
|
except Exception as e:
|
|
print(f"保存报告失败: {str(e)}")
|
|
|
|
def get_recommendations(self):
|
|
"""根据问题提供建议"""
|
|
recommendations = []
|
|
|
|
for problem in self.problems:
|
|
if problem['severity'] in ['CRITICAL', 'HIGH']:
|
|
if "未找到OpenClaw网关进程" in problem['message']:
|
|
recommendations.append("建议: 启动网关 - `openclaw gateway --force &`")
|
|
elif "端口未监听" in problem['message']:
|
|
recommendations.append("建议: 检查网关是否启动，或尝试不同端口")
|
|
elif "DeepSeek API密钥无效" in problem['message']:
|
|
recommendations.append("建议: 更新DeepSeek API密钥 - `openclaw config set provider.deepseek.api_key '新密钥'`")
|
|
elif "文件锁" in problem['message'] and "过多" in problem['message']:
|
|
recommendations.append("建议: 清理锁文件 - `find ~/.openclaw -name '*.lock' -delete`")
|
|
|
|
if recommendations:
|
|
print("\n💡 建议操作:")
|
|
for i, rec in enumerate(recommendations, 1):
|
|
print(f" {i}. {rec}")
|
|
|
|
def main():
|
|
"""主函数"""
|
|
diagnoser = OpenClawDiagnoser()
|
|
|
|
try:
|
|
diagnoser.run_diagnostics()
|
|
diagnoser.get_recommendations()
|
|
except KeyboardInterrupt:
|
|
print("\n\n诊断被用户中断")
|
|
sys.exit(1)
|
|
except Exception as e:
|
|
print(f"诊断过程中发生错误: {str(e)}")
|
|
sys.exit(1)
|
|
|
|
if __name__ == "__main__":
|
|
# 检查Python版本和依赖
|
|
if sys.version_info < (3, 6):
|
|
print("需要Python 3.6或更高版本")
|
|
sys.exit(1)
|
|
|
|
try:
|
|
import psutil
|
|
import requests
|
|
except ImportError as e:
|
|
print(f"缺少依赖: {e}")
|
|
print("请安装: pip install psutil requests")
|
|
sys.exit(1)
|
|
|
|
main()
|