对话式 Linux 运维 Agent：大模型驱动与高危操作人工确认设计实战

我这人比较懒，尤其是碰上重复性的运维操作，能写脚本绝不动手。但脚本有个天生的短板：它只是个执行器，没有判断力。重启个服务、删个日志、改个配置，这些操作本身不难，难的是判断“现在能不能做”“做完之后对不对”。所以当我有机会做一个对话式 Linux 运维 Agent 时，我给自己定了个底线：可以自动，但高危操作必须强制人工确认，绝不做成一把梭的全自动。

这个项目最终落地的是一个基于大模型驱动的运维助手，支持自然语言发起指令，比如“查看 Nginx 最近一小时错误日志并分析原因”“检查 /data 分区空间使用率”，它会自动解析、执行、汇总结果。遇到删除、重启、改配置这类高危操作，会明确拦截并等待人工确认。最麻烦的场景也做了兜底：远程重启导致 SSH 断线后，Agent 会在后台守候，等服务重新上线后自动重连并复检结果。

整个过程做下来，我最大的感触是：运维自动化真正的难点不在“自动”，而在“可控”。这篇文章就把完整的设计思路、关键实现、踩坑过程都拆开讲清楚，希望能给同样在做类似工具的朋友一些参考。

1. 项目定位与技术选型：为什么选大模型驱动，而不是写死的规则脚本

先明确这个项目解决什么问题。传统运维自动化用 Ansible、Shell 脚本就能解决一大半，但这类工具要求操作者具备明确的命令知识，交互方式是“写 Playbook”或者“敲命令”。当运维人员手里的工单堆积如山，很多时候根本没时间分析系统状态，只想知道“现在是什么情况”“我该做什么”“做了之后对不对”。

对话式运维 Agent 的价值就在于，把“理解意图”和“执行操作”这两件事解耦了。你不用记命令，直接用大白话问，Agent 负责翻译成机器能执行的指令，再把结果翻译回人话。这个思路下，技术选型就非常清晰了：

• 对话理解层：大模型负责意图识别、参数抽取、命令生成、结果总结。选用 OpenAI 兼容接口的模型，支持函数调用（Function Calling）以获取结构化输出。部署上为了兼顾数据安全和响应速度，模型可以本地化部署，也可以走云端 API，核心在于构建好 Prompt 上下文。
• 执行层：复用 Paramiko 做 SSH 远程连接，执行命令并采集输出。Paramiko 的优势是纯 Python 实现、生态成熟、支持跳板机配置、断线重连逻辑可以完全自己控制。
• 安全层：自研的高危命令过滤模块。这部分不走大模型，而是用正则加字符串匹配做硬校验，放在执行链路最前段，一票否决。
• 状态管理：用 Python 的 State Machine（状态机）来管理对话流程。为什么要用状态机？因为对话式运维不是单轮问答，操作过程中有“待确认”“执行中”“守候复检”“完成”等状态，状态之间流转关系复杂，用状态机可以避免代码写得像意大利面条。

这里插一句选型上的重要对比：为什么不直接用开源的 AutoGPT 这类通用 Agent？因为它们太“通用”了，没有针对运维场景做安全设计。AutoGPT 解决的是“如何让模型自主完成一个复杂任务”，而我需要的是“如何让模型在严格的边境内完成一个运维操作”。运维场景里，边界和安全远比“自主性”重要。自己做能更好的控制高危命令的人力确认环节，这是通用框架无法直接满足的。

工具链上，日志存储用了 Redis，用户会话和操作记录双写。Redis 选它的原因是：读写在内存中完成，速度快；自带键过期机制，可以自动清理历史会话；而且它在运维系统里基本属于标配，部署成本低。

2. 系统架构与核心模块拆解：Agent 的大脑、手脚和刹车

整个系统拆成四个模块，各司其职。我画不出什么复杂的架构图，但脑子里对它的定位很清楚：

用户的输入到达后，先进入会话管理模块。这一步参考了银行客服机器人那套记忆机制，只保留最近几轮的关键上下文。因为运维指令往往是上下文相关的，比如用户先问“Nginx 昨天日志里有多少 500 错误”，然后问“重启一下”，如果不记得上文，这条“重启一下”就是无意义的。ConversationBufferWindowMemory 采用的窗口策略是保留最近 6 轮对话，既能关联上文，又不会让 Token 占用过载。

任务编排层是大模型发挥作用的核心地带。Function Calling 机制允许大模型返回结构化的 JSON 指令，而不是纯文本回复。例如用户说“最近磁盘快满了，帮我看看哪些文件占空间大”，模型的返回可能是：

{ "thought": "用户需要排查磁盘空间，先看整体使用率，再定位大文件", "commands": [ "df -h", "du -ah /data 2>/dev/null | sort -rh | head -20" ], "risk_level": "low", "need_confirm": false }

命令执行层拿到 JSON 后，逐条做参数合法性校验、权限检查，然后通过 Paramiko 在远端执行。每条命令都有独立的超时时间（默认 30 秒，长任务单独调高），避免一条命令卡死拖垮整个会话。

安全控制层是比较得意的设计。它不依赖大模型判断，而是用规则引擎做硬校验。大模型负责“听懂人话”，但“能不能做”这件事必须由确定性规则决定。所有命令在执行前，都会先过一层包含高危命令特征词的正则匹配，再配合一次 SSH 跳板登录的二次校验，双保险。

3. 高危操作强制人工确认：代码之外的生死线

高危命令的识别与拦截，是这个项目里设计优先级最高的模块。在代码世界里，“误删”和“误操作”付出的代价是真实世界的资产损失。我给自己定的原则是：宁可把安全范围扩大，也不能漏掉一条危险命令。过滤规则直接从等保合规的“最小权限原则”出发，越权操作和不可逆操作全部视为高风险。

实现上分为三层过滤：

第一层，命令关键词硬匹配。这里不是简单做几个关键词的字符串匹配，而是结合正则和空格分词做逻辑判断。比如rm -rf里的-rf参数是删除的关键特征，不仅要匹配到rm，还要判断是否带-f或-r；dd命令本身不算高危，但dd if=/dev/zero of=/dev/sda就是不可逆操作，要匹配of=指向的块设备路径。这一层我维护了一个规则表：

RISK_RULES = [ {"pattern": r"\brm\s+.*-rf\b", "reason": "强制递归删除", "level": "critical"}, {"pattern": r"\bdd\s+.*of=/dev/", "reason": "写入块设备", "level": "critical"}, {"pattern": r"\bshutdown\b|\breboot\b|\binit\s+6\b", "reason": "重启系统", "level": "critical"}, {"pattern": r"\bmkfs\.", "reason": "格式化操作", "level": "critical"}, {"pattern": r":\(\)\s*\{.*\}\s*;", "reason": "危险递归函数", "level": "critical"}, ]

第二层，针对复合命令的语义识别。很多高危操作会藏在复合命令里，比如cd /data && rm -rf ./bak && systemctl restart nginx，如果整体匹配，确实命中了rm -rf，但如果只匹配到末尾的重启命令呢？为了稳妥，我的实现是先用 Shell 词法分析器把整条命令按&&、||、;、|拆分，逐段做规则匹配。只要其中任何一段命中高危规则，整条命令就直接进入人工确认流程。

第三层，人工确认的交互闭环。这一步是拦截的关键，也是体验的分水岭。系统会在检测到高危命令后，立刻将命令详情（完整命令、目标机器、风险评估）推送到管理端的 Webhook 地址。管理端的消息推送包含了“执行”和“拒绝”两个按钮，点击后才能流转到下一步。确认接口的 Token 有效期设了 5 分钟，超时则自动取消操作。

这里要特别说明为什么坚持用人工确认，而不是加一个“二次确认密码”或者“冷却时间”就完事。密码确认只能证明“有权限的人在执行”，不能证明“这个人在认真思考这个操作”。运维事故往往发生在深夜加班、操作疲劳、误把生产环境当测试环境时。人工确认环节的意义在于强制打断一下，让操作者清醒地看一遍自己即将做的事。尤其是rm -rf这类命令，多看 3 秒钟可能就拦下了一次事故。

4. 对话指令解析与任务编排：从自然语言到可执行命令

对话解析是整个系统用户体验的关键。用户说的是人话，机器执行的是命令，中间这一层翻译的准确度直接决定这个工具能不能用。模型选型上我用了支持 Function Calling 的指令微调模型，但在实现上还是保留了很多兜底逻辑，避免出现模型“一本正经胡说八道”的情况。

任务编排的核心是：让大模型输出结构化的 JSON，而不是自由发挥的文本。这需要使用 Function Calling 的特性，在请求中定义好要调用的函数及参数。我定义了这些函数用于承接解析结果：

大模型返回调用意图后，系统会做一次参数合法性校验。比如run_command的command参数，必须通过 Shell 注入防护的正则过滤，把$(...)、反引号、;等特殊字符全部列入黑名单。这一点是很容易忽略的坑：大模型生成的命令是可信的，但大模型生成命令所用的上下文是不可信的。如果用户故意在对话中拼接恶意内容，模型生成的命令可能带上注入代码。所以命令在进入执行层之前，必须做独立的危险字符过滤。

任务排队上，我引入了 Redis 队列做任务发布订阅。所有指令依次进入队列，执行完成后推送结果。这样做有个额外收益：支持异步任务的执行与回调，比如“执行一键巡检脚本，跑完把结果发给我”。在线下场景里非常实用，不用一直在终端前傻等。

编排层还处理了一个很微妙的问题：指令冲突。当用户先要求“查看 Nginx 状态”，还没等结果返回，又要求“检查磁盘空间”，实际上这时会话可能被前一个指令占用了终端锁。系统通过会话 ID 加锁，同一时刻只有一个指令在执行，后续指令进入待处理状态，执行完上一个再继续。这个设计初看是限制，实际用起来反而避免了很多线上误操作。

5. 重启断线自动守候复检：一次设计失误换来的完整解

这是项目里最有意思的一段，也是我最初没考虑到，直到踩了坑才补上的功能。先说事故场景：我在测试环境执行一条systemctl restart docker命令，SSH 会话在服务重启的一瞬间就断开了。这个时候重连会失败，因为服务还没起来；但我的 Agent 代码还在傻等 Paramiko 返回结果，最终抛出超时异常，然后整个会话状态挂死，用户界面上显示的是“重启命令已执行，但结果未知”。这种状态在实际运维里极其危险——命令执行了吗？成功了没？服务状态到底怎样？全部是未知。

补全这个功能花了我大半天时间，思路彻底重构了一遍。重启类命令的执行不能用常规的“连接—执行—返回结果”模型，而要改成“预判断线—断开守护—重连复检—验证状态”的完整链路。

我的实现拆成了几步：

预判断线。在执行命令之前，先检查命令里是否含有reboot、shutdown、systemctl restart、service xxx restart这类关键词，一旦命中，立即通知会话管理器：这条命令执行后 SSH 连接将不可用，转入守候模式。

守候重连。SSH 断开后，启动一个后台线程，以 5 秒为间隔尝试重新连接，重试上限设为 60 次。5 秒这个间隔是最优解，太短会导致频繁握手加重系统负担，太长则复检不及时。断开后的这段时间，Agent 的状态机流转到 WAITING_FOR_RECONNECT。

新 SSH 会话建立后，执行验证命令。验证命令不能是简单的echo ok，要根据重启对象区别对待。比如重启的是 Docker，验证命令就应该是docker ps和systemctl status docker；重启的是网络服务，验证命令就要检查端口连通性。这个验证逻辑我在编排层做了一个映射表：

RESTART_VERIFY_MAP = { "docker": ["systemctl status docker", "docker ps --format '{{.Names}}: {{.Status}}'"], "nginx": ["systemctl status nginx", "ss -tlnp | grep :80", "curl -I http://127.0.0.1/"], "mysql": ["systemctl status mysql", "mysqladmin -uroot ping"], "network": ["ip addr show", "ping -c 2 8.8.8.8"], }

复检通过后，系统把所有结果汇总推送给用户，并且特别标注了“服务已恢复，以下为重启后状态确认”。如果复检失败，则发送告警并进入人工介入模式。整个过程，用户看到的不再是“进程已执行，结果未知”，而是一段完整的生命周期记录。

这个功能的实现代价不高，但价值极大。我把它的核心逻辑进一步抽象成了“不可用期”的概念，不光是重启，凡是会造成 SSH 断开的操作，都适用这个守候复检框架。

6. 多会话管理与并发控制：避免一人操作，全局遭殃

对话式运维系统跑起来后，很快会面临一个现实问题：多人同时使用。我的 Agent 最初只面向单人使用，只有一个全局 SSH 连接池。后来某同学要临时用一下，才发现两个人同时执行命令会互相串台——A 看的磁盘占用率，被 B 的cd /data && du -sh干扰了输出。

为此我重构了连接管理，核心思路是做会话隔离。每个用户拥有独立的 SSH 连接，连接参数包含目标机器、登录用户、会话上下文。连接池按会话 ID 管理，单个会话内串行执行，会话之间并行隔离。这个改造解决的不只是输出串扰，还解决了一个更隐蔽的权限问题：不同用户在同一台机器上操作时，如果共用一个系统账号，那么在日志审计里根本分不清是谁执行的。独立会话配合 sudo 提权命令，可以确保日志里记录的是“哪个用户通过 Agent 做了什么”，审计链路很清晰。

并发控制的另一个场景是目标机器的负载保护。一台生产服务器，如果同时有七八个运维操作在执行，即使每一条单看都正常，累积起来也可能把负载打爆。Agent 的系统级并发上限默认是 3，超出后进入等待队列。单机并发数可以在配置文件中调整，但我的建议是生产环境别超过 5。守候复检线程也会占用连接资源，所以每台机器最多允许一个守候任务在跑，其余操作排队。

7. 部署实操：从项目初始化到完成核心流程

前面讲了设计思路，这一段是完整的部署和实现过程，照做就能跑起来。

环境信息先说清楚，我的部署环境是 CentOS 7.9，Python 3.8，内存 4G，目标机器是内网若干台 Linux 服务器。Agent 本体是单进程架构，运行在一台管理机上，通过 SSH 管理目标机器。

7.1 基础依赖安装

# 更新系统 yum update -y # 安装 Python3.8 及 pip yum install -y python38 python38-devel python38-pip # 项目依赖 pip3 install paramiko redis langchain openai pandas

目录结构是这样的：

agent/ ├── main.py # 主入口 ├── config.py # 配置文件 ├── core/ │ ├── session.py # 会话管理 │ ├── executor.py # 命令执行器 │ ├── security.py # 高危命令识别 │ ├── recovery.py # 断线守候复检 │ └── state_machine.py # 状态机 ├── rules/ │ └── risk_rules.json # 高危规则表 ├── logs/ └── templates/

7.2 核心代码实现

先看命令执行器。这块用了 Paramiko 的连接池管理，避免每次请求都重新握手。

# core/executor.py import paramiko import time from concurrent.futures import ThreadPoolExecutor class SSHExecutor: def __init__(self, host, user, key_path, pool_size=5): self.host = host self.user = user self.key_path = key_path self.pool = ThreadPoolExecutor(max_workers=pool_size) self._conn = self._create_connection() def _create_connection(self): client = paramiko.SSHClient() client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) client.connect( hostname=self.host, username=self.user, key_filename=self.key_path, timeout=15, banner_timeout=30, auth_timeout=30 ) return client def execute(self, command, timeout=30): stdin, stdout, stderr = self._conn.exec_command(command, timeout=timeout) exit_code = stdout.channel.recv_exit_status() output = stdout.read().decode('utf-8', errors='replace') error = stderr.read().decode('utf-8', errors='replace') return {"exit_code": exit_code, "stdout": output, "stderr": error} def reconnect(self): """断线后重新建立连接，带重试""" for attempt in range(60): try: self._close() self._conn = self._create_connection() return True except Exception as e: time.sleep(5) return False def _close(self): try: self._conn.close() except: pass

高危命令识别模块放在执行链路的最前面，这是整个系统安全的第一道闸门：

# core/security.py import re import json class RiskFilter: def __init__(self, rules_path="rules/risk_rules.json"): with open(rules_path, "r", encoding="utf-8") as f: self.rules = json.load(f) def check(self, command): """返回风险的危害级别，无风险返回 None""" # 先拆分复合命令 segments = re.split(r'\s*(?:&&|\|\||;)\s*', command) for segment in segments: for rule in self.rules: if re.search(rule["pattern"], segment, re.IGNORECASE): return { "level": rule["level"], "reason": rule["reason"], "matched": segment } return None def injection_check(self, command): """检查命令注入特征""" dangerous_patterns = [r'\$\(', r'`', r'\bbash\b', r'\bpython\b', r';\s*\w+\s*\('] for pattern in dangerous_patterns: if re.search(pattern, command): return True return False

状态机的设计是整个会话流转的核心。我把状态定义成一组常量，在状态机类中完成状态间的合法迁移与操作绑定。

# core/state_machine.py class AgentState: IDLE = "idle" PARSING = "parsing" WAITING_CONFIRM = "waiting_confirm" EXECUTING = "executing" WAITING_RECONNECT = "waiting_reconnect" VERIFYING = "verifying" COMPLETED = "completed" FAILED = "failed" class StateMachine: """简单的状态机实现，约束状态合法流转""" TRANSITIONS = { AgentState.IDLE: [AgentState.PARSING], AgentState.PARSING: [AgentState.WAITING_CONFIRM, AgentState.EXECUTING], AgentState.WAITING_CONFIRM: [AgentState.EXECUTING, AgentState.IDLE], AgentState.EXECUTING: [AgentState.WAITING_RECONNECT, AgentState.COMPLETED, AgentState.FAILED], AgentState.WAITING_RECONNECT: [AgentState.VERIFYING, AgentState.FAILED], AgentState.VERIFYING: [AgentState.COMPLETED, AgentState.FAILED], AgentState.COMPLETED: [AgentState.IDLE], AgentState.FAILED: [AgentState.IDLE], } def __init__(self): self.state = AgentState.IDLE def transition(self, new_state): if new_state in self.TRANSITIONS[self.state]: print(f"状态迁移: {self.state} -> {new_state}") self.state = new_state return True else: raise ValueError(f"非法状态迁移: {self.state} -> {new_state}")

7.3 对话主流程实现

对话主流程整合上面几个模块。用户输入先过会话记忆，再交给大模型做 Function Calling 解析，然后过安全层，最后进入执行层。

# main.py def handle_message(user_input, session_id): # 1. 上一轮会话状态校验 sm = get_state(session_id) if sm.state in [AgentState.WAITING_CONFIRM]: return "上一条高风险操作仍在等待确认，请先处理后再发送新指令。" # 2. 大模型意图理解 intent = llm.parse(user_input, history=get_history(session_id)) # 3. 高危命令过滤 risk = risk_filter.check(intent["commands"]) if risk: sm.transition(AgentState.WAITING_CONFIRM) send_webhook_notify(session_id, intent["commands"], risk) return f"检测到高风险操作({risk['reason']})，已提交人工确认，等待审批。" # 4. 判断是否需要守候复检 if is_restart_command(intent["commands"]): sm.transition(AgentState.WAITING_RECONNECT) execute_with_recovery(intent["commands"]) # 5. 执行并返回结果 result = executor.execute(intent["commands"]) return format_result(result)

7.4 关键配置与运行参数

配置文件 config.py 中几个重要的参数：

# config.py # SSH 配置 SSH_HOST = "192.168.x.x" SSH_USER = "opsadmin" SSH_KEY_PATH = "/root/.ssh/id_rsa" # Redis 配置 REDIS_HOST = "localhost" REDIS_PORT = 6379 # 大模型接口配置 LLM_API_BASE = "http://localhost:8000/v1" LLM_MODEL_NAME = "qwen2.5-7b-instruct" # 超时配置 CMD_DEFAULT_TIMEOUT = 30 # 普通命令超时 CMD_LONG_TIMEOUT = 120 # 长任务命令超时 RECONNECT_INTERVAL = 5 # 守候重连间隔（秒） RECONNECT_MAX_RETRIES = 60 # 最大重连次数 # 并发控制 MAX_PARALLEL_PER_HOST = 3 # Webhook 确认地址 CONFIRM_WEBHOOK_URL = "http://your-admin-platform.com/webhook/confirm"

7.5 运行日志观察

系统运行时的日志设计也值得一说。日志不仅记录命令和输出，还记录完整的会话流转过程：用户身份、意图解析结果、命令生成结果、安全过滤结果、人工确认状态、执行耗时、守候情况。这样操作审计就有完整的链路。比如这样一段输出：

2025-06-12 14:23:01 [session: a3f8] [user: ops] 用户输入: 看下nginx状态 2025-06-12 14:23:02 [session: a3f8] intent: check_service, service=nginx, action=status 2025-06-12 14:23:02 [session: a3f8] 安全检查: low，无需人工确认 2025-06-12 14:23:02 [session: a3f8] 执行命令: systemctl status nginx --no-pager 2025-06-12 14:23:03 [session: a3f8] 执行成功，耗时0.8s 2025-06-12 14:23:03 [session: a3f8] 结果摘要返回，会话完成

8. 常见问题与排查技巧实录

这里整理项目实施过程中遇到的一批典型问题，每个都有实践经验做支撑，照着排查会少走很多弯路。

排查思路里最核心的一条经验：问题先分“连接层”和“业务层”。SSH 断线、超时、连接被拒这类是连接层问题，直接抓 Paramiko 的异常类型和错误码；命令执行返回错误、退出码非 0、输出内容异常，则是业务层问题，重点看 stdout 和 stderr 的内容。这两层混在一起查会浪费很多时间。

9. 经验心得与迭代计划：自动化程度越高，安全边界要越清晰

整个项目从立项到基本可用，前后两三周，中途推倒重来了两次。一次是对话解析部分，最初尝试用纯 Prompt 让模型自由输出命令，结果可靠性太差，后来换成 Function Calling + 参数校验的组合拳才稳定下来。另一次就是重启断线问题，最初没有做守候机制，导致系统沦为一个“只能看不能用”的半成品。

在实际使用中，我的体会是：运维 Agent 的价值不在于“能执行多复杂的命令”，而在于“在什么范围内能放心地让它自己跑”。现在的系统基本能做到简单巡检全自动、中风险操作提示确认、高风险操作强制确认、断线操作自动守候复检，这个分级控制的思路，比一股脑地把所有操作都推给人工，或者全盘不管直接自动跑，都更符合真实运维环境的需求。

最近在规划的一个迭代方向是“操作影响面分析”。比如执行rm -rf /data/logs/之前，自动扫描该目录下有多少个活跃进程在占用文件句柄，把这些影响面信息一并推到人工确认环节中，让确认的人做更充分的判断。这个功能已经在开发计划里了，研究清楚了再单独写一篇。