Flink Agents 与 AI 流式应用 —— 第 2 章 环境准备与第一个流式 Agent章节定位动手开始。本章从环境搭建到跑通第一个Kafka 告警事件 → Agent 诊断 → 输出处置动作的完整链路目标 30 分钟见效。核心信息已按官方 0.3.x 文档核实但 Flink Agents 为 preview 项目API 标注为 experimental——实际开发时请以官方最新文档为准本章提供的是概念骨架与工程规范细节可能随版本微调。版本基线Apache Flink 2.3.0 Flink Agents 0.3.xJDK 17版本提示Flink Agents 0.3.02026-06-19 发布为 preview 版本API 可能变化。部分商业发行版如阿里云 VVR当前 Agents 支持基于 Flink 1.20本课程以 Apache 开源主线为准Flink 2.3.0 Agents 0.3.x 为开源演进方向。章节导读2.1 环境依赖与版本对齐2.2 模型接入复用内网 LLM 网关2.3 快速上手30 分钟跑通第一个 Agent2.4 开发入口Java API 与装配模式2.5 工程规范延续2.6 调试技巧决策追踪与常见错误2.7 本章小结与下章预告2.1 环境依赖与版本对齐2.1.1 版本矩阵┌──────────────────┬─────────────────┬────────────────────────┐ │ 组件 │ 版本 │ 说明 │ ├──────────────────┼─────────────────┼────────────────────────┤ │ JDK │ 17 │ 前作基线,全栈统一 │ │ Apache Flink │ 2.3.0 │ 运行时 │ │ Flink Agents │ 0.3.x │ preview,API experimental│ │ Maven │ 3.8 │ Java 构建 │ │ Kafka │ 3.x │ 事件源 │ │ Paimon / Doris │ 1.4.x / 4.1 │ 结果落地(可选) │ └──────────────────┴─────────────────┴────────────────────────┘版本兼容说明Flink Agents 0.3.0 官方基于 Flink 2.x 主线开发。部分商业发行版如阿里云 VVR-11.7当前绑定的 Agents 版本对应 Flink 1.20——这是发行版适配节奏差异不影响开源主线演进。本课程以 Apache 开源版为准。2.1.2 Maven 依赖!-- pom.xml 核心依赖 --propertiesflink.version2.3.0/flink.versionflink-agents.version0.3.0/flink-agents.versionjava.version17/java.version/propertiesdependencies!-- Flink 核心 --dependencygroupIdorg.apache.flink/groupIdartifactIdflink-streaming-java/artifactIdversion${flink.version}/versionscopeprovided/scope/dependency!-- Flink Agents 框架 --dependencygroupIdorg.apache.flink/groupIdartifactIdflink-agents-java/artifactIdversion${flink-agents.version}/version/dependency!-- 模型调用:复用前作 openai provider(内建) --dependencygroupIdorg.apache.flink/groupIdartifactIdflink-table-ai/artifactIdversion${flink.version}/version/dependency!-- Kafka 连接器 --dependencygroupIdorg.apache.flink/groupIdartifactIdflink-connector-kafka/artifactIdversion3.2.0-2.3/version/dependency!-- Jackson:事件序列化 --dependencygroupIdcom.fasterxml.jackson.core/groupIdartifactIdjackson-databind/artifactIdversion2.15.4/version/dependency/dependencies2.1.3 部署形态Session 集群(推荐开发/测试): 本地 ./bin/start-cluster.sh 或 YARN Session SQL Client / 程序提交均可 Standalone 集群(生产): 沿用前作 Session/Gateway 规范(第 16 章) jar 进 lib 禁 -j; pipeline.name 命名; checkpoint 30s2.2 模型接入复用内网 LLM 网关2.2.1 为什么复用前作网关前作《Flink 2.3 实时 AI 推理实战》已搭建: 内网 LLM 网关 → 统一鉴权/限流/审计/多模型路由 Flink Agents 内部 LLM 调用 → 同样走该网关: ① 密钥不分散,仍集中在网关层 ② 审计统一,Agent 调用与 ML_PREDICT 调用同一份日志 ③ 限流配额共享,避免两个链路互相击穿2.2.2 模型连接配置// 作业入口:注册 LLM 连接importorg.apache.flink.agents.llm.ChatModel;ChatModelmodelChatModel.builder().provider(openai)// 内建 provider.endpoint(System.getenv(LLM_GATEWAY))// 内网网关.apiKey(System.getenv(LLM_API_KEY))// 环境变量注入.model(qwen-plus)// 推理模型.temperature(0.3)// 低温度,决策稳定.timeout(Duration.ofSeconds(30)).build();// 注意:apiKey 禁止硬编码,envsubst 或 K8s Secret 注入(前作 16.2 规范)2.2.3 与 CREATE MODEL 的关系前作:CREATE MODEL 注册在 Catalog 中,ML_PREDICT 通过 Catalog 查找 本作:Flink Agents 内部 LLM 调用通常直接构造 ChatModel, 不经过 Catalog——Agent 需要更细粒度的参数控制(temperature/top-p 等) 复用建议: 网关 endpoint/api-key 复用同一套环境变量 模型版本命名沿用前作规范: 域.用途_能力_vN Agent 侧模型与前作推理模型可共享网关配额,统一治理2.3 快速上手30 分钟跑通第一个 Agent2.3.1 目标链路Kafka 告警事件 → Agent 接收 → LLM 诊断 → 输出处置结论 → Kafka 结果 topic 输入示例: {event_id:ALM001,vin:LFV...,alert_text:单体压差超限187mV} 输出示例: {event_id:ALM001,diagnosis:电池故障,action:建P1工单,confidence:0.87}2.3.2 事件定义// 输入事件publicclassAlertEvent{publicstaticfinalStringEVENT_TYPEalert.incoming;publicStringeventId;publicStringvin;publicStringalertText;publiclongcollectTime;}// 输出事件publicclassDiagnosisResult{publicstaticfinalStringEVENT_TYPEdiagnosis.completed;publicStringeventId;publicStringdiagnosis;publicStringaction;publicdoubleconfidence;}0.3.x Breaking Change事件类型用字符串常量标识EVENT_TYPE而非事件类本身。Action注解监听的是事件类型字符串。2.3.3 Agent 骨架Javaimportorg.apache.flink.agents.agent.Agent;importorg.apache.flink.agents.agent.AgentContext;importorg.apache.flink.agents.action.Action;importorg.apache.flink.agents.action.ActionContext;publicclassAlertDiagnosisAgentimplementsAgent{privatefinalChatModelmodel;publicAlertDiagnosisAgent(ChatModelmodel){this.modelmodel;}Overridepublicvoidconfigure(AgentContextctx){// 系统提示词:定义 Agent 角色与约束ctx.setSystemPrompt(你是车联网告警诊断专家。基于告警文本给出诊断结论和处置动作。\n输出 JSON: {\diagnosis\:\...\,\action\:\...\,\confidence\:0.0~1.0});}// 监听告警到达事件——0.3.x 用事件类型字符串Action(listenEventTypes{AlertEvent.EVENT_TYPE})publicDiagnosisResultonAlert(ActionContextctx,AlertEventevent){// ① 构建 promptStringpromptString.format(告警: %s\n请诊断并给出处置建议。,event.alertText);// ② 调用 LLM(异步封装,框架托管)Stringrawmodel.chat(prompt);// ③ 解析输出(生产环境需加校验兜底,见 6.4)DiagnosisResultresultparseJson(raw);result.eventIdevent.eventId;// ④ 返回输出事件,框架自动序列化到下游returnresult;}}2.3.4 Pipeline 装配与提交importorg.apache.flink.agents.pipeline.AgentPipeline;importorg.apache.flink.streaming.api.environment.StreamExecutionEnvironment;publicclassAlertAgentJob{publicstaticvoidmain(String[]args)throwsException{StreamExecutionEnvironmentenvStreamExecutionEnvironment.getExecutionEnvironment();// ① 配置 Checkpoint(前作规范延续)env.enableCheckpointing(30_000);env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);// ② 创建 Kafka 输入流KafkaSourceAlertEventsourceKafkaSource.AlertEventbuilder().setBootstrapServers(System.getenv(KAFKA_BROKERS)).setTopics(tsp-alert).setGroupId(System.getenv(KAFKA_GROUP_AGENT_DIAG))// 配置化.setStartingOffsets(OffsetsInitializer.committedOffsets())// group-offsets.setValueOnlyDeserializer(newAlertEventDeserializationSchema()).build();DataStreamAlertEventinputenv.fromSource(source,WatermarkStrategy.noWatermarks(),alert-source);// ③ 装配 Agent PipelineChatModelmodel/* 2.2.2 构造 */;AgentPipelinepipelineAgentPipeline.builder().addAgent(newAlertDiagnosisAgent(model)).build();// ④ 输入接入 Agent,输出到结果 topicDataStreamDiagnosisResultoutputpipeline.process(input);output.addSink(KafkaSink.DiagnosisResultbuilder().setBootstrapServers(System.getenv(KAFKA_BROKERS)).setRecordSerializer(newDiagnosisResultSerializationSchema()).build());// ⑤ 提交env.execute(alert-diagnosis-agent-v1);// pipeline.name}}API 说明以上代码为概念骨架具体类名与装配 API 以 Flink Agents 0.3.x 官方文档为准。核心思想不变定义事件 → 注解 Action → 装配 Pipeline → 接入 Flink 运行时。2.3.5 验证清单□ Kafka 消费者组独立(group.id 配置化),不抢前作业 offset □ 输入事件到达 → WebUI 看到 Agent 算子处理 latency □ 输出 topic 收到 JSON 结果,字段完整 □ 作业重启(手动 cancel 重提) → 事件不丢不重复 □ checkpoint 30s 间隔,大小合理(Agent 状态轻量)2.4 开发入口Java API 与装配模式2.4.1 两种装配模式模式特点适用声明式装配注解驱动Agent、Action框架扫描标准链路快速开发编程式装配显式构造 Agent / Action / 连接复杂编排精细控制本章示例为声明式第 10 章多 Agent 协作时展示编程式。2.4.2 Agent 生命周期与 Flink 算子对齐open() → 构造 Agent 实例,加载 prompt,初始化工具连接 configure() → AgentContext 装配系统提示、工具列表、记忆策略 process() → 事件到达 → Action 方法执行 → 输出事件 close() → 释放连接,flush 日志 故障恢复:Checkpoint → 状态快照 → 重启后 open() 重新初始化, 但事件从 checkpoint 位点重放,决策上下文由 Memory 恢复2.4.3 事件类型字符串规范// 0.3.x 推荐:命名空间.动作.方向publicstaticfinalStringEVENT_TYPEdomain.action.direction;// 示例alert.incoming// 告警到达diagnosis.completed// 诊断完成ticket.created// 工单创建approval.required// 需人工审批// 好处:跨 Agent 协作时,事件路由清晰;日志检索方便2.5 工程规范延续前作全部规范不变Agent 作业没有特权□ jar 进 lib,禁止 -j 传参(双 ClassLoader 报 not a subtype) □ pipeline.name 必命名(WebUI 识别) □ checkpoint 30s,与 Paimon commit 对齐 □ restart-strategyfixed-delay,attempts2147483647(无限重试) □ group.id 配置化,Kafka scan.startup.modegroup-offsets □ api-key 走 envsubst / K8s Secret,禁止硬编码 □ JDK 17 全栈2.5.1 提交命令# 依赖 fat-jar 已含 flink-agents# 注意:flink-agents jar 必须进 lib,不随作业 jar -j 传入./bin/flink run\-Dpipeline.namealert-diagnosis-agent-v1\-Dexecution.checkpointing.interval30s\-Drestart-strategyfixed-delay\-Drestart-strategy.fixed-delay.attempts2147483647\-Drestart-strategy.fixed-delay.delay10s\./lib/alert-agent-job.jar2.6 调试技巧决策追踪与常见错误2.6.1 决策链路日志Flink Agents 框架自动输出(日志级别 INFO/DEBUG): [Agent:alert-diagnosis] Event:alert.incoming idALM001 [Agent:alert-diagnosis] LLM prompt length312 token [Agent:alert-diagnosis] LLM response latency1.2s [Agent:alert-diagnosis] Action output: diagnosis电池故障 action建P1工单 [Agent:alert-diagnosis] Event emitted: diagnosis.completed idALM001 开启:log4j.properties 中 org.apache.flink.agentsDEBUG2.6.2 常见错误排查错误根因排查ClassNotFoundException: flink-agentsjar 未进 lib,用 -j 传入被隔离确认 lib 目录含 agents jar事件到达但 Action 未触发事件类型字符串不匹配检查Action(listenEventTypes)与输入事件EVENT_TYPELLM 调用超时网关延迟或单 subtask 并发过高调 timeout / 确认网关健康 / 看反压输出格式混乱prompt 约束不足加 JSON schema 约束 解析兜底(6.4)Checkpoint 过大Memory 未设 TTL,状态膨胀配置状态 TTL(第 7 章)2.7 本章小结与下章预告本章小结┌──────────────────────────────────────────────────────────────┐ │ 第 2 章 要点回顾 │ └──────────────────────────────────────────────────────────────┘ ✓ 环境:Flink 2.3.0 Agents 0.3.x(preview) JDK 17 Maven 3.8 商业发行版适配可能滞后,以 Apache 开源主线为准 ✓ 模型:复用内网 LLM 网关,ChatModel 直接构造, 密钥环境变量注入,与前作统一治理 ✓ 快速上手:定义 Event(含 EVENT_TYPE 字符串常量) → Action(listenEventTypes) 监听 → Agent 内 LLM 调用 解析 → 返回输出事件 → Pipeline 装配 → Kafka 接入 → 提交 ✓ 规范:jar 进 lib 禁 -j / pipeline.name / 30s ck / 无限重试 / group.id 配置化 / group-offsets —— 前作规范零折扣延续 ✓ 调试:框架自动决策链路日志;常见错误四则速查下章预告第 3 章 核心抽象Event / Action / Tool / Memory深入 Flink Agents 的四大核心抽象——Event 的驱动模型、Action 的执行单元、Tool 的调用契约、Memory 与 Flink State 的映射关系。把第 2 章的骨架填充为血肉建立完整的运行时心智模型。官方参考资料Apache Flink Agents 0.3.0 Release AnnouncementFlink Agents 0.3 DocumentationApache Flink 2.3 DocumentationFlink 2.1 发布公告ML_PREDICT / Model DDL本章实战代码清单序号内容说明1Maven pom.xml 核心依赖2.1.22ChatModel 配置网关复用2.2.23事件定义EVENT_TYPE 字符串常量2.3.24Agent 骨架Action LLM 调用2.3.35Pipeline 装配与提交完整代码2.3.46工程规范清单 提交命令2.57调试日志与常见错误排查表2.6