Flink Agents 与 AI 流式应用 —— 第 16 章 实战一：告警自愈 Agent

Flink Agents 与 AI 流式应用 —— 第 16 章 实战一告警自愈 Agent章节定位第六篇终局实战之一。本章把第 1~15 章全部能力总装成一个生产级告警自愈系统从事件接入到复检闭环的完整链路、工程结构、关键代码、部署配置与上线检查单。目标读完本章你能照着搭出一个可上线的告警自愈 Agent。版本基线Apache Flink 2.3.0 Flink Agents 0.3.xJDK 17 Kafka 3.x Doris 4.1 Paimon 1.4章节导读16.1 需求与目标16.2 总体架构16.3 工程结构与核心代码16.4 关键链路串联16.5 部署配置16.6 上线检查单与验收16.1 需求与目标16.1.1 业务需求现状痛点: 日均 3 万条 TSP 告警,人工运维团队 8 人,人均日处理 300 → 疲态、漏判、处置不一致、经验无法沉淀 自愈目标: ① 70% 告警自动完成诊断建单(不限车型/故障类型) ② 高危告警 100% 转人工,不误判不遗漏 ③ 诊断依据可完整追溯(审计/复盘/合规) ④ 单事件综合成本 ≤ 预算红线(14.2.3)16.1.2 技术目标□ 端到端延迟 p99 ≤ 90s(告警到达 → 工单创建) □ 可用性 99.9%(含降级期间,13.4) □ 零重复工单(event_id 全链路幂等) □ refs 真实率 100% □ 通过三阶段 SOP 上线(15.6)16.2 总体架构┌─────────────────────────────────────────────────────────────────┐ │ 告警自愈系统全景 │ │ │ │ Kafka tsp-alert │ │ │ │ │ ▼ │ │ ┌─────────┐ 同码5分钟窗口 ┌──────────┐ │ │ │ 聚合防抖 │────────────────→│ 分诊 Agent│ 小模型黑板 │ │ │ (11.2) │ │ (10.5.2) │ │ │ └─────────┘ └────┬─────┘ │ │ │自检通过→直出旁路 │ triage.completed │ │ ▼ ▼ │ │ ┌─────────────┐ ┌────────────────┐ │ │ │ ML_PREDICT │ │ 专科 Agent ×3 │ 大模型三道闸 │ │ │ 低严重度直出 │ │ 电池/电机/热管理 │ 记忆召回工具集 │ │ └─────────────┘ └───────┬────────┘ │ │ │ diagnosis.completed │ │ ┌─────────────┼─────────────┐ │ │ ▼ ▼ ▼ │ │ ┌──────────┐ ┌───────────┐ ┌──────────┐ │ │ │置信度≥0.8 │ │0.5~0.8 │ │0.5/降级 │ │ │ │自动建单 │ │建单复核 │ │转人工队列 │ │ │ └────┬─────┘ └─────┬─────┘ └────┬─────┘ │ │ ▼ ▼ ▼ │ │ ┌────────────────────────────────┐ │ │ │ 工单 Agent(零LLM,幂等收口) │ │ │ └───────┬────────────┬───────────┘ │ │ ▼ ▼ │ │ Doris 工单 Paimon 审计(全量决策链) │ │ │ │ │ │ ▼ ▼ │ │ 复检结果回填 ──→ 反馈闭环(14.4) ──→ 病程库(8.3) │ └─────────────────────────────────────────────────────────────────┘ 旁路基建: MCP Server 群(5.4) / 监控五件套(14.5) / 混沌演练(13.6) / 三阶段放权控制(15.6)16.3 工程结构与核心代码16.3.1 模块划分alert-healing-agent/ ├── pom.xml # 2.1 依赖 ├── src/main/java/com/fleet/healing/ │ ├── HealingJob.java # 主入口(16.3.2) │ ├── event/ # 事件定义(第 2 章命名规范) │ │ ├── AlertEvent.java │ │ ├── TriageOutput.java │ │ ├── DiagnosisResult.java │ │ └── ... │ ├── agent/ │ │ ├── TriageAgent.java # 10.5.2 │ │ ├── SpecialistAgents.java # 10.5.3 │ │ └── TicketAgent.java # 10.5.3 │ ├── tool/ # 工具集(第 4 章) │ │ ├── ReadOnlyTools.java │ │ └── ActionTools.java # 幂等三保险 │ ├── memory/ │ │ ├── SessionStore.java # 第 7 章 │ │ └── MedicalRecall.java # 8.3.4 │ ├── guard/ │ │ ├── TokenBudgetGuard.java # 9.2.2 │ │ ├── CostCircuitBreaker.java # 13.5.2 │ │ └── ActionPolicy.java # 15.1.2 │ ├── audit/ │ │ └── AuditSink.java # 12.3.3 │ └── config/ │ └── StageConfig.java # 15.6 阶段控制 └── src/main/resources/ ├── prompt/ # 提示词(配置中心覆盖) └── application.conf16.3.2 主入口publicclassHealingJob{publicstaticvoidmain(String[]args)throwsException{StreamExecutionEnvironmentenvStreamExecutionEnvironment.getExecutionEnvironment();// ── 硬约束配置(7.3.2 全套) ──env.setStateBackend(newForStStateBackend());env.enableCheckpointing(30_000);env.getCheckpointConfig().enableUnalignedCheckpoints(true);env.setRestartStrategy(RestartStrategies.fixedDelayRestart(Integer.MAX_VALUE,Time.seconds(10)));// ── 输入(12.2.1 规范) ──DataStreamAlertEventalertsenv.fromSource(kafkaSource(),watermarkStrategy(),alert-source);// ── 防抖聚合(11.2) ──DataStreamAggregatedAlertdedupalerts.keyBy(e-e.vin#e.faultCode).window(TumblingEventTimeWindows.of(Time.minutes(5))).aggregate(newAlertAggregateFunction(),newAlertWindowProcessFunction());// ── Agent Pipeline(10.5.4) ──AgentPipelinepipelineAgentPipeline.builder().addAgent(newTriageAgent(lightModel())).addAgent(SpecialistAgents.battery(heavyModel(),batteryTools())).addAgent(SpecialistAgents.motor(heavyModel(),motorTools())).addAgent(SpecialistAgents.thermal(heavyModel(),thermalTools())).addAgent(newTicketAgent(ticketClient(),actionPolicy())).build();DataStreamEventresultspipeline.process(dedup.keyBy(AggregatedAlert::getVin));// ── 三出口(12.1) ──results.addSink(kafkaResultSink());// 协同事件results.map(r-(TicketRow)r).addSink(dorisSink());results.addSink(paimonAuditSink());// 全量审计env.execute(alert-healing-agent-v1);// pipeline.name}}16.3.3 专科 Agent 组装样例publicstaticReActAgentbattery(ChatModelmodel,ListTooltools){returnReActAgent.builder().model(model)// 提示词配置中心注入(6.3.3),版本化.systemPrompt(promptCenter.get(specialist.电池_v3)).tools(tools)// MCP 工具(第 5 章).memory(medicalRecall)// 病程召回(8.3.4).guard(ReActAgentConfig.builder()// 三道闸(6.2.2).maxSteps(8).loopDetection(LoopPolicy.SAME_CALL_MAX_2).maxDecisionTime(Duration.ofSeconds(60)).maxTokensPerEvent(15_000).build()).outputParser(Parsers.threeTier())// 6.4.1.semanticValidator(Validators.refsAndLevel())// 6.4.2.budgetGuard(newTokenBudgetGuard())// 9.2.2.circuitBreaker(costBreaker)// 13.5.2.build();}16.4 关键链路串联16.4.1 一条告警的完整旅程T0.0s Kafka: {event_id:A001,vin:LFV...,fault:P0AA6, alert:压差187mV,collect_time:...} T0.1s 防抖窗口: 同 vin 同码 5 分钟内无重复,单条通过 T0.2s 分诊: 小模型判 BATTERY(300 tokens), 黑板带档案dtc 释义 T0.3s 电池专科启动: - 会话开启 → 病程库召回(本车 3 月前同码,复检 OK) - Step1: dtc://P0AA6 → 标准释义(黑板命中,零调用) - Step2: query_history → 同上(记忆命中,零调用) - Step3: query_similar_cases → kb://doc/1024 (0.83) - Step4: 输出 {diagnosis:一致性劣化早期, action:P2 均衡充电,confidence:0.82, refs:[dtc://P0AA6,kb://doc/1024]} T8s 校验: refs 真实(全部有 ToolCall 记录)✓ 置信度 0.82 ≥ 0.8,非降级 ✓ → L1 自动建单 T9s 工单 Agent: 幂等键 A001 查无 → 建单 WO20261010001 T9s 三出口落库: Kafka 通知 / Doris 工单 / Paimon 审计 T9s 完成。trace_idA001 全程可查16.4.2 异常分支速查时刻异常防线T0.15 分钟内第 4 条同码防抖合并11.2T0.2分诊拿不准OTHER → 通用专科 人工复核标记T8refs 校验失败回喂修正一次 → 失败转人工6.4T9建单时 event_id 已存在幂等返回原单号4.4.2任意LLM 网关 500降级 L2 规则兜底13.4任意小时成本超限熔断 → L213.5.216.5 部署配置16.5.1 环境变量清单envsubst 渲染# deploy.env(不进 Git,由 CI 注入)KAFKA_BROKERSkafka:9092KAFKA_GROUP_AGENThealing-agent-prod# 配置化硬约束LLM_GATEWAYhttps://llm-gateway.internalLLM_API_KEYfrom-k8s-secretMCP_FAULT_CODE_URLhttp://mcp-fault-code:8090MEM0_ENDPOINThttp://mem0:8080DORIS_JDBCjdbc:mysql://doris-fe:9030/aiAUTONOMY_STAGESHADOW# 15.6 阶段控制HOURLY_TOKEN_BUDGET5000000# 13.5.2 熔断预算16.5.2 资源规划JobManager: 2C4G(HA 双实例) TaskManager: 4C16G × 4(ForSt 本地 SSD 挂载) 并行度: 16(Kafka 分区数对齐,前作硬约束) Slot 隔离: 专科 Agent slotSharingGroup(llm-heavy) —— 重 Join/重推理算子独立 Slot(前作硬约束) 容量估算(3 万告警/日): 防抖后 ~8 千事件/日,平均 0.1 eps,峰值 2 eps 每事件 ~4k tokens → 日耗 ~3.2M tokens → 预算 5M 小时级红线充足,成本可控16.5.3 启动验证# 连通性先行(前作硬约束: 远程模型调用前必须 curl 验证)curl-s$LLM_GATEWAY/v1/models-HAuthorization: Bearer$LLM_API_KEYcurl-s$MCP_FAULT_CODE_URL/health# 提交(禁 -j,jar 进 lib —— 前作硬约束)flink run-Dpipeline.namealert-healing-agent-v1\-Dtable.dml-synctrue\alert-healing-agent.jar16.6 上线检查单与验收16.6.1 上线检查单按章节溯源□ 第 2 章 环境基线对齐,JDK 17,jar 进 lib,pipeline.name □ 第 4 章 工具描述四要素抽检;动作类全部幂等;高危三保险 □ 第 5 章 MCP Server 独立部署,鉴权三层,审计落 Kafka □ 第 6 章 三道闸配置生效;提示词版本化进配置中心; 解析三层兜底 语义校验 □ 第 7 章 ForSt TTL 48h;状态监控三阈值接入 □ 第 8 章 记忆按 vin 隔离;只写终局结论; 召回过滤 recheck_ok1 □ 第 9 章 token 预算守卫上线;外部文本边界包裹 □ 第 10 章 分诊用小模型;工单 Agent 零 LLM □ 第 11 章 防抖窗口生效;Event Time Watermark 对齐 □ 第 12 章 审计表建妥(毫秒/1000,__DT__ 分区); ML_PREDICT 前置拦截率达标 □ 第 13 章 失败三分配置;降级阶梯演练通过; 成本熔断 白名单生效;混沌五场景演练通过 □ 第 14 章 四维指标暴露;红线告警配置;refs 真实率 P0 告警 □ 第 15 章 权限策略上线;红队样本测试通过; 脱敏三处内置;阶段一 SHADOW 启动16.6.2 三阶段验收数据真实目标值阶段一 SHADOW(第 2~5 周): 一致率 89.2% (达标线 85%) ✓ refs 真实率 100% ✓ 越权企图 0 ✓ → 准予进入阶段二 阶段二 APPROVAL(第 6~13 周): 驳回率 6.8% (达标线 10%) ✓ 驳回归因: 工具描述缺陷 3 例(已修),提示词边界 2 例(v3.1) → 准予进入阶段三(L0/L1 自动) 阶段三 LIMITED_AUTO(长期): 自动处置率 74% (目标 70%) ✓ 复检推翻率 12% (红线 25%) ✓ 端到端 p99 41s (目标 90s) ✓ 月均 LLM 成本 ≈ 预算 65% ✓16.6.3 运营节奏每日: 四维看板巡检(14.5),红线告警值班 每周: 周报 Agent 自动评估(14.6),badcase 评审 每月: 混沌演练(13.6),红队样本更新(15.2.3), 成本归因复盘(14.3),放权范围评审(15.6.3) 每季: 凭证轮换(15.4.2),记忆库换版评估(8.4.3)本章小结┌──────────────────────────────────────────────────────────────┐ │ 第 16 章 要点回顾 │ └──────────────────────────────────────────────────────────────┘ ✓ 架构总装: 防抖→分诊(小模型)→专科(大模型三道闸记忆) →三档分流→工单收口(零LLM);三出口(Kafka/Doris/Paimon) ✓ 能力溯源: 每个组件都对应前 15 章的具体节, 本章的价值在总装图而非新知识点 ✓ 工程落地: 模块结构/主入口/专科组装/部署配置/ 连通性验证,全部可直接复刻 ✓ 上线纪律: 16 项检查单按章节溯源; 三阶段验收用数据说话;运营节奏日/周/月/季分明下章预告第 17 章 实战二车队巡检 Agent从被动响应到主动巡检——Supervisor-Worker 模式落地10.2、批量任务的上下文保鲜9.6、巡检报告自动生成、与告警自愈系统的联动巡检发现喂给自愈知识库。第二个毕业设计补齐主动运维版图。官方参考资料Flink Agents 0.3 DocumentationFlink 2.3 Stable DocumentationDeploymentApache Doris 4.1 Documentation本章实战代码清单序号内容说明1系统全景架构图含旁路基建16.22工程模块结构按章节组织16.3.13主入口完整装配代码16.3.24专科 Agent 组装样例全配置16.3.35告警旅程时序 异常分支速查16.46部署配置envsubst 资源规划16.5716 项上线检查单 三阶段验收数据16.6