当前位置：首页 > news >正文

Spring AI Alibaba 项目源码学习（八）-Flow Agent 分析

news 2026/5/13 16:38:29

Flow Agent 分析

请关注微信公众号：阿呆-bot

概述

本文档分析 Spring AI Alibaba Agent Framework 中的 Flow Agent 系列，包括 FlowAgent 基类、SequentialAgent、ParallelAgent、LoopAgent 和 LlmRoutingAgent 的具体实现，重点分析其关键方法、功能职责和节点能力。

入口类说明

FlowAgent - Flow Agent 基类

FlowAgent 是所有 Flow Agent 的抽象基类，继承自 Agent，提供了多 Agent 编排的基础能力。

核心职责：

管理子 Agent 列表
提供 Graph 构建的抽象方法
委托给 FlowGraphBuilder 构建图

关键代码：

public abstract class FlowAgent extends Agent {protected List<String> interruptBefore;protected List<Agent> subAgents;protected FlowAgent(String name, String description, CompileConfig compileConfig, List<Agent> subAgents)throws GraphStateException {super(name, description);this.compileConfig = compileConfig;this.subAgents = subAgents;}@Overrideprotected StateGraph initGraph() throws GraphStateException {// Use FlowGraphBuilder to construct the graphFlowGraphBuilder.FlowGraphConfig config = FlowGraphBuilder.FlowGraphConfig.builder().name(this.name()).rootAgent(this).subAgents(this.subAgents());// Delegate to specific graph builder based on agent typereturn buildSpecificGraph(config);}@Overridepublic ScheduledAgentTask schedule(ScheduleConfig scheduleConfig) throws GraphStateException {CompiledGraph compiledGraph = getAndCompileGraph();return compiledGraph.schedule(scheduleConfig);}public StateGraph asStateGraph(){return getGraph();}/*** Abstract method for subclasses to specify their graph building strategy. This* method should be implemented by concrete FlowAgent subclasses to define how their* specific graph structure should be built.* @param config the graph configuration* @return the constructed StateGraph* @throws GraphStateException if graph construction fails*/protected abstract StateGraph buildSpecificGraph(FlowGraphBuilder.FlowGraphConfig config)throws GraphStateException;public CompileConfig compileConfig() {return compileConfig;}public List<Agent> subAgents() {return this.subAgents;}/*** Creates a map with messages and input for String message*/private Map<String, Object> createInputMap(String message) {return Map.of("messages", convertToMessages(message), "input", message);}}

关键方法：

buildSpecificGraph()：抽象方法，子类实现具体的图构建策略
subAgents()：获取子 Agent 列表

SequentialAgent - 顺序执行 Agent

SequentialAgent 按顺序执行多个子 Agent，前一个 Agent 的输出作为下一个 Agent 的输入。

关键代码：

public class SequentialAgent extends FlowAgent {protected SequentialAgent(SequentialAgentBuilder builder) throws GraphStateException {super(builder.name, builder.description, builder.compileConfig, builder.subAgents);}public static SequentialAgentBuilder builder() {return new SequentialAgentBuilder();}@Overrideprotected StateGraph buildSpecificGraph(FlowGraphBuilder.FlowGraphConfig config) throws GraphStateException {return FlowGraphBuilder.buildGraph(FlowAgentEnum.SEQUENTIAL.getType(), config);}/*** Builder for creating SequentialAgent instances. Extends the common FlowAgentBuilder* to provide type-safe building.*/public static class SequentialAgentBuilder extends FlowAgentBuilder<SequentialAgent, SequentialAgentBuilder> {@Overrideprotected SequentialAgentBuilder self() {return this;}@Overrideprotected void validate() {super.validate();// Add any SequentialAgent-specific validation here if needed}@Overridepublic SequentialAgent build() throws GraphStateException {validate();return new SequentialAgent(this);}}}

执行模式：

Agent1 → Agent2 → Agent3 → ... → END
每个 Agent 的输出作为下一个 Agent 的输入

ParallelAgent - 并行执行 Agent

ParallelAgent 并行执行多个子 Agent，然后合并结果。

关键代码：

public class ParallelAgent extends FlowAgent {private static final Logger logger = LoggerFactory.getLogger(ParallelAgent.class);private final MergeStrategy mergeStrategy;private String mergeOutputKey;private final Integer maxConcurrency;protected ParallelAgent(ParallelAgentBuilder builder) throws GraphStateException {super(builder.name, builder.description, builder.compileConfig, builder.subAgents);this.mergeStrategy = builder.mergeStrategy != null ? builder.mergeStrategy : new DefaultMergeStrategy();this.maxConcurrency = builder.maxConcurrency;this.mergeOutputKey = builder.mergeOutputKey;}public static ParallelAgentBuilder builder() {return new ParallelAgentBuilder();}@Overrideprotected StateGraph buildSpecificGraph(FlowGraphBuilder.FlowGraphConfig config) throws GraphStateException {// Add parallel-specific properties to configconfig.customProperty("mergeStrategy", this.mergeStrategy);config.customProperty("maxConcurrency", this.maxConcurrency);return FlowGraphBuilder.buildGraph(FlowAgentEnum.PARALLEL.getType(), config);}/*** Gets the merge strategy used by this ParallelAgent.* @return the merge strategy*/public MergeStrategy mergeStrategy() {return mergeStrategy;}public String mergeOutputKey() {return mergeOutputKey;}/*** Gets the maximum concurrency limit for this ParallelAgent.* @return the max concurrency, or null if unlimited*/public Integer maxConcurrency() {

执行模式：

所有子 Agent 同时执行（Fan-Out）
执行完成后合并结果（Gather）
支持最大并发数限制

LoopAgent - 循环执行 Agent

LoopAgent 循环执行一个子 Agent，支持多种循环策略。

关键代码：

public class LoopAgent extends FlowAgent {private final LoopStrategy loopStrategy;public static final String LOOP_STRATEGY = "loopStrategy";private LoopAgent(LoopAgentBuilder builder) throws GraphStateException {super(builder.name, builder.description, builder.compileConfig, builder.subAgents);this.loopStrategy = builder.loopStrategy;}@Overrideprotected StateGraph buildSpecificGraph(FlowGraphBuilder.FlowGraphConfig config) throws GraphStateException {config.customProperty(LOOP_STRATEGY, loopStrategy);return FlowGraphBuilder.buildGraph(FlowAgentEnum.LOOP.getType(), config);}public static LoopAgentBuilder builder() {return new LoopAgentBuilder();}public static class LoopAgentBuilder extends FlowAgentBuilder<LoopAgent, LoopAgentBuilder> {private LoopStrategy loopStrategy = null;@Overrideprotected LoopAgentBuilder self() {return this;}public LoopAgentBuilder subAgent(Agent subAgent) {this.subAgents = List.of(subAgent);return self();}@Overridepublic LoopAgentBuilder subAgents(List<Agent> subAgents) {throw new UnsupportedOperationException("LoopAgent must have only one subAgent, please use subAgent() method.");}public LoopAgentBuilder loopStrategy(LoopStrategy loopStrategy) {this.loopStrategy = loopStrategy;return self();

循环策略：

CountLoopStrategy：固定次数循环
ConditionLoopStrategy：条件循环（条件为 true 时终止）
ArrayLoopStrategy：遍历数组元素

LlmRoutingAgent - LLM 路由 Agent

LlmRoutingAgent 使用 LLM 根据输入内容路由到不同的子 Agent。

关键代码：

public class LlmRoutingAgent extends FlowAgent {private final ChatModel chatModel;protected LlmRoutingAgent(LlmRoutingAgentBuilder builder) throws GraphStateException {super(builder.name, builder.description, builder.compileConfig, builder.subAgents);this.chatModel = builder.chatModel;}public static LlmRoutingAgentBuilder builder() {return new LlmRoutingAgentBuilder();}@Overrideprotected StateGraph buildSpecificGraph(FlowGraphBuilder.FlowGraphConfig config) throws GraphStateException {config.setChatModel(this.chatModel);return FlowGraphBuilder.buildGraph(FlowAgentEnum.ROUTING.getType(), config);}/*** Builder for creating LlmRoutingAgent instances. Extends the common FlowAgentBuilder* and adds LLM-specific configuration.*/public static class LlmRoutingAgentBuilder extends FlowAgentBuilder<LlmRoutingAgent, LlmRoutingAgentBuilder> {private ChatModel chatModel;/*** Sets the ChatModel for LLM-based routing decisions.* @param chatModel the chat model to use for routing* @return this builder instance for method chaining*/public LlmRoutingAgentBuilder model(ChatModel chatModel) {this.chatModel = chatModel;return this;}@Overrideprotected LlmRoutingAgentBuilder self() {return this;}@Overrideprotected void validate() {super.validate();if (chatModel == null) {throw new IllegalArgumentException("ChatModel must be provided for LLM routing agent");}}@Overridepublic LlmRoutingAgent build() throws GraphStateException {validate();return new LlmRoutingAgent(this);}}}

路由机制：