程序包 io.agentscope.core.agent

类 AgentBase

java.lang.Object
io.agentscope.core.agent.AgentBase
所有已实现的接口:
Agent, CallableAgent, ObservableAgent, StreamableAgent, StateModule
直接已知子类:
StructuredOutputCapableAgent, UserAgent
public abstract class AgentBase extends Object implements StateModule, Agent
Abstract base class for all agents in the AgentScope framework.

This class provides common functionality for agents including basic hook integration, MsgHub subscriber management, interrupt handling, tracing, and state management through StateModule. It does NOT manage memory - that is the responsibility of specific agent implementations like ReActAgent.

Design Philosophy:

  • AgentBase provides infrastructure (hooks, subscriptions, interrupt, state) but not domain logic
  • Memory management is delegated to concrete agents that need it (e.g., ReActAgent)
  • State management implements StateModule interface
  • Interrupt mechanism uses reactive patterns: subclasses call checkInterruptedAsync() at appropriate checkpoints, which propagates InterruptedException through Mono chain
  • Observe pattern: agents can receive messages without generating a reply

Thread Safety: Agent instances are NOT designed for concurrent execution. A single agent instance should not be invoked concurrently from multiple threads (e.g., calling call() or stream() simultaneously). The hooks list is mutable and modified during streaming operations without synchronization, which is safe only under single-threaded execution per agent instance.

Interrupt Mechanism: 


 // External call to interrupt
 agent.interrupt(userMsg);

 // Inside agent's Mono chain, at checkpoints:
 return checkInterruptedAsync()
     .then(doWork())
     .flatMap(result -> checkInterruptedAsync().thenReturn(result));

 // AgentBase.call() catches the exception:
 .onErrorResume(error -> {
     if (error instanceof InterruptedException) {
         return handleInterrupt(context, msg);
     }
     ...
 });

  • 构造器详细资料

    • AgentBase

      public AgentBase(String name)
      Constructor for AgentBase.
      参数:
      name - Agent name

    • AgentBase

      public AgentBase(String name, String description)
      Constructor for AgentBase.
      参数:
      name - Agent name
      description - Agent description

    • AgentBase

      public AgentBase(String name, String description, boolean checkRunning, List<Hook> hooks)
      Constructor for AgentBase with hooks.
      参数:
      name - Agent name
      description - Agent description
      checkRunning - Whether to check running state
      hooks - List of hooks for monitoring/intercepting execution

  • 方法详细资料

    • getAgentId

      public final String getAgentId()
      从接口复制的说明: Agent
      Get the unique identifier for this agent.
      指定者:
      getAgentId 在接口中 Agent
      返回:
      Agent ID

    • getName

      public final String getName()
      从接口复制的说明: Agent
      Get the name of this agent.
      指定者:
      getName 在接口中 Agent
      返回:
      Agent name

    • getDescription

      public final String getDescription()
      从接口复制的说明: Agent
      Get the description of this agent.
      指定者:
      getDescription 在接口中 Agent
      返回:
      Agent description

    • call

      public final reactor.core.publisher.Mono<Msg> call(List<Msg> msgs)
      Process a list of input messages and generate a response with hook execution.

      Tracing data will be captured once telemetry is enabled.

      指定者:
      call 在接口中 CallableAgent
      参数:
      msgs - Input messages
      返回:
      Response message

    • call

      public final reactor.core.publisher.Mono<Msg> call(List<Msg> msgs, Class<?> structuredOutputClass)
      Process multiple input messages and generate structured output with hook execution.

      Tracing data will be captured once telemetry is enabled.

      指定者:
      call 在接口中 CallableAgent
      参数:
      msgs - Input messages
      structuredOutputClass - Class defining the structure of the output
      返回:
      Response message with structured data in metadata

    • call

      public final reactor.core.publisher.Mono<Msg> call(List<Msg> msgs, com.fasterxml.jackson.databind.JsonNode schema)
      Process multiple input messages and generate structured output with hook execution.

      Tracing data will be captured once telemetry is enabled.

      指定者:
      call 在接口中 CallableAgent
      参数:
      msgs - Input messages
      schema - com.fasterxml.jackson.databind.JsonNode instance defining the structure of the output
      返回:
      Response message with structured data in metadata

    • doCall

      protected abstract reactor.core.publisher.Mono<Msg> doCall(List<Msg> msgs)
      Internal implementation for processing multiple input messages. Subclasses must implement their specific logic here.
      参数:
      msgs - Input messages
      返回:
      Response message

    • doCall

      protected reactor.core.publisher.Mono<Msg> doCall(List<Msg> msgs, Class<?> structuredOutputClass)
      Internal implementation for processing multiple messages with structured output. Subclasses that support structured output must override this method. Default implementation throws UnsupportedOperationException.
      参数:
      msgs - Input messages
      structuredOutputClass - Class defining the structure
      返回:
      Response message with structured data in metadata

    • doCall

      protected reactor.core.publisher.Mono<Msg> doCall(List<Msg> msgs, com.fasterxml.jackson.databind.JsonNode outputSchema)
      Internal implementation for processing multiple messages with structured output. Subclasses that support structured output must override this method. Default implementation throws UnsupportedOperationException.
      参数:
      msgs - Input messages
      outputSchema - com.fasterxml.jackson.databind.JsonNode instance defining the structure
      返回:
      Response message with structured data in metadata

    • addSystemHook

      public static void addSystemHook(Hook hook)

    • removeSystemHook

      public static void removeSystemHook(Hook hook)

    • interrupt

      public void interrupt()
      Interrupt the current agent execution. Sets an interrupt flag that will be checked by the agent at appropriate checkpoints.
      指定者:
      interrupt 在接口中 Agent

    • interrupt

      public void interrupt(Msg msg)
      Interrupt the current agent execution with a user message. Sets an interrupt flag and associates a user message with the interruption.
      指定者:
      interrupt 在接口中 Agent
      参数:
      msg - User message associated with the interruption

    • interrupt

      public void interrupt(InterruptSource source)
      Interrupt execution with explicit source.
      参数:
      source - interruption source

    • checkInterruptedAsync

      protected reactor.core.publisher.Mono<Void> checkInterruptedAsync()
      Check if the agent execution has been interrupted (reactive version). Returns a Mono that completes normally if not interrupted, or errors with InterruptedException if interrupted.

      Subclasses should call this at appropriate checkpoints in their Mono chains. For simple agents (like UserAgent), checkpoints may not be needed. For complex agents (like ReActAgent), call this at:

      • Start of each iteration
      • Before/after reasoning
      • Before/after each tool execution
      • During streaming (each chunk)

      Example usage: 

      
 return checkInterruptedAsync()
     .then(reasoning())
     .flatMap(result -> checkInterruptedAsync().thenReturn(result))
     .flatMap(result -> executeTools(result));
      返回:
      Mono that completes if not interrupted, or errors if interrupted

    • resetInterruptFlag

      protected void resetInterruptFlag()
      Reset the interrupt flag and associated state. This is called at the beginning of each call() to prepare for new execution.

    • getInterruptFlag

      protected AtomicBoolean getInterruptFlag()
      Get the interrupt flag for access by subclasses. Subclasses can use this flag to implement custom interrupt-checking logic in addition to the standard checkInterruptedAsync() method.
      返回:
      The atomic boolean interrupt flag

    • getInterruptSource

      protected InterruptSource getInterruptSource()
      Get current interruption source.
      返回:
      interruption source

    • doObserve

      protected reactor.core.publisher.Mono<Void> doObserve(Msg msg)
      Observe a message without generating a reply. This allows agents to receive messages from other agents or the environment without responding. It's commonly used in multi-agent collaboration scenarios.

      Common implementation patterns:

      • Stateless agents: Empty implementation if observation is not needed
      • Stateful agents: Store message in memory/context for use in future calls
      • Collaborative agents: Update shared knowledge or trigger side effects
      参数:
      msg - The message to observe
      返回:
      Mono that completes when observation is done

    • handleInterrupt

      protected abstract reactor.core.publisher.Mono<Msg> handleInterrupt(InterruptContext context, Msg... originalArgs)
      Handle an interruption that occurred during execution. Subclasses must implement this to provide recovery logic based on the interrupt context.

      Implementation guidance:

      • Simple agents: Return a basic interrupt acknowledgment message
      • Complex agents: Generate a summary including any pending operations or partial results
      • Stateful agents: Ensure state is saved appropriately before returning
      参数:
      context - The interrupt context containing metadata about the interruption
      originalArgs - The original arguments passed to the call() method (empty, single Msg, or List)
      返回:
      Recovery message to return to the user

    • getHooks

      public List<Hook> getHooks()
      Get the list of hooks for this agent. Protected to allow subclasses to access hooks for custom notification logic.
      返回:
      List of hooks

    • addHook

      protected void addHook(Hook hook)
      Add a hook to this agent dynamically.

      Hooks can be added during agent execution to provide temporary functionality. This is commonly used for structured output handling or other short-lived behaviors.

      参数:
      hook - The hook to add

    • removeHook

      protected void removeHook(Hook hook)
      Remove a hook from this agent dynamically.

      Hooks should be removed when they are no longer needed to avoid memory leaks and unintended side effects.

      参数:
      hook - The hook to remove

    • getSortedHooks

      protected List<Hook> getSortedHooks()
      Get hooks sorted by priority (lower value = higher priority). Hooks with the same priority maintain registration order.
      返回:
      Sorted list of hooks

    • removeSubscribers

      public void removeSubscribers(String hubId)
      Remove all subscribers for a specific MsgHub. This method is typically called when a MsgHub is being destroyed or reset. After calling this method, the agent will no longer receive messages from the specified hub.
      参数:
      hubId - MsgHub identifier

    • resetSubscribers

      public void resetSubscribers(String hubId, List<AgentBase> subscribers)
      Reset the subscriber list for a specific MsgHub. This replaces any existing subscribers for the given hub with the new list. Typically called by MsgHub when the subscription topology changes.
      参数:
      hubId - MsgHub identifier
      subscribers - New list of subscribers (will be copied)

    • hasSubscribers

      public boolean hasSubscribers()
      Check if this agent has any subscribers. Subscribers are agents that will receive messages published through MsgHub.
      返回:
      True if agent has one or more subscribers

    • getSubscriberCount

      public int getSubscriberCount()
      Get the total number of subscribers across all MsgHubs. Subscribers are agents that will receive messages published through MsgHub.
      返回:
      Total count of subscribers

    • observe

      public final reactor.core.publisher.Mono<Void> observe(Msg msg)
      Observe a single message without generating a reply. This is the public API that delegates to doObserve implementation.
      指定者:
      observe 在接口中 ObservableAgent
      参数:
      msg - Message to observe
      返回:
      Mono that completes when observation is done

    • observe

      public final reactor.core.publisher.Mono<Void> observe(List<Msg> msgs)
      Observe multiple messages without generating a reply. This is the public API that delegates to doObserve implementation.
      指定者:
      observe 在接口中 ObservableAgent
      参数:
      msgs - Messages to observe
      返回:
      Mono that completes when all observations are done

    • stream

      public final reactor.core.publisher.Flux<Event> stream(List<Msg> msgs, StreamOptions options)
      Stream with multiple input messages.
      指定者:
      stream 在接口中 StreamableAgent
      参数:
      msgs - Input messages
      options - Stream configuration options
      返回:
      Flux of events emitted during execution

    • stream

      public final reactor.core.publisher.Flux<Event> stream(List<Msg> msgs, StreamOptions options, Class<?> structuredModel)
      Stream with multiple input messages.
      指定者:
      stream 在接口中 StreamableAgent
      参数:
      msgs - Input messages
      options - Stream configuration options
      structuredModel - Optional class defining the structure
      返回:
      Flux of events emitted during execution

    • stream

      public final reactor.core.publisher.Flux<Event> stream(List<Msg> msgs, StreamOptions options, com.fasterxml.jackson.databind.JsonNode schema)
      Stream with multiple input messages using a JSON schema.
      指定者:
      stream 在接口中 StreamableAgent
      参数:
      msgs - Input messages
      options - Stream configuration options
      schema - JSON schema defining the structure of the response
      返回:
      Flux of events emitted during execution

    • toString

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