基于 OpenAI Agents SDK 实现 Linux 运维智能代理系统

from openai import OpenAI
import subprocess
import json
import os
import re
import platform
import psutil
import socket
import time


# 初始化OpenAI客户端
client = OpenAI(api_key="sk-NDYyLTU2LTE3NDE0OTk1MzM2MjM=", base_url="https://api.scnet.cn/api/llm/v1")


class LinuxOpsAgent:
    def __init__(self):
        self.client = client
        self.model = "DeepSeek-R1-Distill-Qwen-7B"
        self.system_info = self.get_system_info()


    def get_system_info(self):
        """获取系统信息"""
        try:
            info = {
                "os": platform.system(),
                "os_version": platform.version(),
                "hostname": socket.gethostname(),
                "cpu_count": psutil.cpu_count(),
                "memory_total": round(psutil.virtual_memory().total  (1024**3), 2),  # GB
                "disk_usage": self.get_disk_usage(),
                "network_interfaces": self.get_network_interfaces()
            }
            return info
        except Exception as e:
            return {"error": str(e)}


    def get_disk_usage(self):
        """获取磁盘使用情况"""
        disk_info = []
        for partition in psutil.disk_partitions():
            try:
                usage = psutil.disk_usage(partition.mountpoint)
                disk_info.append({
                    "device": partition.device,
                    "mountpoint": partition.mountpoint,
                    "total_gb": round(usage.total  (1024**3), 2),
                    "used_gb": round(usage.used  (1024**3), 2),
                    "percent": usage.percent
                })
            except:
                pass
        return disk_info


    def get_network_interfaces(self):
        """获取网络接口信息"""
        interfaces = []
        for iface, addrs in psutil.net_if_addrs().items():
            for addr in addrs:
                if addr.family == socket.AF_INET:
                    interfaces.append({
                        "interface": iface,
                        "ip": addr.address,
                        "netmask": addr.netmask
                    })
        return interfaces


    def execute_command(self, command, safe_mode=True):
        """执行Shell命令"""
        # 安全模式下禁止执行危险命令
        dangerous_commands = ["rm -rf", "mkfs", "dd", ">", "format"]


        if safe_mode:
            for cmd in dangerous_commands:
                if cmd in command:
                    return {"error": f"危险命令被拒绝执行: {command}"}


        try:
            result = subprocess.run(command, shell=True, capture_output=True, text=True)
            return {
                "stdout": result.stdout,
                "stderr": result.stderr,
                "returncode": result.returncode
            }
        except Exception as e:
            return {"error": str(e)}


    def analyze_log(self, log_path):
        """分析日志文件"""
        try:
            if not os.path.exists(log_path):
                return {"error": f"日志文件不存在: {log_path}"}


            # 读取日志文件
            with open(log_path, 'r') as f:
                log_content = f.read()


            # 使用AI分析日志
            response = self.client.chat.completions.create(
                model=self.model,
                messages=[
                    {"role": "system", "content": "你是一个专业的Linux日志分析专家，请分析以下日志内容，找出异常和潜在问题。"},
                    {"role": "user", "content": f"请分析以下日志内容:\n\n{log_content[:4000]}"}  # 限制长度
                ],
                stream=False
            )


            return {
                "analysis": response.choices[0].message.content,
                "log_sample": log_content[:200] + "..." if len(log_content) > 200 else log_content
            }
        except Exception as e:
            return {"error": str(e)}


    def monitor_resources(self, duration=60, interval=5):
        """监控系统资源使用情况"""
        try:
            start_time = time.time()
            end_time = start_time + duration


            metrics = []


            while time.time() < end_time:
                cpu_percent = psutil.cpu_percent(interval=1)
                memory = psutil.virtual_memory()
                disk_io = psutil.disk_io_counters()
                net_io = psutil.net_io_counters()


                metrics.append({
                    "timestamp": time.time(),
                    "cpu_percent": cpu_percent,
                    "memory_percent": memory.percent,
                    "disk_read_mb": round(disk_io.read_bytes  (1024**2), 2),
                    "disk_write_mb": round(disk_io.write_bytes  (1024**2), 2),
                    "net_sent_mb": round(net_io.bytes_sent  (1024**2), 2),
                    "net_recv_mb": round(net_io.bytes_recv  (1024**2), 2)
                })


                time.sleep(interval)


            # 使用AI分析资源使用情况
            response = self.client.chat.completions.create(
                model=self.model,
                messages=[
                    {"role": "system", "content": "你是一个专业的系统性能分析专家，请分析以下系统资源使用数据，找出性能瓶颈和优化建议。"},
                    {"role": "user", "content": f"请分析以下系统资源使用数据:\n\n{json.dumps(metrics, indent=2)}"}
                ],
                stream=False
            )


            return {
                "metrics": metrics,
                "analysis": response.choices[0].message.content
            }
        except Exception as e:
            return {"error": str(e)}


    def process_query(self, query):
        """处理用户查询，返回运维建议和可执行命令"""
        try:
            system_info_str = json.dumps(self.system_info, indent=2)


            response = self.client.chat.completions.create(
                model=self.model,
                messages=[
                    {"role": "system", "content": "你是一个专业的Linux运维专家，请根据用户的查询和系统信息，提供专业的运维建议和可执行的命令。"},
                    {"role": "user", "content": f"系统信息:\n{system_info_str}\n\n用户查询: {query}"}
                ],
                stream=False
            )


            advice = response.choices[0].message.content


            # 提取建议中的命令
            commands = re.findall(r'```bash\n(.*?)\n```', advice, re.DOTALL)


            return {
                "query": query,
                "advice": advice,
                "suggested_commands": commands
            }
        except Exception as e:
            return {"error": str(e)}


# 示例使用
if __name__ == "__main__":
    agent = LinuxOpsAgent()


    print("Linux运维智能代理系统已启动")
    print(f"系统信息: {json.dumps(agent.system_info, indent=2)}")


    while True:
        query = input("\n请输入您的运维问题 (输入'exit'退出): ")


        if query.lower() == 'exit':
            break


        if query.startswith("执行:"):
            # 直接执行命令
            cmd = query[3:].strip()
            print(f"正在执行命令: {cmd}")
            result = agent.execute_command(cmd)
            print(json.dumps(result, indent=2, ensure_ascii=False))


        elif query.startswith("分析日志:"):
            # 分析日志文件
            log_path = query[5:].strip()
            print(f"正在分析日志: {log_path}")
            result = agent.analyze_log(log_path)
            print(json.dumps(result, indent=2, ensure_ascii=False))


        elif query == "监控资源":
            # 监控系统资源
            print("开始监控系统资源 (60秒)...")
            result = agent.monitor_resources(duration=60, interval=5)
            print(json.dumps(result, indent=2, ensure_ascii=False))


        else:
            # 处理普通查询
            print("正在处理您的查询...")
            result = agent.process_query(query)
            print("\n运维建议:")
            print(result["advice"])


            if result["suggested_commands"]:
                print("\n建议执行的命令:")
                for i, cmd in enumerate(result["suggested_commands"], 1):
                    print(f"{i}. {cmd}")