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

用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( defaultdefault, descriptionKubernetes命名空间默认值为default ) service_name: Optional[str] Field( defaultNone, description服务名称Deployment名称用于过滤Pod ) status_filter: Optional[str] Field( defaultNone, descriptionPod状态过滤条件可选: Running, Pending, Failed, CrashLoopBackOff, Unknown ) class LogQueryInput(BaseModel): 日志查询工具的输入参数 pod_name: str Field( description目标Pod名称 ) namespace: str Field( defaultdefault, descriptionKubernetes命名空间 ) tail_lines: int Field( default100, description返回最近的日志行数 ) keyword: Optional[str] Field( defaultNone, 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_filek8s_config_file) else: # 尝试加载集群内配置(运行在Pod中时) try: config.load_incluster_config() except config.ConfigException: # 回退到默认kubeconfig config.load_kube_config() except Exception as e: raise RuntimeError(fKubernetes配置加载失败: {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 fapp{service_name} # 查询Pod列表 pod_list self.v1.list_namespaced_pod( namespacenamespace, label_selectorlabel_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 } fPod{服务: service_name if service_name else }, 状态统计: status_count }, ensure_asciiFalse, indent2) return json.dumps(result, ensure_asciiFalse, indent2) except ApiException as e: error_msg fK8s 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_asciiFalse) except Exception as e: return json.dumps( {错误: f查询失败: {str(e)}}, ensure_asciiFalse ) 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_filek8s_config_file) else: try: config.load_incluster_config() except config.ConfigException: config.load_kube_config() except Exception as e: raise RuntimeError(fKubernetes配置加载失败: {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( namepod_name, namespacenamespace ) # 获取最近的日志行 log_lines self.v1.read_namespaced_pod_log( namepod_name, namespacenamespace, tail_linestail_lines, timestampsTrue ) # 日志内容处理 if not log_lines.strip(): return json.dumps({ Pod名称: pod_name, 命名空间: namespace, 消息: 该Pod没有日志输出 }, ensure_asciiFalse) 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_asciiFalse) 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_asciiFalse, indent2) except ApiException as e: if e.status 404: return json.dumps({ 错误: fPod {pod_name} 在命名空间 {namespace} 中不存在 }, ensure_asciiFalse) return json.dumps({ 错误: f日志查询失败: {e.reason} (状态码: {e.status}) }, ensure_asciiFalse) except Exception as e: return json.dumps( {错误: f日志查询异常: {str(e)}}, ensure_asciiFalse ) 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_asciiFalse, indent2) 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( modelmodel_name, temperature0.1, # 低温度确保输出稳定 api_keyapi_key ) # 初始化工具集 self.tools [ K8sPodQueryTool(k8s_config_filek8s_config), K8sLogQueryTool(k8s_config_filek8s_config), K8sOperationConfirmationTool(), ] # 初始化记忆模块 self.memory ConversationBufferMemory( memory_keychat_history, return_messagesTrue ) # 构建Prompt模板 self.prompt ChatPromptTemplate.from_messages([ (system, SYSTEM_PROMPT), MessagesPlaceholder(variable_namechat_history), (human, {input}), MessagesPlaceholder(variable_nameagent_scratchpad) ]) # 创建Agent self.agent create_openai_tools_agent( llmself.llm, toolsself.tools, promptself.prompt ) # 创建Agent执行器 self.executor AgentExecutor( agentself.agent, toolsself.tools, memoryself.memory, verboseverbose, handle_parsing_errorsTrue, max_iterations10, # 最大工具调用次数 max_execution_time120, # 最大执行时间(秒) early_stopping_methodgenerate # 超时时生成最佳回答 ) 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 fAgent执行异常: {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(verboseFalse) except ValueError as e: print(f初始化失败: {e}) print(请设置环境变量: export OPENAI_API_KEYyour_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 安全护栏设计在生产环境部署时必须实现以下多层安全机制安全层机制实现方式身份认证OAuth2/JWT集成公司SSO权限控制RBACK8s RBAC 自定义权限表操作审计不可篡改日志写入Elasticsearch/审计链敏感操作确认二次确认工具层自动触发确认操作频率限制速率限制Redis实现滑动窗口限流命名空间隔离生产环境只读K8s RBAC: 仅get/list/watch4.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 的操作能力有机结合起来实现了运维场景的自然语言交互。这套方案的核心价值在于降低运维门槛新同事不需要记上百条 kubectl 命令用自然语言就能查询和诊断提升响应速度将理解告警 → 查命令 → 执行 → 分析结果的多步人工操作压缩为一句话交互知识沉淀Agent 的上下文记忆和回复可以作为运维知识库的自动化输入。当前方案的局限性在于LLM 的推理延迟通常 2-5 秒对于实时操作场景仍然偏高对于多集群、跨云的复杂场景需要扩展更多的工具和更精细的权限模型。这些将在后续迭代中逐步完善。下一阶段可以引入 RAG检索增强生成技术将运维 Runbook、历史故障报告、内部 Wiki 等文档集成到 Agent 的知识库中实现更精准的诊断建议。