智能体在 DevOps 中的落地：自动诊断告警、生成预案与执行回滚

副标题：从被动响应到主动自愈，构建AI驱动的现代化运维体系

---

第一部分：引言与基础

1. 摘要/引言

问题陈述

在当今快速迭代的软件开发环境中，DevOps团队面临着前所未有的压力。随着微服务架构的普及、系统复杂性的指数级增长，以及业务对可用性要求的不断提高，传统的运维模式已经显得力不从心。

你是否有过这样的经历：

• 凌晨3点被告警电话吵醒，睡眼惺忪地排查问题
• 面对满屏的告警信息，不知道从何下手
• 故障发生后，需要手动查找文档、执行一系列复杂的恢复操作
• 由于人为操作失误，导致故障扩大化
• 团队精力被大量重复性的运维工作消耗，无法专注于更有价值的创新

这些问题不仅影响了团队的工作效率和生活质量，更重要的是，它们直接关系到业务的连续性和用户体验。据Gartner统计，平均每分钟的IT downtime会给企业带来5,600美元的损失，对于大型企业来说，这个数字可能高达每分钟数十万美元。

核心方案

那么，有没有一种方法能够改变这一现状？答案是肯定的——基于智能体（Agent）的自动化运维系统。

在本文中，我将向大家介绍如何构建一个能够：

1. 自动诊断告警：不仅仅是收集告警，而是理解告警背后的真正原因
2. 智能生成预案：根据历史数据和当前状态，自动生成可行的故障恢复方案
3. 安全执行回滚：在人类监督下，自动执行回滚操作，将系统恢复到健康状态

的智能体系统。我们将使用Python作为主要开发语言，结合LangChain作为智能体框架，Prometheus和Grafana作为监控数据源，以及Kubernetes作为容器编排平台，来构建一个完整的解决方案。

主要成果/价值

读完本文后，你将能够：

• 理解智能体在DevOps领域的应用场景和价值
• 掌握构建DevOps智能体的核心技术栈和架构设计
• 实现一个能够自动诊断、生成预案并执行回滚的原型系统
• 了解如何将智能体安全地集成到现有的DevOps流程中
• 获得关于未来运维自动化发展趋势的洞察

这不仅仅是一篇技术教程，更是一次关于运维未来形态的探索。

文章导览

本文将分为四个主要部分：

1. 引言与基础：介绍问题背景、核心概念和前置知识
2. 核心内容：深入探讨系统架构、关键技术和实现细节
3. 验证与扩展：展示系统运行效果，讨论优化方向和常见问题
4. 总结与附录：总结全文，提供参考资料和补充材料

现在，就让我们开始这段 exciting 的技术之旅吧！

---

2. 目标读者与前置知识

目标读者

本文主要面向以下人群：

• DevOps工程师/SRE：希望通过自动化手段提升工作效率，减少夜间告警
• 平台工程师：负责构建和维护公司内部开发平台的技术人员
• AI应用开发者：对将大语言模型应用于实际业务场景感兴趣的开发者
• 技术负责人/架构师：关注技术趋势，希望在团队中引入创新技术的管理者

前置知识

为了充分理解本文内容，建议你具备以下基础知识：

• Python编程：熟练掌握Python语言，了解异步编程基础
• DevOps基础：了解CI/CD流程、容器化（Docker）、容器编排（Kubernetes）
• 监控系统：了解Prometheus、Grafana等常见监控工具的基本概念
• 大语言模型基础：对LLM（如GPT-4、Claude、Llama等）有基本了解
• API开发：了解RESTful API的设计和使用

如果你对某些概念不太熟悉也没关系，我会在文中对关键概念进行解释。但具备上述基础将帮助你更好地理解和实践本文内容。

---

3. 文章目录

1. 引言与基础

   • 摘要/引言
   • 目标读者与前置知识
   • 文章目录

2. 问题背景与动机

   • 传统运维面临的挑战
   • 智能运维的兴起
   • 为什么选择智能体架构

3. 核心概念与理论基础

   • 智能体（Agent）的定义与分类
   • DevOps全流程中的智能体应用场景
   • 关键技术组件：LLM、RAG、工具使用
   • 系统架构概览

4. 环境准备

   • 技术栈选型
   • 开发环境搭建
   • 依赖安装与配置

5. 分步实现

   • 第一步：构建监控数据采集模块
   • 第二步：实现告警智能诊断引擎
   • 第三步：开发预案生成与评估模块
   • 第四步：构建安全执行与回滚系统
   • 第五步：集成与测试

6. 关键代码解析与深度剖析

   • 智能体的核心决策逻辑
   • RAG系统在运维知识库中的应用
   • 安全机制与权限控制设计
   • 提示工程（Prompt Engineering）最佳实践

7. 结果展示与验证

   • 系统演示：从告警触发到自动恢复
   • 有效性评估：准确率、响应时间、恢复时间
   • 真实场景案例分析

8. 性能优化与最佳实践

   • 如何提升智能体的决策质量
   • 降低系统延迟的技巧
   • 成本控制策略
   • 人机协作模式设计

9. 常见问题与解决方案

   • 智能体"幻觉"问题的应对
   • 安全性与风险控制
   • 如何处理复杂的关联故障
   • 团队接受度与变革管理

10. 未来展望与扩展方向

    • 从"自动恢复"到"预测性运维"
    • 多智能体协作
    • 与混沌工程的结合
    • 行业发展趋势

11. 总结

12. 参考资料

13. 附录

---

第二部分：核心内容

4. 问题背景与动机

在深入技术实现之前，让我们先花些时间理解为什么我们需要这样一个系统，以及它将如何改变我们的工作方式。

传统运维面临的挑战

让我们从一个真实的故事开始——这是我从一位在大型互联网公司工作的SRE朋友那里听来的：

"那是一个周五的晚上，我正准备和家人出去吃饭，手机突然响了——是监控系统的告警。我赶紧打开电脑，发现是核心支付服务的错误率突然飙升。

接下来的两个小时，我就像一只无头苍蝇：

• 首先，我需要在几十个告警中找出哪些是因，哪些是果
• 然后，我要登录多个系统查看日志、指标、链路追踪
• 接着，我要回忆或者查找之前类似的问题是怎么解决的
• 最后，我还需要小心翼翼地执行恢复操作，生怕一不小心把事情搞得更糟

等我终于把问题解决，餐厅也早就关门了。家人虽然没说什么，但我能看出他们的失望。更糟糕的是，周六一整天我都在担心会不会再次出问题，根本没有心情休息。"

相信很多DevOps/SRE同学都有过类似的经历。那么，传统运维模式到底存在哪些问题呢？

1. 告警风暴与疲劳

现代系统产生的告警数量是惊人的。根据我的经验，一个中等规模的微服务架构，每天可能会产生数千甚至数万个告警。在这海量的告警中，真正需要人工干预的可能只占不到1%。

但问题是，我们无法预知哪一个告警是重要的。于是，SRE们不得不处理大量的"噪音"，这导致了严重的告警疲劳——人们开始对告警变得麻木，甚至错过真正重要的告警。

2. 知识孤岛与专家依赖

运维知识往往分散在不同人的头脑中，或者散落在各种文档、wiki、聊天记录里。当一个经验丰富的工程师离职后，他带走的不仅仅是个人，还有宝贵的故障处理经验。

同时，对于复杂的系统，往往只有少数几个"专家"真正理解其内部原理。一旦这些专家不在，遇到问题时团队就会陷入困境。

3. 人为错误与恢复延迟

即使找到了问题的根源，执行恢复操作也往往是一个缓慢且容易出错的过程。根据Uptime Institute的数据，数据中心停机的主要原因中，人为错误占比高达70%。

为什么会这样？因为：

• 故障发生时，工程师往往处于压力之下，容易做出错误的判断
• 恢复操作可能涉及多个步骤，需要在不同系统间切换
• 缺乏标准化的流程，每个人的操作方式可能不同
• 手动执行操作本身就比自动化慢得多

4. 被动响应而非主动预防

传统运维本质上是被动的——我们等待问题发生，然后再去解决。但在当今的业务环境下，这种模式越来越不可接受。用户期望的是永远在线的服务，而不是"我们会尽快修复"的承诺。

更重要的是，被动响应的模式无法从根本上提升系统的可靠性。我们总是在"救火"，却没有时间去"防火"。

智能运维的兴起

为了应对这些挑战，行业开始探索**智能运维（AIOps）**的解决方案。Gartner在2016年首次提出了AIOps的概念，将其定义为：

“AIOps combines big data and machine learning to automate IT operations processes, including event correlation, anomaly detection, and causality determination.”

简单来说，AIOps就是利用大数据和机器学习技术来自动化IT运维流程。

经过几年的发展，AIOps已经经历了几个阶段：

阶段	时间	主要特征	局限性
规则驱动	2010年代初	基于预定义规则的告警和自动化	规则难以维护，无法应对未知场景
传统机器学习	2010年代中	基于统计分析和传统ML的异常检测	需要大量标注数据，可解释性差
大模型时代	2020年代至今	基于LLM的智能体，具备理解和推理能力	仍在探索中，需要解决安全性等问题

我们现在正处于从第二阶段向第三阶段过渡的时期。传统的AIOps解决方案虽然在某些方面取得了成功，但它们往往存在以下问题：

• 需要大量的历史数据和标注工作
• 模型的可解释性差，人们不知道它为什么做出某个决策
• 缺乏泛化能力，只能处理特定类型的问题
• 无法与人类自然地交互和协作

而大语言模型的出现，为解决这些问题带来了新的可能。

为什么选择智能体架构

在众多的AI应用架构中，为什么我们选择**智能体（Agent）**作为我们的核心架构呢？

要回答这个问题，我们首先需要理解什么是智能体。简单来说，智能体是一个能够感知环境、做出决策并采取行动的自主实体。一个典型的智能体通常包含以下组件：

1. 感知模块：获取和理解环境信息
2. 推理/决策模块：基于感知到的信息做出决策
3. 行动模块：执行决策，与环境交互
4. 记忆模块：存储历史信息和知识

为什么这种架构特别适合DevOps场景呢？

1. 应对开放环境的挑战

DevOps环境是一个典型的开放环境——我们永远无法预知会出现什么样的新问题。传统的软件系统是为了解决特定问题而设计的，它们在面对未知情况时往往无能为力。

而智能体架构，特别是结合了大语言模型的智能体，具备一定的泛化能力。它们可以利用从海量数据中学到的知识，去尝试解决从未见过的问题。

2. 工具使用能力

DevOps运维涉及到大量的工具——kubectl、helm、terraform、各种云服务商的CLI等等。让AI直接学会使用这些工具，比重新实现一套相同的功能要高效得多。

智能体架构的一个核心优势就是**工具使用（Tool Use）**能力。通过为智能体提供一组工具，我们可以让它像人类工程师一样，利用现有的工具去解决问题。

3. 迭代式问题解决

故障诊断往往不是一个一蹴而就的过程。真实的情况往往是：

1. 先查看一些指标，有了一个初步的假设
2. 然后去查看相关的日志，验证或修正这个假设
3. 接着可能需要去查一下知识库，看看之前有没有类似的案例
4. 最后才确定问题根源并采取行动

这种迭代式的问题解决方式，正是智能体架构所擅长的。智能体可以根据上一步的结果，决定下一步该做什么，形成一个"感知-推理-行动"的闭环。

4. 人机协作

最后但同样重要的是，智能体架构天然支持人机协作。我们不需要让智能体完全取代人类，而是让它成为人类的"智能副手"。

在这种模式下：

• 智能体负责处理重复性的、机械性的工作
• 人类负责处理需要创造性、价值判断的工作
• 智能体可以向人类请求帮助或确认
• 人类可以监督和干预智能体的决策

这种人机协作的模式，不仅更加安全，也更加符合当前的技术发展水平——我们不需要等待完全自主的AI出现，就可以开始从AI技术中获益。

---

5. 核心概念与理论基础

在开始构建我们的系统之前，让我们先明确一些核心概念，并建立一个共同的理解框架。

智能体（Agent）的定义与分类

在人工智能领域，"智能体"是一个非常宽泛的概念。让我们从一个经典的定义开始：

智能体（Agent）：一个能够通过传感器（Sensor）感知环境，并通过执行器（Actuator）对环境施加影响的实体。

这个定义来自Russell和Norvig的经典教材《人工智能：一种现代方法》。虽然这个定义是在20多年前提出的，但它仍然很好地描述了我们今天所构建的系统。

在DevOps场景中，我们可以将这个定义具体化：

• 环境：我们的IT系统——服务器、网络、应用程序、数据库等
• 传感器：监控系统（Prometheus）、日志系统（ELK）、链路追踪系统（Jaeger）等
• 执行器：各种运维工具——kubectl、terraform、云服务商API等

智能体的分类

根据智能体的复杂程度和能力范围，我们可以将其分为以下几类：

类型	描述	特点	DevOps应用示例
简单反射型智能体	仅基于当前感知做出决策，不考虑历史	简单、快速，但缺乏灵活性	基于固定阈值的告警规则
基于模型的反射型智能体	维护内部状态，利用历史信息	能处理部分可观察环境	基于趋势的异常检测
基于目标的智能体	有明确的目标，会规划达成目标的路径	具备一定的推理和规划能力	自动扩缩容系统
基于效用的智能体	不仅考虑目标，还考虑达成目标的"质量"	能在多个方案中进行权衡选择	多维度优化的资源调度
学习型智能体	能够从经验中学习，不断改进自己的行为	适应能力强，能处理未知情况	我们即将构建的系统

我们将要构建的DevOps智能体，属于最后一类——学习型智能体。它不仅能够基于当前信息做出决策，还能够从历史数据和人类反馈中学习，不断提升自己的能力。

DevOps全流程中的智能体应用场景

智能体在DevOps中的应用远不止故障诊断和恢复。让我们从全流程的角度，看看智能体可以在哪些地方发挥作用：

虽然本文的重点是运行阶段的告警诊断、预案生成和回滚执行，但在设计系统时，我们应该考虑到未来扩展到其他场景的可能性。

现在，让我们聚焦于我们的核心场景——故障的自动诊断、预案生成和回滚执行。这个流程可以抽象为以下几个步骤：

这是一个理想的流程。在实际实现中，我们需要考虑各种边缘情况和异常处理。但这个序列图为我们提供了一个很好的起点。

关键技术组件：LLM、RAG、工具使用

要构建这样一个系统，我们需要三个关键的技术组件：大语言模型（LLM）、检索增强生成（RAG）和工具使用（Tool Use）。让我们逐一了解它们。

1. 大语言模型（LLM）

大语言模型是整个系统的"大脑"。它负责理解告警信息、分析数据、推理根因、生成预案。

在选择LLM时，我们需要考虑以下几个因素：

• 能力：模型的推理能力、领域知识、代码理解能力
• 上下文窗口：能够处理的输入长度（这对处理长日志很重要）
• 速度：响应延迟（这对实时性要求高的场景很重要）
• 成本：API调用的费用
• 安全性：数据隐私和合规性
• 可控性：是否能够进行微调（Fine-tuning）或提示工程

对于DevOps场景，我建议优先考虑以下模型：

• GPT-4：综合能力最强，特别是推理和代码理解
• Claude 3：上下文窗口大，适合处理长日志
• CodeLlama：可以本地部署，适合对数据隐私要求高的场景
• Qwen（通义千问）：阿里开源，在中文和代码方面表现不错

在本文中，我们将使用OpenAI的GPT-4作为示例，但我们的架构设计是模型无关的——你可以根据自己的需求替换成任何其他模型。

2. 检索增强生成（RAG）

虽然LLM具有强大的能力，但它们也有一些固有的局限性：

• 知识截止日期：模型不知道训练数据之后发生的事情
• 幻觉问题：模型可能会编造一些听起来合理但实际上是错误的信息
• 领域知识不足：通用模型可能缺乏特定领域（如你的公司的特定架构）的知识
• 无法访问实时数据：模型无法直接查询当前的系统状态

**检索增强生成（Retrieval-Augmented Generation，RAG）**是解决这些问题的一种有效方法。简单来说，RAG的工作原理是：

1. 将我们的知识（如运维文档、历史故障案例、系统架构图等）进行向量化存储
2. 当需要回答问题时，先从向量数据库中检索相关的知识
3. 将检索到的知识和问题一起传给LLM
4. LLM基于检索到的知识生成回答

在我们的系统中，RAG将主要用于：

• 存储和检索历史故障处理案例
• 存储和检索系统架构和组件信息
• 存储和检索标准操作流程（SOP）
• 存储和检索运维手册和最佳实践

3. 工具使用（Tool Use）

最后一个关键组件是工具使用（Tool Use），也称为函数调用（Function Calling）。这是让智能体能够与真实世界交互的关键。

工具使用的基本思想是：

1. 我们为LLM提供一组工具的描述（包括工具的名称、功能、参数格式等）
2. LLM根据当前任务，决定是否需要使用工具以及使用哪个工具
3. 如果需要使用工具，LLM会生成工具调用的参数
4. 我们的系统负责执行工具调用，并将结果返回给LLM
5. LLM根据工具调用的结果，继续下一步的推理或生成最终回答

在DevOps场景中，我们可以为智能体提供各种工具，例如：

• 查询类工具：查询Prometheus指标、查询Elasticsearch日志、查询Kubernetes资源状态
• 操作类工具：重启Pod、调整副本数、执行回滚、修改配置
• 分析类工具：分析日志中的错误模式、分析性能瓶颈、计算相关指标
• 沟通类工具：发送通知、创建工单、请求人类确认

当然，提供操作类工具需要非常谨慎——我们需要建立完善的安全机制，确保智能体不会做出危险的操作。我们将在后面的章节中详细讨论这个问题。

系统架构概览

现在，让我们将这些组件组合在一起，看看我们的系统整体架构是什么样子的：

这个架构可能看起来有点复杂，但不要担心——我们将在后续的章节中逐步实现每一个组件。现在，你只需要对整体架构有一个大致的了解即可。

这个架构的核心设计原则是：

1. 模块化：每个组件都是独立的，可以单独替换或升级
2. 可扩展：可以轻松添加新的工具、新的知识源、新的智能体能力
3. 安全第一：所有的操作都经过安全层的检查和审计
4. 人机协作：人类可以在任何环节介入，监督和指导智能体的行为

---

6. 环境准备

在开始编写代码之前，让我们先准备好我们的开发环境。

技术栈选型

首先，让我们确定我们将使用的技术栈：

类别	技术选型	版本要求	说明
编程语言	Python	3.10+	生态丰富，AI库支持好
LLM框架	LangChain	0.1.x+	简化LLM应用开发
向量数据库	ChromaDB / Qdrant	最新版	轻量级，适合开发
Web框架	FastAPI	0.100+	高性能，自动生成文档
容器化	Docker	20.10+	环境一致性
容器编排	Kubernetes (Minikube)	1.28+	本地测试环境
监控	Prometheus + Grafana	最新版	监控和告警
日志	Loki + Promtail	最新版	日志收集和查询
消息队列	Redis / RabbitMQ	最新版	任务队列和缓存
前端	React + TypeScript	最新版	Web控制台（可选）

这是一个适合开发和原型验证的技术栈。在生产环境中，你可能需要根据实际情况进行调整——例如，使用更强大的向量数据库（如Pinecone或Weaviate），使用更完善的消息队列（如Kafka）等。

开发环境搭建

现在，让我们一步步搭建我们的开发环境。

步骤1：安装Python 3.10+

首先，确保你安装了Python 3.10或更高版本。你可以使用pyenv来管理多个Python版本：

# 安装pyenv（如果你还没有的话）
curl https://pyenv.run | bash

# 安装Python 3.10
pyenv install 3.10.12
pyenv global 3.10.12

步骤2：创建虚拟环境

接下来，创建一个新的虚拟环境：

# 创建项目目录
mkdir devops-agent
cd devops-agent

# 创建虚拟环境
python -m venv venv

# 激活虚拟环境
# Linux/Mac
source venv/bin/activate
# Windows
venv\Scripts\activate

步骤3：安装基础依赖

现在，安装一些基础的依赖：

# 创建requirements.txt文件
cat > requirements.txt << 'EOF'
# LLM相关
langchain==0.1.10
langchain-openai==0.0.5
langchain-community==0.0.25

# 向量数据库
chromadb==0.4.24
qdrant-client==1.8.0

# Web框架
fastapi==0.109.2
uvicorn==0.27.1
pydantic==2.6.1

# Kubernetes客户端
kubernetes==29.0.0

# Prometheus客户端
prometheus-api-client==0.54.0

# 日志与监控
python-json-logger==2.0.7
structlog==24.1.0

# 异步任务
celery==5.3.6
redis==5.0.1

# 工具库
python-dotenv==1.0.1
pydantic-settings==2.2.1
tenacity==8.2.3
httpx==0.27.0
EOF

# 安装依赖
pip install -r requirements.txt

步骤4：设置Minikube（可选但推荐）

如果你没有一个现成的Kubernetes集群，可以使用Minikube来创建一个本地集群：

# 安装Minikube（以Linux为例，其他操作系统请参考官方文档）
curl -LO https://storage.googleapis.com/minikube/releases/latest/minikube-linux-amd64
sudo install minikube-linux-amd64 /usr/local/bin/minikube

# 启动Minikube
minikube start --cpus=4 --memory=8192

# 安装Prometheus和Grafana（使用Helm）
helm repo add prometheus-community https://prometheus-community.github.io/helm-charts
helm repo update
helm install prometheus prometheus-community/kube-prometheus-stack

# 安装Loki（用于日志收集）
helm repo add grafana https://grafana.github.io/helm-charts
helm install loki grafana/loki-stack

步骤5：配置环境变量

最后，创建一个.env文件来存储我们的配置：

# 创建.env文件
cat > .env << 'EOF'
# 应用配置
APP_NAME=DevOpsAgent
APP_ENV=development
APP_DEBUG=true

# API配置
API_HOST=0.0.0.0
API_PORT=8000

# OpenAI配置
OPENAI_API_KEY=your-openai-api-key-here
OPENAI_MODEL_NAME=gpt-4-turbo-preview
OPENAI_TEMPERATURE=0.1

# 向量数据库配置
VECTOR_DB_TYPE=chroma  # 或 qdrant
CHROMA_DB_PATH=./data/chroma
# QDRANT_URL=http://localhost:6333

# Kubernetes配置
KUBE_CONFIG_PATH=~/.kube/config

# Prometheus配置
PROMETHEUS_URL=http://localhost:9090

# Grafana配置
GRAFANA_URL=http://localhost:3000
GRAFANA_API_KEY=your-grafana-api-key-here

# Loki配置
LOKI_URL=http://localhost:3100

# Redis配置
REDIS_URL=redis://localhost:6379/0

# 安全配置
SAFETY_MODE=audit  # audit, approval, disabled
MAX_ACTION_RISK_LEVEL=medium  # low, medium, high

# 日志配置
LOG_LEVEL=INFO
LOG_FORMAT=json
EOF

记得将其中的占位符（如your-openai-api-key-here）替换为你自己的实际配置。

---

7. 分步实现

好了，环境准备完毕！现在，让我们开始编写代码，实现我们的DevOps智能体系统。

我们将按照以下顺序进行：

1. 第一步：构建监控数据采集模块
2. 第二步：实现告警智能诊断引擎
3. 第三步：开发预案生成与评估模块
4. 第四步：构建安全执行与回滚系统
5. 第五步：集成与测试

让我们开始吧！

第一步：构建监控数据采集模块

首先，我们需要构建一个模块，用于从各种监控系统中采集数据。这个模块将作为智能体的"眼睛"和"耳朵"。

1.1 项目结构组织

首先，让我们组织一下我们的项目结构：

devops-agent/
├── .env
├── requirements.txt
├── README.md
├── src/
│   └── devops_agent/
│       ├── __init__.py
│       ├── config.py          # 配置管理
│       ├── models/            # 数据模型
│       │   ├── __init__.py
│       │   ├── alert.py
│       │   ├── incident.py
│       │   └── action.py
│       ├── collectors/        # 数据采集模块
│       │   ├── __init__.py
│       │   ├── base.py
│       │   ├── prometheus.py
│       │   ├── loki.py
│       │   └── kubernetes.py
│       ├── agents/            # 智能体模块
│       │   ├── __init__.py
│       │   ├── base.py
│       │   ├── diagnostic.py
│       │   ├── planner.py
│       │   └── executor.py
│       ├── knowledge/         # 知识库模块
│       │   ├── __init__.py
│       │   ├── base.py
│       │   ├── vector_store.py
│       │   └── case_manager.py
│       ├── tools/             # 工具模块
│       │   ├── __init__.py
│       │   ├── registry.py
│       │   ├── query_tools.py
│       │   └── action_tools.py
│       ├── safety/            # 安全模块
│       │   ├── __init__.py
│       │   ├── authorization.py
│       │   ├── risk_assessment.py
│       │   └── audit_log.py
│       ├── api/               # API模块
│       │   ├── __init__.py
│       │   ├── main.py
│       │   └── routes/
│       └── utils/             # 工具函数
│           ├── __init__.py
│           └── logging.py
├── tests/                     # 测试代码
├── scripts/                   # 脚本文件
└── data/                      # 数据目录

这是一个相对完整的项目结构。在本文中，我们将重点关注其中的核心模块。

1.2 配置管理

首先，让我们创建一个配置管理模块，用于管理我们的应用配置：

# src/devops_agent/config.py
from typing import Optional
from pydantic_settings import BaseSettings
from functools import lru_cache


class Settings(BaseSettings):
    """应用配置"""
    # 应用配置
    app_name: str = "DevOpsAgent"
    app_env: str = "development"
    app_debug: bool = True
    
    # API配置
    api_host: str = "0.0.0.0"
    api_port: int = 8000
    
    # OpenAI配置
    openai_api_key: str
    openai_model_name: str = "gpt-4-turbo-preview"
    openai_temperature: float = 0.1
    openai_max_tokens: int = 4096
    
    # 向量数据库配置
    vector_db_type: str = "chroma"
    chroma_db_path: str = "./data/chroma"
    qdrant_url: Optional[str] = None
    
    # Kubernetes配置
    kube_config_path: Optional[str] = None
    
    # Prometheus配置
    prometheus_url: str = "http://localhost:9090"
    
    # Grafana配置
    grafana_url: Optional[str] = None
    grafana_api_key: Optional[str] = None
    
    # Loki配置
    loki_url: Optional[str] = None
    
    # Redis配置
    redis_url: str = "redis://localhost:6379/0"
    
    # 安全配置
    safety_mode: str = "audit"  # audit, approval, disabled
    max_action_risk_level: str = "medium"  # low, medium, high
    
    # 日志配置
    log_level: str = "INFO"
    log_format: str = "json"
    
    class Config:
        env_file = ".env"
        case_sensitive = False


@lru_cache()
def get_settings() -> Settings:
    """获取配置单例"""
    return Settings()

这个配置类使用了pydantic-settings，它会自动从环境变量和.env文件中读取配置。

1.3 数据模型

接下来，让我们定义一些核心的数据模型：

# src/devops_agent/models/alert.py
from datetime import datetime
from typing import Dict, Any, Optional, List
from enum import Enum
from pydantic import BaseModel, Field


class AlertSeverity(str, Enum):
    """告警级别"""
    CRITICAL = "critical"
    WARNING = "warning"
    INFO = "info"
    DEBUG = "debug"


class AlertStatus(str, Enum):
    """告警状态"""
    FIRING = "firing"
    PENDING = "pending"
    RESOLVED = "resolved"
    ACKNOWLEDGED = "acknowledged"


class Alert(BaseModel):
    """告警模型"""
    id: str = Field(..., description="告警唯一标识")
    name: str = Field(..., description="告警名称")
    severity: AlertSeverity = Field(..., description="告警级别")
    status: AlertStatus = Field(..., description="告警状态")
    message: str = Field(..., description="告警消息")
    labels: Dict[str, str] = Field(default_factory=dict, description="告警标签")
    annotations: Dict[str, str] = Field(default_factory=dict, description="告警注解")
    starts_at: datetime = Field(..., description="告警开始时间")
    ends_at: Optional[datetime] = Field(None, description="告警结束时间")
    generator_url: Optional[str] = Field(None, description="告警生成器URL")
    fingerprint: Optional[str] = Field(None, description="告警指纹")
    metadata: Dict[str, Any] = Field(default_factory=dict, description="额外元数据")
    
    class Config:
        json_encoders = {
            datetime: lambda v: v.isoformat()
        }

# src/devops_agent/models/incident.py
from datetime import datetime
from typing import Dict, Any, Optional, List
from enum import Enum
from pydantic import BaseModel, Field
from .alert import Alert, AlertSeverity


class IncidentStatus(str, Enum):
    """事件状态"""
    OPEN = "open"
    INVESTIGATING = "investigating"
    IDENTIFIED = "identified"
    RESOLVING = "resolving"
    RESOLVED = "resolved"
    CLOSED = "closed"


class Incident(BaseModel):
    """事件模型"""
    id: str = Field(..., description="事件唯一标识")
    title: str = Field(..., description="事件标题")
    description: str = Field(..., description="事件描述")
    severity: AlertSeverity = Field(..., description="事件级别")
    status: IncidentStatus = Field(..., description="事件状态")
    alerts: List[Alert] = Field(default_factory=list, description="关联的告警")
    root_cause: Optional[str] = Field(None, description="根因分析结果")
    resolution_plan: Optional[str] = Field(None, description="解决方案")
    created_at: datetime = Field(default_factory=datetime.utcnow, description="创建时间")
    updated_at: datetime = Field(default_factory=datetime.utcnow, description="更新时间")
    resolved_at: Optional[datetime] = Field(None, description="解决时间")
    metadata: Dict[str, Any] = Field(default_factory=dict, description="额外元数据")

这些数据模型将用于在不同模块之间传递数据。

1.4 Prometheus采集器

现在，让我们创建一个Prometheus采集器，用于从Prometheus中查询指标：

# src/devops_agent/collectors/base.py
from abc import ABC, abstractmethod
from typing import Any, Dict, Optional
from datetime import datetime, timedelta


class BaseCollector(ABC):
    """采集器基类"""
    
    @abstractmethod
    def connect(self) -> None:
        """连接到数据源"""
        pass
    
    @abstractmethod
    def disconnect(self) -> None:
        """断开与数据源的连接"""
        pass
    
    @abstractmethod
    def query(self, query: str, **kwargs) -> Any:
        """执行查询"""
        pass
    
    @abstractmethod
    def query_range(
        self, 
        query: str, 
        start: datetime, 
        end: datetime, 
        step: timedelta,
        **kwargs
    ) -> Any:
        """执行范围查询"""
        pass

# src/devops_agent/collectors/prometheus.py
from datetime import datetime, timedelta
from typing import Any, Dict, List, Optional
import logging
from prometheus_api_client import PrometheusConnect
from prometheus_api_client.utils import parse_datetime

from .base import BaseCollector
from ..config import get_settings

logger = logging.getLogger(__name__)
settings = get_settings()


class PrometheusCollector(BaseCollector):
    """Prometheus采集器"""
    
    def __init__(self, url: Optional[str] = None):
        self.url = url or settings.prometheus_url
        self._client: Optional[PrometheusConnect] = None
        logger.info(f"Initializing PrometheusCollector with URL: {self.url}")
    
    def connect(self) -> None:
        """连接到Prometheus"""
        try:
            self._client = PrometheusConnect(url=self.url, disable_ssl=True)
            # 测试连接
            self._client.all_metrics()
            logger.info("Successfully connected to Prometheus")
        except Exception as e:
            logger.error(f"Failed to connect to Prometheus: {e}")
            raise
    
    def disconnect(self) -> None:
        """断开与Prometheus的连接"""
        self._client = None
        logger.info("Disconnected from Prometheus")
    
    def query(self, query: str, time: Optional[datetime] = None, **kwargs) -> List[Dict[str, Any]]:
        """
        执行即时查询
        
        Args:
            query: PromQL查询语句
            time: 查询时间点，默认为当前时间
            
        Returns:
            查询结果
        """
        if not self._client:
            self.connect()
        
        try:
            params = {}
            if time:
                params["time"] = parse_datetime(str(time))
            
            result = self._client.custom_query(query, params=params)
            logger.debug(f"Executed Prometheus query: {query}, got {len(result)} results")
            return result
        except Exception as e:
            logger.error(f"Failed to execute Prometheus query: {e}")
            raise
    
    def query_range(
        self, 
        query: str, 
        start: datetime, 
        end: datetime, 
        step: timedelta,
        **kwargs
    ) -> List[Dict[str, Any]]:
        """
        执行范围查询
        
        Args:
            query: PromQL查询语句
            start: 开始时间
            end: 结束时间
            step: 时间步长
            
        Returns:
            查询结果
        """
        if not self._client:
            self.connect()
        
        try:
            result = self._client.custom_query