基于多Agent协同的Kubernetes故障自动修复

学习目标

学完本章你应该能够：

1. 讲清多 Agent 的三种协同方式——协作模式、管理员模式、分层代理模式——各自的调度思路、适用场景与实现难度。
2. 说出多 Agent 的典型应用场景（数学问题、检索增强、现实世界问题、多 Agent 编码、动态群组对话）以及各自拆分了哪些角色。
3. 引用微软论文的多 Agent 效果数据，解释为什么把任务拆给多个 Agent 能明显提升准确率。
4. 画出 Supervisor 模式下 K8s 故障自动修复的整体架构：管理员 Agent 如何把问题分流给 Researcher / Code / AutoFixK8s / HumanHelp。
5. 用 LangGraph 写出一个「节点 + 条件边的有环图」，并讲清 finish 标识如何让循环终止。
6. 写一个能根据 K8s 事件生成 patch JSON、再调用 K8s API 自动修复镜像版本错误 / OOM 的 AutoFix 工具。
7. 通过实时监听 K8s Warning 事件，把让「发现 → 分流 → 修复 → 确认 → 退出」的闭环自动化。

前置知识：

• Python 基础与 Jupyter Notebook 使用习惯。
• Kubernetes 基础：Deployment / Pod / Event、kubectl apply / kubectl patch。
• LangChain / LangGraph 基本概念（Chain、Tool、Node、State）；对 Agent 有初步认知（可先看第 4 章 AIOps Agent 入门）。
• 一个 OpenAI 兼容的 API Key 和一个 Tavily 搜索 Key。

本章你会动手做的事：

1. 搭建一个 LangGraph 工作流，把 Researcher / Code / AutoFixK8s / HumanHelp 四个 Agent 加上一个 Supervisor 串成有环图。
2. 用 OpenAI 让 AutoFixK8s 根据事件生成 patch JSON，调用 Kubernetes Python SDK 自动修正 nginx:lates 这样的镜像错误。
3. 实时监听 K8s Warning 事件，验证集群自动修复镜像错误与内存 OOM 两类故障，并在无法自动修复时收到飞书通知。

为什么需要多 Agent 协同

先问一个问题：单个大模型 Agent 能不能直接把 K8s 故障修好？

能，但很勉强。原因在于单 Agent 在一条回复里要同时干好几件事：判断"这个报错是什么"、决定"要不要用工具"、动手执行修复、检查结果。每一步都要消耗"推理能力"，而大模型的幻觉会让它在某一步出错，错误还会一路往下累积。

多 Agent 的思路是：把一件复杂任务拆成几个角色，每个角色只负责自己擅长的一小块，再想办法让他们协作起来。就好比看病，你不会让一个人既当分诊台护士、又当主治医生、又当手术师、又当发药窗口——分工之后，每一环都能做得更专业，也更容易追踪到底是哪一环出了问题。

多 Agent 的三种协同模式

这是本模块第一个重点。多 Agent 要自动化处理任务，就得回答三个问题：每个 Agent 是什么角色、怎么和其他 Agent 协作、任务怎么分配调度。不同的答案，就形成了三种协同模式。

模式一：协作模式（Collaboration）

协作模式用一个路由器（Router）把任务路由给不同的专家 Agent，形成循环，直到拿到 finish 才退出。

以"生成某个地区过去十年平均气温图表"为例：

1. 用户请求先进到 researcher（研究员）Agent，它去搜索过去十年的气温数据。
2. 信息进入 router（路由 Agent），由它判断下一步是继续调用工具（function calling）、还是信息不够需要再回到 researcher，还是一个循环。
3. 数据收集齐全后，router 调用 chart generator（代码执行 Agent），由它真正运行一段 Python 代码把图表画出来——因为图表是大模型"生成"不出来的，必须借代码运行。
4. 大模型判断当前任务已进入 finish（终止）标识，退出整个循环，把结果返回用户。

协作模式的核心思想是任而置之 / 就近指派：为每个任务或垂直领域创建一个专门代理，router 的功能就是"把任务路由给正确的专家"。

flowchart LR
    U[用户请求] --> R[Researcher
网络搜索]
    R --> RT{Router
判断下一步}
    RT -->|信息不够| R
    RT -->|需要执行| C[Chart Generator
代码执行]
    C --> RT
    RT -->|finish| E[结束, 返回结果]

模式二：管理员模式（Supervisor）

管理员模式引入一个主管 Agent（Supervisor）。它先把用户的请求做意图识别 / 拆解，然后把任务分发给 A1 / A2 / A3 多个 Agent。与协作模式不同，Supervisor 是中心化调度：

• 它会根据 A1 的返回结果再调度 A2，根据 A2 的结果再调度 A3，串行推进。
• 调每个 Agent 时都是一个循环——反复调用到它返回正确结果为止，然后才考虑下一条路径。

管理员模式最大的特点是主管 Agent 的能力决定了整体效果：因为它多了一次"先决策、再执行"的推理，把"决策"和"执行"解耦开，准确率通常更高。

模式三：分层代理模式（Hierarchical）

分层代理模式有点像把模式一和模式二组合成多层级的组织结构：层级很深，有顶层主管，也有中间层主管，每个 Agent 还被分成不同的 team / 职能。你可以把它理解成一家公司的组织架构——按职能拆得很细、多级多层的代理体系。

• 实现难度：分层代理模式最复杂，实际工作里用得相对少（管理员模式基本已满足日常需求）。
• 调度器：由多层的多个主管构成，中间还有控制节点衔接。

三种模式对比

这张图把三种模式的"调度形态"画到一起对比——协作模式是"自己人互相转"，管理员模式是"一个老大指挥一串兵"，分层模式是"多级组织层层上报"。看边和层级就能分清谁简单谁复杂。

flowchart TB
    subgraph 协作[协作模式 就近指派]
        U1[用户] --> R1[Router] --> E1[专家Agent]
        E1 --> R1
    end
    subgraph 管理员[管理员模式 中心化调度]
        U2[用户] --> S[Supervisor]
        S -->|调度| A1[Agent A1] --> S
        S -->|调度| A2[Agent A2] --> S
        S -->|调度| A3[Agent A3] --> S
    end
    subgraph 分层[分层代理模式 多级多主管]
        T1[顶层主管] --> M1[中间主管] --> L1[团队Agent]
        T1 --> M2[中间主管] --> L2[团队Agent]
    end

多 Agent 的典型应用场景

多 Agent 不只是理论，它在很多真实场景里都用得上。下面五个是视频里提到的典型例子。

场景一：解决数学问题

ChatGPT 刚出来时，你问它简单数学题它经常算错——因为它本质是「预测下一个字符的概率」，而不是真在做运算。解决方法是引入一个数学运算专家 Agent：让大模型先生成一段解答代码（通常是 Python），再交给专家 Agent 在 Sandbox 环境里把代码真正运行起来，把运行结果作为答案。

因为是真正跑代码的结果，所以准确率几乎是 100%（除非前置生成的代码本身有误）。我们要做的往往只是加一层简单错误判断：代码有异常就跑不起来，那就反馈回去让它重新生成代码再跑，运行结果符合预期才返回给用户。

场景二：检索增强生成（RAG）聊天

单个检索 Agent 做 RAG，准确率往往不够。于是引入用户代理 + 检索代理的协作：

• 用户代理做判断：检索到的内容如果不包含问题的答案，就控制检索代理检索更多上下文，逐步递增，直到满足条件（是个循环）。
• 用户代理还会把检索代理生成的答案做再验证：如果答案没真正解决用户的提问，就让它重新检索。
• 通过这种多重校验循环，可以极大提升检索增强生成的准确率。

场景三：解决现实世界问题（ALF-World）

大模型在解决现实物理世界问题时，常常缺乏常识认知。比如：

• 你让它"检查一个物体"，它可能忽略"必须先找到并拿走这个物体"才能检查。
• 你让它"使用某个目标物体"，它可能忽略"必须先找到它所在位置"。

用多 Agent 解决这个问题的方法是拆成两个角色：**助理代理（Assistant Agent）**负责做常识问题的规划，**执行代理（Executor Agent）**负责实际操作执行。把"懂常识做规划"和"实际动手"分开，就能避免常识遗漏。

场景四：多 Agent 编码

直接让大模型写好一段代码，常常在性能、安全性上有可优化点。多 Agent 编码的思路是：除了写代码的 Agent，额外加一个安全 / 评审 Agent：

1. 写代码 Agent 先产出代码。
2. 安全 Agent 负责 review 这份代码，评估它的性能、效果、安全性。
3. 根据评审结果，要求写代码的 Agent 进一步改写。

这很像"大模型反思"的过程——先写出来，再单独用一个流程去检查和重写。它与吴恩达开源的 translation-agent 类似：翻译出一个结果后，用另一个 Agent 去检查翻译是否符合当地习惯和语法，不符就重写。

场景五：动态群组对话

动态群组对话与调度强相关。它由管理员 Agent + 多个发言人 Agent 组成：

• 管理员根据任务要求选择发言人，收集发言人生成的结果。
• 管理员把结果广播（broadcast）给其他发言人，让每个发言人掌握全局上下文（global context）。
• 这样每个 Agent 在做下一步任务时，都知道上一步哪个 Agent 做了什么、产生了什么结果，可以基于前人的结果继续思考优化。

多 Agent 的效果提升（微软论文）

多 Agent 到底有没有用？视频引用了微软的一篇论文，直接看数据。

几点解读：

• 数学问题在多 Agent + AutoGen 的框架下能达到 69.48%，远高于 ReAct（23.3%）等其他写法。
• 现实问题用 3 个 Agent 达到 77%，有效解决了前面说的"大模型缺常识、容易遗漏前置条件"的问题。论文里没写用更多 Agent 的情况，但把问题拆得更细、用 4 个甚至更多 Agent，理论上可能进一步提高准确率。
• 结论很直接：把任务拆给多个 Agent，能显著提升准确率——这正是多 Agent 协同的价值所在。

实战设计：Supervisor 模式下的 K8s 故障自动修复

理论讲完，进入实战。目标是：多个 Agent 协同、自动发现并自动修复 K8s 故障。整体采用前面说的管理员模式（模式二）。

整体架构由 5 个 Agent 构成：

flowchart LR
    K[K8s 集群 Event] --> S[Supervisor
管理员Agent]
    S -->|分流| R[Researcher
网络搜索]
    S -->|分流| C[Code
代码执行]
    S -->|分流| F[AutoFixK8s
自动修复]
    S -->|分流| H[HumanHelp
寻求人工帮助]
    F --> R
    F --> C
    F --> H
    H --> FF[飞书通知/建工单]
    S -->|finish| E[返回用户]

故障怎么分类？K8s 事件类型很多，比如：

• CrashLoopBackOff：容器反复崩溃重启。
• 镜像版本错误 / 拉不到镜像（ImagePullBackOff / ErrImagePull）。
• OOMKilled：内存溢出。
• 集群级问题（如资源不足、节点异常）——这类 AutoFixK8s 修不了，转给 HumanHelp。

管理员 Supervisor 的核心提示词

Supervisor 的提示词其实很简单，核心是让大模型在高阶成员里选一个 Agent 去调度，再根据返回结果判断是否完成：

您是一个管理员 Agent。
您负责管理下面这些成员 Agent 之间的对话，并且对给定的用户请求作出响应，
把任务派给不同的 Agent 执行，执行完后思考下一步行动。

成员：{members}      # research / code / autofix_k8s / humanhelp

每个 Agent 执行一项任务。您需要响应最终的状态：
如果问题已解决，请返回 {finish} 结束符；否则继续调度合适的 Agent。

关键点：Supervisor 拿到 AutoFix 的返回后，判断"修复成功 → 返回 finish，退出循环并返回用户；修复失败 → 继续调度其他 Agent 或再次循环，直到解决，或转给 HumanHelp"。

AutoFixK8s 的核心提示词

AutoFix 提示词的核心是让大模型扮演云原生技术专家，根据 K8s 报错事件 + Deployment 现有 YAML，生成一段能直接用于 kubectl patch 的 JSON：

你是一位云原生技术专家。
请根据 K8s 的报错事件信息（Event），结合下方的 Deployment 清单，生成一段全面的 patch JSON 用于修复问题。

例如：镜像版本错误则修正镜像；OOM 则调高内存 request/limit。
注意：不要提供其他无用的建议，直接返回 JSON；不要把 JSON 放在代码块里，以便直接获取。

拿到 patch JSON 后，直接调用 K8s 的 patch 命令把这段 JSON 应用到集群，就完成了自动修复。

核心实现（上）：依赖、内置工具与消息节点

下面进入真正的编码，视频里用 Jupyter Notebook 演示。标题就叫《多 Agent 协同自动修复 K8s 故障》。

第一步：安装依赖

需要几个关键包：langgraph、langchain、langchain-openai、langchain-community，以及 langchain-experimental（提供代码执行工具）和 kubernetes（K8s SDK）。

pip install langgraph langchain langchain-openai langchain-community langchain-experimental kubernetes

第二步：配置环境变量

定义一个函数，若环境变量未设置则弹窗提示输入：

from getpass import getpass
import os

def _set_if_undefined(var: str):
    if not os.environ.get(var):
        os.environ[var] = getpass(f"Please provide {var}")

# OpenAI API Key
_set_if_undefined("OPENAI_API_KEY")
# Tavily 搜索 API Key
_set_if_undefined("TAVILY_API_KEY")

第三步：导入内置工具（Researcher 和 Code 不需要自己写）

两个内置 Agent 直接从 LangChain 里导入即可：

from langchain_community.tools import TavilySearchResults
from langchain_experimental.tools import PythonREPLTool

# 网络搜索工具（Tavily）
tavily_tool = TavilySearchResults(max_results=5)

# 执行 Python 代码的工具
python_repl_tool = PythonREPLTool()

第四步：定义消息代理节点

把 Agent 的结果封装成 HumanMessage，方便后续传给 Supervisor 作为上下文：

from typing import Literal, Annotated, Optional
from langchain_core.messages import HumanMessage, BaseMessage
from typing_extensions import TypedDict
import operator

def agent_node(state, agent, name):
    result = agent.invoke(state)
    # 组装成 HumanMessage，取最后一条消息的内容
    return {
        "messages": [HumanMessage(content=result["messages"][-1].content, name=name)]
    }

核心实现（上）：管理员 Supervisor Agent

定义成员与系统提示词

Supervisor 需要管理 4 个工作 Agent + 1 个自身（共 5 个）：research、code、autofix_k8s、human_help。

members = ["research", "code", "autofix_k8s", "human_help"]

# 系统提示词：告诉 supervisor 负责调度多个 Agent
system_prompt = (
    "您是一个管理员 Agent，负责管理下面这些成员 Agent 之间的对话：\n"
    "{members}。\n"
    "对于给定的用户请求，请把任务派给合适的成员执行，然后思考下一步行动。\n"
    "每个 Agent 执行一项任务。如果问题已解决，请返回 {finish}；"
    "否则继续调度合适的 Agent。"
)
options = ["FINISH"] + members

定义 Supervisor 的响应结构与 Agent

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_openai import ChatOpenAI
from pydantic import BaseModel, Field

class RouteResponse(BaseModel):
    next: Literal["FINISH", "research", "code", "autofix_k8s", "human_help"]

llm = ChatOpenAI(model="gpt-4o")

def supervisor_agent(state):
    prompt = ChatPromptTemplate.from_messages(
        [
            ("system", system_prompt + "\n请根据上面的对话选择下一个要执行的 Agent 或返回 FINISH。"),
            MessagesPlaceholder(variable_name="messages"),
            ("system", "选择一个下一步行动。{options}（选项用逗号分隔）。"),
        ]
    ).partial(options_str=", ".join(options), members=", ".join(members))

    # 用结构化输出让模型只返回"下一步该调用谁"
    supervisor_chain = (
        prompt
        | llm.with_structured_output(RouteResponse)
    )
    return supervisor_chain.invoke(state)

这里最值得注意的就是 with_structured_output(RouteResponse)：让大模型只输出一个"下一步要调用哪个 Agent"的枚举，而不是一大段自由文本。大模型的调度决策就这么被"结构化"地拿到手。

定义 AgentState

存放会话状态——消息列表 + 下一步路由：

class AgentState(TypedDict):
    messages: Annotated[list[BaseMessage], operator.add]
    next: str

核心实现（上）：AutoFixK8s 自动修复工具

这是真正需要自己实现的 Agent（前面的 research / code 只是引用内置工具）。

构造工具函数

import json, yaml
from kubernetes import config, client
from openai import OpenAI

# 加载本地 kubeconfig
config.load_kube_config()  # 或 config.load_kube_config_from_dict(...)
apps_v1 = client.AppsV1Api()

def autofix_k8s(deployment_name: str, namespace: str, event_message: str) -> str:
    """自动修复 K8s 问题。"""
    # 1. 根据工作负载名称获取 Deployment 的清单
    deployment = apps_v1.read_namespaced_deployment(name=deployment_name, namespace=namespace)

    # 2. 转成字典，去掉无用字段（state、metadata 等会占用上下文）
    deployment_dict = deployment.to_dict()
    if deployment_dict.get("metadata"):
        deployment_dict["metadata"].pop("resourceVersion", None)
    if "status" in deployment_dict:
        deployment_dict.pop("status", None)

    # 3. 请求 OpenAI 生成修复用的 patch JSON
    client = OpenAI()
    resp = client.chat.completions.create(
        model="gpt-4o",
        response_format={"type": "json_object"},
        messages=[
            {"role": "system", "content": "你是一个用于输出 JSON 的助手。"},
            {"role": "user", "content": (
                "你是一位云原生技术专家，请根据 K8s 的报错事件 Event 信息 + 下方 Deployment 清单，"
                "生成全面的 patch JSON 以修复问题，不要提供其他无用建议，直接返回 JSON，"
                "不要把 JSON 放在代码块里。" +
                f"\nEvent 信息：{event_message}\nDeployment 清单：{yaml.dump(deployment_dict)}"
            )},
        ],
    )
    patch_text = resp.choices[0].message.content
    patch_json = json.loads(patch_text)

    # 4. 把 patch JSON 直接改成 Kubernetes patch 应用到集群
    try:
        apps_v1.patch_namespaced_deployment(name=deployment_name, namespace=namespace, body=patch_json)
        print("TRAER 已自动修复")
        return "工作负载自动修复成功"
    except Exception as e:
        # 修复失败，把错误返回给大模型，大模型会继续调度或转 HumanHelp
        return f"修复失败: {e}"

为什么这个函数能做到自动修复？ 核心就一句话：LLM 生成的这段 patch JSON，本身就是用来修复问题的那段 patch。我们只要把它应用到集群即可。

• 镜像版本错误 → 生成的 patch 把 image 修正为正确 tag。
• OOM → 生成的 patch 把内存 request/limit 调大。

而当 patch 失败（如 fix 失败），函数会把错误返回给大模型——大模型收到"调用 autofix 失败"的错误后，就会自主去调用 HumanHelp，告诉人类"这个问题我搞不定，需要你介入"。

核心实现（上）：HumanHelp 人工帮助通知

用于在无法自动修复时，通过飞书机器人 Webhook 发送消息通知：

import requests

def human_help(event_message: str) -> str:
    """无法修复问题时，寻求人工帮助。发送飞书消息通知。"""
    url = "https://open.feishu.cn/open-apis/bot/v2/hook/xxxx"  # 你的飞书机器人 Webhook
    headers = {"Content-Type": "application/json"}
    data = {"msg_type": "text", "content": {"text": event_message}}
    resp = requests.post(url, headers=headers, json=data)
    return str(resp.status_code)

实际工程里，这里可以做很多额外工作：创建运维工单、给问题分类、艾特对应负责人等等。消息内容会包含 K8s 事件，或它解决过程中产生的错误。

核心实现（下）：用 LangGraph 搭建工作流

LangGraph 的核心是有向有环图（Graph），把上面那些 Agent 连线起来。工作流定义涉及三件事：创建 Agent、添加节点、添加路由条件边。

创建 4 个工作 Agent

research 和 code 用 LangGraph 内置的 create_react_agent 一行创建，autofix_k8s 和 human_help 是上面写的带 @tool 装饰的函数：

from langgraph.graph import END, StateGraph, START
from langgraph.prebuilt import create_react_agent
from langchain_core.tools import tool

# research agent（内置，传入 Tavily 搜索工具）
research_agent = create_react_agent(llm, [tavily_tool])

# code agent（内置，传入 Python REPL 工具）
code_agent = create_react_agent(llm, [python_repl_tool])

# autofix_k8s agent（自定义工具，函数带 @tool 注解后可直接作为工具传入）
@tool
def autofix_k8s(deployment_name: str, namespace: str, event_message: str) -> str:
    """自动修复 K8s 问题。"""
    return _do_autofix(deployment_name, namespace, event_message)

autofix_agent = create_react_agent(llm, [autofix_k8s])

# human_help agent
human_help_agent = create_react_agent(llm, [human_help])

为每个 Agent 创建节点

from langgraph.prebuilt import ToolNode

# 节点命名，后面连线时要用，比如 autofix_k8s 节点
research_node = ToolNode(research_agent)
code_node = ToolNode(code_agent)
autofix_node = ToolNode(autofix_k8s)
human_help_node = ToolNode(human_help)

构建有环图并添加节点

workflow = StateGraph(AgentState)

# 依次添加所有节点
workflow.add_node("research", research_node)
workflow.add_node("code", code_node)
workflow.add_node("autofix_k8s", autofix_node)
workflow.add_node("human_help", human_help_node)
workflow.add_node("supervisor", supervisor_agent)

# 初始边：从 START 进入 supervisor
workflow.add_edge(START, "supervisor")

添加路由条件边（关键！）

这一步是形成"环"的核心。Supervisor 返回 next 之后，图要决定：跳到对应的 Agent，还是跳到 END。

# 条件映射：除 FINISH 外，其余 route 到对应 Agent（形成环）
conditional_map = {k: k for k in members}

def route_supervisor(state):
    if state["next"] == "FINISH":
        return "END"
    return state["next"]

# supervisor 之后的边：根据 next 走对应 Agent 或 END
workflow.add_conditional_edges(
    "supervisor", route_supervisor, {"END": END, **conditional_map}
)

# 每个工作 Agent 完成后，都回到 supervisor
for member in members:
    workflow.add_edge(member, "supervisor")

graph = workflow.compile()

这里的逻辑：

• conditional_map 把 research→research、code→code、autofix_k8s→autofix_k8s、human_help→human_help 一一对应，还把 FINISH→END 接上。
• add_conditional_edges 是虚线（条件判断），围绕 supervisor 形成一个条件环；所有工作 Agent 修复完都 add_edge 回到 supervisor（实线）。
• 因此每个 Agent 与 supervisor 之间都有边：Agent 干完 → 回 supervisor → supervisor 再决定下个 Agent 或 finish。环由此产生。

把绘制出来的图打印一下，可以看到：从 START 进来，给每个 Agent 都加了边，每个都是循环，最后还有一个 FINISH 到 END——和架构设计图完全一致。

这张图把 LangGraph 的"有环图"结构画清楚：实线是 Agent 干完必回 supervisor，虚线是 supervisor 根据 next 条件判断往哪走。环就在"supervisor → 某 Agent → 回 supervisor"这条回路上。

flowchart LR
    ST[START] --> SUP[supervisor 管理员]
    SUP -->|next=research| RS[research]
    SUP -->|next=code| CD[code]
    SUP -->|next=autofix| AF[autofix_k8s]
    SUP -->|next=human_help| HH[human_help]
    SUP -->|next=FINISH| EN[END 结束]
    RS --> SUP
    CD --> SUP
    AF[autofix_k8s] --> SUP
    HH --> SUP

核心实现（下）：监听真实 K8s 事件并跑全流程

最后把"手动传入事件"升级为"自动监听 K8s 事件"。思路是：监控 K8s Warning 事件 → 取出对应工作负载 → 调用 graph 处理 → 修复成功则 Supervisor 判 finish。

from kubernetes import client, config, watch

config.load_kube_config()

# 需要 apps/v1（Deployment）和 core/v1（Pod / Event）
apps_v1 = client.AppsV1Api()
core_v1 = client.CoreV1Api()

# 封装 graph 运行
def run_stream(deployment_name, namespace, event_message):
    inputs = {"messages": [HumanMessage(content=f"{deployment_name}|{namespace}|{event_message}")]}
    for s in graph.stream(inputs, stream_mode="values"):
        if "END" not in s:
            print(s)
    print("===== 本轮 Agent 处理完成 =====")

事件监听主循环（只处理 Warning 级别事件）：

namespace = "default"
w = watch.Watch()
for event in w.stream(core_v1.list_namespaced_event, namespace=namespace, timeout_seconds=180):
    event_obj = event["object"]
    if event_obj.type == "Warning":           # 只关注报错的 Warning 事件
        pod_name = event_obj.involved_object.name
        reason = event_obj.reason
        message = event_obj.message
        print(f"WARNING event: pod={pod_name}, reason={reason}, message={message}")

        try:
            # 事件产生时 Pod 可能已不存在（正在被杀/重建），要处理异常
            pod = core_v1.read_namespaced_pod(name=pod_name, namespace=namespace)
        except client.exceptions.ApiException:
            # Pod 已不存在，忽略该事件
            continue

        # 取 Deployment 名：简单起见用 labels 里的 app 字段
        deployment_name = pod.metadata.labels.get("app")
        if not deployment_name:
            continue

        try:
            run_stream(deployment_name, namespace, f"{reason}|{message}")
            time.sleep(10)                    # 每轮休眠 10 秒再继续监听
        except Exception as e:
            print(e)
            continue

关于"取工作负载名"这点视频里有专门提示：生产环境最好用 ownerReference 字段逐级解析（ReplicaSet → Deployment），得到真正的 Deployment 名；演示里图省事用 labels["app"]。这也是后面 OOM 演示里为什么特意给工作负载打上 app 标签的原因。

案例演示一：硬编码事件自动修复（镜像版本错误）

先不接真实集群事件，手动硬编码一个事件验证 AutoFixK8s 能不能修好。构造一个错误的 Deployment（镜像 tag 故意写成 lates，正确应是 latest），并把 Back-off pulling image 错误手动传进去：

# 制造一个真实错误：镜像 tag 故意写错
kubectl run nginx --image=nginx:lates        # 故意写错

# 手动传入攒好的参数（先不解析集群事件）
inputs = {
  "messages": [HumanMessage(content="deployment_name=nginx, namespace=default, event=Back-off pulling image nginx:lates")]
}

运行 graph，观察流程：

1. Supervisor 说：“我要调用 autofix_k8s”。
2. AutoFix 识别出镜像标签写错了，生成修复 patch JSON。
3. 调用 K8s patch 应用，镜像被更正为 latest，Deployment 恢复 Running。
4. Supervisor 收到成功结果，返回 FINISH，整个 Agent 流程结束。

视频里也演示了不稳定的地方：有时 100Mi 的过程里 Supervisor 会中途调到 Code Agent（因为 AutoFix 之间有循环），也会因为提示词没写好而出现 must write at least messages and next 之类的报错。这些都是结构 / 提示词的调优过程——优化提示词后，最后稳定输出：

“the issue with nginx in default namespace (Back-off pulling image error) has been successfully auto-fixed” → 进入 FINISH，Agent 流程全部完成，Deployment 自动恢复正常。

案例演示二：无法修复时寻求人工帮助

构造一个"Cluster 级 / 修不了"的问题——比如一个叫 unknown 的 Deployment，事件是 “CPU 和内存不足，无法调度”。这种问题 AutoFixK8s 解决不了，就应该转给人类。

运行观察：

1. 先调 autofix_k8s，AutoFix 返回"这个问题是 K8s 集群级问题，解决不了"。
2. Supervisor 收到失败后，自主调度 human_help，把"unknown 的 CPU 和内存资源不足"发到飞书。
3. HumanHelp 发完飞书消息，返回成功 → Supervisor 判 finish，流程结束。

这里视频遇到一个典型调优点：HumanHelp 提示词写得不好时，会重复调用多次、连发好几条飞书消息（因为大模型认为自己还没真正完成任务）。解决办法是在 HumanHelp 的提示词里明确加上一句：“已经寻求了人类帮助，请等待人类介入” / “结束任务”。这样大模型就知道这一步已经"成功完成"，不会再重复调用。改完之后，最终只调一次 HumanHelp，正确把事件发到飞书。

案例演示三：监听真实事件自动修复闭环

把前面的硬编码注释掉，用真正的事件监听循环跑起来。Agent 一直在 Watch，一旦有新的 Warning 事件就处理。

演示 A：真实镜像版本错误（nginx:lates）

先在集群创建一个 nginx:lates（镜像 tag 写错）的 Deployment。监听循环立刻捕获到 Warning 事件 → 自动调 autofix_k8s → 成功修正为 latest → Supervisor 收到成功结果判 finish。去集群确认，Deployment 17 秒前已被自动修复为正确镜像并恢复 Running。

演示 B：真实 OOM 内存溢出

提供一个 OOM 压力工作负载（om_stress），它启动就消耗 200MB 内存，而 request/limit 只有 100Mi——一启动必然 OOMKilled、进入 CrashLoopBackOff。

监听循环捕获 OOMKilled 事件 → 自动调 autofix_k8s → LLM 生成 patch JSON：把内存 request/limit 调大（比如 request 200Mi / limit 400Mi）→ 应用 patch → Deployment 恢复 Running。

第一次修复可能不稳定（会反复尝试、续修），视频里还遇到一个干扰项：Docker Hub 网络问题让 AutoFix 误判成"镜像拉取错误"。这也是为什么先打好 app 标签能帮助正确解析出 Deployment 名，稳定修复。最终生成的 patch 显式修改了 container.resources 领域： request 和 limit 都被提高了 → patch 直接应用成功 → OOM 故障被自动修复。

这些都做完后，就实现了视频开头的目标：监听事件 → Supervisor 决策 → 把任务调度给合适的 Agent → 生成 patch JSON → 应用修复成功 → Supervisor 收到消息判 finish → 退出循环。

这张图把"发现 → 分流 → 修复 → 确认 → 退出"的闭环画清楚：事件进来先由 Supervisor 分流，能自动修的走 AutoFixK8s，修不了或集群级问题转给 HumanHelp，最终都以 FINISH 退出循环。

flowchart LR
    K[K8s Warning 事件] --> S[Supervisor
分流决策]
    S -->|可自动修复| A[AutoFixK8s
生成 patch JSON]
    A --> P[K8s API
应用修复]
    P -->|成功| FF[修复成功 → FINISH 退出]
    P -->|失败| H[HumanHelp
飞书通知人工]
    S -->|集群级问题| H
    H --> FH[等待人工介入 → FINISH]

踩坑与优化备忘

• Python REPL 代码执行有安全风险：务必用 Docker / Sandbox 隔离，别让大模型生成的代码直接在本机跑删除类危险操作。
• with_structured_output 决定调度质量：Supervisor 的 next 一定用结构化输出（枚举），别让它自由发挥成一段话。
• HumanHelp 会重复调用：提示词里要明确"已寻求人类帮助、等待介入 / 结束任务"，否则会连发好几条飞书通知。
• 事件产生时 Pod 可能已不存在：读取 Pod 要加 try/except，Pod 不存在就忽略该事件。
• 取工作负载名：演示用 labels["app"] 图省事；生产应该用 ownerReference 逐级解析到真实 Deployment 名。
• 监听是长循环：每处理一轮 time.sleep(10)，避免高频占用；Agent 只在有 Warning 事件时才真正跑起来。

总结

本章聚焦多 Agent 协同在 AIOps 里的价值，并用一个完整实战验证了这套思路。

• 三种协同模式：协作模式（router 就近指派）、管理员模式（中心化 supervisor 调度，本章实战采用）、分层代理模式（多层多级组织，最复杂）。管理员模式通过"把决策与执行解耦 + 多一次推理"，在准确率上更优且实现成本可控。
• 典型应用场景：数学问题、检索增强、现实世界问题、多 Agent 编码、动态群组对话——本质都是"按角色拆分 + 让角色协作"。
• 效果数据：微软论文显示多 Agent（AutoGen）能让数学问题准确率到 69.48%、检索增强召回 66.65%、现实问题 3-Agent 达 77%，显著高于单 Agent / ReAct。
• 实战架构：Supervisor + Researcher + Code + AutoFixK8s + HumanHelp 五个角色；AutoFixK8s 用 LLM 生成可 patch 的 JSON 再调 K8s API 实现修复，HumanHelp 在修复失败 / 集群级问题时发飞书通知。
• 关键技术：LangGraph 有向有环图 + 条件边 + finish 结束符；结构化输出驱动调度决策；实时监听 Warning 事件实现"发现 → 分流 → 修复 → 确认 → 退出"的自动闭环。

真正让这套系统"可落地"的关键，是清晰的 Agent 分工 + 稳定的结构化调度 + 明确的结束条件（finish）——三者缺一，多 Agent 就会沦为"多个大模型在乱喊"。

自测题与动手练习

自测题（合上书能答出来，才算懂）：

动手练习（建议真做一遍）：

1. 用 LangGraph 搭一个最小 Supervisor 有环图，只放 Research + Code 两个 Agent，验证 add_node / add_conditional_edges / FINISH 的完整流程，并把绘制的图打印出来对照设计。
2. 把 autofix_k8s 函数里"取工作负载名"的逻辑从 labels["app"] 改为用 ownerReference 逐级解析（ReplicaSet → Deployment），跑通后再用真实镜像错误 / OOM 场景验证。
3. 把 human_help 从"发飞书消息"扩展为"创建工单 + 按错误类型艾特负责人"，并优化提示词让它对同一故障只通知一次。

本章小结