Python 运维 Agent 工程实现详解

整体架构逻辑：
用户自然语言提问 → 厂商大模型API思考 → 判断是否调用运维工具(K8s/GPU/Shell) → 执行工具拿到真实集群数据 → 大模型整理结果返回给用户
依赖：通义千问/DeepSeek/智谱等兼容OpenAI格式的云API，无需本地GPU

项目结构

ops_agent/
├── .env                # 全局配置：API密钥、命令白名单
├── prompts.py          # 系统提示词：定义Agent身份、工具规范、输出格式
├── llm_client.py       # 统一封装大模型API调用，屏蔽厂商差异
├── tools/
│   ├── __init__.py     # 工具注册表，映射工具名和执行函数
│   ├── k8s_tool.py     # K8s集群查询工具（Pod/Node/事件）
│   ├── gpu_tool.py     # DCGM GPU硬件监控工具（温度/显存/XID故障）
│   └── shell_tool.py   # 安全受限Shell命令执行工具（白名单拦截危险指令）
├── agent.py            # ReAct核心循环：对话管理、工具调用、多轮思考
└── run.py              # 控制台交互入口，直接运行和Agent对话

1. 配置文件 .env（全局变量）

# 大模型厂商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

字段说明

1. LLM_API_KEY：厂商后台申请的密钥，鉴权用；
2. LLM_BASE_URL：兼容OpenAI的接口地址，DeepSeek/智谱替换即可；
3. LLM_MODEL：调用的模型名称；
4. SHELL_WHITELIST：Shell命令白名单，只有命令开头匹配列表才允许执行。

2. llm_client.py 大模型统一调用封装

作用：统一对接厂商API，上层业务代码无需关心底层接口细节，解耦。

import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI

# 加载.env环境变量，项目启动仅执行一次
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

逐行解析

1. load_dotenv()：读取项目根目录.env文件，把配置注入环境变量os.environ；
2. ChatOpenAI：LangChain封装的标准OpenAI兼容客户端，所有云厂商API通用；
3. temperature=0.05：运维场景必须调低，降低模型随机发挥，输出格式稳定、不编造故障；
4. max_tokens：限制单次返回文本长度，防止超长回答触发API截断报错；
5. 函数封装：整个项目只调用get_llm()获取模型，切换厂商仅修改.env，无需改动业务代码。

3. prompts.py 系统提示词（Agent行为规则核心）

大模型所有能力、工具格式、安全限制全部由这段文本约束，是整个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、格式化磁盘、修改系统内核配置等危险指令。
"""

核心约束说明

1. 强制模型不能编造数据，所有硬件/集群信息必须调用工具；
2. 固定3个工具+入参规范，模型必须严格按照JSON格式输出工具调用指令；
3. 安全规则：高危操作人工确认，屏蔽破坏性系统命令；
4. 格式强制：工具调用只能返回纯JSON，方便代码解析，不会混入多余文字导致解析失败。

4. tools 工具层详解

4.1 tools/init.py 工具注册表

作用：建立工具名字符串与执行函数的映射，上层agent只需要通过字符串调用工具，不需要import一堆函数。

# 导入三个工具执行函数
from .k8s_tool import k8s_tool
from .gpu_tool import gpu_tool
from .shell_tool import shell_tool

# 工具映射字典，key是模型输出的tool名称，value是对应执行函数
tool_map = {
    "k8s_tool": k8s_tool,
    "gpu_tool": gpu_tool,
    "shell_tool": shell_tool
}

4.2 tools/k8s_tool.py K8s集群查询工具

依赖官方kubernetes Python SDK，读取集群Pod、节点、事件，仅查询，无删除/修改高危操作。

from kubernetes import client, config

# 加载kubeconfig配置，分两种环境：本地开发 / K8s容器内运行
try:
    # 本地机器读取 ~/.kube/config
    config.load_kube_config()
except:
    # 容器内部自动读取挂载的ServiceAccount权限
    config.load_incluster_config()

# 实例化K8s核心API客户端
core_v1 = client.CoreV1Api()

def k8s_tool(params: dict):
    """
    入参params：模型生成的参数字典
    返回值：文本格式集群数据
    """
    # 解析入参
    action = params.get("action")
    ns = params.get("namespace", "default")
    name = params.get("name", "")

    # 1. 查询Pod列表/单个Pod详情
    if action == "get_pod":
        if name:
            # 查询指定Pod详情
            pod = core_v1.read_namespaced_pod(name, ns)
            return f"Pod详情：名称{pod.metadata.name} 状态{pod.status.phase} 镜像{pod.spec.containers[0].image}"
        else:
            # 查询命名空间下全部Pod
            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)

    # 2. 查询所有节点状态
    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)

    # 3. 查询最近集群事件（崩溃、调度失败等告警）
    elif action == "get_event":
        events = core_v1.list_namespaced_event(ns)
        res = []
        # 只取最后10条事件，避免文本过长
        for e in events.items[-10:]:
            res.append(f"[{e.type}] {e.message}")
        return "\n".join(res)

    # 不支持的操作返回提示
    return "不支持的k8s操作"

关键细节

1. 双模式kubeconfig加载：本地调试、容器部署都兼容；
2. 仅实现只读查询，无删除、重启Pod等写操作，降低风险；
3. 事件只取最后10条，控制返回文本长度，减少大模型API消耗。

4.3 tools/gpu_tool.py DCGM GPU硬件监控工具

调用官方pydcgm，读取显卡硬件指标：利用率、显存、温度、XID硬件故障码。

import sys
# 导入DCGM官方Python绑定路径
sys.path.insert(0, "/usr/local/dcgm/bindings")
import pydcgm

def gpu_tool(params: dict):
    # 获取入参gpu_id，为空则查询全部显卡
    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()
        # 如果指定了gpu_id，只遍历对应显卡
        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)

关键细节

1. sys.path.insert：手动加载DCGM自带Python绑定；
2. 嵌入模式dcgmStartEmbedded：无需后台启动nv-hostengine，脚本直接读取硬件；
3. 读取XID故障码：用于AI自动识别显卡硬件损坏、算力报错。

4.4 tools/shell_tool.py 安全Shell执行工具（白名单防护核心）

防止大模型输出rm -rf /、chmod 777等高危命令，做第一层拦截。

import os
import subprocess
from dotenv import load_dotenv

# 加载.env中的命令白名单
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:
        # 执行shell命令，超时10秒防止卡死
        result = subprocess.check_output(
            cmd,
            shell=True,
            timeout=10,
            stderr=subprocess.STDOUT,
            text=True
        )
        return result
    except subprocess.CalledProcessError as e:
        # 命令执行返回非0错误码
        return f"命令执行失败: {e.output}"
    except Exception as e:
        # 超时、权限不足等通用异常捕获
        return f"执行异常: {str(e)}"

安全核心逻辑

1. 拆分命令首程序，仅允许白名单内程序；
   例：kubectl get pods 首程序kubectl放行；rm -rf /data首程序rm直接拦截；
2. timeout=10：防止死循环命令占用服务；
3. stderr=subprocess.STDOUT：标准错误和标准输出合并，完整返回给模型分析报错。

5. agent.py ReAct 智能体核心循环（整套程序调度中枢）

实现多轮思考：提问 → 调用工具 → 带回数据 → 汇总回答，限制最大循环次数避免死循环。

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}
    ]
    # 限制最多连续调用3次工具，防止无限循环思考
    max_loop = 3

    for _ in range(max_loop):
        # 1. 调用厂商大模型API思考
        resp = llm.invoke(messages)
        content = resp.content.strip()

        # 2. 尝试解析模型输出是否为工具调用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]
            # 3. 执行本地运维工具，拿到真实集群/硬件数据
            tool_result = tool_func(tool_params)

            # 4. 将工具调用记录、工具返回结果追加到对话上下文
            messages.append({"role": "assistant", "content": content})
            messages.append({
                "role": "user",
                "content": f"工具执行结果:\n{tool_result}\n根据以上数据回答用户原始问题：{user_query}"
            })
        # JSON解析失败 = 模型不需要调用工具，直接输出最终答案
        except json.JSONDecodeError:
            return content

    # 达到最大工具调用次数，强制生成最终总结回答
    final_resp = llm.invoke(messages)
    return final_resp.content

逐段核心逻辑拆解

1. 对话上下文 messages

遵循OpenAI对话格式：system(系统规则)、user(用户提问)、assistant(模型输出)，每次工具执行结果追加进上下文，模型能记住之前查询的数据。

2. ReAct循环流程

1. 把当前对话传给大模型；
2. 尝试解析返回内容：

• 成功解析JSON：代表模型需要调用工具，执行对应运维函数，把工具返回数据塞进对话，进入下一轮循环；
• JSON解析报错：代表模型已有足够数据，直接返回回答；

3. max_loop=3 防护：防止模型无限循环反复调用工具，超过3次强制输出总结。

3. 工具执行链路

模型输出JSON字符串 → json.loads解析字典 → tool_map匹配函数 → 执行工具获取集群真实数据 → 数据回传给大模型做二次整理。

6. run.py 控制台交互入口（程序启动入口）

简易终端交互，输入问题即可对话，输入exit退出。

from agent import run_agent

if __name__ == "__main__":
    print("==== 运维智能Agent 已启动，输入exit退出 ====")
    # 持续接收用户输入
    while True:
        question = input("\n请输入运维问题：")
        # 退出判断
        if question.lower() == "exit":
            break
        # 调用Agent核心逻辑
        ans = run_agent(question)
        # 打印最终回答
        print("\nAgent回答：")
        print(ans)

运行方式

终端执行：

python run.py

示例交互：

==== 运维智能Agent 已启动，输入exit退出 ====

请输入运维问题：查看所有GPU温度和XID故障码
Agent回答：
GPU0 利用率:12% 显存占用:8192MB 温度:62℃ XID故障码:0
GPU1 利用率:5% 显存占用:4096MB 温度:58℃ XID故障码:0
所有显卡无硬件报错，负载正常。

7. 完整执行流程串联演示

用户输入：查看default命名空间所有Pod状态

1. run_agent 接收问题，初始化对话上下文；
2. 调用厂商大模型API，模型根据提示词输出纯JSON：
   {"tool": "k8s_tool", "params": {"action":"get_pod", "namespace":"default"}}
3. 代码json.loads解析成功，匹配k8s_tool函数；
4. 执行K8s API查询全部Pod，返回Pod状态文本；
5. 将工具调用记录、Pod列表追加到messages；
6. 再次调用大模型，传入Pod真实数据；
7. 模型无需再调用工具，直接返回整理后的中文Pod状态总结；
8. run.py打印回答展示给用户。

8. 扩展改造方向（代码可直接升级）

1. 增加多轮记忆持久化：把messages存入redis，实现跨会话连续对话；
2. FastAPI接口封装：给run_agent套一层接口，对接钉钉/企业微信机器人；
3. 高危操作确认机制：在shell/k8s工具中增加判断，删除资源时返回确认提示，二次输入yes才执行；
4. RAG知识库接入：提问前先检索运维故障文档，把故障案例送入大模型，排障更精准；
5. 异常捕获增强：给agent.py增加try-except捕获API超时、K8s权限不足、DCGM初始化失败等异常，友好提示用户。