Linux 运维如何转型？掌握 AIOps 大模型，开启云计算智能运维新方向

传统运维技能（保留，但不再是核心价值）
├── Linux系统管理（内核参数、性能调优）  → 转向“定义SLO/SLI”
├── 网络基础（TCP/IP、DNS、负载均衡）   → 转向“混沌工程实验设计”
├── Shell/Python脚本编写               → 转向“运维智能体编排”
└── 监控系统搭建（Prometheus/Zabbix）   → 转向“AIOps平台选型与集成”

新增智能运维技能（核心价值提升点）
├── AI/ML基础：理解异常检测、时序预测、NLP原理
├── 大模型应用：Prompt Engineering、RAG、Agent开发（结合运维场景）
├── 数据分析能力：SQL、日志分析、因果推断
├── 平台工程：将AI能力集成到可观测性平台
└── 成本意识：FinOps + 模型调用成本优化

# 使用Facebook Prophet进行CPU使用率异常检测
from prophet import Prophet
import pandas as pd

# 读取过去30天的CPU数据
df = pd.read_csv('cpu_usage.csv')
df.columns = ['ds', 'y']  # ds: 时间戳, y: 指标值

# 训练模型
model = Prophet()
model.fit(df)

# 预测未来2小时 + 检测当前点是否为异常
future = model.make_future_dataframe(periods=120, freq='min')
forecast = model.predict(future)

# 当前真实值与预测区间比较
current_value = get_current_cpu()
upper_bound = forecast.iloc[-1]['yhat_upper']
if current_value > upper_bound * 1.2:
    alert("CPU使用率异常偏离预测区间")

# 告警关联伪代码
def correlate_alerts(alerts: List[Alert], topology: Graph):
    # 1. 提取时间窗口内所有告警
    window = alerts.filter(lambda a: a.time > now - 5min)
    
    # 2. 按服务拓扑层级分组
    groups = defaultdict(list)
    for alert in window:
        service = alert.service
        # 在依赖图中向上溯源
        upstreams = topology.get_upstream(service)
        groups[upstreams[-1]].append(alert)  # 最上游作为根因候选
    
    # 3. 排序：选择出现时间最早、依赖最上游的作为根因
    root_cause = min(groups.keys(), key=lambda s: min(s.alerts, key=lambda a: a.time))
    return root_cause

topk(5, sum(rate(http_requests_total[1h])) by (service))

输入： "查一下过去1小时哪个服务请求量最大"
输出： PromQL自动生成 + 结果可视化

def nl_to_promql(query: str) -> str:
    # 加载指标元数据字典
    metadata = load_metric_metadata()  # 包含所有metric名称、标签
    
    prompt = f"""
    你是一个PromQL专家。现有指标元数据如下：
    {metadata}
    
    请将用户的自然语言查询转换为PromQL表达式：
    用户查询："{query}"
    
    只输出PromQL表达式，不要解释。
    """
    
    response = llm.invoke(prompt)
    return response

# 根因定位 + LLM报告生成
def generate_rca_report(incident_data):
    # 1. 图遍历获取候选路径
    paths = knowledge_graph.find_paths(
        from_node=incident_data['impacted_service'],
        relation='DEPENDS_ON'
    )
    
    # 2. 基于时间戳和变化事件（变更记录）排序
    candidates = rank_by_timeline(paths, incident_data['start_time'])
    
    # 3. LLM生成自然语言报告
    prompt = f"""
    根据以下故障排查路径，生成一份面向运维团队的根因分析报告：
    {candidates}
    
    报告结构要求：
    1. 故障现象
    2. 排查过程
    3. 根因结论
    4. 修复建议
    5. 预防措施
    """
    
    return llm.invoke(prompt)

# 预测Pod内存耗尽风险
from sklearn.ensemble import RandomForestRegressor

def predict_oom_risk(pod_name: str) -> float:
    # 特征：过去7天的内存使用趋势、部署新版本次数、流量增长
    features = extract_features(pod_name)
    
    # 加载预先训练的模型（基于历史OOM事件）
    model = joblib.load('oom_predictor.pkl')
    risk_score = model.predict_proba([features])[0][1]  # 概率值
    
    if risk_score > 0.7:
        alert(f"Pod {pod_name} 在24小时内OOM风险: {risk_score:.0%}")
    return risk_score

# 运维知识库构建
import chromadb
from chromadb.utils import embedding_functions

class OpsKnowledgeBase:
    def __init__(self):
        self.client = chromadb.PersistentClient(path="./ops_kb")
        self.embed_fn = embedding_functions.SentenceTransformerEmbeddingFunction(
            model_name="BAAI/bge-large-zh-v1.5"  # 中文运维场景效果好
        )
        self.collection = self.client.get_or_create_collection(
            name="ops_docs",
            embedding_function=self.embed_fn
        )
    
    def index_incident(self, incident: dict):
        """索引历史故障"""
        doc = f"""
        故障标题：{incident['title']}
        现象：{incident['symptom']}
        根因：{incident['root_cause']}
        解决方案：{incident['solution']}
        受影响服务：{incident['services']}
        """
        self.collection.add(
            documents=[doc],
            metadatas=[{"service": incident['services'], "date": incident['date']}],
            ids=[incident['id']]
        )
    
    def retrieve_similar_incidents(self, symptom: str, top_k=3):
        """根据症状检索相似历史故障"""
        results = self.collection.query(
            query_texts=[symptom],
            n_results=top_k
        )
        return results['documents'][0]

# 运维智能体核心逻辑
class OpsAgent:
    def __init__(self, llm, tools, knowledge_base):
        self.llm = llm
        self.tools = tools  # 工具集: query_logs, restart_pod, scale_deployment, etc.
        self.kb = knowledge_base
        
    def handle_incident(self, user_report: str):
        """处理运维工单"""
        # Step 1: 检索历史相似故障
        similar = self.kb.retrieve_similar_incidents(user_report)
        
        # Step 2: 构建系统提示词
        system_prompt = f"""
        你是一位资深SRE工程师。请根据用户描述，执行以下流程：
        1. 首先判断是否有历史相似案例，参考相似案例中的解决方案
        2. 如果需要查看系统状态，使用tools查询
        3. 给出诊断结论和修复建议
        
        历史相似案例参考：
        {similar}
        
        可用工具：
        - query_logs(service, time_range): 查询服务日志
        - check_pod_status(service): 查看Pod状态
        - query_metrics(metric_name, service): 查询指标
        - restart_pod(pod_name): 重启Pod（需审批）
        - scale_deployment(deployment, replicas): 扩容（需审批）
        """
        
        # Step 3: 多轮对话，直到问题解决或需要人工介入
        messages = [
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": user_report}
        ]
        
        for _ in range(5):  # 最多5轮迭代
            response = self.llm.invoke(messages)
            
            if "需要人工介入" in response:
                return self.escalate_to_human(messages, response)
            
            if self.has_tool_call(response):
                tool_result = self.execute_tool(response)
                messages.append({"role": "tool", "content": tool_result})
                continue
            
            # 无工具调用，输出最终方案
            return self.generate_final_report(messages, response)
    
    def execute_tool(self, tool_call):
        tool_name = tool_call['name']
        params = tool_call['parameters']
        
        if tool_name == "restart_pod":
            # 高危操作：需要二次确认
            if not self.confirm_with_user(f"确认重启Pod {params['pod_name']}?"):
                return "操作已取消，等待用户确认"
        
        # 执行真实操作
        result = self.tools[tool_name](**params)
        
        # 记录审计日志
        audit_log.info({
            "tool": tool_name,
            "params": params,
            "result": result,
            "user": get_current_user(),
            "timestamp": datetime.now()
        })
        return result

# 飞书机器人消息处理
@app.route('/feishu/webhook', methods=['POST'])
def handle_feishu_message():
    data = request.json
    user_input = data['event']['message']['content']
    
    # 调用运维智能体
    result = ops_agent.handle_incident(user_input)
    
    # 返回给飞书
    return {
        "msg_type": "post",
        "content": {
            "post": {
                "zh_cn": {
                    "title": "🤖 运维助手回复",
                    "content": [
                        [{"tag": "text", "text": result['answer']}],
                        [{"tag": "text", "text": f"\n📊 置信度: {result['confidence']}"}]
                    ]
                }
            }
        }
    }