Flink Agents 与 AI 流式应用 —— 第 7 章 Agent 状态管理

Flink Agents 与 AI 流式应用 —— 第 7 章 Agent 状态管理章节定位第 6 章解决了单事件内会思考本章解决跨事件记得住。生产刚需场景一辆车三天前开始排查的告警今天新告警到达时 Agent 必须记得排查进度一次远程诊断分了五轮交互每轮都要基于上轮结论继续。本章深入 Flink 状态体系在 Agent 场景的落地Keyed State 按 VIN 隔离会话、ForSt 状态后端配置、TTL 48h 防膨胀、Checkpoint 恢复语义实战渐进式诊断 Agent。版本基线Apache Flink 2.3.0 Flink Agents 0.3.xJDK 17章节导读7.1 为什么 Agent 需要状态三类跨事件记忆7.2 Keyed State按 VIN 隔离的诊断会话7.3 状态后端与 CheckpointForSt 配置实战7.4 状态恢复语义重启后 Agent 还记得什么7.5 实战渐进式诊断 Agent7.6 本章小结与下章预告7.1 为什么 Agent 需要状态三类跨事件记忆无状态 Agent(第 6 章): 每个事件独立推理 → 每次都从零开始 → 重复调用工具烧钱 例: 同一辆车 3 天内 5 次压差告警,每次都重新查故障码历史工单 有状态 Agent 记得三类事: ① 会话进度: 这辆车的诊断已走到第 3 步,在等复检数据 ② 已收集证据: 故障码释义查过了,历史工单拉过了,别重复 ③ 已做决策: 上次判 P2 观察期,这次指标恶化,升级 P17.1.1 状态 vs 记忆的边界本章的状态(State) 第 8 章的记忆(Memory) ───────────────────── ───────────────────────── 运行时工作记忆 可检索的经验沉淀 随会话结束而 TTL 过期 长期保留,跨会话复用 Keyed State,强一致 向量库/Mem0,近似检索 排查到哪一步了 类似的车当年怎么修的 一句话: State 管进行中的事, Memory 管已沉淀的事。7.1.2 状态的成本意识状态不是免费的: 每 KB 状态 × 100 万车辆 Key 状态体积 状态大 → Checkpoint 慢 → 恢复慢 → 反压敏感 设计纪律(延续前作硬约束): TTL 48h 强制过期;只存结论与指针,不存原始报文; 大对象放 Paimon/对象存储,State 里只放引用 ID7.2 Keyed State按 VIN 隔离的诊断会话7.2.1 Key 选择为什么是 VINFlink 状态按 Key 分区,Agent 场景的 Key 选择: ✅ vin → 同一辆车的所有告警路由到同一子任务, 状态自然隔离,互不串味 ❌ event_id → 每事件一个 Key,等于没状态 ❌ alert_type → 同类告警挤一个 Key,车辆间证据混淆(事故) 状态物理布局: KeyGroup(vinLFV001) ── TM1 ── ForSt LSM KeyGroup(vinLFV002) ── TM2 ── ForSt LSM ...百万级 Key,扩容重均衡由 Flink 托管7.2.2 会话状态结构定义publicclassDiagnosisSession{publicStringvin;publicStringsessionId;// 沿用 ActiveSession 思想: 配对键publicSessionPhasephase;// COLLECTING / WAITING_DATA / CONCLUDEDpublicMapString,Stringevidence;// 已收集证据: 工具名→结论摘要(非原文!)publicListStringdecisions;// 已做决策链publicintllmCalls;// 本会话 LLM 调用数(成本熔断用)publiclonglastUpdateTs;}// Agent 内声明状态privateValueStateDiagnosisSessionsessionState;Action(listenEventTypes{AlertEvent.EVENT_TYPE})publicEventonAlert(ActionContextctx,AlertEvente)throwsException{DiagnosisSessionssessionState.value();if(snull)sDiagnosisSession.fresh(e.vin);// ① 防重复: 证据已收集的工具不重复调(省钱的核心)if(!s.evidence.containsKey(dtc:e.faultCode)){StringdtcdtcResource.read(e.faultCode);s.evidence.put(dtc:e.faultCode,summarize(dtc));}// ② 基于会话进度继续推理,而非从零开始EventoutcontinueDiagnosis(s,e);// ③ 成本熔断: 单会话 LLM 调用上限(第 13 章展开)if(s.llmCalls20)returnnewEscalateEvent(s,会话成本超限);s.lastUpdateTsctx.currentEventTime();sessionState.update(s);returnout;}7.2.3 状态里存什么、不存什么┌────────────────────┬──────────────────────────────────┐ │ ✅ 存 │ ❌ 不存 │ ├────────────────────┼──────────────────────────────────┤ │ 证据结论摘要(百字节) │ 工具返回原文(可能数十KB) │ │ 决策链(枚举时间戳) │ 完整 LLM 上下文(第 8/9 章另设) │ │ 会话阶段与计数器 │ 原始 CAN 报文/大 JSON │ │ 大对象的引用 ID │ 大对象本体(放 Paimon/对象存储) │ └────────────────────┴──────────────────────────────────┘ 经验红线: 单 Key 状态 10KB;超出说明设计出了问题7.3 状态后端与 CheckpointForSt 配置实战7.3.1 为什么必须 ForSt百万车辆 Key × 会话状态 GB~TB 级状态 HashMapStateBackend: 状态全在 JVM 堆 → OOM,直接排除 EmbeddedRocksDB: 可用,但 Flink 2.x 官方主推继任者 ForSt(推荐): 前作硬约束延续——大状态窗口必用 ForSt 增量 Checkpoint 原生,本地恢复更快7.3.2 作业配置延续前作硬约束StreamExecutionEnvironmentenv...;// ① ForSt 状态后端env.setStateBackend(newForStStateBackend());env.getConfig().set(ForStOptions.PRIMARY_DIRECTORY,file:///data/flink/forst);// TM 本地 SSD// ② Checkpoint 组合(前作规范原样延续)env.enableCheckpointing(30_000);// 30sCheckpointConfigckenv.getCheckpointConfig();ck.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);ck.setMinPauseBetweenCheckpoints(10_000);ck.setCheckpointTimeout(120_000);ck.setMaxConcurrentCheckpoints(1);// 增量 本地恢复 Unaligned(背压时) Buffer Debloatck.enableUnalignedCheckpoints(true);// ③ 重启策略: fixed-delay 无限重试(硬约束)env.setRestartStrategy(RestartStrategies.fixedDelayRestart(Integer.MAX_VALUE,Time.seconds(10)));7.3.3 TTL48h 防膨胀StateTtlConfigttlStateTtlConfig.newBuilder(Time.hours(48))// 前作硬约束: 48h.setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite).setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired).cleanupFullSnapshot()// 快照时清理过期 Key.build();ValueStateDescriptorDiagnosisSessiondescnewValueStateDescriptor(diagnosis-session,DiagnosisSession.class);desc.enableTimeToLive(ttl);sessionStategetRuntimeContext().getState(desc);TTL 语义确认: 48h 无新告警的车 → 会话视为终结 → 状态过期清理 第 49h 又来告警 → 全新会话(这是业务上可接受的: 两天没动静的告警,大概率已线下处置完毕)7.3.4 状态监控三阈值前作延续□ 状态体积 TM 内存 80% → 查 Key 分布倾斜/大对象混入 □ 日增长 10% → 查 TTL 是否生效 □ Checkpoint 时长 30s → 查 ForSt 目录 IO/增量失效7.4 状态恢复语义重启后 Agent 还记得什么7.4.1 恢复全景作业取消/崩溃 → 从最近 Checkpoint 重启: ✅ 恢复: Keyed State(会话进度/证据摘要/计数器) ✅ 恢复: Kafka offset(从 checkpoint 位点继续消费) ❌ 不恢复: JVM 内存对象、线程池、HTTP 连接 → open() 重建 ❌ 不恢复: 进行中的 LLM 调用(见 7.4.2)7.4.2 进行中的 LLM 调用恢复时的关键缝隙崩溃时序: 事件消费 → LLM 调用中(还没回填) → 崩溃 → 重启 会发生什么: offset 未提交 → 事件重放 → LLM 再调一次 后果分级: 只读工具重放 → 多花一次钱,可接受 LLM 重放 → 多花一次钱,可接受 动作类工具重放 → 重复建单! 必须幂等(第 4 章红线在此兑现) 结论: Exactly-Once 管得住状态,管不住外部副作用。 副作用幂等 event_id 去重,没有第二条路。7.4.3 状态 Schema 演进改表规则延续: 停作业 → DDL/代码变更 → 无状态重启 (状态结构不兼容时,Savepoint 恢复会 OutOfRangeException) 加字段(如 DiagnosisSession 增加 remark): 兼容性取决于序列化器——POJO 加字段通常可恢复 但生产仍推荐无状态重启 会话自然重建(48h TTL 兜底)7.5 实战渐进式诊断 Agent7.5.1 场景疑难电池故障往往无法一次定性: Day1 告警压差超限 → 初诊:建观察会话,等更多数据 Day2 告警压差扩大温升 → 复诊:证据恶化,召远程诊断 Day3 诊断数据回流 → 终诊:输出根因工单 三次事件,一个会话,Agent 全程记得——这就是状态的价值。7.5.2 状态机设计┌────────────┐ 初诊结论不确定 ┌──────────────┐ │ COLLECTING │──────────────→│ WAITING_DATA │ │ 收集证据 │ │ 等复检/回流 │ └─────┬──────┘ └──────┬───────┘ │ 证据充分 │ 数据回流/超时 ▼ ▼ ┌────────────┐ ┌──────────────┐ │ CONCLUDED │←──────────────│ RE_ASSESS │ │ 输出终诊 │ 直接终诊 │ 复诊评估 │ └────────────┘ └──────────────┘ 状态迁移 代码控制(确定性),迁移时机 LLM 建议(灵活性) ——又是主干 Workflow 局部 ReAct的混合思想7.5.3 核心实现Action(listenEventTypes{AlertEvent.EVENT_TYPE,DiagDataEvent.EVENT_TYPE})publicEventonEvent(ActionContextctx,BaseEvente)throwsException{DiagnosisSessionsloadOrFresh(e.vin);switch(s.phase){caseCOLLECTING-{// 复用第 6 章 ReAct 诊断,但结果分流:DiagnosisResultrdiagnose(s,e);if(r.confidence0.8){s.phaseCONCLUDED;returnfinalize(s,r);// 终诊出单}s.phaseWAITING_DATA;s.waitUntilctx.currentEventTime()24H;// 等 24h 复检returnnewObserveEvent(s,r);// 观察中事件}caseWAITING_DATA-{// 新告警或诊断回流到达 → 复诊s.evidence.putAll(collectNew(e));DiagnosisResultrdiagnose(s,e);// 带历史证据复诊s.phaseCONCLUDED;returnfinalize(s,r);}caseCONCLUDED-{// 会话已终结又来告警 → 新会话(新 sessionId)returnstartNewSession(e);}}}7.5.4 定时器WAITING_DATA 超时兜底// 等了 24h 复检数据没来 → 定时器触发,降级处置OverridepublicvoidonTimer(longts,ActionContextctx)throwsException{DiagnosisSessionssessionState.value();if(s!nulls.phaseWAITING_DATA){ctx.emit(newEscalateEvent(s,复检数据 24h 未回流,转人工跟进));// 事件永远有出口(6.4)sessionState.clear();}}7.5.5 验收指标□ 同车重复告警的工具调用量下降 ≥ 40%(证据复用生效) □ 状态体积 p99 单 Key 10KB □ Checkpoint 时长 p99 30s □ 手动 cancel重启 → 会话进度无损,无重复建单 □ 48h TTL 到期清理率 100%7.6 本章小结与下章预告本章小结┌──────────────────────────────────────────────────────────────┐ │ 第 7 章 要点回顾 │ └──────────────────────────────────────────────────────────────┘ ✓ 三类跨事件记忆: 会话进度/已收集证据/已做决策; State 管进行中的事,Memory 管已沉淀的事 ✓ Key 选 vin: 同车同 Key 状态隔离;存结论摘要与引用 ID, 不存原文,单 Key 10KB 红线 ✓ ForSt 增量 CK 本地恢复 Unaligned Buffer Debloat; TTL 48h(OnCreateAndWrite NeverReturnExpired); 重启 fixed-delay 无限重试 ✓ 恢复语义: 状态与 offset 可恢复,LLM 调用不恢复 → 事件重放 → 动作类工具幂等是唯一防线 ✓ 实战: COLLECTING→WAITING_DATA→CONCLUDED 状态机, 迁移代码控制、时机 LLM 建议,定时器兜底超时转人工下章预告第 8 章 短期与长期记忆状态记住了排查进度但真正的智能来自经验沉淀——记忆分层架构对话缓冲/会话摘要/长期知识、摘要压缩策略、Mem0 与向量库长期记忆接入、记忆写入时机与质量控制实战随车医生 Agent让 Agent 记得这辆车的全部病史越用越懂车。官方参考资料Flink Agents 0.3 DocumentationMemory 章节Flink 2.3 Stable DocumentationState Backends / ForStMem0: Building Production-Ready AI Agents with Scalable Long-Term Memory本章实战代码清单序号内容说明1State vs Memory 边界判定7.1.12DiagnosisSession 状态结构与证据去重7.2.23状态存取红线存摘要不存原文7.2.34ForSt Checkpoint TTL 完整配置7.35恢复语义与幂等防线分析7.46渐进式诊断状态机 定时器兜底7.5