自然语言转运维指令 Python Agent 搭建

从零搭建「自然语言转运维指令」Python 运维Agent（基于云厂商大模型API，无需本地GPU）

整体流程：

调用通义千问/DeepSeek/智谱等厂商API做思考+工具调用；
内置3套运维工具：K8s查询、DCGM GPU查询、安全Shell白名单；
ReAct逻辑：自然语言→判断是否调用工具→执行工具→汇总结果回答；
安全机制：高危操作二次确认、shell命令白名单拦截危险指令。

一、环境依赖安装

pip install openai python-dotenv kubernetes pydantic langchain langchain-openai

二、项目目录

ops_agent/
├── .env                # API密钥、配置
├── prompts.py          # 运维Agent系统提示词
├── llm_client.py       # 统一封装厂商API调用
├── tools/
│   ├── __init__.py
│   ├── k8s_tool.py     # K8s资源查询工具
│   ├── gpu_tool.py     # DCGM GPU监控工具
│   └── shell_tool.py   # 安全受限shell工具
├── agent.py            # ReAct智能体主逻辑
└── run.py              # 交互入口，控制台对话

三、配置文件 .env

以阿里通义千问兼容OpenAI接口举例，更换base_url即可切换DeepSeek/智谱：

# 大模型API配置
LLM_API_KEY=sk-xxxxxxxxxxxxxxxxxxxx
LLM_BASE_URL=https://dashscope.aliyuncs.com/compatible-mode/v1
LLM_MODEL=qwen-turbo

# 安全白名单shell，只允许这些命令
SHELL_WHITELIST = kubectl,nvidia-smi,dcgmi,df,free,top,ps,lsof,uptime
# 高危操作强制二次确认
CONFIRM_DANGER=true

四、llm_client.py 封装厂商API

所有厂商兼容OpenAI格式，统一调用入口，后续切换模型只改env：

import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI

load_dotenv()

def get_llm():
    llm = ChatOpenAI(
        api_key=os.getenv("LLM_API_KEY"),
        base_url=os.getenv("LLM_BASE_URL"),
        model=os.getenv("LLM_MODEL"),
        temperature=0.05,  # 运维场景降低随机性，输出稳定
        max_tokens=1500
    )
    return llm

五、prompts.py 核心系统提示词（定义Agent能力、工具格式）

SYSTEM_PROMPT = """
你是专业K8s+GPU集群运维智能助手，只能使用提供的3种工具获取真实数据，禁止编造信息。
可用工具列表：
1. k8s_tool
  作用：查询K8s集群Pod、节点、事件、Deployment
  参数：
    action: str 可选值 get_pod / get_node / get_event
    namespace: str 命名空间，默认default
    name: str 资源名称，可选
2. gpu_tool
  作用：调用DCGM读取GPU温度、显存、利用率、XID故障码
  参数：gpu_id: int，显卡编号，不传默认查全部GPU
3. shell_tool
  作用：执行服务器运维shell命令，仅允许白名单内程序
  参数：cmd: str 完整命令字符串

执行规则：
1. 用户输入自然语言，先判断是否需要调用工具获取数据；无数据直接回答；
2. 需要调用工具时，**仅输出纯JSON**，无多余解释、无换行、无markdown：
{"tool": "工具名称", "params": {键值对参数}}
3. 拿到工具返回结果后，整理简洁中文结论反馈用户；
4. 若用户需求是删除Pod、驱逐节点、清理磁盘、杀死占用GPU进程等高危操作，必须主动询问用户确认后再执行；
5. 禁止执行rm -rf、chmod 777、格式化磁盘、修改系统内核配置等危险指令。
"""

六、工具层代码实现

6.1 tools/init.py

from .k8s_tool import k8s_tool
from .gpu_tool import gpu_tool
from .shell_tool import shell_tool

tool_map = {
    "k8s_tool": k8s_tool,
    "gpu_tool": gpu_tool,
    "shell_tool": shell_tool
}

6.2 tools/k8s_tool.py（K8s查询工具）

from kubernetes import client, config

# 加载kubeconfig
try:
    config.load_kube_config()
except:
    config.load_incluster_config()

core_v1 = client.CoreV1Api()
apps_v1 = client.AppsV1Api()

def k8s_tool(params: dict):
    action = params.get("action")
    ns = params.get("namespace", "default")
    name = params.get("name", "")

    if action == "get_pod":
        if name:
            pod = core_v1.read_namespaced_pod(name, ns)
            return f"Pod详情：名称{pod.metadata.name} 状态{pod.status.phase} 镜像{pod.spec.containers[0].image}"
        else:
            pods = core_v1.list_namespaced_pod(ns)
            res = []
            for p in pods.items:
                res.append(f"{p.metadata.name} | 状态:{p.status.phase}")
            return "\n".join(res)

    elif action == "get_node":
        nodes = core_v1.list_node()
        res = []
        for n in nodes.items:
            res.append(f"节点:{n.metadata.name} 就绪状态:{n.status.conditions[-1].status}")
        return "\n".join(res)

    elif action == "get_event":
        events = core_v1.list_namespaced_event(ns)
        res = []
        for e in events.items[-10:]:
            res.append(f"[{e.type}] {e.message}")
        return "\n".join(res)

    return "不支持的k8s操作"

6.3 tools/gpu_tool.py（DCGM读取GPU信息）

import sys
sys.path.insert(0, "/usr/local/dcgm/bindings")
import pydcgm

def gpu_tool(params: dict):
    gpu_id_input = params.get("gpu_id")
    # 初始化DCGM
    dcgm_handle = pydcgm.DcgmHandle(pydcgm.dcgmStartEmbedded(0))
    system = dcgm_handle.GetSystem()
    gpus = system.GetAllGpus()
    output = []
    for gpu in gpus:
        gid = gpu.GetGpuId()
        if gpu_id_input is not None and gid != gpu_id_input:
            continue
        util = gpu.GetFieldValue(pydcgm.dcgm_structs.DCGM_FI_DEV_GPU_UTIL).value
        mem_used = gpu.GetFieldValue(pydcgm.dcgm_structs.DCGM_FI_DEV_MEM_USED).value // 1024
        temp = gpu.GetFieldValue(pydcgm.dcgm_structs.DCGM_FI_DEV_GPU_TEMP).value
        xid = gpu.GetFieldValue(pydcgm.dcgm_structs.DCGM_FI_DEV_XID_ERRORS).value
        output.append(f"GPU{gid} 利用率:{util}% 显存占用:{mem_used}MB 温度:{temp}℃ XID故障码:{xid}")
    return "\n".join(output)

6.4 tools/shell_tool.py（安全白名单shell）

import os
import subprocess
from dotenv import load_dotenv

load_dotenv()
white_list = os.getenv("SHELL_WHITELIST").split(",")

def shell_tool(params: dict):
    cmd = params.get("cmd", "")
    # 校验命令白名单
    cmd_head = cmd.strip().split()[0]
    if cmd_head not in white_list:
        return f"拒绝执行，{cmd_head}不在运维命令白名单中，禁止运行"
    try:
        result = subprocess.check_output(cmd, shell=True, timeout=10, stderr=subprocess.STDOUT, text=True)
        return result
    except subprocess.CalledProcessError as e:
        return f"命令执行失败: {e.output}"
    except Exception as e:
        return f"执行异常: {str(e)}"

七、agent.py ReAct 智能体核心逻辑

循环：用户提问 → LLM思考 → 判断调用工具/直接回答 → 执行工具带回数据 → 二次总结输出

import json
from llm_client import get_llm
from prompts import SYSTEM_PROMPT
from tools import tool_map

llm = get_llm()

def run_agent(user_query: str):
    messages = [
        {"role": "system", "content": SYSTEM_PROMPT},
        {"role": "user", "content": user_query}
    ]
    max_loop = 3  # 最多连续调用3次工具，防止死循环
    for _ in range(max_loop):
        resp = llm.invoke(messages)
        content = resp.content.strip()
        # 判断输出是否是工具调用JSON
        try:
            tool_call = json.loads(content)
            tool_name = tool_call.get("tool")
            tool_params = tool_call.get("params", {})
            # 执行对应工具
            tool_func = tool_map[tool_name]
            tool_result = tool_func(tool_params)
            # 把工具返回数据送入对话，让模型汇总答案
            messages.append({"role": "assistant", "content": content})
            messages.append({"role": "user", "content": f"工具执行结果:\n{tool_result}\n根据以上数据回答用户原始问题：{user_query}"})
        except json.JSONDecodeError:
            # 不是JSON，说明模型拿到足够数据，直接输出结论
            return content
    # 循环上限，返回最后一次回答
    final_resp = llm.invoke(messages)
    return final_resp.content

八、run.py 控制台交互入口

from agent import run_agent

if __name__ == "__main__":
    print("==== 运维智能Agent 已启动，输入exit退出 ====")
    while True:
        question = input("\n请输入运维问题：")
        if question.lower() == "exit":
            break
        ans = run_agent(question)
        print("\nAgent回答：")
        print(ans)

九、使用示例

示例1：自然语言查GPU状态

输入：

帮我看看所有GPU的温度、显存占用和故障码

流程：

LLM输出JSON调用gpu_tool；
代码执行DCGM查询，返回硬件数据；
模型根据硬件数据整理中文总结输出。

示例2：查询K8s异常Pod

输入：

查看default命名空间下所有Pod状态

Agent自动调用k8s_tool列出Pod，汇总异常容器。

示例3：执行服务器基础运维命令

输入：

查看服务器内存使用情况

模型自动生成shell命令free -h，调用shell工具执行并返回结果。

十、扩展与生产优化

对接企业微信/钉钉：封装FastAPI接口，接收消息调用run_agent，把回答推送聊天群；
高危操作确认：在shell_tool/k8s_tool内增加判断，删除/驱逐类操作先返回确认提示；
历史记忆：用列表持久化messages，实现多轮连续对话；
RAG知识库增强：接入运维故障文档向量库，提问前检索历史故障案例再送入大模型；
错误捕获：增加try-except捕获API超时、K8s权限不足、DCGM读取失败等异常。

十一、切换其他厂商API仅需修改.env

DeepSeek

LLM_BASE_URL=https://api.deepseek.com/v1
LLM_MODEL=deepseek-chat

智谱AI

LLM_BASE_URL=https://open.bigmodel.cn/api/paas/v4
LLM_MODEL=glm-4-flash

YancheBlog

热门标签

自然语言转运维指令 Python Agent 搭建

自然语言转运维指令 Python Agent 搭建

从零搭建「自然语言转运维指令」Python 运维Agent（基于云厂商大模型API，无需本地GPU）

一、环境依赖安装

二、项目目录

三、配置文件 .env

四、llm_client.py 封装厂商API

五、prompts.py 核心系统提示词（定义Agent能力、工具格式）

六、工具层代码实现

6.1 tools/init.py

6.2 tools/k8s_tool.py（K8s查询工具）

6.3 tools/gpu_tool.py（DCGM读取GPU信息）

6.4 tools/shell_tool.py（安全白名单shell）

七、agent.py ReAct 智能体核心逻辑

八、run.py 控制台交互入口

九、使用示例

示例1：自然语言查GPU状态

示例2：查询K8s异常Pod

示例3：执行服务器基础运维命令

十、扩展与生产优化

十一、切换其他厂商API仅需修改.env

目录