企业官网建设流程全解析

Flink Agents 与 AI 流式应用 —— 第 11 章 流式 Agent 设计模式

章节定位：前四章把单 Agent 和多 Agent 的构件做齐了，本章提供经过生产验证的"套路"——那些在车联网告警处置场景中反复出现、足以沉淀为模式的设计方案。事件路由、聚合防抖、定时窗口、人机协同、CEP 结合五大模式，再加一份反模式清单。学完本章，面对新需求时你能直接说"这个适合用定时窗口模式"。

版本基线：Apache Flink 2.3.0 + Flink Agents 0.3.x（JDK 17）

章节导读

• 11.1 事件路由模式
• 11.2 聚合防抖模式
• 11.3 定时窗口模式
• 11.4 人机协同模式
• 11.5 CEP 结合模式
• 11.6 反模式清单：生产上炸过的坑
• 11.7 本章小结与下章预告

11.1 事件路由模式

11.1.1 场景

同一条告警事件,不同特征走不同链路: 压差告警 → 电池专科(大模型) 绝缘告警 → 热管理专科(大模型) 自检通过 → 无 LLM 直出(零成本) 批量召回 → Supervisor 批量链路

11.1.2 实现

// 基于事件字段的分流,不是基于主题(topic)——// 避免 N 个主题 × N 个消费者组的复杂度@Action(listenEventTypes={AlertEvent.EVENT_TYPE})publicEventroute(ActionContextctx,AlertEvente){if(e.category.equals("SELF_CHECK_OK"))returnnewPassThroughEvent(e);// 零成本旁路if(isMassRecall(e.vin))returnmassRecallSupervisor.handle(e);// 批量链路returnnewTriageOutput(e);// 常规分诊→专科}

11.1.3 变体：动态路由表

路由规则不硬编码,从配置中心读: 新告警类型上线 → 改配置即可,不发版 灰度某链路 → 路由表写比例参数(如 battery=80%旧+20%新) 纪律: 路由决策必须是确定性代码,不用 LLM—— 否则每事件多花一笔判断钱,ROI 为负

11.2 聚合防抖模式

11.2.1 场景

车辆抖动发送了 30 条同码告警/分钟 → 不处理: Agent 疯了,30 次 LLM 调用,结论全一样 处理: 合并为 1 条事件,结论挂 30 个原始告警 ID 与第 7 章"同会话证据复用"的区别: 复用: 逐条到达,逐条查但不重复查工具 防抖: 先把 N 条聚成 1 条,再进 Agent

11.2.2 实现

// 同 vin + 同 faultCode + 5 分钟滚动窗口 = 一条聚合事件input.keyBy(e->e.vin+"#"+e.faultCode).window(TumblingEventTimeWindows.of(Time.minutes(5))).aggregate(newAlertAggregateFunction(),newAlertWindowProcessFunction());// AggregateFunction(前作硬约束: 轻量,推荐用它而非全量收集)classAlertAggregateFunctionimplementsAggregateFunction<AlertEvent,AlertAccumulator,AggregatedAlert>{publicAlertAccumulatorcreateAccumulator(){returnnewAlertAccumulator();}publicAlertAccumulatoradd(AlertEvente,AlertAccumulatoracc){acc.eventIds.add(e.eventId);acc.maxValue=Math.max(acc.maxValue,e.metricValue);acc.lastTs=e.collectTime;returnacc;}publicAggregatedAlertgetResult(AlertAccumulatoracc){returnnewAggregatedAlert(acc.vin,acc.faultCode,acc.eventIds,acc.maxValue,acc.lastTs);// 只传聚合值}}

11.2.3 防抖纪律

□ 保留列表: 全部 eventIds(审计追溯) □ 保留极值: max/min(漏了真实恶化值就是事故) □ 保留时间: first/last(抖动时长是诊断线索) □ 丢弃详情: 重复的原始报文、相同工具结果

11.3 定时窗口模式

11.3.1 场景

"等复检数据 24h" 的 WAITING_DATA(第 7 章): 不是等一条特定事件,而是等一段时间窗口内 可能到达的多条事件(复检数据、新告警、维修报告) → 需要定时窗口: 等待 + 超时 + 多源输入

11.3.2 实现

// ProcessFunction + 定时器: 比 CEP 轻,比窗口灵活classAwaitDataFunctionextendsKeyedProcessFunction<String,BaseEvent,Event>{privateValueState<AwaitingState>state;publicvoidprocessElement(BaseEvente,Contextctx,Collector<Event>out){AwaitingStates=state.value();if(s==null){// 首次进入等待s=AwaitingState.start(e);state.update(s);ctx.timerService().registerEventTimeTimer(e.collectTime*1000L+24*3600*1000L);// 24h 后触发}s.merge(e);// 收集窗口内事件if(s.hasEnough()){// 数据够了提前触发out.collect(s.finalizeAndClear(state));}}publicvoidonTimer(longts,OnTimerContextctx,Collector<Event>out){AwaitingStates=state.value();if(s!=null){out.collect(s.timeoutAndClear(state));// 24h 未回 → 转人工}}}

11.3.3 模式变体

11.4 人机协同模式

11.4.1 场景

Agent 能处理 70% 场景,但 30% 必须人工: - 置信度 < 0.5 的疑难(6.2.3 降级) - 高危操作未审批(4.5.3 三保险) - 行业首例故障(知识库召回为空) 不是"做不出来就抛异常",而是"优雅交出控制权"。

11.4.2 协作事件流

Agent 人工运维台 │ approval.required │ │ ───────────────────────→│ │ │ │ approval.granted │ │ ←───────────────────────│ (Kafka,带 approval_id) │ │ │ 继续执行 / 改判 │ │ ───────────────────────→│ 整个协作 = 事件流,Agent 不挂等,运维台不轮询

11.4.3 实现要点

@Action(listenEventTypes={"approval.granted"})publicEventonApproval(ActionContextctx,ApprovalEvente){DiagnosisSessions=state.value();if(s==null||!s.awaitingApprovalId.equals(e.approvalId))returnnull;// 过期审批,静默丢弃// 双保险: 审批号 + 审批人权限,服务端再验(第 15 章)if(!approvalService.verify(e.approvalId,e.approver))returnnull;// 伪造审批returncontinueWithApproval(s,e.approvedAction);}

纪律: □ 转人工事件带完整上下文(诊断理由/证据/建议动作), 不能让人工从零开始读 □ 人工回复也走事件,Agent 靠监听事件恢复, 不走回调/接口等同步方式 □ 审批超时(如 4h 未回) → 默认策略自动执行或升级, 不能无限等待

11.5 CEP 结合模式

11.5.1 场景

Agent 适合推理,但不适合模式匹配: "同一辆车 5 分钟内出现压差+温升+绝缘异常三个告警" → 这是 CEP 的拿手好戏,不是 LLM 该干的事 分工: CEP 管"发生了什么模式" Agent 管"发生了之后怎么决策"

11.5.2 实现

// CEP 定义模式(纯 Flink CEP,无 LLM)Pattern<AlertEvent,?>cascade=Pattern.<AlertEvent>begin("diff").where(e->e.category.equals("PRESSURE_DIFF")).next("temp").where(e->e.category.equals("TEMP_RISE")).within(Time.minutes(5));// CEP 输出复合事件 → Agent 输入PatternStream<AlertEvent>ps=CEP.pattern(input,cascade);ps.process(newPatternHandler(){publicvoidonMatch(Map<String,AlertEvent>match,Contextctx,Collector<ComplexAlert>out){out.collect(newComplexAlert(match.get("diff").vin,"CASCADE_BATTERY_THERMAL",match.get("diff"),match.get("temp")));// 打包给 Agent}});// Agent 侧: 收到复合事件 → 触发热失控专章推理@Action(listenEventTypes={"alert.cascade"})publicEventonCascade(ActionContextctx,ComplexAlerte){ctx.appendSystemContext(urgent.热失控);// 9.3.2 高危专章returndiagnose(e);// 第 6 章 ReAct}

11.5.3 选型边界

用 CEP: 时序约束明确(先后/同时/窗口) 例: A 告警后 5 分钟内出现 B 且没有出现 C 用 Agent: 需要语义理解、跨文本推理 例: 综合故障码释义+历史工单+召回案例做根因判断 黄金搭档: CEP 过滤 → Agent 决策 CEP 便宜(Flink 原生),Agent 贵(LLM), 用 CEP 把 90% 无意义事件拦住,Agent 只处理值得推理的 10%

11.6 反模式清单：生产上炸过的坑

11.6.1 反模式一：Agent 做流处理

❌ 用 LLM 判断"这 5 条告警是不是同一类" → 每批次都调 LLM,token 烧光,延迟飙高 ✅ CEP / 窗口聚合做归类,Agent 只做归类后的决策

11.6.2 反模式二：无限等待人工

❌ 转人工后 Agent 挂起等回调,不超时 → 人工 2h 未回 → 状态膨胀 → Checkpoint 失败 ✅ 审批事件 + 定时器兜底(11.4.3),到点默认策略

11.6.3 反模式三：全量事件塞 LLM

❌ 每条告警原文直接进 prompt,不做防抖不做聚合 → 抖动事件 × LLM 单价 = 预算黑洞 ✅ 聚合防抖(11.2) + 预算守卫(9.2) + CEP 过滤(11.5)

11.6.4 反模式四：Agent 之间同步调用

❌ Agent A 直接 await Agent B 的结果 → 阻塞事件循环 → 反压 → 延迟连锁恶化 ✅ 事件驱动异步传递(10.3.1),Agent 间永远走事件流

11.6.5 反模式五：忽视事件时间乱序

❌ 用处理时间(Processing Time)判断告警先后 → Kafka 重平衡/重放导致乱序 → "先温升后压差" → CEP 匹配失效或错误匹配 ✅ 严格 Event Time + Watermark(5s 乱序容忍) + 前作 Watermark 对齐(20s 偏差,60s idle timeout)

11.7 本章小结与下章预告

本章小结

┌──────────────────────────────────────────────────────────────┐ │ 第 11 章 要点回顾 │ └──────────────────────────────────────────────────────────────┘ ✓ 事件路由: 确定性代码分流,不用 LLM;配置中心驱动灰度 ✓ 聚合防抖: AggregateFunction 聚合同码告警, 保留极值/事件 ID 列表,丢弃重复详情 ✓ 定时窗口: ProcessFunction+定时器,等复检/复合评估, 多源到达可提前触发,超时兜底 ✓ 人机协同: 转人工不是抛异常,是事件流; 审批走事件+定时器,人工回复也是事件 ✓ CEP+Agent: CEP 管模式匹配(便宜),Agent 管推理(贵), 黄金搭档拦住 90% 无效事件 ✓ 五大反模式: Agent 做流处理/无限等人/全量塞 LLM/ 同步调用/忽视 Event Time

下章预告

第 12 章 与湖仓生态集成

Agent 是数据流的算子，天然需要读写 Kafka/Paimon/Doris。本章讲透 Agent 的数据进出口：决策审计落湖、Agent 驱动数仓任务、与 ML_PREDICT 的分工边界，实战"数据质量巡检 Agent"——让 Agent 自己监控自己产出的数据质量。

官方参考资料

• Flink Agents 0.3 Documentation
• Flink CEP Documentation
• Flink 2.3 Stable Documentation（ProcessFunction / Timers）

本章实战代码清单

序号	内容	说明
1	事件路由（含动态路由表）	11.1
2	聚合防抖 AggregateFunction 实现	11.2
3	定时窗口 ProcessFunction+定时器	11.3
4	人机协同事件流与审批超时兜底	11.4
5	CEP 复合事件 + Agent 专章推理	11.5
6	五大反模式与正面对照	11.6