程序包 io.agentscope.core.pipeline

类 MsgHub

java.lang.Object
io.agentscope.core.pipeline.MsgHub
所有已实现的接口:
AutoCloseable
public class MsgHub extends Object implements AutoCloseable
MsgHub is designed to share messages among a group of agents.

MsgHub manages message broadcasting and subscription in multi-agent conversations. When agents are added to a MsgHub, they automatically observe each other's messages without explicit message passing code.

Features:

  • Automatic Broadcasting: Messages from any participant are automatically broadcast to all other participants
  • Dynamic Participants: Add or remove agents during conversation
  • Manual Broadcasting: Broadcast messages manually when needed
  • Announcement Support: Send initial messages when entering the hub
  • Lifecycle Management: Automatic cleanup with try-with-resources

Usage Example: 


 // Create agents
 ReActAgent alice = ReActAgent.builder()
     .name("Alice")
     .model(model)
     .formatter(new DashScopeMultiAgentFormatter())
     .build();

 ReActAgent bob = ReActAgent.builder()
     .name("Bob")
     .model(model)
     .formatter(new DashScopeMultiAgentFormatter())
     .build();

 // Use MsgHub for multi-agent conversation
 Msg announcement = new Msg("system", "Let's start the discussion", MsgRole.SYSTEM);

 try (MsgHub hub = MsgHub.builder()
         .participants(alice, bob)
         .announcement(announcement)
         .build()) {

     hub.enter().block();

     // Alice's reply will be automatically broadcast to Bob
     alice.call().block();

     // Bob's reply will be automatically broadcast to Alice
     bob.call().block();
 }

This is much simpler than manual message passing: 


 // Without MsgHub (verbose and error-prone)
 Msg x1 = alice.call().block();
 bob.observe(x1).block();

 Msg x2 = bob.call().block();
 alice.observe(x2).block();

Thread Safety: MsgHub uses CopyOnWriteArrayList for thread-safe participant management. However, individual agent instances should not be invoked concurrently.

另请参阅:

  • 方法详细资料

    • getName

      public String getName()
      Get the name of this MsgHub.
      返回:
      MsgHub name

    • getParticipants

      public List<AgentBase> getParticipants()
      Get the list of current participants.
      返回:
      Unmodifiable view of participants

    • isAutoBroadcastEnabled

      public boolean isAutoBroadcastEnabled()
      Check if auto-broadcast is enabled.
      返回:
      True if auto-broadcast is enabled

    • enter

      public reactor.core.publisher.Mono<MsgHub> enter()
      Enter the MsgHub context. This method initializes subscriber relationships and broadcasts announcement messages.

      Must be called before using the hub. Typically called in a try-with-resources block or explicitly managed with enter()/exit().

      返回:
      Mono containing this MsgHub instance

    • exit

      public reactor.core.publisher.Mono<Void> exit()
      Exit the MsgHub context. This method cleans up subscriber relationships.

      Called automatically when using try-with-resources via close().

      返回:
      Mono that completes when cleanup is done

    • close

      public void close()
      Close the MsgHub and cleanup resources. This is the AutoCloseable implementation for try-with-resources support.
      指定者:
      close 在接口中 AutoCloseable

    • add

      public reactor.core.publisher.Mono<Void> add(AgentBase... newParticipants)
      Add new participants to this hub. Automatically updates subscriber relationships if the hub is active.
      参数:
      newParticipants - Agents to add (varargs)
      返回:
      Mono that completes when participants are added

    • add

      public reactor.core.publisher.Mono<Void> add(List<AgentBase> newParticipants)
      Add new participants to this hub. Automatically updates subscriber relationships if the hub is active.
      参数:
      newParticipants - List of agents to add
      返回:
      Mono that completes when participants are added

    • delete

      public reactor.core.publisher.Mono<Void> delete(AgentBase... toRemove)
      Remove participants from this hub. Automatically updates subscriber relationships if the hub is active.
      参数:
      toRemove - Agents to remove (varargs)
      返回:
      Mono that completes when participants are removed

    • delete

      public reactor.core.publisher.Mono<Void> delete(List<AgentBase> toRemove)
      Remove participants from this hub. Automatically updates subscriber relationships if the hub is active.
      参数:
      toRemove - List of agents to remove
      返回:
      Mono that completes when participants are removed

    • broadcast

      public reactor.core.publisher.Mono<Void> broadcast(Msg msg)
      Broadcast a message to all participants. All participants will receive the message via their observe() method.
      参数:
      msg - Message to broadcast
      返回:
      Mono that completes when all participants have observed the message

    • broadcast

      public reactor.core.publisher.Mono<Void> broadcast(List<Msg> msgs)
      Broadcast multiple messages to all participants. All participants will receive all messages via their observe() method.
      参数:
      msgs - Messages to broadcast
      返回:
      Mono that completes when all participants have observed all messages

    • setAutoBroadcast

      public void setAutoBroadcast(boolean enable)
      Enable or disable automatic broadcasting.

      When enabled, each participant's reply will be automatically broadcast to all other participants. When disabled, the MsgHub only serves as a manual broadcaster.

      参数:
      enable - True to enable auto-broadcast, false to disable

    • builder

      public static MsgHub.Builder builder()
      Create a new MsgHub builder.
      返回:
      New builder instance