程序包 io.agentscope.core

类 ReActAgent

java.lang.Object
io.agentscope.core.agent.AgentBase
io.agentscope.core.ReActAgent
所有已实现的接口:
Agent, CallableAgent, ObservableAgent, StreamableAgent, AutoCloseable
public class ReActAgent extends AgentBase implements AutoCloseable
ReAct (Reasoning and Acting) Agent implementation.

ReAct is an agent design pattern that combines reasoning (thinking and planning) with acting (tool execution) in an iterative loop. The agent alternates between these two phases until it either completes the task or reaches the maximum iteration limit.

Key Features:

  • Reactive Streaming: Uses Project Reactor for non-blocking execution
  • Hook System: Extensible hooks for monitoring and intercepting agent execution
  • HITL Support: Human-in-the-loop via stopAgent() in PostReasoningEvent/PostActingEvent
  • Structured Output: per-call generate_response tool provides type-safe output

Usage Example: 


 import io.agentscope.extensions.model.dashscope.DashScopeChatModel;

 // Create a model (requires dependency: agentscope-extensions-model-dashscope)
 DashScopeChatModel model = DashScopeChatModel.builder()
     .apiKey(System.getenv("DASHSCOPE_API_KEY"))
     .modelName("qwen-plus")
     .build();

 // Create a toolkit with tools
 Toolkit toolkit = new Toolkit();
 toolkit.registerObject(new MyToolClass());

 // Build the agent
 ReActAgent agent = ReActAgent.builder()
     .name("Assistant")
     .sysPrompt("You are a helpful assistant.")
     .model(model)
     .toolkit(toolkit)
     .maxIters(10)
     .build();

 // Use the agent
 Msg response = agent.call(Msg.builder()
     .name("user")
     .role(MsgRole.USER)
     .content(TextBlock.builder().text("What's the weather?").build())
     .build()).block();

Thread Safety: ReActAgent is not thread-safe. A single instance processes exactly one call() at a time; a concurrent invocation on the same instance throws IllegalStateException. For web services or other concurrent scenarios, create one agent instance per request via a factory method. Model, Toolkit (as a template — build() deep-copies it), and AgentStateStore are all safe to share across instances.

  • 字段详细资料

    • STRUCTURED_OUTPUT_TOOL_NAME

      public static final String STRUCTURED_OUTPUT_TOOL_NAME
      Tool name used for the per-call structured-output generate_response tool.
      另请参阅:

    • CONFIRM_SINK_KEY

      @Deprecated public static final String CONFIRM_SINK_KEY
      已过时。
      Permission HITL no longer uses a Reactor Sink. Confirm results are now delivered via a second agent.call(msgs) carrying a ConfirmResult payload — see applyConfirmResults. This constant is retained as a compile-time marker and will be removed in a future release.
      另请参阅:

  • 方法详细资料

    • callSerializationKey

      protected Object callSerializationKey(RuntimeContext rc)
      从类复制的说明: AgentBase
      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.
      覆盖:
      callSerializationKey 在类中 AgentBase
      参数:
      rc - the caller-supplied per-call RuntimeContext (may be null)
      返回:
      the serialization key, or null to run without serialization

    • beforeAgentExecution

      protected Object beforeAgentExecution(List<Msg> msgs, RuntimeContext rc)
      从类复制的说明: AgentBase
      Invoked at the start of a call / stream-backed call, after AgentBase.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 AgentBase.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.
      覆盖:
      beforeAgentExecution 在类中 AgentBase
      参数:
      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

    • seedSystemMsg

      protected reactor.core.publisher.Mono<Msg> seedSystemMsg(Object callExectution)
      从类复制的说明: AgentBase
      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.

      覆盖:
      seedSystemMsg 在类中 AgentBase
      参数:
      callExectution - 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)
      从类复制的说明: AgentBase
      Returns the AgentState whose conversation buffer should seed the pre-call memory snapshot for this invocation. The default reads the agent-level AgentBase.getAgentState(); subclasses with per-call scope (e.g. ReActAgent) return the scope's session state so the snapshot is concurrency-correct.
      覆盖:
      stateForCall 在类中 AgentBase
      参数:
      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)
      从类复制的说明: AgentBase
      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).

      覆盖:
      consumeSystemMsgAfterPreCall 在类中 AgentBase
      参数:
      systemMsg - the system message produced by all PreCall hooks (may be null)
      callScope - the per-call scope captured at call entry (see AgentBase.beforeAgentExecution(List, RuntimeContext)); may be null

    • afterAgentExecution

      protected void afterAgentExecution()
      从类复制的说明: AgentBase
      Invoked in Mono.using cleanup, before clearing the running state. Pairs with AgentBase.beforeAgentExecution(List, RuntimeContext). The default is a no-op.
      覆盖:
      afterAgentExecution 在类中 AgentBase

    • call

      public reactor.core.publisher.Mono<Msg> call(List<Msg> msgs, RuntimeContext context)
      Calls the agent with a per-call RuntimeContext (metadata for hooks and tools, not persisted).

    • call

      public reactor.core.publisher.Mono<Msg> call(List<Msg> msgs, Class<?> structuredOutputClass, RuntimeContext context)

    • call

      public reactor.core.publisher.Mono<Msg> call(List<Msg> msgs, com.fasterxml.jackson.databind.JsonNode outputSchema, RuntimeContext context)

    • call

      public reactor.core.publisher.Mono<Msg> call(String text, RuntimeContext context)
      Calls the agent with a plain text input and per-call RuntimeContext.
      参数:
      text - input text (wrapped into a UserMessage)
      context - per-call runtime context
      返回:
      response message

    • call

      public reactor.core.publisher.Mono<Msg> call(String text, Class<?> structuredOutputClass, RuntimeContext context)
      Calls the agent with a plain text input, structured output class, and per-call RuntimeContext.
      参数:
      text - input text (wrapped into a UserMessage)
      structuredOutputClass - class defining the structure
      context - per-call runtime context
      返回:
      response message with structured data in metadata

    • interrupt

      @Deprecated public void interrupt()
      已过时。
      Use interrupt(String, String) with explicit userId/sessionId.
      从接口复制的说明: 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 在类中 AgentBase

    • interrupt

      @Deprecated public void interrupt(Msg msg)
      已过时。
      Use interrupt(String, String, Msg) with explicit userId/sessionId.
      从接口复制的说明: 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
      覆盖:
      interrupt 在类中 AgentBase
      参数:
      msg - User message associated with the interruption

    • interrupt

      @Deprecated public void interrupt(InterruptSource source)
      已过时。
      Use interrupt(String, String) with explicit userId/sessionId.
      覆盖:
      interrupt 在类中 AgentBase

    • interrupt

      public void interrupt(RuntimeContext ctx)
      Interrupts the in-flight call identified by the given RuntimeContext. Uses ctx.getUserId() and ctx.getSessionId() to locate the session.
      参数:
      ctx - the runtime context identifying the session to interrupt

    • interrupt

      public void interrupt(RuntimeContext ctx, Msg msg)
      Interrupts the in-flight call identified by the given RuntimeContext with an associated user message.
      参数:
      ctx - the runtime context identifying the session to interrupt
      msg - optional user message to attach to the interrupt signal

    • interrupt

      public void interrupt(String userId, String sessionId)
      Interrupts the in-flight call for a specific (userId, sessionId) session.
      参数:
      userId - the user id (null = anonymous / single-tenant)
      sessionId - the session id

    • interrupt

      public void interrupt(String userId, String sessionId, Msg msg)
      Interrupts the in-flight call for a specific (userId, sessionId) session with an associated user message.

    • stream

      @Deprecated(since="2.0.0", forRemoval=true) public reactor.core.publisher.Flux<Event> stream(List<Msg> msgs, StreamOptions options, RuntimeContext context)
      已过时, 待删除: 此 API 元素将从以后的版本中删除。
      since 2.0.0, for removal. Use streamEvents(List) (and overloads that accept RuntimeContext via call()-driven lifecycle) for the fine-grained AgentEvent stream.
      从接口复制的说明: StreamableAgent
      Stream execution events with a caller-supplied per-call runtime context.

      Implementations that do not consume RuntimeContext can rely on this default bridge and continue using the original two-argument stream method.

      指定者:
      stream 在接口中 StreamableAgent
      参数:
      msgs - Input messages
      options - Stream configuration options
      context - Runtime metadata for this call
      返回:
      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, RuntimeContext context)
      已过时, 待删除: 此 API 元素将从以后的版本中删除。
      since 2.0.0, for removal. Use streamEvents(List) for the fine-grained AgentEvent stream.

    • 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, RuntimeContext context)
      已过时, 待删除: 此 API 元素将从以后的版本中删除。
      since 2.0.0, for removal. Use streamEvents(List) for the fine-grained AgentEvent stream.

    • callInternal

      protected reactor.core.publisher.Mono<Msg> callInternal(List<Msg> msgs, RuntimeContext context, Function<List<Msg>,reactor.core.publisher.Mono<Msg>> doCallFn)
      Overrides the base-class hook so that every call() variant — including structured output and context-bearing overloads — runs through the same buildAgentStream(java.util.List<io.agentscope.core.message.Msg>, io.agentscope.core.agent.RuntimeContext, java.util.function.Function<java.util.List<io.agentscope.core.message.Msg>, reactor.core.publisher.Mono<io.agentscope.core.message.Msg>>) core as streamEvents(). This guarantees that the onAgent middleware chain fires on all invocation paths, not only on the streaming path.

      The result is extracted from the AgentResultEvent emitted by buildAgentStream(java.util.List<io.agentscope.core.message.Msg>, io.agentscope.core.agent.RuntimeContext, java.util.function.Function<java.util.List<io.agentscope.core.message.Msg>, reactor.core.publisher.Mono<io.agentscope.core.message.Msg>>) before AgentEndEvent.

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

    • streamEvents

      public reactor.core.publisher.Flux<AgentEvent> streamEvents(List<Msg> msgs)
      Stream fine-grained AgentEvents from the full agent lifecycle.

      Both call() and streamEvents() share the same internal buildAgentStream(java.util.List<io.agentscope.core.message.Msg>, io.agentscope.core.agent.RuntimeContext, java.util.function.Function<java.util.List<io.agentscope.core.message.Msg>, reactor.core.publisher.Mono<io.agentscope.core.message.Msg>>) core, so the onAgent middleware chain fires on all paths. The stream includes AgentResultEvent (carrying the final Msg) immediately before AgentEndEvent.

      参数:
      msgs - input messages
      返回:
      event stream covering the full agent invocation lifecycle

    • streamEvents

      public reactor.core.publisher.Flux<AgentEvent> streamEvents(Msg msg)
      Stream fine-grained AgentEvents for a single input message.
      参数:
      msg - input message
      返回:
      event stream covering the full agent invocation lifecycle

    • streamEvents

      public reactor.core.publisher.Flux<AgentEvent> streamEvents(List<Msg> msgs, RuntimeContext context)
      Stream fine-grained AgentEvents with a caller-supplied RuntimeContext.

      Delegates directly to buildAgentStream(java.util.List<io.agentscope.core.message.Msg>, io.agentscope.core.agent.RuntimeContext, java.util.function.Function<java.util.List<io.agentscope.core.message.Msg>, reactor.core.publisher.Mono<io.agentscope.core.message.Msg>>) — the same core used by call(). Concurrent invocations do not share any state; each subscription gets its own event sink and lifecycle execution.

      参数:
      msgs - input messages
      context - runtime context to propagate into the call
      返回:
      event stream covering the full agent invocation lifecycle

    • streamEvents

      public reactor.core.publisher.Flux<AgentEvent> streamEvents(Msg msg, RuntimeContext context)
      Stream fine-grained AgentEvents for a single input message with a caller-supplied RuntimeContext.
      参数:
      msg - input message
      context - runtime context to propagate into the call
      返回:
      event stream covering the full agent invocation lifecycle

    • streamEvents

      public reactor.core.publisher.Flux<AgentEvent> streamEvents(String text)
      Stream fine-grained AgentEvents for a plain text input.
      参数:
      text - input text (wrapped into a UserMessage)
      返回:
      event stream covering the full agent invocation lifecycle

    • streamEvents

      public reactor.core.publisher.Flux<AgentEvent> streamEvents(String text, RuntimeContext context)
      Stream fine-grained AgentEvents for a plain text input with a caller-supplied RuntimeContext.
      参数:
      text - input text (wrapped into a UserMessage)
      context - runtime context to propagate into the call
      返回:
      event stream covering the full agent invocation lifecycle

    • doCall

      protected reactor.core.publisher.Mono<Msg> doCall(List<Msg> msgs)
      从类复制的说明: AgentBase
      Internal implementation for processing multiple input messages. Subclasses must implement their specific logic here.
      指定者:
      doCall 在类中 AgentBase
      参数:
      msgs - Input messages
      返回:
      Response message

    • doCall

      protected reactor.core.publisher.Mono<Msg> doCall(List<Msg> msgs, Class<?> structuredOutputClass)
      从类复制的说明: AgentBase
      Internal implementation for processing multiple messages with structured output. Subclasses that support structured output must override this method. Default implementation throws UnsupportedOperationException.
      覆盖:
      doCall 在类中 AgentBase
      参数:
      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)
      从类复制的说明: AgentBase
      Internal implementation for processing multiple messages with structured output. Subclasses that support structured output must override this method. Default implementation throws UnsupportedOperationException.
      覆盖:
      doCall 在类中 AgentBase
      参数:
      msgs - Input messages
      outputSchema - com.fasterxml.jackson.databind.JsonNode instance defining the structure
      返回:
      Response message with structured data in metadata

    • buildGenerateOptions

      protected GenerateOptions buildGenerateOptions()

    • handleInterrupt

      protected reactor.core.publisher.Mono<Msg> handleInterrupt(InterruptContext context, Msg... originalArgs)
      从类复制的说明: AgentBase
      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
      指定者:
      handleInterrupt 在类中 AgentBase
      参数:
      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

    • doObserve

      protected reactor.core.publisher.Mono<Void> doObserve(Msg msg)
      从类复制的说明: AgentBase
      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
      覆盖:
      doObserve 在类中 AgentBase
      参数:
      msg - The message to observe
      返回:
      Mono that completes when observation is done

    • getToolkit

      public Toolkit getToolkit()
      Returns this agent's toolkit (a per-instance deep copy made at build time).
      指定者:
      getToolkit 在接口中 Agent

    • getSysPrompt

      public String getSysPrompt()

    • getModel

      public Model getModel()

    • getMaxIters

      public int getMaxIters()

    • getGenerateOptions

      public GenerateOptions getGenerateOptions()
      Gets the configured generation options for this agent.
      返回:
      The generation options, or null if not configured

    • getAgentState

      @Deprecated public AgentState getAgentState()
      已过时。
      Use getAgentState(RuntimeContext) or getAgentState(String, String) with explicit session identity. This method delegates to the default session slot.
      从类复制的说明: AgentBase
      Returns the agent's mutable runtime state, or null if this agent type does not maintain an AgentState.
      指定者:
      getAgentState 在接口中 Agent
      覆盖:
      getAgentState 在类中 AgentBase

    • getRuntimeContext

      public RuntimeContext getRuntimeContext()
      从类复制的说明: AgentBase
      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.
      覆盖:
      getRuntimeContext 在类中 AgentBase

    • getAgentState

      public AgentState getAgentState(RuntimeContext ctx)
      Returns the AgentState for the session identified by the given RuntimeContext.
      参数:
      ctx - the runtime context (uses getUserId() and getSessionId())
      返回:
      the agent state for the identified session

    • getAgentState

      public AgentState getAgentState(String userId, String sessionId)
      Returns the AgentState for the given (userId, sessionId) slot, loading it from the configured AgentStateStore on first access and caching it for subsequent calls within this JVM.

      Note: in distributed deployments the authoritative reload happens at call start inside activateSlotForContext. This method returns the locally cached instance (suitable for the "get → mutate → save" pattern used by admin APIs and tests).

    • setPermissionMode

      public void setPermissionMode(String userId, String sessionId, PermissionMode mode)
      Switches the PermissionMode for the given (userId, sessionId) session at runtime and rebuilds that session's cached PermissionEngine so the change takes effect on the next tool evaluation. The configured rules and working directories are preserved; only the mode changes. The change is persisted so the next call on that session sees it.

      Use this to implement a deliberate, user-initiated "bypass permissions" toggle (pass PermissionMode.BYPASS) or to restore stricter enforcement afterwards (pass PermissionMode.DEFAULT). BYPASS disables all rule evaluation, so it should be an explicit, per-session action and is best paired with a sandboxed environment.

      An in-flight call keeps the engine it started with; the new mode applies to subsequent calls on the slot.

      参数:
      userId - user identity for the slot (may be null)
      sessionId - session identity (falls back to the default session id when null)
      mode - the permission mode to switch to

    • setPermissionMode

      public void setPermissionMode(RuntimeContext ctx, PermissionMode mode)
      Switches the PermissionMode for the session identified by the given RuntimeContext. See setPermissionMode(String, String, PermissionMode).
      参数:
      ctx - the runtime context identifying the session
      mode - the permission mode to switch to

    • getPermissionMode

      public PermissionMode getPermissionMode(String userId, String sessionId)
      Returns the current PermissionMode for the given (userId, sessionId) session.
      参数:
      userId - user identity for the slot (may be null)
      sessionId - session identity (falls back to the default session id when null)
      返回:
      the session's current permission mode

    • saveAgentState

      public void saveAgentState(RuntimeContext ctx)
      Persists the cached AgentState for the session identified by the given RuntimeContext. No-op when no store is configured or the session has never been loaded.
      参数:
      ctx - the runtime context identifying the session to save

    • saveAgentState

      public void saveAgentState(String userId, String sessionId)
      Persists the cached AgentState for the given (userId, sessionId) slot via the configured AgentStateStore. No-op when no store is configured or the slot has never been loaded into the cache.

    • getStateStore

      public AgentStateStore getStateStore()
      Returns the AgentStateStore configured for state persistence, or null.

    • getDefaultSessionId

      public String getDefaultSessionId()
      Returns the builder-time fallback sessionId, used when a call's RuntimeContext carries no sessionId.

    • getModelConfig

      public ModelConfig getModelConfig()
      Returns the model-call configuration (retries, timeouts).

    • getReactConfig

      public ReactConfig getReactConfig()
      Returns the reasoning-loop configuration (maxIters, stopOnReject).

    • getPermissionEngine

      @Deprecated public PermissionEngine getPermissionEngine()
      已过时。
      Use getAgentState(userId, sessionId).getPermissionContext() instead.

    • getPermissionContext

      @Deprecated public PermissionContextState getPermissionContext()
      已过时。
      Use getAgentState(userId, sessionId).getPermissionContext() instead.

    • getMiddlewares

      public List<MiddlewareBase> getMiddlewares()
      Returns the immutable list of registered middlewares.

    • getModelExecutionConfig

      public ExecutionConfig getModelExecutionConfig()
      Returns the per-model-call ExecutionConfig, or null if none was set.

    • getToolExecutionConfig

      public ExecutionConfig getToolExecutionConfig()
      Returns the per-tool-call ExecutionConfig, or null if none was set.

    • getToolExecutionContext

      public ToolExecutionContext getToolExecutionContext()
      Returns the ToolExecutionContext bound at build time, or null if none.

    • isPendingToolRecoveryEnabled

      public boolean isPendingToolRecoveryEnabled()
      Returns whether pending-tool-call recovery is enabled (HITL: resume tool calls left in a pending state across call() invocations).

    • getSystemPrompt

      public String getSystemPrompt()
      Returns the system prompt (alias for getSysPrompt()).

    • builder

      public static ReActAgent.Builder builder()

    • close

      public void close()
      指定者:
      close 在接口中 AutoCloseable