分类 默认分类 下的文章

基于 Agent Skill 的自动化问题诊断框架

IMG_3493.jpg

最近在小规模实践一个问题:大模型时代,如何做生产环境的自动问题分析诊断。本文是一个阶段性思考,因为裁员的问题, 只能先记录下来。

前提是生产环境的数据收集、分析、整理、告警等基础设施已经比较完善。在此基础上,用 Agent 代理人工分析诊断问题的过程,并持续优化和迭代诊断能力。

终极信念当然是 AI 自动运维、排除 human-in-the-loop。但现阶段这仍是理想,尤其在生产环境:一次误操作的风险,可能比诊断不出来更大。因此,当前更务实的方式是让 Agent 多数部署在本地 corp 环境,而不是直接放进生产环境。

核心设计理念

  1. 大模型具备通用分析诊断能力,尤其对开源产品和通用技术栈。
  2. 公司内部相对静态的信息,通过 RAG 或 skill 获取,并持续添加、迭代更新。
  3. 公司内部相对动态的信息,由 skill 调用公司 API 或 MCP 工具获取,主要对接 metrics、logs、traces、topology、config 等。

基本流程可以概括为:

告警 -> Agent -> Skill/RAG -> API/MCP -> metrics/logs/traces/topology/config

工程上会涉及 Agent 编排、Memory 共享、skill 层次结构与更新、RAG 文档迭代,以及 API/MCP 的权限和认证。

本质:把人的诊断过程迁移给 Agent

使用 Agent + Skill,本质上是把原来人做的事情转成 Agent 要做的事情:

  • 人的判断分析,改成大模型的分析判断;
  • 人的信息收集整理,改成 Agent 的收集整理,并且可以并行、更快;
  • 大模型不知道的内部静态信息,通过 RAG 或 skill 中的信息获取;
  • 大模型不知道的内部动态信息,通过 skill 调用的 API 或 MCP 工具获取。

人做得慢,思维也慢;Agent 可以大规模并行收集、分析、验证,这是核心优势。

主要难点

第一,人喜欢“一图胜千言”,快速浏览一批 dashboard 就能发现可疑异常点。但 Agent 不擅长读 metrics 图表;即使喂原始数据,也需要借助其他工具分析。目前缺少现成的数据 insights 工具,更谈不上并行看多图的效率。

第二,要把原先给人看的成熟 dashboard 转成 Agent 能理解的形式。短文本问题不大,但图表需要很大改动。

第三,授权和认证。过去的登录、认证、授权都以人为基础;Agent 可以继承人的身份,但人的 UI 双因素认证与 API/MCP 认证完全不同。

第四,操作边界。仅在 skill 里约束不够,边界要清晰,并且要从 API 和 MCP 层面做强制鉴权和授权。

事后记录与 Skill 迭代

人通常只记录关键结论,并主动漏掉走弯路的部分;事情解决后,也缺少动力记录过程。Agent 可以把整个过程记录下来,并在后续分析中使用。

事后整理的 trajectory 可能需要当时的实时数据。这些数据要同步保留,供以后重跑、验证 skill 能力。

实施上,需要一个地方保留有用的 skill,并支持持续升级迭代;同时按 skill 的 scope 做分层、分类整理,便于管理和调用。至于 Agent 的并行能力,目前 Claude 和 OpenCode 提供的已在一定程度上够用。

现阶段的核心不是让大模型直接替代人,而是把人的诊断过程拆成可复用的 skill 和可强制授权的工具调用,先做一个可审计、可回退、可迭代的诊断助手。

提问,是AI时代最硬的计算机技能

IMG_3648.jpg

前阵子和朋友聊起他孩子读大专计算机的事。我们最后的结论很一致:让AI当你的私人导师。现在大模型讲很多知识点,比课堂更细、比教材更贴你的水平。但前提是——你得会提问。

这一点,中国学生普遍吃亏。大家习惯等别人提问,或者自己闷头搜索。太慢了。提问本身就是一种信号:说明你对现状已经花了心思去理解,只是还有卡点。这恰恰是真正理解一个问题的起点。

那提问能力怎么来?本质是一个反复确认的过程:你以为的状态、实际的状态、你不确定的部分,三者不断碰撞。大脑在这个循环里整理、分析,最后生成新的认知。有了AI之后,获取知识变便宜了,但想把一个问题挖透,靠的是不断往下追问。

我最近用大模型学PyPots,深有体会。以前看书、看文档,写的人默认你懂某些前置知识,可那些我恰恰没概念。于是我先让模型给我讲整体框架,包含哪些模块;再逐个拆开,遇到没接触过的就让它展开,多举例子,讲清楚"为什么发明这个概念、前提是什么"。最后它帮我整理成一份适合我自己的文档——每一条从"我不会"开始,直到我把疑问全部划掉,才算真懂。

当然,分支无限,不可能全追到底,有些地方要懂得打住。

这套方法我反复用,总结成一句:从框架讲起,逐层深入到细节,消化完再往下走。这是一个不断提问、不断深挖的迭代过程。过程中还能让它画图、从最基础讲起、举大量例子——以前没有哪个老师能为你一个人讲到这个程度。

所以在我看来,会提问,就是AI时代最重要的自学能力。

诊断 Python 占用一个 CPU 的问题

早上刚打开 MAC 笔记本 5分钟, 就提示我电池快没电了, 于是感觉给它供电. 紧接着, 就听到风扇呼呼作响, 于是查看 Activity Monitor, 发现有个 Python 进程几乎占用一个CPU, 感觉有 bug, 也许是项目的那个代码没写好. hmm, 生产环境不会也是这样吧?

Activity_Monitor.png

先确定进程信息

有了进程号, 进程很容易确定.

% ps aux  | grep 24315
xiatian          24315  99.3  0.0 36938856   9804   ??  R    Mon10PM 1044:13.73 /usr/local/Cellar/[email protected]/3.11.11/Frameworks/Python.framework/Versions/3.11/Resources/Python.app/Contents/MacOS/Python /Users/supra/work/projects/agent-ui/main.py

再做 CPU profiling

py-spy 是一个非常流行的采样 profiler,可以在不修改代码的情况下分析任何正在运行的 Python 程序。

pip install py-spy
py-spy record -o profile.svg --pid 24315

火焰图如下:
profile_svg.png

前面三行都是 Python 自带 python3.11/threading.py 里面的代码. 最后一行是 concurrent/futures/thread.py) 的代码, 代码如下:

    try:
        while True:
            work_item = work_queue.get(block=True)
            if work_item is not None:
                work_item.run()
                # Delete references to object. See issue16284
                del work_item

                # attempt to increment idle count
                executor = executor_reference()
                if executor is not None:
                    executor._idle_semaphore.release()
                del executor
                continue

            executor = executor_reference()
            if _shutdown or executor is None or executor._shutdown:
                # Flag the executor as shutting down as early as possible if it
                # is not gc-ed yet.
                if executor is not None:
                    executor._shutdown = True
                # Notice other workers
                work_queue.put(None)
                return
            del executor
    except BaseException:
        _base.LOGGER.critical('Exception in worker', exc_info=True)

看上去从 work_queue.get(block=True) 拿 item , 每次拿到的都是 None. 那么下一个问题就是这个 work_queue 从哪里来的, 为什么会以极快的速度返回一个 None?

交叉验证

要回答上面的问题, 就要深入内存去看这个 work_queue 从哪里来的, 它引用了那些对象, 哪些对象还在引用它. 这样才能确定这个 work_queue 有什么问题. 但是, 在此之前, 我们可以去production 看一下, 看看在prod 的 Linux 上是不是有类似的问题, 结果发现完全没问题, 并且它运行的时间更长.

于是猜测, 这个问题难道只是个 MAC 有关? 或者和 MAC 休眠有关?

python Async Sync 函数

这篇是比较好的一篇文章: https://docs.chainlit.io/guides/sync-async

  1. sync -> sync -> ok
  2. sync -> async -> asyncio.run() | asyncio.get_event_loop().run_until_complete(async_func())
  3. async -> sync -> ok | await asyncio.to_thread(sync_task)
  4. async -> async -> await
  5. Python 的 asyncio 库是标准库中用于编写异步 I/O 代码的核心框架,自 Python 3.4 引入后成为构建高性能网络应用和并发程序的重要工具. asyncio 是 Python 异步编程的基石,适合需要高效处理 I/O 并发的场景。通过事件循环和协程机制,开发者能以同步风格编写异步代码,显著提升程序性能.
  6. Asyncer 是一个用于简化 Python 异步编程的第三方库,它通过更优雅的语法和高级功能让异步代码更易编写和维护. 相比标准库 asyncio,Asyncer 通过更高层次的抽象降低了异步编程门槛,适合需要快速开发高并发应用的场景.
  7. Python 的 anyio 库是一个​​跨异步框架的统一接口库​​,旨在简化异步编程的复杂性,同时兼容主流异步模型(如 asyncio 和 trio)。

证书的生成和验证

1. 安装 OpenSSL

首先,确保您的系统上安装了 OpenSSL。在大多数 Linux 发行版中,您可以使用包管理器来安装它。

例如,在 Ubuntu 上:

sudo apt update
sudo apt install openssl

2. 生成私钥

私钥是安全证书的核心,应该保密。使用以下命令生成私钥:

openssl genpkey -algorithm RSA -out private.key -pkeyopt rsa_keygen_bits:2048

这里,-algorithm RSA 指定了使用 RSA 算法,-out private.key 指定了输出文件名,rsa_keygen_bits:2048 设置了密钥长度。

3. 生成 CSR(证书签名请求)

CSR 是发送给证书颁发机构以获取签名证书的请求。生成 CSR 时,系统会提示您输入一些信息,如国家、州、组织等。

openssl req -new -key private.key -out certificate.csr

4. 生成自签名证书

在测试环境中,您可以使用 OpenSSL 生成自签名证书。以下是命令:

openssl x509 -req -days 365 -in certificate.csr -signkey private.key -out certificate.crt

这里,-days 365 设置了证书的有效期限为一年,-in certificate.csr 指定了 CSR 文件,-signkey private.key 指定了用于签名的私钥,-out certificate.crt 指定了输出证书文件。

5. 查看证书信息

您可以使用以下命令来查看证书的内容:

openssl x509 -in certificate.crt -text -noout

可以看到之前的设置内容:

 Signature Algorithm: sha256WithRSAEncryption
Issuer: C=CN, ST=SH, L=Shanghai, O=tianxiaohui.com, CN=10.236.90.154, [email protected]
Subject: C=CN, ST=SH, L=Shanghai, O=tianxiaohui.com, CN=10.236.90.154, [email protected]

6. 使用证书

生成的 certificate.crt(证书)和 private.key(私钥)可以用于配置 HTTPS 服务器,如 Apache 或 Nginx。

安装证书到 Flask server app

一个基于 Flask 到本地app:

from flask import Flask

app = Flask(__name__)


@app.route('/', methods=['GET', 'POST'])
def index():
    return 'Hello, World!'
# ... 你的应用代码 ...


if __name__ == '__main__':
    # 指定证书和私钥的路径
    app.run(host='0.0.0.0', port=443, ssl_context=('certificate.crt', 'private.key'))

本地访问

本地浏览器访问的到下面的错误:
Your connection is not private
使用 python requests 访问:

requests.exceptions.SSLError: HTTPSConnectionPool(host='10.236.90.154', port=443): Max retries exceeded with url: / (Caused by SSLError(SSLCertVerificationError(1, '[SSL: CERTIFICATE_VERIFY_FAILED] certificate verify failed: self signed certificate (_ssl.c:1129)')))