程序包 io.agentscope.core.pipeline

类 FanoutPipeline

java.lang.Object
io.agentscope.core.pipeline.FanoutPipeline
所有已实现的接口:
Pipeline<List<Msg>>
public class FanoutPipeline extends Object implements Pipeline<List<Msg>>
Fanout pipeline implementation for parallel agent execution. This pipeline distributes the same input to multiple agents and executes them either concurrently or sequentially, collecting all results. Execution flow: Input -> [Agent1, Agent2, ..., AgentN] -> [Output1, Output2, ..., OutputN] Features: - Fan-out pattern execution (one input, multiple outputs) - Configurable concurrent vs sequential execution - Input isolation (each agent gets a copy of the input) - Result aggregation into a list - Enhanced error handling with detailed agent failure information - Composite exception collection for multiple agent failures - Individual agent error isolation without affecting others

  • 构造器详细资料

    • FanoutPipeline

      public FanoutPipeline(List<AgentBase> agents, boolean enableConcurrent)
      Create a fanout pipeline with the specified agents and execution mode. Uses boundedElastic scheduler by default for concurrent execution.
      参数:
      agents - List of agents to execute in parallel
      enableConcurrent - True for concurrent execution, false for sequential

    • FanoutPipeline

      public FanoutPipeline(List<AgentBase> agents, boolean enableConcurrent, reactor.core.scheduler.Scheduler scheduler)
      Create a fanout pipeline with the specified agents, execution mode and scheduler.
      参数:
      agents - List of agents to execute in parallel
      enableConcurrent - True for concurrent execution, false for sequential
      scheduler - Custom scheduler for execution

    • FanoutPipeline

      public FanoutPipeline(List<AgentBase> agents)
      Create a fanout pipeline with concurrent execution enabled by default.
      参数:
      agents - List of agents to execute in parallel

  • 方法详细资料

    • execute

      public reactor.core.publisher.Mono<List<Msg>> execute(Msg input)
      从接口复制的说明: Pipeline
      Execute the pipeline with the given input message.
      指定者:
      execute 在接口中 Pipeline<List<Msg>>
      参数:
      input - Input message to process through the pipeline
      返回:
      Mono containing the pipeline result

    • execute

      public reactor.core.publisher.Mono<List<Msg>> execute(Msg input, Class<?> structuredOutputClass)
      从接口复制的说明: Pipeline
      Execute the pipeline with the given input message and structured output.
      指定者:
      execute 在接口中 Pipeline<List<Msg>>
      参数:
      input - Input message to process through the pipeline
      structuredOutputClass - The class type for structured output
      返回:
      Mono containing the pipeline result with structured output

    • getAgents

      public List<AgentBase> getAgents()
      Get the list of agents in this pipeline.
      返回:
      Copy of the agents list

    • size

      public int size()
      Get the number of agents in this pipeline.
      返回:
      Number of agents

    • isEmpty

      public boolean isEmpty()
      Check if this pipeline is empty (has no agents).
      返回:
      True if pipeline has no agents

    • isConcurrentEnabled

      public boolean isConcurrentEnabled()
      Check if concurrent execution is enabled.
      返回:
      True if agents execute concurrently

    • getDescription

      public String getDescription()
      从接口复制的说明: Pipeline
      Get a description of this pipeline.
      指定者:
      getDescription 在接口中 Pipeline<List<Msg>>
      返回:
      Human-readable description of the pipeline

    • stream

      public reactor.core.publisher.Flux<Event> stream(Msg input)
      Stream execution events from all agents with default options.

      Events from multiple agents are merged (concurrent mode) or concatenated (sequential mode) based on the pipeline configuration.

      参数:
      input - Input message to distribute to all agents
      返回:
      Flux of events emitted during execution from all agents

    • stream

      public reactor.core.publisher.Flux<Event> stream(Msg input, StreamOptions options)
      Stream execution events from all agents with specified options.

      Events from multiple agents are merged (concurrent mode) or concatenated (sequential mode) based on the pipeline configuration.

      参数:
      input - Input message to distribute to all agents
      options - Stream configuration options
      返回:
      Flux of events emitted during execution from all agents

    • stream

      public reactor.core.publisher.Flux<Event> stream(Msg input, StreamOptions options, Class<?> structuredOutputClass)
      Stream execution events from all agents with structured output support.

      Events from multiple agents are merged (concurrent mode) or concatenated (sequential mode) based on the pipeline configuration.

      参数:
      input - Input message to distribute to all agents
      options - Stream configuration options
      structuredOutputClass - The class type for structured output (optional)
      返回:
      Flux of events emitted during execution from all agents

    • toString

      public String toString()
      覆盖:
      toString 在类中 Object

    • builder

      public static FanoutPipeline.Builder builder()
      Create a builder for constructing fanout pipelines.
      返回:
      New pipeline builder