运维AI助手
一个成熟应用的第一步是目录结构清晰，各层职责分明。推荐采用 后端（FastAPI）+ CLI 客户端 + 共享模型 的结构：
text
ops-ai/
├── backend/
│ ├── app/
│ │ ├── main.py # FastAPI 入口
│ │ ├── config.py # 环境变量、模型配置
│ │ ├── api/
│ │ │ └── v1/
│ │ │ └── commands.py # 命令生成/确认/反馈端点
│ │ ├── core/
│ │ │ ├── rag.py # RAG 检索服务
│ │ │ ├── llm.py # LLM 调用封装（含结构化输出）
│ │ │ ├── safety.py # 安全规则引擎
│ │ │ └── env.py # 环境指纹采集
│ │ ├── models/
│ │ │ └── schemas.py # Pydantic 请求/响应模型
│ │ └── db/
│ │ ├── audit.py # 审计日志写入
│ │ └── feedback.py # 用户反馈存储
│ ├── requirements.txt
│ └── Dockerfile
├── cli/
│ ├── opsai/
│ │ ├── __init__.py
│ │ ├── main.py # CLI 入口（Typer/Click）
│ │ ├── client.py # HTTP 客户端封装
│ │ └── env_fingerprint.py # 本地环境指纹采集
│ └── pyproject.toml
├── knowledge/
│ ├── man_pages/ # 预处理后的 man 页面 / --help 文本
│ └── ingest.py # 知识库导入脚本
├── docker-compose.yml
└── README.md
CLI 作为入口，后端作为“命令生成与安全审计”的服务端，知识库离线构建、在线检索。这样 CLI 可以轻量分发，后端可以独立升级模型和规则。
后端核心代码：FastAPI + RAG + 安全引擎
数据模型（schemas.py）
先定义好“生成结果”的结构，后续所有模块围绕它工作：
from pydantic import BaseModel, Field
from typing import Literal, Optional
class CommandRequest(BaseModel):
query: str = Field(..., description="用户的自然语言描述或报错日志")
env: dict = Field(default_factory=dict, description="环境指纹，如 os, version")
session_id: Optional[str] = None
class CommandCandidate(BaseModel):
command: str
explanation: str
risk_level: Literal["readonly", "change", "high_risk"]
rollback: Optional[str] = None
retrieved_sources: list[str] = []
class CommandResponse(BaseModel):
candidates: list[CommandCandidate]
blocked: bool = False
block_reason: Optional[str] = None
audit_id: str
risk_level 是安全层分流的依据：readonly 直接返回，change 需要用户确认，high_risk 直接拦截并给出替代建议。
安全规则引擎（safety.py）
这是运维 AI 区别于通用 ChatBot 的核心模块，必须独立于 LLM 运行。即使模型输出了“看起来合理”的命令，也要经过这一层：
import re
from dataclasses import dataclass
@dataclass
class SafetyRule:
pattern: str
level: str # "block" 或 "warn"
message: str
replacement: str = ""
DEFAULT_RULES = [
SafetyRule(
pattern=r"rm\s+-rf\s+/(\s|$)",
level="block",
message="禁止递归删除根目录。",
replacement="请明确指定要删除的子目录路径。",
),
SafetyRule(
pattern=r"dd\s+.*of=/dev/(sd|nvme|vd)",
level="block",
message="禁止直接写入磁盘设备。",
),
SafetyRule(
pattern=r"mkfs(\.\w+)?\s+/dev/",
level="block",
message="禁止对块设备执行 mkfs。",
),
SafetyRule(
pattern=r"iptables\s+-F",
level="warn",
message="清空防火墙规则可能导致连接中断，确认前请先保存规则。",
),
SafetyRule(
pattern=r"kill\s+-9",
level="warn",
message="SIGKILL 无法被进程捕获，可能造成数据丢失。",
),
SafetyRule(
pattern=r"(shutdown|reboot|poweroff)",
level="warn",
message="系统关机/重启命令，请确认影响范围。",
),
]
def evaluate(command: str) -> dict:
for rule in DEFAULT_RULES:
if re.search(rule.pattern, command, re.IGNORECASE):
return {
"blocked": rule.level == "block",
"level": rule.level,
"reason": rule.message,
"replacement": rule.replacement,
}
return {"blocked": False, "level": "readonly", "reason": "", "replacement": ""}
规则库应该配置化（YAML/数据库），方便后续扩展而不用改代码。高风险命令必须在返回给用户之前被拦截，而不是依赖模型自我审查。
LLM 调用与结构化输出（llm.py）
不要直接让模型返回自由文本。要求模型输出 JSON，并且强制包含 risk_level 和 rollback：
import json
from openai import OpenAI
from backend.app.models.schemas import CommandCandidate
SYSTEM_PROMPT = """你是一个运维命令生成助手。你的输出必须是 JSON 数组，每个元素包含：
- command: 完整的 shell 命令
- explanation: 一句话解释命令的作用
- risk_level: "readonly" | "change" | "high_risk"
- rollback: 如果命令产生变更，提供回滚命令；只读命令可为 null
规则：
1. 只生成当前环境可用的命令（参考用户提供的 env 和检索到的文档）。
2. 如果不确定，risk_level 至少标为 "change"。
3. 不要编造不存在的参数。
"""
def generate_candidates(query: str, context: str, env: dict) -> list[CommandCandidate]:
client = OpenAI()
user_prompt = f"""用户需求：{query}
环境信息：{json.dumps(env, ensure_ascii=False)}
参考文档：
{context}
请生成 1-3 个候选命令。"""
resp = client.chat.completions.create(
model="gpt-4o",
messages=[
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": user_prompt},
],
response_format={"type": "json_object"},
temperature=0.1,
)
raw = json.loads(resp.choices[0].message.content)
items = raw if isinstance(raw, list) else raw.get("commands", [])
return [CommandCandidate(**item) for item in items]
关键点：temperature=0.1 降低随机性；response_format 强制 JSON；系统提示词中明确“不确定就标 change”。
RAG 服务（rag.py）
使用 LangChain + Chroma 做离线知识库检索。knowledge/ingest.py 负责把 man 页面和 --help 输出切块、向量化并写入 Chroma：
# knowledge/ingest.py
from langchain_community.document_loaders import TextLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_chroma import Chroma
import os
def ingest(man_dir: str, persist_dir: str):
docs = []
for fname in os.listdir(man_dir):
if fname.endswith(".txt"):
loader = TextLoader(os.path.join(man_dir, fname), encoding="utf-8")
docs.extend(loader.load())
splitter = RecursiveCharacterTextSplitter(chunk_size=800, chunk_overlap=120)
chunks = splitter.split_documents(docs)
Chroma.from_documents(
chunks,
OpenAIEmbeddings(),
persist_directory=persist_dir,
)
在线检索时，rag.py 暴露一个 retrieve(query, k=5)，返回拼接后的上下文文本和来源列表。RAG 的作用是把模型“锚定”在真实文档上，而不是让它自由发挥。
API 端点（commands.py）
将 RAG → LLM → 安全层串起来：
from fastapi import APIRouter, HTTPException
from backend.app.core import rag, llm, safety
from backend.app.models.schemas import CommandRequest, CommandResponse, CommandCandidate
from backend.app.db import audit
router = APIRouter()
@router.post("/commands/generate", response_model=CommandResponse)
async def generate(req: CommandRequest):
context, sources = rag.retrieve(req.query)
raw_candidates = llm.generate_candidates(req.query, context, req.env)
final: list[CommandCandidate] = []
blocked_any = False
block_reason = None
for c in raw_candidates:
verdict = safety.evaluate(c.command)
if verdict["blocked"]:
blocked_any = True
block_reason = verdict["reason"]
continue
# 用安全层的判定覆盖模型自评，防止模型低估风险
c.risk_level = verdict["level"] if verdict["level"] != "readonly" else c.risk_level
c.retrieved_sources = sources
final.append(c)
audit_id = audit.write(req, final, blocked_any)
return CommandResponse(
candidates=final,
blocked=blocked_any,
block_reason=block_reason,
audit_id=audit_id,
)
这里有一个重要的设计决策：安全层的判定结果覆盖模型的 risk_level。即使模型认为某条 kill -9 是只读的，规则引擎也会把它提升为 warn，触发用户确认。
CLI 客户端：让运维真正用起来
后端 API 准备好后，CLI 是最高效的交付形态。使用 Typer 实现：
# cli/opsai/main.py
import typer
import httpx
from rich.console import Console
from rich.prompt import Confirm
from cli.opsai.env_fingerprint import collect
app = typer.Typer()
console = Console()
API_BASE = "http://127.0.0.1:8000/api/v1"
@app.command()
def ask(query: str, confirm: bool = False):
env = collect()
with httpx.Client(timeout=60) as client:
resp = client.post(f"{API_BASE}/commands/generate", json={
"query": query,
"env": env,
})
data = resp.json()
if data.get("blocked") and not data["candidates"]:
console.print(f"[red]已拦截：[/red]{data['block_reason']}")
raise typer.Exit(1)
for c in data["candidates"]:
console.print(f"\n[bold]命令：[/bold]{c['command']}")
console.print(f"[dim]{c['explanation']}[/dim]")
console.print(f"风险等级：{c['risk_level']}")
if c.get("rollback"):
console.print(f"回滚：{c['rollback']}")
if confirm and data["candidates"]:
cmd = data["candidates"][0]["command"]
if data["candidates"][0]["risk_level"] != "readonly":
if not Confirm.ask(f"确认执行 `{cmd}`？"):
raise typer.Exit()
import subprocess
subprocess.run(cmd, shell=True)
if __name__ == "__main__":
app()
env_fingerprint.py 采集 platform.system()、platform.version()、当前用户、是否 root 等信息，作为上下文发给后端。CLI 只负责“问”和“展示/执行”，所有生成和安全逻辑集中在后端，便于统一升级。
部署与“成熟”所需的工程配置
docker-compose.yml 把后端和向量库编排在一起，开箱即用：
services:
api:
build: ./backend
ports:
- "8000:8000"
environment:
- OPENAI_API_KEY=${OPENAI_API_KEY}
- CHROMA_PERSIST_DIR=/data/chroma
volumes:
- chroma_data:/data/chroma
chroma:
image: chromadb/chroma:latest
ports:
- "8001:8000"
volumes:
- chroma_data:/chroma/chroma
volumes:
chroma_data:
backend/Dockerfile：
FROM python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]
审计日志建议写入 SQLite（轻量）或 PostgreSQL（多人使用），记录 audit_id、原始 query、候选命令、用户是否采纳、是否被拦截。这是后续做 Bad Case 分析和模型迭代的数据基础。
让应用真正“成熟”的三个关键点
上面的代码是一个可运行的起点，但要达到生产可用，需要在三个地方做持续加固：
第一，规则引擎的覆盖率和可维护性。 把 safety.py 中的规则迁移到 YAML 或数据库，增加版本管理和生效范围（比如区分“生产环境规则”和“测试环境规则”）。可以参考 shellfirm 的规则分类思路，但必须自己维护一套面向你团队环境的规则集。
第二，RAG 知识库的质量决定生成准确性的上限。 只导入 man 页面是不够的。你还需要导入：不同 OS 发行版的命令差异表、内部 SOP 脚本、常见故障的“现象→诊断命令→修复命令”案例库。ingest.py 应该支持按来源打标签，检索时根据 env 过滤（例如 Ubuntu 环境只检索 Ubuntu 相关的文档块）。
第三，反馈闭环。 CLI 的 ask 命令可以增加一个 --feedback 参数，把用户对候选命令的评价（采纳/修改/拒绝）回传到后端。后端定期导出这些数据，分析被拒绝或修改的命令，补充到规则库或 RAG 知识库中。没有反馈数据，模型和规则都不会自己变好。
启动顺序
# 1. 准备知识库
python knowledge/ingest.py --man-dir ./knowledge/man_pages --persist-dir ./chroma
# 2. 启动后端
docker-compose up -d
# 3. 安装并运行 CLI
pip install -e cli/
opsai ask "磁盘满了，帮我看看哪个目录占用最大"
这套代码骨架的价值在于：它已经把“生成—校验—拦截—审计”的主链路完整跑通了。你可以先替换成自己的模型 API key 和知识库，跑起来看到效果，再逐步加固规则、扩充文档、接入反馈数据。成熟度是在这个骨架上迭代出来的，而不是一开始就设计出来的。