How to Implement Human-in-the-Loop Interruption and Resumption in AgentScope

AgentScope enables human-in-the-loop interruption by cancelling the active reply coroutine via asyncio.CancelledError, triggering the handle_interrupt() method to generate a tagged response, and allowing resumption by sending a new message to the same agent instance.

AgentScope is a multi-agent platform that treats agent execution as asynchronous tasks, making it straightforward to pause and resume workflows. Implementing human-in-the-loop interruption and resumption in AgentScope requires understanding how the framework handles task cancellation, interruption metadata, and context preservation. This guide walks through the core mechanisms, source code locations, and practical implementation patterns.

Understanding the Interruption Architecture

AgentScope's interruption system is built on Python's asyncio primitives. When an agent is running, its reply method executes inside a coroutine wrapped by AgentBase.__call__.

The Async Task Cancellation Mechanism

The interruption begins when you call await agent.interrupt(). In src/agentscope/agent/_agent_base.py (lines 48-60), this method cancels the active _reply_task:

# Conceptual flow from _agent_base.py

async def interrupt(self):
    if self._reply_task and not self._reply_task.done():
        self._reply_task.cancel()

When the task is cancelled, asyncio.CancelledError propagates to the AgentBase.__call__ wrapper. The base class catches this exception and immediately forwards execution to handle_interrupt().

The handle_interrupt Method

Each concrete agent implements handle_interrupt() to define its interruption behavior. In src/agentscope/agent/_react_agent.py (lines 94-104), the ReActAgent overrides this method to return a structured message:

def handle_interrupt(self, msg: Msg) -> Msg:
    return Msg(
        name=self.name,
        content="[Interrupted] The agent was interrupted by the user.",
        role="assistant",
        metadata={"_is_interrupted": True},
    )

The metadata={"_is_interrupted": True} flag is critical. Downstream components in src/agentscope/message/_message_base.py can inspect this metadata to trigger UI updates or workflow pauses.

Detecting Interruptions in Your Application

Your application layer must check for the interruption flag to coordinate the human-in-the-loop pause. After awaiting an agent call, inspect the returned Msg object:

response = await agent()
if response.metadata.get("_is_interrupted"):
    print("Agent interrupted. Waiting for user input...")
    # Pause workflow, update UI, or trigger notification

This pattern appears in src/agentscope/agent/_user_agent.py, which implements a human proxy agent that naturally handles these pauses by waiting for actual user input.

Resuming Agent Execution

Resumption in AgentScope is stateless by default—the agent simply processes a new message. However, for long-running tasks, you must preserve context.

Basic Resumption

To resume after an interruption, send a new message to the same agent instance:

# After detecting interruption

new_instruction = Msg("User", "Please continue where you left off.", "user")
resumed_response = await agent(new_instruction)

The agent starts a fresh reply cycle, treating the new input as the next turn in the conversation.

Resuming Long-Running Tasks with Compression

For agents with extensive memory, use CompressionConfig to generate continuation summaries. In src/agentscope/agent/_react_agent.py (lines 29-38), the ReActAgent supports automatic memory compression:

from agentscope.agent import ReActAgent, CompressionConfig
from agentscope.memory import InMemoryMemory
from agentscope.token import TokenCounter

compress_cfg = CompressionConfig(
    enable=True,
    agent_token_counter=TokenCounter(),
    trigger_threshold=3000,  # Compress when exceeding 3000 tokens

    keep_recent=3,         # Keep last 3 messages uncompressed

)

agent = ReActAgent(
    name="LongTaskAgent",
    sys_prompt="You are processing a lengthy document.",
    model=model,
    memory=InMemoryMemory(),
    compression_config=compress_cfg,
)

When the token count exceeds trigger_threshold, the agent creates a summary that remains in memory. After an interruption, this summary ensures the resumed context contains the essential information without exceeding token limits.

Complete Implementation Example

The following runnable example demonstrates the full interruption and resumption cycle using ReActAgent:

import asyncio
import os
from agentscope.agent import ReActAgent
from agentscope.model import DashScopeChatModel
from agentscope.formatter import DashScopeChatFormatter
from agentscope.message import Msg
from agentscope.tool import Toolkit

async def main():
    # Initialize agent

    agent = ReActAgent(
        name="Friday",
        sys_prompt="You are a helpful assistant.",
        model=DashScopeChatModel(
            model_name="qwen-max",
            api_key=os.getenv("DASHSCOPE_API_KEY"),
        ),
        formatter=DashScopeChatFormatter(),
        toolkit=Toolkit(),
    )
    
    # Start agent task

    reply_task = asyncio.create_task(agent())
    
    # Simulate user interrupt after 1 second

    await asyncio.sleep(1)
    await agent.interrupt()
    
    # Get interruption message

    interrupt_msg = await reply_task
    print(f"Status: {interrupt_msg.get_text_content()}")
    print(f"Interrupted flag: {interrupt_msg.metadata.get('_is_interrupted')}")
    
    # Resume with new instruction

    new_msg = Msg("User", "Please continue where you left off.", "user")
    resumed_reply = await agent(new_msg)
    print(f"Resumed: {resumed_reply.get_text_content()}")

if __name__ == "__main__":
    asyncio.run(main())

This script demonstrates:

• Creating an async task for the agent reply
• Calling interrupt() to trigger cancellation
• Detecting the _is_interrupted metadata flag
• Resuming execution with a new message

Summary

• Interruption mechanism: AgentScope uses asyncio.CancelledError triggered by agent.interrupt() to halt the active reply coroutine, as implemented in src/agentscope/agent/_agent_base.py.
• Handle interruption: Concrete agents override handle_interrupt() to return a message with metadata={"_is_interrupted": True}, signaling downstream components to pause.
• Detection: Check msg.metadata.get("_is_interrupted") in your application layer to coordinate UI updates or workflow pauses.
• Resumption: Send a new Msg to the same agent instance to start a fresh reply cycle; for long-running tasks, enable CompressionConfig in ReActAgent to preserve context via automatic summarization.
• Key files: src/agentscope/agent/_agent_base.py (cancellation), src/agentscope/agent/_react_agent.py (interruption handling and compression), src/agentscope/agent/_user_agent.py (HITL reference), src/agentscope/message/_message_base.py (metadata).

Frequently Asked Questions

How does AgentScope technically cancel an agent's reply when interrupted?

AgentScope treats each agent reply as an asynchronous task stored in self._reply_task. When you call await agent.interrupt(), the method invokes self._reply_task.cancel(), which raises asyncio.CancelledError inside the running coroutine. The AgentBase.__call__ wrapper in src/agentscope/agent/_agent_base.py catches this exception and routes execution to handle_interrupt(), allowing the agent to return a structured response rather than terminating abruptly.

What is the purpose of the _is_interrupted metadata flag?

The _is_interrupted metadata flag is a boolean value set in the metadata dictionary of the Msg object returned when an agent handles an interruption. Defined in concrete implementations like src/agentscope/agent/_react_agent.py, this flag signals to downstream components—such as UI layers, message hubs, or workflow orchestrators—that the agent's output represents an interruption event rather than a completed reply. Your application should check msg.metadata.get("_is_interrupted") to trigger appropriate pause or notification logic.

Can I resume an agent from exactly where it left off after interruption?

AgentScope's resumption model is stateless by default: when you send a new message after an interruption, the agent starts a fresh reply cycle based on its current memory state. To achieve seamless resumption after a long-running task, enable the CompressionConfig feature available in ReActAgent (located in src/agentscope/agent/_react_agent.py). When the token count exceeds your configured trigger_threshold, the agent automatically generates a continuation summary that remains in memory. After an interruption, this summary ensures the agent retains the essential context needed to continue logically from where the task was paused.

Which agent classes support the interruption mechanism?

The interruption mechanism is implemented in the abstract AgentBase class (src/agentscope/agent/_agent_base.py), making it available to all concrete agent implementations. The ReActAgent (src/agentscope/agent/_react_agent.py) provides a full implementation of handle_interrupt() with optional compression support. The UserAgent (src/agentscope/agent/_user_agent.py) also implements interruption handling, serving as a reference for building human-in-the-loop interfaces. Any custom agent inheriting from AgentBase can override handle_interrupt() to define specialized interruption behavior.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}