跳至主要內容

Cactus li...大约 11 分钟

一、本章介绍

采用新的设计思路，提升 AgentWorkflowNode 的流转管理能力，使 LoopAgentNode、ParallelAgentNode 与 SequentialAgentNode 不再承担流转判断的职责。每个节点在完成自身处理后，都会返回 AgentWorkflowNode，由它统一进行流转决策。这种方式能够支持更灵活、复杂的智能体编排和组合，实现多样化的流程控制。

二、流程设计

如图，增强 AgentWorkflowNode 流转能力，每个节点流转完都重新回到 AgentWorkflowNode 节点进行决策；

![AI Agent 脚手架-增强装配-AgentWorkflowNode.drawio](https://beauties.eu.org/blogimg/main/img1/AI Agent 脚手架-增强装配-AgentWorkflowNode.drawio.png)

  • 左侧展示的是旧版流程设计，其中 LoopAgentNode、ParallelAgentNode、SequentialAgentNode 之间存在相互跳转和交叉流转的逻辑，节点既承担业务处理，又负责流程决策，职责边界相对模糊。

  • 新版流程则将所有流转控制统一收敛到 AgentWorkflowNode 中，由它负责整个工作流的调度与决策。这样一来，LoopAgentNode、ParallelAgentNode、SequentialAgentNode 只专注于自身业务逻辑处理，不再参与流程判断，使各节点职责更加单一、清晰。

  • 经过调整后，AgentWorkflowNode 相当于整个流程的调度中心。三个智能体节点在完成各自任务后，统一返回到 AgentWorkflowNode，再由其根据当前状态和配置决定下一步执行路径，从而实现更加灵活且易于扩展的智能体编排能力。

三、功能实现

1. 工程结构

  • 调整 LoopAgentNode、ParallelAgentNode、SequentialAgentNode 三个节点的流程控制逻辑，将原本分散在各节点中的流转处理统一收敛到 AgentWorkflowNode 中，由其负责后续节点的路由与调度。

  • 在 AgentWorkflowNode 中，需要根据当前 agentWorkflows 配置列表，定位到执行步骤中的第 N 个节点，并将对应的节点信息写入上下文。后续无论流转到哪个业务节点，都只需从上下文中获取当前节点配置并完成处理，无需再关心流程编排逻辑。

  • 通过这种设计，流程控制与业务处理职责得到有效分离。AgentWorkflowNode 专注于工作流编排和节点调度，而 LoopAgentNode、ParallelAgentNode、SequentialAgentNode 则只负责各自的业务执行，从而提升整体架构的可维护性与扩展能力。

2. 核心模块

2.1 定义上下文

@Data
@Builder
@AllArgsConstructor
@NoArgsConstructor
public static class DynamicContext {

    /**
     * LLM API
     */
    private OpenAiApi openAiApi;

    /**
     * 对话模型
     */
    private ChatModel chatModel;

    /**
     * 原子安全的递进步骤
     */
    private AtomicInteger currentStepIndex = new AtomicInteger(0);

    /**
     * 当前的智能体
     */
    private AiAgentConfigTableVO.Module.AgentWorkflow currentAgentWorkflow;

    /**
     * 智能体组
     */
    private Map<String, BaseAgent> agentGroup = new HashMap<>();

    private Map<String, Object> dataObjects = new HashMap<>();
}

  • 首先，对上下文中的工作流数据结构进行调整，不再直接使用 agentWorkflows 列表中的当前元素作为处理对象，而是新增一个 currentAgentWorkflow 属性，用于保存当前正在执行的工作流节点配置。

  • 随后，增加 currentStepIndex 作为当前步骤索引标识。每当一个节点执行完成后，便将索引值递增，通过 currentStepIndex 从 agentWorkflows 集合中获取对应的工作流配置，并更新到 currentAgentWorkflow 中。

  • 采用这种方式后，后续节点只需关注 currentAgentWorkflow 即可获取当前执行信息，无需频繁从工作流列表中查找目标对象。同时，流程推进逻辑也能够通过 currentStepIndex 统一管理，使执行状态更加清晰，代码实现更加简洁，后续扩展和维护也会更加方便。

2.2 增强流转

@Slf4j
@Service
public class AgentWorkflowNode extends AbstractArmorySupport {

    @Resource
    private LoopAgentNode loopAgentNode;

    @Resource
    private ParallelAgentNode parallelAgentNode;

    @Resource
    private SequentialAgentNode sequentialAgentNode;

    @Resource
    private RunnerNode runnerNode;

    @Override
    protected AiAgentRegisterVO doApply(ArmoryCommandEntity requestParameter, DefaultArmoryFactory.DynamicContext dynamicContext) throws Exception {
        log.info("Ai Agent 装配操作 - AgentWorkflowNode");

        AiAgentConfigTableVO aiAgentConfigTableVO = requestParameter.getAiAgentConfigTableVO();
        List<AiAgentConfigTableVO.Module.AgentWorkflow> agentWorkflows = aiAgentConfigTableVO.getModule().getAgentWorkflows();

        // 如果未配置 agentWorkflows 则直接流转到 RunnerNode
        if (null == agentWorkflows || agentWorkflows.isEmpty() || dynamicContext.getCurrentStepIndex() >= agentWorkflows.size()) {
            // 设置结果值
            dynamicContext.setCurrentAgentWorkflow(null);
            // 路由下节点
            return router(requestParameter, dynamicContext);
        }

        // 设置当前判断流程对象
        dynamicContext.setCurrentAgentWorkflow(agentWorkflows.get(dynamicContext.getCurrentStepIndex()));

        // 步骤值增加
        dynamicContext.addCurrentStepIndex();

        return router(requestParameter, dynamicContext);
    }

    @Override
    public StrategyHandler<ArmoryCommandEntity, DefaultArmoryFactory.DynamicContext, AiAgentRegisterVO> get(ArmoryCommandEntity requestParameter, DefaultArmoryFactory.DynamicContext dynamicContext) throws Exception {

        AiAgentConfigTableVO.Module.AgentWorkflow currentAgentWorkflow = dynamicContext.getCurrentAgentWorkflow();

        // 没有下一个节点，流转到结束节点
        if (null == currentAgentWorkflow) {
            return runnerNode;
        }

        String type = currentAgentWorkflow.getType();
        AgentTypeEnum agentTypeEnum = AgentTypeEnum.fromType(type);

        if (null == agentTypeEnum) {
            throw new RuntimeException("agentWorkflow type is error!");
        }

        String node = agentTypeEnum.getNode();

        return switch (node) {
            case "loopAgentNode" -> loopAgentNode;
            case "parallelAgentNode" -> parallelAgentNode;
            case "sequentialAgentNode" -> sequentialAgentNode;
            default -> runnerNode;
        };

    }

}

  • doApply 方法的核心职责，是判断是否配置了 agentWorkflows，并按步骤持续获取当前要执行的节点配置，整体逻辑类似于一次循环遍历。如果已经执行到最后一个步骤，则将上下文中的 currentAgentWorkflow 设置为 null，并继续向下路由。否则，就根据当前步骤索引从 agentWorkflows 中取出对应对象，写入到上下文中，同时将步骤索引执行 +1，为下一次流转做好准备。

  • get 方法主要负责节点路由判断。若当前节点为 null，说明已经没有需要继续处理的子智能体节点，此时直接流转到 RunnerNode。如果当前节点不为空，则根据配置中的节点类型，分别路由到对应的子 Agent 节点进行处理。

2.3 子智能体节点

2.3.1 LoopAgentNode
@Service
public class LoopAgentNode extends AbstractArmorySupport {

    @Override
    protected AiAgentRegisterVO doApply(ArmoryCommandEntity requestParameter, DefaultArmoryFactory.DynamicContext dynamicContext) throws Exception {
        log.info("Ai Agent 装配操作 - LoopAgentNode");

        AiAgentConfigTableVO.Module.AgentWorkflow currentAgentWorkflow = dynamicContext.getCurrentAgentWorkflow();

        List<BaseAgent> subAgents = dynamicContext.queryAgentList(currentAgentWorkflow.getSubAgents());

        LoopAgent loopAgent =
                LoopAgent.builder()
                        .name(currentAgentWorkflow.getName())
                        .description(currentAgentWorkflow.getDescription())
                        .subAgents(subAgents)
                        .maxIterations(currentAgentWorkflow.getMaxIterations())
                        .build();

        dynamicContext.getAgentGroup().put(currentAgentWorkflow.getName(), loopAgent);

        return router(requestParameter, dynamicContext);
    }

    @Override
    public StrategyHandler<ArmoryCommandEntity, DefaultArmoryFactory.DynamicContext, AiAgentRegisterVO> get(ArmoryCommandEntity requestParameter, DefaultArmoryFactory.DynamicContext dynamicContext) throws Exception {
        return getBean("agentWorkflowNode");
    }

}

  • doApply 方法需要改为从上下文中的 dynamicContext.getCurrentAgentWorkflow() 获取当前节点配置，并基于该配置完成对应 Agent 的构建。构建完成后，再继续执行路由流程。

  • get 方法则统一流转回 getBean("agentWorkflowNode")，把流程控制权重新交给 AgentWorkflowNode。这样后续节点如何选择、流程是否继续推进，都由 AgentWorkflowNode 统一判断处理。

  • 通过这种调整，子 Agent 节点只负责当前节点的业务构建，不再关心后续流转逻辑，整体职责划分会更加清晰。

2.3.2 ParallelAgentNode
@Service
public class ParallelAgentNode extends AbstractArmorySupport {

    @Override
    protected AiAgentRegisterVO doApply(ArmoryCommandEntity requestParameter, DefaultArmoryFactory.DynamicContext dynamicContext) throws Exception {
        log.info("Ai Agent 装配操作 - ParallelAgentNode");

        AiAgentConfigTableVO.Module.AgentWorkflow currentAgentWorkflow = dynamicContext.getCurrentAgentWorkflow();

        List<BaseAgent> subAgents = dynamicContext.queryAgentList(currentAgentWorkflow.getSubAgents());

        ParallelAgent parallelAgent =
                ParallelAgent.builder()
                        .name(currentAgentWorkflow.getName())
                        .subAgents(subAgents)
                        .description(currentAgentWorkflow.getDescription())
                        .build();

        dynamicContext.getAgentGroup().put(currentAgentWorkflow.getName(), parallelAgent);

        return router(requestParameter, dynamicContext);
    }

    @Override
    public StrategyHandler<ArmoryCommandEntity, DefaultArmoryFactory.DynamicContext, AiAgentRegisterVO> get(ArmoryCommandEntity requestParameter, DefaultArmoryFactory.DynamicContext dynamicContext) throws Exception {
        return getBean("agentWorkflowNode");
    }
}
  • 代码修改方式同 LoopAgentNode
2.3.3 SequentialAgentNode
@Service
public class SequentialAgentNode extends AbstractArmorySupport {

    @Override
    protected AiAgentRegisterVO doApply(ArmoryCommandEntity requestParameter, DefaultArmoryFactory.DynamicContext dynamicContext) throws Exception {
        log.info("Ai Agent 装配操作 - SequentialAgentNode");

        AiAgentConfigTableVO.Module.AgentWorkflow currentAgentWorkflow = dynamicContext.getCurrentAgentWorkflow();

        List<BaseAgent> subAgents = dynamicContext.queryAgentList(currentAgentWorkflow.getSubAgents());

        SequentialAgent sequentialAgent =
                SequentialAgent.builder()
                        .name(currentAgentWorkflow.getName())
                        .description(currentAgentWorkflow.getDescription())
                        .subAgents(subAgents)
                        .build();

        dynamicContext.getAgentGroup().put(currentAgentWorkflow.getName(), sequentialAgent);

        return router(requestParameter, dynamicContext);
    }

    @Override
    public StrategyHandler<ArmoryCommandEntity, DefaultArmoryFactory.DynamicContext, AiAgentRegisterVO> get(ArmoryCommandEntity requestParameter, DefaultArmoryFactory.DynamicContext dynamicContext) throws Exception {
        return getBean("agentWorkflowNode");
    }

}
  • 代码修改方式同 LoopAgentNode

四、测试验证

1. 修改配置文件

parallel_research_app.yml

ai:
  agent:
    config:
      tables:
        testAgent02:
          app-name: ResearchAndSynthesisPipeline
          agent:
            agent-id: 100002
            agent-name: 测试智能体02
            agent-desc: 并行检索多个主题，然后由汇总智能体生成结构化报告
          module:
            ai-api:
              base-url: https://api1.oaipro.com
              api-key: sk-FtZfc3Lyr2RjMkT1TDMTuY73t2NDvyU5dDgqjt4ZqwdMAdNk3xAW
              completions-path: v1/chat/completions
              embeddings-path: v1/embeddings
            chat-model:
              model: gpt-4.1
              tool-mcp-list:
                - sse:
                    name: baidu-search
                    base-uri: http://appbuilder.baidu.com/v2/ai_search/mcp/
                    sse-endpoint: sse?api_key=bce-v3/ALTAK-6SyOtHHftyPsWLoIpz3WL/e7a0d8c2e39e9596b0f1e183cd985da1c32a0d67
                    request-timeout: 50000
            agents:
              - name: RenewableEnergyResearcher
                description: Researches renewable energy sources.
                instruction: |
                  You are an AI Research Assistant specializing in energy.
                  Research the latest advancements in 'renewable energy sources'.
                  Use the Google Search tool provided.
                  Summarize your key findings concisely (1-2 sentences).
                  Output *only* the summary.
                output-key: renewable_energy_result
              - name: EVResearcher
                description: Researches electric vehicle technology.
                instruction: |
                  You are an AI Research Assistant specializing in transportation.
                  Research the latest developments in 'electric vehicle technology'.
                  Use the Google Search tool provided.
                  Summarize your key findings concisely (1-2 sentences).
                  Output *only* the summary.
                output-key: ev_technology_result
              - name: CarbonCaptureResearcher
                description: Researches carbon capture methods.
                instruction: |
                  You are an AI Research Assistant specializing in climate solutions.
                  Research the current state of 'carbon capture methods'.
                  Use the Google Search tool provided.
                  Summarize your key findings concisely (1-2 sentences).
                  Output *only* the summary.
                output-key: carbon_capture_result
              - name: SynthesisAgent
                description: Combines research findings into a structured report.
                instruction: |
                  You are an AI Assistant responsible for combining research findings into a structured report.
                  Your primary task is to synthesize the following research summaries, clearly attributing findings to their source areas. Structure your response using headings for each topic. Ensure the report is coherent and integrates the key points smoothly.
                  **Crucially: Your entire response MUST be grounded *exclusively* on the information provided in the 'Input Summaries' below. Do NOT add any external knowledge, facts, or details not present in these specific summaries.**
                  **Input Summaries:**

                  *   **Renewable Energy:**
                      {renewable_energy_result}

                  *   **Electric Vehicles:**
                      {ev_technology_result}

                  *   **Carbon Capture:**
                      {carbon_capture_result}

                  **Output Format:**

                  ## Summary of Recent Sustainable Technology Advancements

                  ### Renewable Energy Findings
                  (Based on RenewableEnergyResearcher's findings)
                  [Synthesize and elaborate *only* on the renewable energy input summary provided above.]

                  ### Electric Vehicle Findings
                  (Based on EVResearcher's findings)
                  [Synthesize and elaborate *only* on the EV input summary provided above.]

                  ### Carbon Capture Findings
                  (Based on CarbonCaptureResearcher's findings)
                  [Synthesize and elaborate *only* on the carbon capture input summary provided above.]

                  ### Overall Conclusion
                  [Provide a brief (1-2 sentence) concluding statement that connects *only* the findings presented above.]

                  Output *only* the structured report following this format. Do not include introductory or concluding phrases outside this structure, and strictly adhere to using only the provided input summary content.
            agent-workflows:
              - type: parallel
                name: ParallelWebResearchAgent
                description: Runs multiple research agents in parallel to gather information.
                sub-agents:
                  - RenewableEnergyResearcher
                  - EVResearcher
                  - CarbonCaptureResearcher
              - type: sequential
                name: ResearchAndSynthesisPipeline
                description: Coordinates parallel research and synthesizes the results.
                sub-agents:
                  - ParallelWebResearchAgent
                  - SynthesisAgent
            runner:
              agent-name: ResearchAndSynthesisPipeline

  • 在 agent-workflows 配置中，依次添加 ParallelWebResearchAgent 和 ResearchAndSynthesisPipeline 两个智能体节点，以此验证工作流按步骤循环流转的处理效果，以及 AgentWorkflowNode 对流程调度逻辑的正确性。

  • 同时，在 runner 配置中，将 agent-name 指定为 ResearchAndSynthesisPipeline，用于完成运行时执行对象的装配。这样在工作流执行结束后，RunnerNode 便会根据配置获取对应的智能体实例，并构建运行所需的执行环境。

  • 通过这组配置，可以完整验证从工作流编排、节点流转到 Runner 装配和运行的整个处理链路是否符合预期。

2. 测试方法

    @Test
    public void test_handlerMessage_02(){
        AiAgentRegisterVO aiAgentRegisterVO = applicationContext.getBean("100002", AiAgentRegisterVO.class);

        String appName = aiAgentRegisterVO.getAppName();
        InMemoryRunner runner = aiAgentRegisterVO.getRunner();

        Session session = runner.sessionService()
                .createSession(appName, "cactusli")
                .blockingGet();

        Content userMsg = Content.fromParts(Part.fromText("你具备哪些能力?"));
        Flowable<Event> events = runner.runAsync("cactusli", session.id(), userMsg);

        List<String> outputs = new ArrayList<>();
        events.blockingForEach(event -> outputs.add(event.stringifyContent()));

        log.info("测试结果:{}", JSON.toJSONString(outputs));
    }
26-06-04.17:10:10.557 [main            ] INFO  SpringAIObservabilityHandler - Request completed successfully: model=openai, type=chat, duration=4100ms, tokens=1131
26-06-04.17:10:13.368 [main            ] INFO  SpringAIObservabilityHandler - Request completed successfully: model=openai, type=chat, duration=2805ms, tokens=1244
26-06-04.17:10:15.879 [main            ] INFO  SpringAIObservabilityHandler - Request completed successfully: model=openai, type=chat, duration=2510ms, tokens=1235
26-06-04.17:10:21.825 [main            ] INFO  SpringAIObservabilityHandler - Request completed successfully: model=openai, type=chat, duration=5941ms, tokens=2544
26-06-04.17:10:21.828 [main            ] INFO  AiAgentAutoConfigTest  - 测试结果:["我可以帮助你检索和总结最新的可再生能源领域的前沿进展，包括但不限于太阳能、风能、生物质能、地热能、海洋能等，同时能够分析技术趋势、产业动态、政策变化以及相关科研论文和市场信息。","我具备以下能力：\n\n1. 电动汽车技术研究：可以为你搜索并汇总最新的电动汽车技术动态，包括电池创新、新一代电机、充电技术、智能网联等领域的进展。\n2. 快速信息检索：能够利用互联网搜索功能，查找权威来源的最新研究、产业动态、政策法规等相关信息。\n3. 信息整合与总结：擅长将复杂的信息和数据进行简明扼要的汇总，便于快速理解和决策。\n4. 科技趋势解读：能分析主流电动汽车领域的技术发展趋势和未来前景。\n5. 多语种能力：支持中英文信息的理解与整理。\n\n如果你有关于电动汽车技术或相关领域的问题，欢迎随时提问！","我是一个专注于气候解决方案的AI研究助理，主要擅长以下能力：\n\n1. 研究与梳理碳捕集方法（如直接空气捕集、生物碳捕集、工业碳捕集等）及其最新发展动态；\n2. 利用高效的网络搜索工具，获取和总结碳捕集相关的最新资讯、技术进展和科学研究成果；\n3. 提供碳捕集领域相关的简要综述、技术对比、应用场景说明等；\n4. 回答有关碳捕集基本原理、实际应用、政策、市场前景等方面的问题。\n\n如需碳捕集方法的最新研究动态或技术对比等信息，请随时提问！","## Summary of Recent Sustainable Technology Advancements\n\n### Renewable Energy Findings\n(Based on RenewableEnergyResearcher's findings)\n\nRecent advancements in the renewable energy sector cover a broad range of sources, including solar, wind, biomass, geothermal, and ocean energy. Key capabilities include the ability to retrieve and summarize the latest progress in these areas, analyze technology trends, track industry developments, observe policy changes, and review related scientific literature and market information. This comprehensive approach enables in-depth understanding of both the technological and market landscapes in renewable energy.\n\n### Electric Vehicle Findings\n(Based on EVResearcher's findings)\n\nCurrent research into electric vehicles (EVs) involves several critical aspects: the latest breakthroughs in battery innovation, next-generation motors, charging technologies, and intelligent connectivity systems. Additional strengths include rapid information retrieval from authoritative sources on industry trends, research, and policy updates, as well as the ability to succinctly integrate and summarize complex data for efficient understanding and decision-making. The analysis also encompasses interpretation of major technology trends and future outlooks in the EV sector, all with support for multilingual information processing.\n\n### Carbon Capture Findings\n(Based on CarbonCaptureResearcher's findings)\n\nIn the field of carbon capture, expertise includes research and synthesis of various capture methods such as direct air capture, biogenic capture, and industrial carbon capture, as well as tracking their latest advancements. Capabilities extend to efficiently sourcing and summarizing up-to-date information, technological progress, and scientific findings using advanced search tools. The research also provides concise overviews, technical comparisons, and explanations of different application scenarios, and addresses questions about basic principles, practical implementation, policies, and market prospects in carbon capture.\n\n### Overall Conclusion\n\nThe summarized findings indicate robust progress and information synthesis capabilities across renewable energy, electric vehicles, and carbon capture, enabling comprehensive analysis of technologies, trends, and market dynamics within the sustainable technology landscape."]
默认页脚
Copyright © 2018-2026 Cactus li