AI运维

内网运维助手系统技术实现

项目介绍

内网运维助手是一个基于FastAPI和SSH的内网服务器管理工具，通过Web界面实现对远程Linux服务器的监控和管理。该系统提供了类似Coze的聊天界面，用户可以通过自然语言指v令执行各种运维操作，如查看CPU、内存、磁盘使用情况等。

初级版本展示

aiops_demo

技术栈选择

后端技术

  • FastAPI：现代化的Python Web框架，提供自动API文档生成和类型提示
  • Paramiko：Python的SSH实现，用于连接和操作远程服务器
  • Uvicorn：ASGI服务器，用于运行FastAPI应用

前端技术

  • HTML5：页面结构
  • CSS3：页面样式，采用深色主题
  • JavaScript：前端交互逻辑

系统架构

整体架构

┌─────────────────┐     HTTP     ┌─────────────────┐     SSH     ┌─────────────────┐
│    前端页面     │ ───────────> │    后端API      │ ──────────> │  远程Linux服务器  │
└─────────────────┘ <─────────── └─────────────────┘ <────────── └─────────────────┘
         ↑                         ↑                         ↑
         │                         │                         │
         │                         │                         │
     聊天界面                   命令处理                  命令执行

模块划分

  1. 前端模块：负责用户界面展示和用户交互D
  2. 后端API模块：处理前端请求，执行SSH命令
  3. SSH客户端模块：负责与远程服务器建立连接并执行命令
  4. 配置模块：存储服务器连接信息

核心功能实现

1. 后端API实现

健康检查接口

@app.get("/health")
async def health():
    """
    健康检查接口：
    - 方便在 Coze 中配置 HTTP 工具前先测试服务是否可达。
    """
    return {"status": "ok", "service": "intranet-ops-tools"}

通用SSH命令执行接口

@app.post("/tools/ssh/run")
async def tools_ssh_run(req: CommandRequest):
    """
    通用 SSH 命令执行接口。
    建议在 Coze 中只对白名单命令使用此接口。
    """
    if not req.command.strip():
        raise HTTPException(status_code=400, detail="command 不能为空")

    try:
        code, out, err = run_ssh_command(req.command)
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"连接或执行失败: {e}")

    return {
        "success": code == 0,
        "exit_code": code,
        "stdout": out,
        "stderr": err,
    }

系统指标监控接口

@app.get("/tools/metrics/cpu")
async def tools_metrics_cpu():
    """
    查看 CPU 情况：
    - 可在 Coze 中配置为"cpu_check"工具。
    """
    cmd = "top -b -n 1 | head -n 5"
    try:
        code, out, err = run_ssh_command(cmd)
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"连接或执行失败: {e}")

    return {
        "success": code == 0,
        "command": cmd,
        "exit_code": code,
        "stdout": out,
        "stderr": err,
    }

AI聊天接口

@app.post("/api/chat")
async def api_chat(req: ChatRequest):
    """
    AI聊天接口：
    - 接收聊天消息历史，根据最新消息内容决定调用哪个工具
    - 将工具执行结果包装成AI回复
    """
    if not req.messages:
        raise HTTPException(status_code=400, detail="messages 不能为空")

    # 获取最新的用户消息
    latest_message = req.messages[-1]
    if latest_message.role != "user":
        raise HTTPException(status_code=400, detail="最新消息必须是用户消息")

    user_input = latest_message.content.lower()
    response = ""

    try:
        # 根据用户输入的关键词决定调用哪个工具
        if any(keyword in user_input for keyword in ["cpu", "处理器", "cpu使用率"]):
            # 调用CPU工具
            cmd = "top -b -n 1 | head -n 5"
            code, out, err = run_ssh_command(cmd)
            response = f"我已查询服务器CPU使用情况：\n\n```\n{out}\n```"
        elif any(keyword in user_input for keyword in ["内存", "mem", "memory"]):
            # 调用内存工具
            cmd = "free -h"
            code, out, err = run_ssh_command(cmd)
            response = f"我已查询服务器内存使用情况：\n\n```\n{out}\n```"
        elif any(keyword in user_input for keyword in ["磁盘", "disk", "空间"]):
            # 调用磁盘工具
            cmd = "df -h"
            code, out, err = run_ssh_command(cmd)
            response = f"我已查询服务器磁盘使用情况：\n\n```\n{out}\n```"
        elif any(keyword in user_input for keyword in ["负载", "load", "系统负载"]):
            # 调用系统负载工具
            cmd = "uptime"
            code, out, err = run_ssh_command(cmd)
            response = f"我已查询服务器系统负载情况：\n\n```\n{out}\n```"
        elif any(keyword in user_input for keyword in ["hello", "你好", "hi"]):
            response = "你好！我是内网运维助手，有什么可以帮你查询的吗？例如：\n- 查询CPU使用情况\n- 查看内存使用情况\n- 检查磁盘使用情况\n- 查看系统负载"
        else:
            response = "抱歉，我不太理解你的问题。你可以尝试询问关于CPU、内存、磁盘或系统负载的情况。"
    except Exception as e:
        response = f"抱歉，执行命令时出现错误：{e}"

    return {"reply": response}

2. SSH客户端实现

import paramiko
from typing import Tuple

from config import SSH_HOST, SSH_PORT, SSH_USER, SSH_PASSWORD, SSH_TIMEOUT


def run_ssh_command(
    command: str,
    host: str = SSH_HOST,
    port: int = SSH_PORT,
    username: str = SSH_USER,
    password: str = SSH_PASSWORD,
    timeout: int = SSH_TIMEOUT,
) -> Tuple[int, str, str]:
    """
    在远程 Linux 服务器上执行命令，返回 (exit_code, stdout, stderr)。
    """
    client = paramiko.SSHClient()
    client.set_missing_host_key_policy(paramiko.AutoAddPolicy())

    try:
        client.connect(
            hostname=host,
            port=port,
            username=username,
            password=password,
            timeout=timeout,
            look_for_keys=False,
            allow_agent=False,
        )
        stdin, stdout, stderr = client.exec_command(command)

        exit_code = stdout.channel.recv_exit_status()
        out = stdout.read().decode("utf-8", errors="ignore")
        err = stderr.read().decode("utf-8", errors="ignore")

        return exit_code, out, err
    except Exception as e:
        # 打印详细的错误信息
        print(f"SSH连接错误: {type(e).__name__}: {e}")
        print(f"连接信息: {host}:{port}, 用户名: {username}")
        print(f"密码长度: {len(password)}")
        raise
    finally:
        client.close()

3. 前端实现

页面结构

<!DOCTYPE html>
<html lang="zh-CN">
  <head>
    <meta charset="UTF-8" />
    <title>内网运维助手 Intranet Ops Assistant</title>
    <!-- 样式代码 -->
  </head>
  <body>
    <div class="header">
      <h1>内网运维助手</h1>
      <div class="status">
        <div class="status-indicator"></div>
        服务运行中
      </div>
    </div>

    <div class="main-content">
      <div class="sidebar">
        <h3>快捷操作</h3>
        <button class="quick-action" id="quickCpu">查看 CPU 使用情况</button>
        <button class="quick-action" id="quickMemory">查看内存使用情况</button>
        <button class="quick-action" id="quickDisk">查看磁盘使用情况</button>
        <button class="quick-action" id="quickLoad">查看系统负载</button>

        <h3>常用命令</h3>
        <button class="quick-action" id="quickTop">执行 top 命令</button>
        <button class="quick-action" id="quickPs">查看进程列表</button>
      </div>

      <div class="chat-container">
        <div class="chat-header">
          运维助手对话
        </div>

        <div class="chat-window" id="chatWindow">
          <!-- 消息会插入到这里 -->
        </div>

        <div class="chat-input-area">
          <div class="chat-input-container">
            <textarea 
              class="chat-input" 
              id="chatInput" 
              placeholder="例如：查询一下当前 CPU 使用率"></textarea>
            <button class="send-btn" id="sendBtn">发送</button>
          </div>
        </div>
      </div>
    </div>

    <!-- JavaScript代码 -->
  </body>
</html>

前端交互逻辑

async function sendChat(text) {
  appendMessage("user", text);
  showLoading();

  sendBtn.disabled = true;

  try {
    const resp = await fetch("http://localhost:8007/api/chat", {
      method: "POST",
      headers: { "Content-Type": "application/json" },
      body: JSON.stringify({ messages }),
    });

    const data = await resp.json();
    hideLoading();

    if (!resp.ok) {
      appendMessage("assistant", "请求失败：" + (data.detail || resp.status));
    } else {
      appendMessage("assistant", data.reply || "(无回复内容)");
    }
  } catch (e) {
    hideLoading();
    appendMessage("assistant", "调用异常：" + e);
  } finally {
    sendBtn.disabled = false;
  }
}

配置管理

配置文件结构

"""
项目名称：内网运维助手（Intranet Ops Assistant）

说明：
- 这里只是配置信息，真实的测试服务器 IP / 账号 / 密码请按实际情况修改。
- 目前你的网段是 192.168.108.x，这里先放一个占位 IP。
"""

SSH_HOST = "192.168.1.1"  # 需替换真实测试服务器 IP
SSH_PORT = 22
SSH_USER = "admin"    # 需替换真实用户名
SSH_PASSWORD = "123456"  # 需替换真实密码
SSH_TIMEOUT = 10

部署与使用

环境准备

  1. 安装依赖
    pip install -r requirements.txt

  2. 配置服务器信息

修改 config.py 文件，填入真实的服务器 IP、用户名和密码。

  1. 启动服务器
    python -m uvicorn app:app –host 0.0.0.0 –port 8007

  2. 访问前端页面

在浏览器中打开 http://localhost:8007/static/index.html。

使用方法

  1. 通过快捷操作：点击左侧的快捷操作按钮，如”查看 CPU 使用情况”。
  2. 通过自然语言：在聊天输入框中输入自然语言指令，如”帮我查看内存使用情况”。
  3. 执行常用命令：点击左侧的常用命令按钮，如”执行 top 命令”。

技术难点与解决方案

1. SSH连接认证问题

问题：SSH连接认证失败，无法执行命令。

解决方案：

  • 确保配置文件中的用户名和密码正确
  • 在SSH客户端代码中添加详细的错误信息输出
  • 禁用密钥查找和代理，只使用密码认证

2. 跨域访问问题

问题：前端页面无法访问后端API，出现跨域错误。

解决方案：

  • 在FastAPI中添加CORS中间件
  • 允许所有来源的请求（生产环境中应限制为具体域名）

3. 端口占用问题

问题：启动服务器时出现端口被占用的错误。

解决方案：

  • 使用不同的端口启动服务器
  • 检查并关闭占用端口的进程

未来改进方向

1. 多机器管理系统

  • 添加机器列表页面，显示所有可管理的服务器
  • 按部门、功能或位置对服务器进行分组
  • 为每台机器添加详细信息（配置、IP、负责人等）
  • 在聊天界面中可以快速切换要操作的机器

2. 功能扩展

  • 更多监控指标：网络流量、进程状态、服务运行状态等
  • 文件管理：支持文件上传、下载、编辑等操作
  • 服务管理：启动、停止、重启系统服务
  • 定时任务：设置定时执行的运维任务
  • 脚本管理：保存和管理常用的运维脚本

3. 安全性增强

  • 用户认证：添加登录系统，支持不同用户权限
  • 操作审计：记录所有操作日志
  • 命令白名单：限制可执行的命令
  • 加密传输：确保敏感数据的传输安全

4. 智能运维助手

  • 集成Coze智能体：利用Coze的能力，实现更智能的对话交互
  • 自然语言处理：支持更复杂的自然语言指令
  • 故障诊断：根据系统状态自动诊断潜在问题
  • 智能建议：基于系统状态提供优化建议

5. 数据可视化

  • 实时监控仪表盘：展示关键系统指标的实时变化
  • 历史趋势分析：查看系统性能的历史趋势
  • 异常检测：自动标记异常的系统状态
  • 报表生成：定期生成系统状态报表

总结

内网运维助手系统是一个轻量级的服务器管理工具，通过Web界面实现对远程Linux服务器的监控和管理。系统采用FastAPI作为后端框架，Paramiko实现SSH连接，前端采用HTML5、CSS3和JavaScript构建类似Coze的聊天界面。

该系统的主要特点包括：

  1. 简单易用：通过自然语言指令执行运维操作
  2. 功能丰富：支持查看CPU、内存、磁盘等系统指标
  3. 安全可靠：通过SSH协议连接服务器，确保操作安全
  4. 可扩展性：模块化设计，便于添加新功能
  5. 跨平台：可以在任何有浏览器的设备上使用

未来，我们将继续完善系统功能，添加多机器管理、智能运维、数据可视化等特性，使其成为一个更加全面和智能的内网运维工具。