程序包 io.agentscope.core.agent

类 AgentBase

java.lang.Object
io.agentscope.core.agent.AgentBase
所有已实现的接口:
Agent, CallableAgent, ObservableAgent, StreamableAgent
直接已知子类:
ReActAgent, UserAgent
public abstract class AgentBase extends Object implements 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

      @Deprecated public AgentBase(String name, String description, boolean checkRunning, List<Hook> hooks)
      已过时。
      Use AgentBase(String, String, List) instead.

    • AgentBase

      public AgentBase(String name, String description, List<Hook> hooks)
      Constructor for AgentBase with hooks.
      参数:
      name - Agent name
      description - Agent description
      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

    • isCheckRunning

      @Deprecated public final boolean isCheckRunning()
      已过时。
      No longer enforced; per-session serialization handles concurrency.

    • 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

    • callInternal

      protected reactor.core.publisher.Mono<Msg> callInternal(List<Msg> msgs, RuntimeContext context, Function<List<Msg>,reactor.core.publisher.Mono<Msg>> doCallFn)
      Extension point called by every call() overload, allowing subclasses to wrap the entire invocation in additional middleware (e.g. the onAgent chain in ReActAgent).

      The default implementation attaches context to the Reactor Context (when non-null) and delegates straight to runLifecycle(java.util.List<io.agentscope.core.message.Msg>, java.util.function.Function<java.util.List<io.agentscope.core.message.Msg>, reactor.core.publisher.Mono<io.agentscope.core.message.Msg>>). Subclasses that override this method must eventually invoke runLifecycle(msgs, doCallFn) to run the standard lifecycle (shutdown guard, serialization gate, pre/post hooks, tracing).

      参数:
      msgs - input messages
      context - caller-supplied per-call RuntimeContext, or null
      doCallFn - the concrete call implementation (doCall(java.util.List<io.agentscope.core.message.Msg>) or a structured-output variant)
      返回:
      response message

    • runLifecycle

      protected reactor.core.publisher.Mono<Msg> runLifecycle(List<Msg> msgs, Function<List<Msg>,reactor.core.publisher.Mono<Msg>> doCallFn)
      Shared call() lifecycle: acquire execution, then (inside deferContextual so the caller-supplied RuntimeContext is read per-subscription) run beforeAgentExecution(List, RuntimeContext) (which returns this call's per-call scope), carry that scope on the Reactor Context, and run the preCall → doCall → postCall chain with error handling, releasing execution on terminate.

    • callSerializationKey

      protected Object callSerializationKey(RuntimeContext rc)
      Returns a key that serializes concurrent calls sharing it: calls with an equal, non-null key run one-at-a-time in FIFO order, while calls with different keys (or a null key) run concurrently. The default returns null (no serialization). ReActAgent returns its (userId, sessionId) slot key so same-session calls do not corrupt shared conversation state while distinct sessions run in parallel.
      参数:
      rc - the caller-supplied per-call RuntimeContext (may be null)
      返回:
      the serialization key, or null to run without serialization

    • 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

      @Deprecated public void interrupt()
      已过时。
      Subclasses should implement per-session interrupt via RuntimeContext.
      从接口复制的说明: Agent
      Interrupt the current agent execution. This method sets an interrupt flag that will be checked by the agent at appropriate checkpoints during execution. The interruption is cooperative and may not take effect immediately.
      指定者:
      interrupt 在接口中 Agent

    • interrupt

      @Deprecated public void interrupt(Msg msg)
      已过时。
      Subclasses should implement per-session interrupt via RuntimeContext.
      从接口复制的说明: Agent
      Interrupt the current agent execution with a user message. This method sets an interrupt flag and associates a user message with the interruption. The interruption is cooperative and may not take effect immediately.
      指定者:
      interrupt 在接口中 Agent
      参数:
      msg - User message associated with the interruption

    • interrupt

      @Deprecated public void interrupt(InterruptSource source)
      已过时。
      Subclasses should implement per-session interrupt via RuntimeContext.

    • checkInterruptedAsync

      @Deprecated protected reactor.core.publisher.Mono<Void> checkInterruptedAsync()
      已过时。
      No longer needed; ReActAgent uses per-session InterruptControl.

    • resetInterruptFlag

      @Deprecated protected void resetInterruptFlag()
      已过时。
      No-op; per-session interrupt state is managed by AgentState.interruptControl().

    • getInterruptFlag

      @Deprecated protected AtomicBoolean getInterruptFlag()
      已过时。
      No-op stub.

    • getInterruptSource

      @Deprecated protected InterruptSource getInterruptSource()
      已过时。
      Returns USER. Per-session interrupt source is on AgentState.interruptControl().

    • 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

    • getAgentState

      public AgentState getAgentState()
      Returns the agent's mutable runtime state, or null if this agent type does not maintain an AgentState.
      指定者:
      getAgentState 在接口中 Agent

    • getRuntimeContext

      public RuntimeContext getRuntimeContext()
      Returns the current per-call RuntimeContext, or null when the agent keeps no per-call scope. The base implementation returns null; agents with per-call state (e.g. ReActAgent) override this to return their active call scope's context. Because the value is sourced from the agent's most-recently-activated scope, under concurrent calls on one instance this reflects the latest call — middlewares/tools that need their own call's context should read it from the per-subscription RuntimeContext they are handed.

    • beforeAgentExecution

      protected Object beforeAgentExecution(List<Msg> msgs, RuntimeContext rc)
      Invoked at the start of a call / stream-backed call, after acquireExecution() and before any hooks. ReActAgent uses this to activate the per-call session slot from the supplied RuntimeContext and returns the freshly-built per-call scope object, which is carried on the Reactor Context under CALL_SCOPE_KEY. Returning the scope here (rather than via a separate accessor) avoids any window in which a concurrent call could overwrite a shared field between construction and capture.
      参数:
      msgs - the messages passed by the caller to call()
      rc - the caller-supplied per-call RuntimeContext, or null when none was provided (read from the Reactor Context, so concurrency-safe)
      返回:
      this call's per-call scope object, or null if this agent type keeps none

    • afterAgentExecution

      protected void afterAgentExecution()
      Invoked in Mono.using cleanup, before clearing the running state. Pairs with beforeAgentExecution(List, RuntimeContext). The default is a no-op.

    • bindRuntimeContextToHooks

      protected void bindRuntimeContextToHooks(RuntimeContext ctx)
      Pushes ctx to all RuntimeContextAware hooks registered for this agent. The per-call RuntimeContext itself is no longer stored on a shared instance field; it lives on the agent's per-call scope (see getRuntimeContext()).

    • unbindRuntimeContextFromHooks

      protected void unbindRuntimeContextFromHooks()
      Clears the RuntimeContext previously pushed to all RuntimeContextAware hooks.

    • 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

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

    • seedSystemMsg

      protected reactor.core.publisher.Mono<Msg> seedSystemMsg(Object callScope)
      Returns the initial system message to seed into PreCallEvent before hooks run.

      The default implementation returns null. Subclasses (e.g. ReActAgent) override this to build a system message from their configured sysPrompt.

      参数:
      callScope - the per-call scope captured at call entry (may be null); lets subclasses resolve the call's RuntimeContext for system-prompt middlewares without reading a shared instance field
      返回:
      the seed system message; empty Mono if none

    • stateForCall

      protected AgentState stateForCall(Object callScope)
      Returns the AgentState whose conversation buffer should seed the pre-call memory snapshot for this invocation. The default reads the agent-level getAgentState(); subclasses with per-call scope (e.g. ReActAgent) return the scope's session state so the snapshot is concurrency-correct.
      参数:
      callScope - the per-call scope captured at call entry (may be null)
      返回:
      the state to snapshot, or null if none

    • consumeSystemMsgAfterPreCall

      protected void consumeSystemMsgAfterPreCall(Msg systemMsg, Object callScope)
      Called after PreCallEvent hooks have run, with the final system message value.

      The default implementation is a no-op. Subclasses (e.g. ReActAgent) override this to persist the system message into a per-call AtomicReference so it is available to subsequent events (PreReasoningEvent, PreSummaryEvent).

      参数:
      systemMsg - the system message produced by all PreCall hooks (may be null)
      callScope - the per-call scope captured at call entry (see beforeAgentExecution(List, RuntimeContext)); may be null

    • 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

      @Deprecated(since="2.0.0", forRemoval=true) public reactor.core.publisher.Flux<Event> stream(List<Msg> msgs, StreamOptions options)
      已过时, 待删除: 此 API 元素将从以后的版本中删除。
      since 2.0.0, for removal. Use ReActAgent#streamEvents(List) for the fine-grained AgentEvent stream.
      Stream with multiple input messages.
      指定者:
      stream 在接口中 StreamableAgent
      参数:
      msgs - Input messages
      options - Stream configuration options
      返回:
      Flux of events emitted during execution

    • stream

      @Deprecated(since="2.0.0", forRemoval=true) public reactor.core.publisher.Flux<Event> stream(List<Msg> msgs, StreamOptions options, Class<?> structuredModel)
      已过时, 待删除: 此 API 元素将从以后的版本中删除。
      since 2.0.0, for removal. Use ReActAgent#streamEvents(...) for the fine-grained AgentEvent stream.
      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

      @Deprecated(since="2.0.0", forRemoval=true) public reactor.core.publisher.Flux<Event> stream(List<Msg> msgs, StreamOptions options, com.fasterxml.jackson.databind.JsonNode schema)
      已过时, 待删除: 此 API 元素将从以后的版本中删除。
      since 2.0.0, for removal. Use ReActAgent#streamEvents(...) for the fine-grained AgentEvent stream.
      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