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

> Implement human-in-the-loop interruption and resumption in AgentScope. Learn how to cancel replies for manual input and seamlessly resume agent conversations with a simple code example.

- Repository: [AgentScope-AI/agentscope](https://github.com/agentscope-ai/agentscope)
- Tags: how-to-guide
- Published: 2026-03-09

---

**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`](https://github.com/agentscope-ai/agentscope/blob/main/src/agentscope/agent/_agent_base.py) (lines 48-60), this method cancels the active `_reply_task`:

```python

# 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`](https://github.com/agentscope-ai/agentscope/blob/main/src/agentscope/agent/_react_agent.py) (lines 94-104), the `ReActAgent` overrides this method to return a structured message:

```python
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`](https://github.com/agentscope-ai/agentscope/blob/main/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:

```python
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`](https://github.com/agentscope-ai/agentscope/blob/main/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:

```python

# 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`](https://github.com/agentscope-ai/agentscope/blob/main/src/agentscope/agent/_react_agent.py) (lines 29-38), the `ReActAgent` supports automatic memory compression:

```python
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`:

```python
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`](https://github.com/agentscope-ai/agentscope/blob/main/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`](https://github.com/agentscope-ai/agentscope/blob/main/src/agentscope/agent/_agent_base.py) (cancellation), [`src/agentscope/agent/_react_agent.py`](https://github.com/agentscope-ai/agentscope/blob/main/src/agentscope/agent/_react_agent.py) (interruption handling and compression), [`src/agentscope/agent/_user_agent.py`](https://github.com/agentscope-ai/agentscope/blob/main/src/agentscope/agent/_user_agent.py) (HITL reference), [`src/agentscope/message/_message_base.py`](https://github.com/agentscope-ai/agentscope/blob/main/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`](https://github.com/agentscope-ai/agentscope/blob/main/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`](https://github.com/agentscope-ai/agentscope/blob/main/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`](https://github.com/agentscope-ai/agentscope/blob/main/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`](https://github.com/agentscope-ai/agentscope/blob/main/src/agentscope/agent/_agent_base.py)), making it available to all concrete agent implementations. The `ReActAgent` ([`src/agentscope/agent/_react_agent.py`](https://github.com/agentscope-ai/agentscope/blob/main/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`](https://github.com/agentscope-ai/agentscope/blob/main/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.