在生产环境TKE上部署AIOps智能体：基于开源LLM+RAG的故障根因定位实战

import os
from tencentcloud.common import credential
from tencentcloud.cls.v20201016 import cls_client, models
import json
from datetime import datetime, timedelta

def fetch_cls_node_logs(node_ip, minutes=15):
    """
    从腾讯云日志服务（CLS）拉取指定节点的关键错误日志
    """
    cred = credential.Credential(os.getenv("TENCENT_SECRET_ID"), os.getenv("TENCENT_SECRET_KEY"))
    client = cls_client.ClsClient(cred, "ap-guangzhou")
    
    # 构造检索请求（假设日志主题已提前创建）
    req = models.SearchLogRequest()
    req.TopicId = "your-topic-id"  # 替换为您的CLS Topic ID
    end_time = datetime.now()
    start_time = end_time - timedelta(minutes=minutes)
    req.From = int(start_time.timestamp() * 1000)
    req.To = int(end_time.timestamp() * 1000)
    req.Query = f"node:{node_ip} AND (ERROR OR Failed OR timeout)"  # CLS检索语法
    req.Limit = 50
    req.Sort = "desc"
    
    resp = client.SearchLog(req)
    logs = []
    for item in resp.Results:
        logs.append(json.loads(item.LogJson).get("__CONTENT__", ""))
    return "\n".join(logs[:20])  # 只取前20条关键日志，防止Token溢出

from langchain_community.document_loaders import TextLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.embeddings import HuggingFaceEmbeddings
from langchain_community.vectorstores import Chroma

# 加载内部运维WIKI（包含已解决的故障案例）
loader = TextLoader("./ops_knowledge.txt")  # 内部文档
docs = loader.load()
text_splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=50)
splits = text_splitter.split_documents(docs)

# 使用BGE嵌入模型（国产高性能）
embeddings = HuggingFaceEmbeddings(model_name="BAAI/bge-small-zh")
vectorstore = Chroma.from_documents(documents=splits, embedding=embeddings, persist_directory="./chroma_db")
retriever = vectorstore.as_retriever(search_kwargs={"k": 3})

from langchain_community.llms import Ollama
from langchain_core.prompts import ChatPromptTemplate

llm = Ollama(model="qwen2.5:7b", base_url="http://localhost:11434")

template = """
你是一位资深的Linux内核与K8s排障专家。请严格遵循以下步骤思考（Chain-of-Thought）：

【当前节点状态】
节点IP: {node_ip}
K8s状态: {node_status}
最近15分钟CLS关键日志摘要:
{cls_logs}

【历史相似故障参考】
{retrieved_docs}

任务要求：
1. 分析根本原因（Root Cause）。
2. 输出3条紧急恢复的Linux命令（需区分危险等级）。
3. 输出必须为**严格的JSON格式**，不得包含任何markdown或额外解释。

JSON Schema示例：
{
  "root_cause": "简洁准确的根因",
  "commands": [
    {"cmd": "具体的shell/kubectl命令", "risk": "高/中/低", "expect": "执行后预期效果"}
  ],
  "long_term_advice": "长期规避方案"
}
"""

prompt = ChatPromptTemplate.from_template(template)

import re
import subprocess

ALLOWED_COMMANDS = ["kubectl drain", "kubectl uncordon", "systemctl restart", "df -h", "top -bn1"]

def execute_commands_with_audit(json_output):
    cmds = json_output.get("commands", [])
    for item in cmds:
        cmd_str = item.get("cmd")
        # 安全检查：仅允许白名单开头
        if not any(cmd_str.startswith(allowed) for allowed in ALLOWED_COMMANDS):
            print(f"❌ 危险命令被拦截: {cmd_str}，需要人工介入")
            continue
        # 模拟执行（实际生产建议输出为Ansible剧本或工单）
        print(f"✅ 执行命令: {cmd_str}")
        # subprocess.run(cmd_str, shell=True, check=False)