用LangChain构建运维Agent：实现自然语言驱动的K8s资源查询与操作自动化

"帮我查一下生产环境 payment-service 的 Pod 状态，把 CPU 超过 80% 的实例列出来"——这句话如果能让 AI Agent 理解并自动执行，运维工程师就不再需要在凌晨三点强撑着敲 kubectl 命令了。

一、Agent设计总览

1.1 为什么选择Agent架构

传统的 ChatOps 本质上是"命令别名映射"，用户仍然需要记住精确的命令格式。而基于大语言模型的 Agent 模式可以实现：

• 意图理解：从自然语言推断用户想执行的操作；
• 工具编排：将多个原子操作组合为复杂任务；
• 上下文记忆：在多轮对话中保持运维上下文；
• 安全护栏：对危险操作进行二次确认和权限校验。

1.2 系统架构

graph TB
    subgraph 用户交互层
        UI[企业微信/钉钉/Slack]
        CLI[命令行工具]
        WebUI[Web Dashboard]
    end
    
    subgraph Agent核心
        LLM[LLM模型<br/>GPT-4/通义千问]
        MEM[上下文记忆<br/>ConversationBufferMemory]
        ROUTER[意图路由器<br/>RouterChain]
    end
    
    subgraph 工具集
        T1[kubectl查询<br/>Pod/Service/Deploy]
        T2[日志检索<br/>Loki/ELK]
        T3[告警查询<br/>Prometheus/PagerDuty]
        T4[变更记录<br/>GitOps/CI/CD]
        T5[操作执行<br/>重启/扩缩容]
    end
    
    subgraph 安全层
        AUTH[RBAC权限校验]
        AUDIT[操作审计日志]
        CONFIRM[危险操作确认]
    end
    
    UI --> Agent核心
    CLI --> Agent核心
    WebUI --> Agent核心
    
    ROUTER --> T1
    ROUTER --> T2
    ROUTER --> T3
    ROUTER --> T4
    ROUTER --> T5
    
    T5 --> AUTH
    AUTH --> AUDIT
    AUDIT --> CONFIRM
    CONFIRM --> K8s[Kubernetes API]
    
    MEM -.-> ROUTER
    LLM -.-> ROUTER
    
    style LLM fill:#E6A23C,color:#fff
    style ROUTER fill:#409EFF,color:#fff
    style AUTH fill:#F56C6C,color:#fff

二、核心工具定义与实现

2.1 K8s查询工具

#!/usr/bin/env python3
"""
LangChain K8s运维Agent - 工具集定义

依赖:
    pip install langchain langchain-openai kubernetes
"""

from langchain.tools import BaseTool, Tool
from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain.memory import ConversationBufferMemory

from kubernetes import client, config
from kubernetes.client.rest import ApiException

from typing import List, Dict, Optional, Any, Type
from pydantic import BaseModel, Field
from datetime import datetime

import json


# ===================== 工具输入模型定义 =====================

class PodQueryInput(BaseModel):
    """Pod查询工具的输入参数"""
    namespace: str = Field(
        default="default",
        description="Kubernetes命名空间，默认值为default"
    )
    service_name: Optional[str] = Field(
        default=None,
        description="服务名称（Deployment名称），用于过滤Pod"
    )
    status_filter: Optional[str] = Field(
        default=None,
        description="Pod状态过滤条件，可选: Running, Pending, Failed, "
                    "CrashLoopBackOff, Unknown"
    )


class LogQueryInput(BaseModel):
    """日志查询工具的输入参数"""
    pod_name: str = Field(
        description="目标Pod名称"
    )
    namespace: str = Field(
        default="default",
        description="Kubernetes命名空间"
    )
    tail_lines: int = Field(
        default=100,
        description="返回最近的日志行数"
    )
    keyword: Optional[str] = Field(
        default=None,
        description="日志关键字过滤，如 error、timeout"
    )


# ===================== K8s Pod查询工具 =====================

class K8sPodQueryTool(BaseTool):
    """
    K8s Pod查询工具
    
    支持查询Pod列表、状态、资源使用情况等信息
    """
    name: str = "k8s_pod_query"
    description: str = (
        "查询Kubernetes集群中的Pod信息。"
        "可以按命名空间和服务名称过滤。"
        "返回Pod的名称、状态、IP、节点、容器信息。"
        "当用户询问Pod状态、查看Pod列表或检查服务运行时使用此工具。"
    )
    args_schema: Type[BaseModel] = PodQueryInput
    
    def __init__(self, k8s_config_file: Optional[str] = None):
        super().__init__()
        # 加载K8s配置
        try:
            if k8s_config_file:
                config.load_kube_config(config_file=k8s_config_file)
            else:
                # 尝试加载集群内配置(运行在Pod中时)
                try:
                    config.load_incluster_config()
                except config.ConfigException:
                    # 回退到默认kubeconfig
                    config.load_kube_config()
        except Exception as e:
            raise RuntimeError(f"Kubernetes配置加载失败: {e}")
        
        self.v1 = client.CoreV1Api()
    
    def _run(
        self,
        namespace: str = "default",
        service_name: Optional[str] = None,
        status_filter: Optional[str] = None
    ) -> str:
        """
        执行Pod查询
        
        Args:
            namespace: 命名空间
            service_name: 服务名称过滤
            status_filter: 状态过滤
        
        Returns:
            JSON格式的Pod信息列表
        """
        try:
            # 构建标签选择器
            label_selector = None
            if service_name:
                label_selector = f"app={service_name}"
            
            # 查询Pod列表
            pod_list = self.v1.list_namespaced_pod(
                namespace=namespace,
                label_selector=label_selector
            )
            
            # 格式化输出
            result = {
                "查询时间": datetime.now().isoformat(),
                "命名空间": namespace,
                "服务名称": service_name or "全部",
                "Pod总数": len(pod_list.items),
                "Pod列表": []
            }
            
            # 状态统计
            status_count = {}
            
            for pod in pod_list.items:
                pod_status = pod.status.phase
                
                # 检查容器状态（区分Running和CrashLoopBackOff等情况）
                if pod_status == "Running":
                    container_statuses = pod.status.container_statuses or []
                    for cs in container_statuses:
                        if cs.state.waiting and \
                           cs.state.waiting.reason == "CrashLoopBackOff":
                            pod_status = "CrashLoopBackOff"
                            break
                
                status_count[pod_status] = status_count.get(pod_status, 0) + 1
                
                # 如果指定了状态过滤，跳过不匹配的Pod
                if status_filter and pod_status != status_filter:
                    continue
                
                pod_info = {
                    "名称": pod.metadata.name,
                    "状态": pod_status,
                    "节点": pod.spec.node_name or "未调度",
                    "IP": pod.status.pod_ip or "未分配",
                    "创建时间": pod.metadata.creation_timestamp.isoformat() 
                        if pod.metadata.creation_timestamp else "未知",
                    "容器列表": [],
                    "重启次数": sum(
                        cs.restart_count or 0 
                        for cs in (pod.status.container_statuses or [])
                    )
                }
                
                # 容器详细信息
                for container in pod.spec.containers:
                    container_info = {
                        "名称": container.name,
                        "镜像": container.image,
                        "资源请求": {
                            "CPU": container.resources.requests.get("cpu", "未设置")
                                if container.resources.requests else "未设置",
                            "内存": container.resources.requests.get("memory", "未设置")
                                if container.resources.requests else "未设置"
                        }
                    }
                    pod_info["容器列表"].append(container_info)
                
                result["Pod列表"].append(pod_info)
            
            result["状态统计"] = status_count
            
            # 如果没有Pod或全部被过滤
            if not result["Pod列表"]:
                return json.dumps({
                    "消息": f"在命名空间 {namespace} 中未找到"
                           f"{'状态为' + status_filter + '的' if status_filter else ''}"
                           f"Pod{'（服务: ' + service_name + '）' if service_name else ''}",
                    "状态统计": status_count
                }, ensure_ascii=False, indent=2)
            
            return json.dumps(result, ensure_ascii=False, indent=2)
            
        except ApiException as e:
            error_msg = f"K8s API调用失败 (状态码: {e.status}): {e.reason}"
            if e.status == 403:
                error_msg += "\n可能原因: RBAC权限不足，请检查ServiceAccount权限"
            elif e.status == 404:
                error_msg += f"\n可能原因: 命名空间 {namespace} 不存在"
            return json.dumps({"错误": error_msg}, ensure_ascii=False)
        
        except Exception as e:
            return json.dumps(
                {"错误": f"查询失败: {str(e)}"},
                ensure_ascii=False
            )
    
    async def _arun(self, **kwargs) -> str:
        """异步执行（委托给同步方法）"""
        return self._run(**kwargs)


# ===================== 日志查询工具 =====================

class K8sLogQueryTool(BaseTool):
    """
    K8s Pod日志查询工具
    
    支持查询指定Pod的日志，可以按关键字过滤
    """
    name: str = "k8s_log_query"
    description: str = (
        "查询Kubernetes Pod的日志。"
        "可以指定Pod名称、命名空间和返回行数。"
        "支持通过关键字过滤日志内容（如 error、timeout、exception）。"
        "当用户需要排查问题、查看错误日志或追踪请求时使用此工具。"
    )
    args_schema: Type[BaseModel] = LogQueryInput
    
    def __init__(self, k8s_config_file: Optional[str] = None):
        super().__init__()
        try:
            if k8s_config_file:
                config.load_kube_config(config_file=k8s_config_file)
            else:
                try:
                    config.load_incluster_config()
                except config.ConfigException:
                    config.load_kube_config()
        except Exception as e:
            raise RuntimeError(f"Kubernetes配置加载失败: {e}")
        
        self.v1 = client.CoreV1Api()
    
    def _run(
        self,
        pod_name: str,
        namespace: str = "default",
        tail_lines: int = 100,
        keyword: Optional[str] = None
    ) -> str:
        """
        查询Pod日志
        
        Args:
            pod_name: Pod名称
            namespace: 命名空间
            tail_lines: 返回近期行数
            keyword: 关键字过滤
        
        Returns:
            日志内容（文本格式）
        """
        # 参数校验: tail_lines范围限制
        if tail_lines < 1:
            tail_lines = 10
        elif tail_lines > 5000:
            tail_lines = 5000  # 防止拉取过多日志
        
        try:
            # 先验证Pod是否存在
            pod = self.v1.read_namespaced_pod(
                name=pod_name,
                namespace=namespace
            )
            
            # 获取最近的日志行
            log_lines = self.v1.read_namespaced_pod_log(
                name=pod_name,
                namespace=namespace,
                tail_lines=tail_lines,
                timestamps=True
            )
            
            # 日志内容处理
            if not log_lines.strip():
                return json.dumps({
                    "Pod名称": pod_name,
                    "命名空间": namespace,
                    "消息": "该Pod没有日志输出"
                }, ensure_ascii=False)
            
            lines = log_lines.strip().split("\n")
            
            # 关键字过滤
            if keyword:
                filtered_lines = [
                    line for line in lines 
                    if keyword.lower() in line.lower()
                ]
                total_lines = len(lines)
                matched_lines = len(filtered_lines)
                
                if matched_lines == 0:
                    return json.dumps({
                        "Pod名称": pod_name,
                        "命名空间": namespace,
                        "关键字": keyword,
                        "扫描行数": total_lines,
                        "消息": f"在最近{total_lines}行日志中未找到包含'"
                               f"{keyword}'的内容"
                    }, ensure_ascii=False)
                
                lines = filtered_lines
            
            # 提取关键信息
            error_count = sum(
                1 for line in lines 
                if any(err in line.lower() 
                       for err in ['error', 'exception', 'fatal', 'panic'])
            )
            
            result = {
                "Pod名称": pod_name,
                "命名空间": namespace,
                "返回行数": len(lines),
                "日志级别统计": {
                    "错误": error_count,
                    "正常": len(lines) - error_count
                } if not keyword else None,
                "日志内容": lines[-20:]  # 只返回最后20行以避免过长
            }
            
            # 移除None值的键
            result = {k: v for k, v in result.items() if v is not None}
            
            return json.dumps(result, ensure_ascii=False, indent=2)
            
        except ApiException as e:
            if e.status == 404:
                return json.dumps({
                    "错误": f"Pod '{pod_name}' 在命名空间 '{namespace}' 中不存在"
                }, ensure_ascii=False)
            return json.dumps({
                "错误": f"日志查询失败: {e.reason} (状态码: {e.status})"
            }, ensure_ascii=False)
        
        except Exception as e:
            return json.dumps(
                {"错误": f"日志查询异常: {str(e)}"},
                ensure_ascii=False
            )
    
    async def _arun(self, **kwargs) -> str:
        return self._run(**kwargs)


# ===================== 危险操作确认工具 =====================

class K8sOperationConfirmationTool(BaseTool):
    """
    危险操作二次确认工具
    
    在执行Pod删除、Deployment重启等操作前，要求用户确认
    """
    name: str = "k8s_operation_confirm"
    description: str = (
        "在执行危险的Kubernetes操作（删除Pod、重启Deployment、扩缩容）"
        "之前，必须调用此工具向用户确认并记录审计日志。"
        "当Agent准备执行写操作时必须使用此工具。"
    )
    
    def _run(self, operation_detail: str) -> str:
        """
        参数通过ToolInput自动传入为字符串格式
        """
        # 生产环境中，这里应该：
        # 1. 将操作记录写入审计日志
        # 2. 检查操作者权限
        # 3. 返回确认请求给前端
        
        audit_record = {
            "时间": datetime.now().isoformat(),
            "操作": operation_detail,
            "状态": "等待确认"
        }
        
        return json.dumps({
            "确认请求": operation_detail,
            "说明": "该操作需要二次确认后才执行",
            "审计记录": audit_record
        }, ensure_ascii=False, indent=2)
    
    async def _arun(self, **kwargs) -> str:
        return self._run(**kwargs)

2.2 Agent系统提示词

# Agent系统提示词定义
SYSTEM_PROMPT = """你是一个专业的Kubernetes运维Agent助手。你的职责是：
1. 理解用户的运维需求，使用工具查询K8s资源状态
2. 分析查询结果，给出专业的运维建议
3. 在执行危险操作前，必须先向用户确认

基本原则:
- 优先使用k8s_pod_query查询Pod状态，再使用k8s_log_query查看日志
- 当发现Pod异常时，先分析日志和状态，给出诊断建议
- 不要直接执行删除、重启等操作，需要先说明影响面和风险
- 回答保持简洁专业，使用运维领域标准术语

当前集群信息:
- 命名空间需要用户明确指定，默认为default
- Pod命名规范: {service-name}-{replicaset-hash}-{pod-hash}
- 日志路径: 通过kubectl logs访问，不需要指定文件路径
"""

三、Agent组装与执行

3.1 Agent初始化

#!/usr/bin/env python3
"""
LangChain K8s运维Agent主程序
"""

from typing import Optional
import os


class K8sOpsAgent:
    """
    Kubernetes运维Agent
    
    集成LangChain Agent框架，提供自然语言驱动的K8s运维能力
    """
    
    def __init__(
        self,
        model_name: str = "gpt-4o",
        openai_api_key: Optional[str] = None,
        k8s_config: Optional[str] = None,
        verbose: bool = True
    ):
        """
        初始化K8s运维Agent
        
        Args:
            model_name: LLM模型名称
            openai_api_key: OpenAI API密钥（也可通过环境变量设置）
            k8s_config: K8s配置文件路径
            verbose: 是否输出详细日志
        """
        # 初始化LLM
        api_key = openai_api_key or os.getenv("OPENAI_API_KEY")
        if not api_key:
            raise ValueError(
                "请设置OPENAI_API_KEY环境变量或传入openai_api_key参数"
            )
        
        self.llm = ChatOpenAI(
            model=model_name,
            temperature=0.1,  # 低温度确保输出稳定
            api_key=api_key
        )
        
        # 初始化工具集
        self.tools = [
            K8sPodQueryTool(k8s_config_file=k8s_config),
            K8sLogQueryTool(k8s_config_file=k8s_config),
            K8sOperationConfirmationTool(),
        ]
        
        # 初始化记忆模块
        self.memory = ConversationBufferMemory(
            memory_key="chat_history",
            return_messages=True
        )
        
        # 构建Prompt模板
        self.prompt = ChatPromptTemplate.from_messages([
            ("system", SYSTEM_PROMPT),
            MessagesPlaceholder(variable_name="chat_history"),
            ("human", "{input}"),
            MessagesPlaceholder(variable_name="agent_scratchpad")
        ])
        
        # 创建Agent
        self.agent = create_openai_tools_agent(
            llm=self.llm,
            tools=self.tools,
            prompt=self.prompt
        )
        
        # 创建Agent执行器
        self.executor = AgentExecutor(
            agent=self.agent,
            tools=self.tools,
            memory=self.memory,
            verbose=verbose,
            handle_parsing_errors=True,
            max_iterations=10,     # 最大工具调用次数
            max_execution_time=120, # 最大执行时间(秒)
            early_stopping_method="generate"  # 超时时生成最佳回答
        )
    
    def chat(self, message: str) -> str:
        """
        与Agent进行一轮对话
        
        Args:
            message: 用户的自然语言消息
        
        Returns:
            Agent的回复
        
        Raises:
            RuntimeError: Agent执行失败时抛出
        """
        try:
            result = self.executor.invoke({
                "input": message
            })
            return result.get("output", "Agent未返回有效输出")
            
        except TimeoutError:
            return "操作执行超时，请简化您的查询重试"
        except Exception as e:
            error_msg = f"Agent执行异常: {str(e)}"
            return error_msg
    
    def reset_conversation(self):
        """重置对话记忆"""
        self.memory.clear()


# ===================== 使用示例 =====================

def main():
    """运维Agent交互式命令行"""
    print("=" * 60)
    print("  K8s 运维Agent - 自然语言驱动集群管理")
    print("  输入 'quit' 退出, 'reset' 重置对话")
    print("=" * 60)
    
    try:
        agent = K8sOpsAgent(verbose=False)
    except ValueError as e:
        print(f"初始化失败: {e}")
        print("请设置环境变量: export OPENAI_API_KEY=your_key")
        return 1
    
    # 交互式循环
    print("\nAgent就绪，请输入您的运维查询:")
    
    while True:
        try:
            user_input = input("\n🧑 运维>> ").strip()
            
            if not user_input:
                continue
            
            if user_input.lower() == 'quit':
                print("再见！")
                break
            
            if user_input.lower() == 'reset':
                agent.reset_conversation()
                print("对话已重置")
                continue
            
            # 执行查询
            print("\n🤖 Agent>> 思考中...")
            response = agent.chat(user_input)
            print(f"\n{response}")
            
        except KeyboardInterrupt:
            print("\n\n再见！")
            break
        except EOFError:
            break
    
    return 0


if __name__ == "__main__":
    exit(main())

四、安全与实践考量

4.1 安全护栏设计

在生产环境部署时，必须实现以下多层安全机制：

4.2 提示词注入防护

# 用户输入安全检查
def sanitize_user_input(user_input: str) -> str:
    """
    清理用户输入，防止提示词注入攻击
    
    检测并过滤以下模式:
    - 试图覆盖系统提示词
    - 试图调用未授权的工具
    - 恶意SQL/命令注入
    """
    dangerous_patterns = [
        "ignore previous instructions",
        "ignore the above",
        "you are now",
        "system prompt",
        "as a different role",
        "pretend to be",
        "disregard all",
    ]
    
    input_lower = user_input.lower()
    
    for pattern in dangerous_patterns:
        if pattern in input_lower:
            raise ValueError(f"检测到潜在的提示词注入攻击")
    
    # 限制输入长度
    if len(user_input) > 2000:
        raise ValueError("输入超过最大长度限制(2000字符)")
    
    return user_input

五、总结

基于 LangChain 构建运维 Agent 将大语言模型的语义理解能力与 Kubernetes API 的操作能力有机结合起来，实现了运维场景的自然语言交互。这套方案的核心价值在于：

1. 降低运维门槛：新同事不需要记上百条 kubectl 命令，用自然语言就能查询和诊断；
2. 提升响应速度：将"理解告警 → 查命令 → 执行 → 分析结果"的多步人工操作压缩为一句话交互；
3. 知识沉淀：Agent 的上下文记忆和回复可以作为运维知识库的自动化输入。

当前方案的局限性在于：LLM 的推理延迟（通常 2-5 秒）对于实时操作场景仍然偏高；对于多集群、跨云的复杂场景，需要扩展更多的工具和更精细的权限模型。这些将在后续迭代中逐步完善。

下一阶段可以引入 RAG（检索增强生成）技术，将运维 Runbook、历史故障报告、内部 Wiki 等文档集成到 Agent 的知识库中，实现更精准的诊断建议。