对话式Linux运维Agent：高危操作强制确认与断线自动守候复检实践

做运维的应该都经历过这种时刻：凌晨三点，业务告警，人还没完全清醒，手已经放在键盘上，对着屏幕敲命令。一边怕敲错，一边又必须快。我做的这个对话式 Linux 运维 Agent，就是冲着这个场景去的——用聊天的方式下发运维指令，说“把 nginx 重启一下”它就去执行，但凡是高危操作，系统会强制要求人工确认；重启过程中 SSH 断了，它会自动守候、重连、复检，直到确认服务恢复。

这个项目不是花架子，已经在我们内部环境跑了一段时间，处理过配置变更、服务重启、日志排查、批量巡检这些日常活儿。我尽量把当初的设计思路、踩过的坑、关键代码都写出来。适合正在做运维自动化的同学参考，也适合开发兼运维、半夜要值班的人拿来当灵感。

1. 做一个运维 Agent，先解决哪个痛点

1.1 我最初遇到的三类麻烦场景

最早想做这个项目，是因为手头重复性运维工作太多了。典型的三类场景，每一类都让人头疼。

第一类是操作链路长。排查一次 nginx 404，要先看端口监听、再看进程状态、翻错误日志、定位配置问题、改配置、reload，最后还得验证是否生效。这一串命令平时打熟了不觉得难，但半夜迷迷糊糊的时候，漏掉一步就要重来。把这一长串操作浓缩成一句“帮我看下 nginx 为什么 404”，体验完全不一样。

第二类是高危操作的心理压力大。重启数据库、清理磁盘、修改内核参数，敲下去之前你不知道后果有多重，但业务又不等人。数据库重启类操作尤其明显，执行前要确认连接池、确认主从状态、确认有没有长事务，漏一项就是事故。这活儿让人心累，因为要求你时刻保持“全神贯注”状态。

第三类是重启之后的无人区。服务重启、服务器重启期间 SSH 会断开，你只能人肉等待，反复 ping、反复重连、反复看启动日志。这个过程非常磨人，有时候服务起来了，但业务端口还没就绪；有时候进程在，但已经陷入死锁。等发现问题时，已经过去了十几分钟。

这三类场景指向同一个需求：要有一个能听人话、能自动编排、能守住安全底线、还能在断线后继续把活干完的运维助手。这就是我决定做这个对话式 Agent 的直接原因。

1.2 我给这个项目划定的能力边界

做这种 Agent，最大的坑不是做不出来，而是忍不住让它“什么都干”。我一开始也追求全能，后来差点出过几次误操作，才老老实实地把边界画清楚。

目前能力清单分三档：

• 只读操作：查状态、查日志、查端口、统计进程、磁盘水位，默认放行，属于最高频也最安全的一类。
• 常规操作：启停服务、reload 配置、清理非根目录下的缓存文件，需要权限校验加操作审计，执行前展示真实命令。
• 高危操作：删除文件或目录、格式化磁盘、修改网络配置、重启数据库或核心业务、调整内核参数，强制人工确认，确认不通过绝不执行。

为什么这样划？因为 Agent 的可信度，不取决于它能执行多少命令，而取决于它拒绝了多少危险操作。只读操作可以放开让模型折腾，常规操作要留痕，高危操作必须上强制确认机制。后面整套安全设计，都是围绕这个边界展开的。

2. 对话式交互与高危操作强制确认，安全骨架怎么搭

2.1 对话不是聊天：从自然语言到受控指令

很多人以为“对话式 Agent”就是把用户说的话直接丢给 shell 执行，这是大忌。我的设计里，整个链路是这样走的：

用户输入 → 意图识别 → 结构化指令 → 权限与危险等级校验 → 确认或直接执行 → 结果反馈

其中 LLM 只干“翻译官”的活：把自然语言转成结构化的工具调用参数。真正执行动作的是独立工具函数，不是模型本身。这样做的好处是，就算模型抽风乱说，工具层还能兜底拦截，不会直接产生破坏性操作。

举例来说，用户说“帮我把 /data/logs 下超过 500M 的日志清一下”。模型解析后得到的不是一句 shell 命令，而是一个 JSON 结构：

{ "tool": "clean_logs", "path": "/data/logs", "size_threshold": "500M", "delete": false }

注意这里的delete字段默认是 false。也就是说，默认走试删模式，只列出将要删除的文件，不真的删。只有人工确认后，才把 delete 置为 true。这个设计让“清理日志”这类常见但容易闯祸的操作，多了一层缓冲。

我实际测试下来，让模型直接生成 shell 命令再执行的方案，误判率高得吓人。尤其涉及通配符、管道、重定向的时候，模型很容易生成语法对但语义完全错误的命令。改成“工具调用 + 参数填充”之后，模型只需要理解意图并填几个字段，稳定性和安全性都明显提升。

2.2 高危操作强制人工确认：服务端校验，不是前端弹窗

高危操作确认机制是整个项目里我花心思最多的地方。市面上一堆工具也号称有“确认”，实际只是前端弹个窗，懂点技术的人绕过很轻松。我这个方案里的强制确认，是服务端级别的校验，流程如下：

1. 危险等级评估：根据操作类型、命令行模式匹配、路径或对象特征判断是否属于高危。
2. 生成确认请求：返回给用户的信息包括目标主机、操作名称、对象路径、影响范围、以及一次性确认码。
3. 用户回传确认码 + 个人 PIN：确认码默认 30 秒过期，只能用于当前这条已经序列化的操作指令。
4. 服务端二次校验：校验通过后才真正下发执行。

关键点在于：确认码和“当前指令”是绑定的。如果用户在确认期间修改了指令参数，哪怕只改了一个字母，校验也会直接失败。另外确认码不能复用，操作完成后立即失效。这个设计堵住了“换个参数偷跑”“拿着旧码复现”这两个最常见的绕过方式。

代码层面，我把确认服务封装成一个独立模块，核心逻辑大概是这样的：

import os import time import hashlib import secrets class SafeConfirm: def __init__(self, ttl=30): self.ttl = ttl self._store = {} def create_confirmation(self, task_id, operation_snapshot): # 确认码和操作快照绑定，快照是序列化后的完整操作参数 code = secrets.token_hex(4) self._store[task_id] = { "code_hash": self._hash(code), "snapshot_hash": self._hash(operation_snapshot), "expires_at": time.time() + self.ttl } return code def verify(self, task_id, code, operation_snapshot): item = self._store.get(task_id) if not item: return False, "确认请求不存在" if time.time() > item["expires_at"]: del self._store[task_id] return False, "确认码已过期" if item["code_hash"] != self._hash(code): return False, "确认码错误" if item["snapshot_hash"] != self._hash(operation_snapshot): del self._store[task_id] return False, "操作内容已变化，请重新确认" del self._store[task_id] return True, "ok" @staticmethod def _hash(value): return hashlib.sha256(value.encode()).hexdigest()

这段代码看起来很简朴，但它把“确认”这件事的关键语义都锁死了：短时效、一次性、快照绑定。我建议任何做自动化操作平台的人，都参考这个思路做确认机制，不要停留在“二次输入 yes”的水平。

2.3 权限分级与会话管控

只有确认机制还不够，权限分级必须跟上。我的方案里把使用者分成三类角色：

• 只读角色：能查询、能看报告，不能发起任何变更操作。
• 普通运维：能执行常规操作和已确认的高危操作，但高危操作必须由管理员复核。
• 管理员：能发起高危操作、能复核其他普通运维的确认请求、能接管守候超时的任务。

这里有个容易忽略的点：确认人和操作人必须分离。比如我自己发起一个高危操作，不能自己给自己确认通过，必须由另一个有权限的人来确认。这样一来，就算账号被偷或者误操作，也不会一个人完成整个危险动作。

每个会话在建立时就绑定操作者身份，所有操作日志都记录：谁、在什么时间、在哪台机器、执行了什么指令、确认码是否通过、执行结果如何。这些日志在复盘事故时价值极大，我后面会详细讲。

3. 重启断线自动守候复检，关键实现与代码

3.1 什么情况会触发守候机制

守候机制是这个项目区别于普通命令执行器的地方。普通工具执行完命令就结束了，遇上重启这种“执行过程中连接必然断掉”的场景，直接抓瞎。我的方案里有两类触发条件：

第一类是主动触发。Agent 执行了重启服务、重启机器、修改网络配置这类会导致连接断开的操作后，主动进入守候状态，而不是傻等用户下一步指令。

第二类是被动触发。执行常规命令过程中，SSH 连接意外中断，且任务没有收到明确完成标志。这时候也要进入守候，重连后继续探测任务状态。

为此我设计了一个简单的状态机：

pending（待执行） → running（执行中） → connection_lost（连接中断） → waiting_retry（重连等待中，指数退避） → checking（复检中，连续探测） → recovered（恢复完成） 或 lost（异常转人工）

这套状态机是守候机制的地基，每个状态对应固定的处理逻辑和事件钩子。新加一个操作类型时，只需要告诉状态机“这个操作会断连”还是“不会断连”，剩下的流程全部复用。

3.2 断线重连与探活参数设计

断线重连最忌讳“无限重试”。我见过有些脚本用死循环重连，机器起不来就一直刷屏，除了自我安慰没有任何意义。我的设计是：

• 重连间隔采用指数退避：10 秒、20 秒、40 秒、60 秒，最多尝试 6 次。
• 超过尝试阈值后，自动转人工接管，同时保留现场信息。
• 探活维度不只看 SSH 端口，还要组合业务端口、HTTP 健康检查、进程存活多种维度。

“为什么不能无限重试”这个想明白很重要。服务器重启之后如果迟迟连不上，大概率是起不来了，这时候狂刷重连日志只会掩盖问题。正确的做法是尽快告警，让值班的人介入，而不是让机器自己骗自己。

守候复检的核心协程大概是这样的：

import asyncio class Watchdog: def __init__(self, probe, max_retries=6, max_checks=3): self.probe = probe # 探活函数，返回 True/False self.max_retries = max_retries self.max_checks = max_checks # 连续判定成功次数 self.retry_delays = [10, 20, 40, 60, 60, 60] async def run(self, task_ctx): retries = 0 while retries <= self.max_retries: if await self.try_connect(task_ctx): ok = await self.verify_recovered(task_ctx) if ok: return "recovered" if retries < len(self.retry_delays): await asyncio.sleep(self.retry_delays[retries]) retries += 1 return "lost" async def try_connect(self, task_ctx): # 尝试 SSH 连接，并记录连接状态 return await task_ctx.ssh_client.ping() async def verify_recovered(self, task_ctx): # 连续 max_checks 次探测都通过，才算真的恢复 passed = 0 for i in range(self.max_checks): if await self.probe(task_ctx): passed += 1 else: passed = 0 if passed >= self.max_checks: return True await asyncio.sleep(10) return False

这里verify_recovered的逻辑尤其重要：一次探测成功不算数，必须连续三次都通过，中间任何一次失败都会把计数清零。这个设计直接治好了“服务起来又立刻挂掉”的误判问题。

3.3 复检报告：操作前后差异才是关键

守候复检完成后，我不会只给一句“服务恢复了”，而是生成一份对比报告。执行操作之前，Agent 会先记录前置状态快照，包括 CPU、内存、磁盘水位、端口监听列表、关键进程、最后若干行日志。复检通过后自动做 diff，重点展示变化项。

举个实际例子：重启某个 Java 服务后，端口和进程都正常了，但 JVM 内存参数被改掉，堆大小比之前小了，这个信息在普通“端口通了”的判断里是看不到的。有了前后对比，一眼就能发现异常。

如果复检不通过，Agent 会做三件事：抓取现场信息（进程状态、端口、日志片段）、通知对应负责人、暂停守候等待人工接管。我个人的习惯是，现场日志至少抓取最后 200 行，存成独立文件，方便事后排查。

4. 从零实操：技术选型、核心代码与部署配置

4.1 技术选型：为什么是 Python + asyncio

技术栈选择上，我主要考虑了三件事：异步能力、SSH 生态、模型接口接入的便利性。

• 运行时：Python 3.11+，配合 asyncio 做并发控制。守候多个任务时，异步的优势非常明显，不会因为一个任务阻塞其他任务。
• SSH 层：选一个成熟好用的异步 SSH 库，重点看它是否支持连接断开的回调事件。我自己的封装层会把“连接断开”和“命令执行完成”这两个事件明确区分开。
• 对话接入：用 WebSocket 作为交互入口，这样既能接入网页聊天框，也能在后面接 IM 机器人。
• 模型推理：用一个支持结构化输出的模型服务接口，要求返回 JSON 格式，方便直接解析为工具调用参数。
• 数据存储：任务记录、操作日志、确认码状态，直接用 SQLite 起步，量大后再换 PostgreSQL。

这个技术栈的好处是结构简单、调试方便、部署成本低。一台 2C4G 的机器就能跑完整套 Agent 服务，非常适合内部工具场景。

4.2 核心模块划分与目录结构

项目代码我按职责拆成几个模块，每个模块只干一件事：

agent/ ├── llm.py # 意图识别与参数抽取 ├── router.py # 工具路由与参数校验 ├── guard.py # 高危操作检测与人工确认 ├── ssh_client.py # SSH 连接管理与命令执行 ├── watchdog.py # 断线守候与复检 ├── audit.py # 操作日志与审计 └── main.py # 入口与 WebSocket 服务

模块之间单向依赖：main → router → guard/watchdog → ssh_client。llm 模块只负责返回结构化指令，不碰任何执行逻辑。这种分层设计让我后来加新功能非常轻松，比如新增“重启 K8s 节点”操作时，只需要在 router 里注册新工具并告诉 watchdog“该操作会断连”即可。

4.3 关键配置项与部署清单

部署这套 Agent，我用一个 YAML 配置搞定主要参数：

hosts: web-prod-01: ip: 10.10.1.21 role: ops services: [nginx, php-fpm] probe: tcp_port: 80 http_url: "http://127.0.0.1/healthz" rules: dangerous_patterns: - "rm -rf /" - "mkfs" - "dd if=" - "shutdown" - "reboot" watchdog: retry_delays: [10, 20, 40, 60, 60, 60] max_checks: 3 check_interval: 10 confirm: ttl: 30 require_pin: true

这里有个容易踩坑的细节：危险命令模式库不能只匹配完整命令，因为用户可能在命令中间插入路径或参数。我用的方式是同时做“正则匹配”和“关键路径安全校验”。比如检测到rm命令时，会额外校验删除路径是否在/data、/tmp等明确允许的目录内，否则一律按高危处理。

5. 避坑实录：高频问题与排查技巧

5.1 高频问题速查表

我在实际运行中遇到过不少问题，整理成表格给大家参考：

这些问题是迭代过程中一点点踩出来的。最典型的是幂等性那条，守候重连本身没问题，但重连后如果错误地重复执行操作，后果可能比断线本身更严重。所以我强烈建议，所有可能重试的操作都必须有任务幂等设计，这是刚需。

5.2 模型误判的兜底策略

LLM 再怎么调，也难免抽风。我的经验是不要把模型当成可信计算单元，而是把它当做一个“有点笨但大多数时候靠谱”的新手。所以我在 router 层套了三层防护：

第一层是参数强校验。比如路径参数必须在配置的白名单目录内，服务名必须在真实服务列表中，数值参数必须在合理范围内。校验不过直接拒绝，并提示用户“参数不合法，已拦截”。

第二层是意图置信度阈值。模型返回意图时附带置信度，低于 0.8 的，不直接执行，而是先让用户确认“你是想执行这个操作吗”。这个设计能拦截很多模棱两可的说法。

第三层是执行前展示真实命令。不管模型翻译得多好，真正要下发的 shell 命令一定会展示给用户看。用户可以直观地检查命令符不符合预期。有时候模型参数填错了，用户看到命令立刻就能发现问题。

5.3 怎么判断服务“真的起来了”

这是守候复检里最微妙的部分。单纯测 TCP 端口会漏掉很多情况，比如端口监听已经建立，但业务进程还在初始化、数据库连接还没就绪、流量还没切换到新实例。我后来把探活分成四档：

• 第一档：SSH 22 端口通，说明机器层面活过来了。
• 第二档：业务 TCP 端口通，说明监听已经建立。
• 第三档：HTTP 健康检查返回 2xx/3xx，说明请求链路基本通。
• 第四档：关键进程存在且启动日志出现“ready/started/healthy”关键字，说明业务真正就绪。

完整恢复判定必须第三档和第四档同时满足。如果只看前两档，很容易把“半恢复”状态误判成“完全恢复”，然后报告一发出，过五分钟又要拉群救火。

我还加了一个技巧：复检通过后，再额外抓取一次当前进程的启动时间。如果这个启动时间比操作时间还要早，说明进程可能是旧进程，根本不是重启后的新进程，这时候要警惕操作是否真正生效。

5.4 两个实用小技巧分享

最后分享两个我至今觉得非常有用的小工具思路。

第一个是“试删模式”。所有删除类操作，不管你多确定，第一步都只做“列出将删除的文件 + 统计大小”，实际删除必须走人工确认。这个习惯帮我们避免过至少三次“路径参数填错导致差点删掉线上数据”的事故。即使不考虑 Agent，我建议所有运维脚本里的删除操作都默认带dry-run。

第二个是“前置快照比对”。执行任何变更前，自动保存一份主机状态快照，内容包括磁盘占用 Top10、端口监听列表、关键进程列表、当前负载。复检报告出来后，直接把前后差异列出来。这个能力排查问题时价值极大，尤其是“操作后磁盘突然满了”“多监听了一个端口”这类隐蔽变化，一眼就能看到。

说实话，做完这个项目，我最深的感受是：运维 Agent 的核心竞争力不是谁的“口才”好，而是谁更让人放心。自然语言交互谁都能做，但高危操作强制人工确认、重启断线自动守候复检这两件事，才是真正值得下功夫的地方。如果你也准备做类似的系统，我的建议是先别急着堆模型能力，把安全边界和异常闭环画清楚，再慢慢加功能。这样即使将来底层模型换了一茬，Agent 的核心骨架依然稳。