我这人比较懒尤其是碰上重复性的运维操作能写脚本绝不动手。但脚本有个天生的短板它只是个执行器没有判断力。重启个服务、删个日志、改个配置这些操作本身不难难的是判断“现在能不能做”“做完之后对不对”。所以当我有机会做一个对话式 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 的大脑、手脚和刹车整个系统拆成四个模块各司其职。我画不出什么复杂的架构图但脑子里对它的定位很清楚模块作用技术要点会话管理层承载多轮对话上下文LangChain 的 ConversationBufferWindowMemory控制上下文窗口任务编排层将用户意图拆解为可执行步骤大模型 Function Calling 状态机状态流转命令执行层实际执行系统命令Paramiko SSH 连接池命令白名单校验安全控制层高危操作识别与预警正则 敏感词表 人工确认回调用户的输入到达后先进入会话管理模块。这一步参考了银行客服机器人那套记忆机制只保留最近几轮的关键上下文。因为运维指令往往是上下文相关的比如用户先问“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或-rdd命令本身不算高危但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\s6\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_commandcommand, timeout, need_confirm执行单条命令check_serviceservice_name, action服务类操作的封装analyze_loglog_path, pattern, lines日志分析专用入口send_alertmessage, level异常信息主动上报大模型返回调用意图后系统会做一次参数合法性校验。比如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.9Python 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_size5): self.host host self.user user self.key_path key_path self.pool ThreadPoolExecutor(max_workerspool_size) self._conn self._create_connection() def _create_connection(self): client paramiko.SSHClient() client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) client.connect( hostnameself.host, usernameself.user, key_filenameself.key_path, timeout15, banner_timeout30, auth_timeout30 ) return client def execute(self, command, timeout30): stdin, stdout, stderr self._conn.exec_command(command, timeouttimeout) exit_code stdout.channel.recv_exit_status() output stdout.read().decode(utf-8, errorsreplace) error stderr.read().decode(utf-8, errorsreplace) 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_pathrules/risk_rules.json): with open(rules_path, r, encodingutf-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, historyget_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/confirm7.5 运行日志观察系统运行时的日志设计也值得一说。日志不仅记录命令和输出还记录完整的会话流转过程用户身份、意图解析结果、命令生成结果、安全过滤结果、人工确认状态、执行耗时、守候情况。这样操作审计就有完整的链路。比如这样一段输出2025-06-12 14:23:01 [session: a3f8] [user: ops] 用户输入: 看下nginx状态 2025-06-12 14:23:02 [session: a3f8] intent: check_service, servicenginx, actionstatus 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. 常见问题与排查技巧实录这里整理项目实施过程中遇到的一批典型问题每个都有实践经验做支撑照着排查会少走很多弯路。问题现象根因定位解决方案Paramiko 连接偶发超时服务器 MaxSessions 限制SSH 连接数超限连接池复用连接限制每台最大会话数大模型解析指令出现幻觉命令模型对上下文理解偏差生成了不存在的参数增加 Function Calling 结构化约束并在执行前做参数白名单校验重启命令执行后Agent端一直显示“执行中”等待 SSH 返回超时但服务已重启实现守候复检机制预判断线操作并提前切换状态多条命令在同一 SSH 通道执行时输出串扰Paramiko 的 Channel 并发非线程安全同一会话串行执行仅会话间并行特殊字符触发命令注入误判正则误伤比如python3被bash规则拦截将命令上下文和参数进行了更细粒度的拆分模型反馈速度太慢平均 3-5 秒模型推理在 CPU 环境执行增加并发请求缓冲队列将模型服务部署到单张显卡的 GPU 环境的确认操作超时自动取消但用户没收到原因确认 Token 有效期 5 分钟超过即失效强化消息提示注明确认超时自动取消逻辑守候复检总是重连成功但服务未就绪服务启动耗时较长重连成功不代表服务可用复检命令分两步先判断进程存在再判断端口监听或健康接口响应执行df -h偶发卡顿 30 秒NFS 挂载点挂起导致 df 阻塞命令拼接--direct或用/bin/df -x nfs排除网络文件系统排查思路里最核心的一条经验问题先分“连接层”和“业务层”。SSH 断线、超时、连接被拒这类是连接层问题直接抓 Paramiko 的异常类型和错误码命令执行返回错误、退出码非 0、输出内容异常则是业务层问题重点看 stdout 和 stderr 的内容。这两层混在一起查会浪费很多时间。9. 经验心得与迭代计划自动化程度越高安全边界要越清晰整个项目从立项到基本可用前后两三周中途推倒重来了两次。一次是对话解析部分最初尝试用纯 Prompt 让模型自由输出命令结果可靠性太差后来换成 Function Calling 参数校验的组合拳才稳定下来。另一次就是重启断线问题最初没有做守候机制导致系统沦为一个“只能看不能用”的半成品。在实际使用中我的体会是运维 Agent 的价值不在于“能执行多复杂的命令”而在于“在什么范围内能放心地让它自己跑”。现在的系统基本能做到简单巡检全自动、中风险操作提示确认、高风险操作强制确认、断线操作自动守候复检这个分级控制的思路比一股脑地把所有操作都推给人工或者全盘不管直接自动跑都更符合真实运维环境的需求。最近在规划的一个迭代方向是“操作影响面分析”。比如执行rm -rf /data/logs/之前自动扫描该目录下有多少个活跃进程在占用文件句柄把这些影响面信息一并推到人工确认环节中让确认的人做更充分的判断。这个功能已经在开发计划里了研究清楚了再单独写一篇。