摘要: 大模型时代,传统的“脚本自动化”正在向“认知自动化”转变。本文将教你利用 Python、LangChain 和 OpenAI API,构建一个能“看懂”报错、自动查询知识库并给出解决建议的智能运维 Agent。


前言

你是否遇到过这种情况:报警响了,你得去 Wiki 搜文档,去 StackOverflow 搜报错,然后去服务器敲命令查状态。 如果有一个机器人,你发给它报错日志,它能自动:

  1. 分析错误原因。

  2. 自动检索公司内部的运维文档(RAG)。

  3. 自动执行 Python 脚本查询服务器状态。

  4. 给出修复建议。

这就是 AI Native 的运维方式。今天我们用 LangChain 实现一个雏形。

1. 核心技术栈

  • Python 3.10+

  • LangChain: 大模型应用编排框架。

  • OpenAI API (或 Ollama 本地模型): 提供推理能力。

  • FAISS: 向量数据库,用于存储运维文档。

2. 场景一:构建运维知识库(RAG)

当报错发生时,我们希望 AI 基于公司的内部文档(比如《MySQL 常见故障处理手册》)来回答,而不是瞎编。

import os
from langchain_community.document_loaders import TextLoader
from langchain_text_splitters import CharacterTextSplitter
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_community.vectorstores import FAISS
from langchain.chains import RetrievalQA

# 1. 设置 API Key
os.environ["OPENAI_API_KEY"] = "sk-xxxxxx" # 替换为你的 Key

# 2. 模拟加载内部运维文档
# 假设 runbook.txt 内容: "当遇到错误代码 1045 时,通常是 MySQL 密码过期,请执行 update user set..."
loader = TextLoader("runbook.txt", encoding='utf-8')
documents = loader.load()

# 3. 文本切分与向量化存储
text_splitter = CharacterTextSplitter(chunk_size=1000, chunk_overlap=0)
texts = text_splitter.split_documents(documents)
embeddings = OpenAIEmbeddings()
db = FAISS.from_documents(texts, embeddings)

# 4. 创建检索问答链
qa = RetrievalQA.from_chain_type(
    llm=ChatOpenAI(temperature=0), 
    chain_type="stuff", 
    retriever=db.as_retriever()
)

# 5. 测试提问
query = "生产环境报 Error 1045 Access denied 怎么处理?"
print(qa.run(query))

效果:AI 会精准地根据你的 runbook.txt 内容回答,而不是给出通用的互联网建议。

3. 场景二:赋予 AI 执行命令的能力(Agent)

只回答不够,我们希望 AI 能帮我们查一下服务器负载。这就需要 ToolsAgents

from langchain.agents import load_tools, initialize_agent, AgentType
from langchain.tools import tool

# 定义一个自定义工具:查询服务器状态
@tool
def check_server_health(server_ip: str) -> str:
    """当用户询问服务器健康状态或负载时使用此工具。输入必须是IP地址。"""
    # 这里可以是真实的 paramiko 远程执行代码
    # 模拟返回
    if server_ip == "192.168.1.10":
        return "CPU Load: 95% (High), Memory: 40%"
    else:
        return "Connection Timeout"

# 初始化 LLM
llm = ChatOpenAI(temperature=0)

# 加载工具集
tools = [check_server_health]

# 初始化 Agent
agent = initialize_agent(
    tools, 
    llm, 
    agent=AgentType.OPENAI_FUNCTIONS, 
    verbose=True # 打印思考过程
)

# 让 Agent 干活
response = agent.run("请帮我检查一下 192.168.1.10 这台机器的状态,如果负载高,告诉我原因。")
print(response)

AI 的思考过程 (Verbose Output)

  1. Thought: 用户让我查 IP 192.168.1.10。

  2. Action: 调用 check_server_health,参数 server_ip='192.168.1.10'

  3. Observation: 收到返回 "CPU Load: 95% (High)..."。

  4. Final Answer: 192.168.1.10 的 CPU 负载高达 95%,属于异常状态,建议立即排查高占用进程。

4. 安全警示

在运维中使用 Agent 必须极其谨慎:

  • Read-Only 原则:初期只给 AI 赋予“查询类”工具(Get/Check),严禁赋予 Delete/Restart 权限。

  • Human-in-the-loop:关键操作必须经过人工点击“确认”才能执行。

总结

AI+Python 正在重塑运维的边界。通过 RAG,我们将经验数字化;通过 Agent,我们将操作智能化。这不仅仅是效率的提升,更是运维工程师从“救火队员”转型为“系统架构师”的最佳契机。

Logo

Agent 垂直技术社区,欢迎活跃、内容共建。

更多推荐