把 Flink Agents 的源码一路读到第 6 篇我终于遇到了这个系列里第一个真正“下手干活”的类ActionTask。前面几篇我们聊了 Agent 的整体骨架、规划器怎么拆解意图、上下文和记忆怎么维护那些都还停留在“想”的层面。到了 ActionTask事情开始变得具体——它要调用外部系统、执行一个实际动作、把结果带回给 Agent。如果这一层没吃透前面所有关于规划和上下文的讨论都是飘的。这篇我把 ActionTask 的定位、主流程、状态恢复、异常处理和可复用的设计思路拆开讲适合正在读 Flink Agents 源码的同学也适合想做基于 Flink 的 AI Agent 或大数据任务编排平台的工程师。我会尽量按源码阅读的顺序来而不是按教科书顺序来。1. 从“计划”到“动作”ActionTask 在 Agent 执行链里的准确位置在拆代码之前先得搞明白一件事Flink Agents 为什么要有 ActionTask 这个东西而不是像写普通 Flink 作业那样直接在 map 或者 flatMap 里调个接口完事这个问题想清楚了后面所有字段和方法的逻辑都顺了。我把 Agent 的一次完整执行拆成四段意图接收、计划生成、动作执行、结果聚合。ActionTask 就是第三段的载体。它接收到的是“经过规划和校验后的动作指令”而不是原始的自然语言。换句话说ActionTask 不需要理解用户想干什么它只需要理解一件事这条指令对应的动作叫什么、参数是什么、在哪个目标上执行、执行完结果往哪里交。这个定位决定了它的责任边界。ActionTask 不是给大模型用的 Prompt 模板也不是规则引擎它是一个非常务实的执行器把结构化的动作描述变成结构化的执行结果。我在读这版源码时最大的感受是这个类把“动作”这个词当作一种一等公民来建模而不是藏在字符串里到处传。1.1 它和普通 Flink 算子到底差在哪普通 Flink 算子是被动处理数据的。数据流进算子算子做变换数据流出算子整个过程没有“副作用”。ActionTask 完全不同它是要主动调用外部系统的比如写 Hive 表、调远程接口、更新元数据、触发某个外部 API。这就带来一个本质区别普通算子可以从 Flink 的重启机制里获得一致性保证但 ActionTask 必须自己处理外部系统的语义。我读下来的体会是ActionTask 本质上是一个“副作用边界”。Flink 的容错机制保证的是流的状态一致性它并不保证外部系统一定只被执行一次。所以 ActionTask 里必须出现幂等键、审计状态、重试策略这些东西就是为了弥补 Flink 恢复机制覆盖不到的那部分缝隙。另外一个差异是错误处理。普通算子里你抛一个异常Flink 会把作业重启或者把这条数据打入侧输出。但 ActionTask 处理的是动作动作失败不一定是数据问题可能是外部服务超时、权限不足、目标系统临时不可用。这时候如果也直接抛异常重启整个作业代价就太大了。所以 ActionTask 的异常路径不是简单 throw而是把错误分类、包装成结果数据交给上层的 Agent 逻辑去做决策。1.2 动作描述与执行器分离的设计我读到的 ActionTask 里最核心的数据结构是 ActionSpec 和 ActionHandler。ActionSpec 是动作描述它是一段可序列化的元信息包含动作名称、参数 Schema、目标资源标识、超时阈值、重试策略。它不是执行逻辑它就是一段配置。ActionHandler 是真正干活的执行器它知道怎么连接某个具体外部系统、怎么传参、怎么解析返回。这两个东西通过 ActionRegistry 绑定。ActionRegistry 就是一个注册表根据动作名称找到对应的 Handler。这个设计我特别喜欢因为它把“做什么”和“怎么做”彻底拆开了。Agent 在规划阶段只需要生成 ActionSpec它不需要知道 Handler 的细节ActionTask 在执行阶段只需要根据动作名称去注册表里取 Handler它不需要关心 Agent 是怎么想的。中间耦合被砍到最薄。对这个设计工程上的直接好处是新增一个动作类型不需要改 ActionTask 的代码只需要注册一个新的 Handler。如果动作列表来自配置中心甚至可以做到不重启作业就更新 Agent 的能力范围。这一点在后面第 5 章我会再展开。1.3 从字段清单反推作者的设计取舍源码阅读有个技巧先看类的字段字段往往暴露出作者最在意的事情。我读的 ActionTask 主要字段大致可以归成几组。public class ActionTask extends KeyedProcessFunctionString, ActionCall, AgentEvent { // 动作描述与执行器绑定 private final ActionRegistry actionRegistry; private final ActionSpec defaultSpec; // 超时与重试策略 private final Duration connectTimeout; private final Duration readTimeout; private final Duration actionTimeout; private final RetryPolicy retryPolicy; // 状态后端 private transient ValueStateIdempotentKeySet idempotentState; private transient ListStateInFlightAction inflightState; private transient ListStateActionResult auditState; // 侧输出 private final OutputTagActionResult resultTag; private final OutputTagFailedAction deadLetterTag; }一眼看过去重点非常清楚超时、幂等、审计、侧输出。没有花哨的缓存也没有复杂的并发模型。作者在类设计上是非常克制的它只保留了让一个动作可靠执行的最小字段集。有一点值得注意ActionTask 继承的是 KeyedProcessFunction而不是普通的 RichFunction。这意味着它天然可以按 key 来隔离状态。在我看的这个版本里key 用的就是 traceId也就是一次 Agent 会话的标识。这个选择在后面第 3 章讲状态恢复时会显得特别重要。2. execute() 的四段式主流程校验、幂等、执行、回传ActionTask 的核心方法自然是 execute()。我读代码的习惯是先把方法体里的主干抽出来再去看每个分支。抽完之后发现它的主流程其实特别规矩就是四段解析校验、幂等判断、执行动作、结果回传。每一段之间都有明确的返回点没有把逻辑搅在一起。2.1 入口校验要把脏数据挡在门外execute() 的入参是 ActionCall它包含了 traceId、动作名称、动作参数、请求序号这些字段。ActionTask 做的第一件事不是执行而是校验。校验分两层。第一层是基本校验traceId 是否为空、动作名称是否在注册表里。这些属于结构校验任何一条不满足直接抛一个 IllegalArgumentException 或者返回一个带错误码的结果。第二层是参数校验动作参数是否满足 ActionSpec 里声明的 Schema。这块在源码里往往是生成出来的代码不是手写 if-else但它承担了一个很重要的职责——让错误尽早暴露。我一开始觉得这层校验很多余因为上游规划器已经做过一遍了。后来想明白ActionTask 不能信任上游。原因很简单Flink 作业可能从 checkpoint 恢复恢复时上游的状态可能不是最初规划时的状态另外并行子任务重试时某些消息可能被重新处理。如果参数不合法最理想的结果就是“赶紧失败、别碰外部系统”而不是带着脏数据去调接口。所以这层校验的真实价值不是防错而是“降本”。一次外部系统调用可能耗时几十秒而一次本地校验只需几毫秒。拿毫秒级的成本换几十秒的成本这笔账非常划算。2.2 幂等键重试和去重是同一件事校验通过之后ActionTask 会计算一个幂等键。这段逻辑在源码里是一个独立方法名字大概类似于 buildIdempotentKey它做的事情是组合 traceId、动作名称、请求体的内容哈希生成一个稳定的字符串。这个计算有个很重要的原则必须基于业务内容而不是基于消息本身。因为 Flink 在故障恢复时可能把消息反序列化重放消息内部的一些临时字段可能变了但业务内容应该保持不变。如果幂等键里带了时间戳或者随机 UUID那每次重算出来的键都不一样幂等就形同虚设了。幂等键算出来后ActionTask 会查询状态里的 IdempotentKeySet。如果这个键已经存在说明这条动作请求之前已经处理过了直接返回上一次的结果不再执行外部调用。如果不存在就把它写入状态然后继续往下走。这里有个细节值得展开幂等键写入状态和外部动作执行不是一个原子操作。也就是说可能存在“键写进去了、外部动作根本没执行”或者“外部动作执行了、键还没来得及写”的情况。我在阅读时也纠结过这个问题但后来发现源码的处理方式是接受这个窗口而不是试图消除它。因为真正消除需要跨 Flink 状态和外部系统做分布式事务成本太高。接受窗口之后再靠后续的审计结果和重试策略去兜底工程上更现实。2.3 异步执行与在途请求队列幂等判断通过后ActionTask 去 ActionRegistry 里取出对应的 ActionHandler然后开始执行。真正让我觉得这版源码有水平的地方是它没有把外部调用做成同步阻塞。在 Flink 算子里做同步阻塞调用最直接的问题是一个并行子任务被一个慢请求卡住整个 subtask 的资源都被占着checkpoint 屏障也过不去最后引发反压和超时连锁反应。ActionTask 的做法是把执行封装成 CompletableFuture 或者类似异步任务同时维护一个 inFlight 队列记录哪些请求还在途。异步化之后会带来一个新的问题并发控制。如果外部系统 QPS 上限很低而 ActionTask 一股脑把所有请求发出去肯定会把外部系统打挂。所以源码里肯定有一个信号量或者限流器控制同时执行的动作数。我在这个版本里看到的做法是在 ActionTask 内部维护了一个简单的计数器和等待队列超过并发上限的请求会先待在本地而不是直接打到外部系统。这里要给一句提醒异步不等于没有副作用。同样一个动作异步执行时如果线程被 cancel外部系统到底执行没执行你是不确定的。所以 ActionTask 里的 inflightState 必须跟着请求一起持久化。这是下一章要展开的内容。2.4 结果回传为什么走侧输出流而不直接 emit动作执行完成之后ActionTask 会把结果包装成 ActionResult然后写入 auditState。但真正对外回传的时候它走的不是主输出流而是一个侧输出流。这个设计初看有点绕。为什么不直接把结果 emit 出去我后来想明白了因为动作结果和普通的数据流语义不一样。主输出流是给下游数据处理用的它参与 Flink 的背压和检查点语义而动作结果更多是给 Agent 的上下文记忆用的它不应该反过来影响业务数据流的处理速度。如果一个外部系统响应特别慢结果迟迟不产出我们不希望这个慢动作把整个 Flink 数据管道都拖住。侧输出流在这时候就成了天然的隔离带动作结果可以单独被下游消费也可以配上单独的处理逻辑。另外结果回传时要做截断。外部系统返回的 detail 字段可能非常大比如查询结果有几万行这些内容如果全量塞进 Agent 上下文很快就把上下文撑爆。我读到的版本里默认会给结果设置一个字节上限超过部分只保留摘要和截断标记。这个细节看起来小但非常重要。它决定了 Agent 的记忆系统能不能长期稳定运行。3. 状态与检查点执行到一半的 Action 如何恢复ActionTask 不是简单地执行完就完了。由于它运行在 Flink 里所有关于“执行到哪一步”的信息都必须落在状态里否则作业一旦重启外部系统和 Flink 内部会对不上账。这一章专门说状态。3.1 三个状态区的职责划分ActionTask 里我梳理出三类最重要的状态它们各管一件事。第一类是 idempotentState管的是“这个动作已经处理过没有”。它的内容是幂等键集合。恢复的时候如果某个请求的键还在这里面就直接返回历史结果不重复执行。第二类是 inflightState管的是“哪些动作还在执行中”。它的内容是在途请求的元数据包括幂等键、动作参数、发起时间。恢复的时候ActionTask 要判断这些在途请求到底执行完没有。如果外部系统确凿地返回了失败可以清理如果结果未知就要重新发起或者等待。第三类是 auditState管的是“执行过的动作结果”。它的内容是 ActionResult 的列表用于审计以及给 Agent 上下文回放。这三类状态用一个表格来看会更清楚。状态项内容故障恢复时的作用典型保留策略idempotentState幂等键集合防止重复执行外部动作TTL 24 小时起按外部系统事务保留期调整inflightState在途请求元数据判断未完成的动作是重新触发还是标记失败跟随 checkpoint原则上不清除auditState执行成功动作的结果回放给上层 Agent 做上下文拼装按会话或时间窗口裁剪避免无限增长这三类状态缺一不可。只看其中任意一类都没法在故障后自洽地恢复。3.2 检查点屏障与在途请求的协调Flink 做 checkpoint 时会往流里注入屏障。屏障到达 ActionTask 时它会等待自己内部的算子状态对齐然后快照。问题来了如果此时有一个外部调用正在异步执行ActionTask 要不要等它返回如果等检查点时间会被拖长如果不等快照里记录的是“没有这个在途请求”恢复时这个请求就丢了。我看到的处理方式是组合拳。ActionTask 会把运行中的 Future 注册到检查点钩子里在快照执行前给出一个短暂的等待窗口让那些很快就能返回的请求落地。对于超过等待窗口的请求它不会一直等而是把该请求标记为 inFlight在快照里记录下来。恢复时状态里只有真正的 inflight 请求还留在队列里已经完成的请求则进入 auditState。这里有一个非常关键的取舍它没有试图让“外部调用结果”和“Flink 状态快照”达到严格的同时性而是允许一个很小的不一致窗口。这个窗口靠幂等键和审计记录来补偿。我觉得这就是 Flink 上做副作用操作的标准答案不要幻想强一致要想清楚不一致窗口里发生了什么然后设计补救机制。3.3 并行度设计与 keyBy 要求由于 ActionTask 是 KeyedProcessFunction它的所有状态都是按 key 隔离的。在我的理解里key 应该取 traceId这样同一条 Agent 会话里的动作都会落到同一个并行子任务。这个设计不是随便选的。如果 key 取了别的字段比如动作名称那么同一条会话里的多个动作会被分发到不同子任务每个子任务各自维护一部分幂等键。恢复时幂等键的查询范围就分裂了动作 A 在一个子任务里处理动作 B 在另一个子任务里处理它们的审计状态也是分开的这样会给 Agent 的上下文聚合造成很多麻烦。并行度怎么定我给的参考公式是并发数 min(算子并行度外部系统最大 QPS / (单次调用平均耗时 × 权重))。比如外部系统最大支持 10 QPS单次调用平均耗时 2 秒理论上一个子任务每秒最多完成 0.5 次调用那么要支撑满 10 QPS 大约需要 20 个并发子任务。当然这是理论值实际还要留出余量一般按 60% 到 70% 的利用率来估算。这一条属于我个人的工程经验补充不是源码直接给出的公式。4. 异常路径ActionTask 把错误做成了可决策的数据读源码的时候我最喜欢看异常路径。主流程大家都写得差不多异常路径才见真章。ActionTask 的异常处理给我最大的启发是它把错误当成一种数据来对待而不是当成一种状态来抛出。4.1 超时不是一种是三种ActionTask 里至少区分三种超时连接超时、读取超时、整体动作超时。连接超时是建立连接时等待的时间。读取超时是连接建立之后等待响应数据的时间。整体动作超时是整个动作从开始到结束的硬性上限不管前面两个超时怎么设整体超时到点就必须中止。为什么要把超时拆开因为三者对应的故障类型不同。连接超时大概率是网络不通或者服务没起来读取超时大概率是服务活但处理慢整体超时则可能是请求本身被卡在某个排队环节。这三种情况的重试策略和告警指标都应该不一样。如果只用一个统一的超时时间你只能知道“它超时了”没法快速判断该扩容还是该检查网络。我常用的配置是 connectTimeout 3 秒readTimeout 30 秒actionTimeout 60 秒。这样配是因为大多数内部系统在正常情况下响应都在几秒内超过 30 秒还在读数据基本可以断定是卡住了。当然这个值要按你自己的业务去调但思路是三个超时各管一段别用一个大值糊弄。4.2 网络错误与业务错误分离ActionTask 里对异常做了一个很重要的分类网络层错误和业务层错误。网络层错误指的是连接断开、DNS 解析失败、SSL 握手失败这类问题它有一个特点目标系统可能根本没收到请求或者收到了但不确定是否处理。这类错误重试相对安全。业务层错误指的是请求成功到达目标系统但目标系统返回了明确的业务错误比如权限不足、参数不合法、数据版本冲突。这类错误重试大概率是浪费时间甚至可能加重问题。源码里会用不同的错误码来标记这两种情况。我建议所有依赖外部动作的系统都建立一套类似的错误码体系至少包含成功、业务拒绝、业务失败重试、网络不可用、超时、未知异常。不要笼统地返回一个 failed。4.3 退避重试不要放大故障ActionTask 的重试策略也是我重点看的部分。它没有用固定间隔重试而是用了指数退避加抖动。固定间隔重试最怕的场景是外部服务已经出现问题然后 ActionTask 的几十个并行子任务同时失败同时等 5 秒又同时重试把本来还能扛一下的服务直接打崩。这就是重试导致的故障放大。指数退避的公式一般是 delay base * 2^attemptbase 可以设成 1 秒或者 2 秒。但这还不够因为同一批并行子任务的 attempt 次数往往一致计算出来的 delay 也一致还是会造成同步冲击。所以源码里会再加一个随机抖动比如 delay base * 2^attempt random(0, 500ms)让每个子任务的重试时间错开一点。最大重试次数我也建议少一点默认 3 次就够。重试 3 次之后还不成功大概率不是瞬时抖动继续重试只会把问题放大。这时候应该把动作转入失败处理而不是在循环里硬磨。4.4 死信队列不是垃圾桶是决策输入ActionTask 把无法成功执行的动作写入 deadLetterTag 这个侧输出。死信队列在这里不是简单地“丢掉”而是给上层 Agent 一个决策参考。我在源码阅读里注意到的做法是死信里不只记录原始请求还会带上失败阶段、最后一次错误码、已经重试的次数。这些信息会被组装成一条结构化的失败摘要回到 Agent 上下文里。大模型看到这条摘要后可以决定是换一个动作执行、还是调整参数重试、或者直接宣告任务失败。这个设计比传统的“失败就抛异常”更适合 Agent 场景。因为 Agent 的规划是有上下文和语义的它需要知道“为什么失败”才能做出下一步决策。如果只是抛一个乱码异常给大模型它大概率会猜错原因。5. 从 ActionTask 反推一套可复用的 Flink Agent 工程范式代码读到一定量我一般会做一次提纯把作者的设计抽象成可迁移的工程范式。ActionTask 虽然只是一个执行类但它背后体现的几套思路完全可以复制到其他 Flink 项目里。5.1 元数据与实现解耦动作列表可以热更新ActionSpec 是元数据ActionHandler 是实现ActionRegistry 负责绑定。这套三层结构让 Agent 的能力扩展变成了一个注册动作。如果你在一个数据平台上做 Agent我希望你认真考虑这个设计不要让大模型直接拼一个函数名去调代码一定要让大模型输出结构化的动作描述再由一个注册表去解析和分发。这样做有几个好处第一动作列表可以被外部配置中心管理动态增减能力第二动作参数可以被 Schema 校验避免大模型幻觉产生非法参数第三不同动作之间天然隔离不容易互相污染。我实际做过一个类似系统最开始就是让大模型直接调用内部函数结果经常出现大模型编造参数名的情况。改成 ActionSpec 注册表之后参数校验前置错误率明显下降。5.2 一个 Task 只做一层事恢复粒度才可控ActionTask 只是 Agent 执行链路里的一个小环节它的前后还有其他类型的 Task。我读这个系列源码时最大的感受就一句话它没有尝试把所有逻辑塞进一个大算子而是拆成了很多细粒度的小 Task。这个做法的工程收益在故障恢复时会体现出来。Flink 作业的 restart 策略一般是最小恢复单位如果一个超大算子里既有规划逻辑又有调用外部系统的逻辑那么任何一处出错整个大算子都要重启状态也都要对齐。拆成多个 Task 之后可以针对不同 Task 配置不同的并行度、不同的状态保留策略、甚至不同的重试策略。当然不是拆得越细越好。拆得太细作业图会变得很长每次重启要恢复的链条也变长。关键是按照“一层只做一件事”的标准来拆而不是按代码行数来拆。5.3 可以直接抄走的八条实现细节最后我把这次阅读里觉得最值得抄走的细节整理成一个清单方便你对照自己的项目。异步执行外部调用用 inFlight 队列记录在途请求不要把算子槽位卡在同步等待上。幂等键基于业务内容生成不要带时间戳和随机数否则重试就失去意义。幂等键的写入和外部执行不是原子的接受这个窗口用审计结果兜底。超时至少分连接、读取、整体三层每层有独立的指标和告警。网络错误和业务错误分开重试策略根据错误类型来定不能一刀切。重试用指数退避加随机抖动最大重试次数默认 3 次。动作结果要截断超长细节只保留摘要避免撑爆 Agent 上下文。错误码要结构化让上层 Agent 能基于失败原因做下一步决策而不是只看到一个 failed。最后说一个我读这段源码时绕了弯子的地方。我一开始很执着于让 ActionTask 做到“外部系统调用”和“Flink 状态”严格一致总想找到某个魔法锁或者事务机制。后来发现这版源码根本没有这么做它就是承认了不一致窗口然后用幂等、审计和重试去平滑它。这个思路我在自己项目里也用上了效果确实比追求强一致要稳得多。希望这篇对你有用后面我再接着拆这个系列里其他几个 Task 的实现。
