Realtime steering lets you interrupt an agent mid-reply. Call agent.interrupt() to cancel the current task. The agent then runs handle_interrupt() for post-processing.
import asynciofrom agentscope.agent import ReActAgentasync def main(): # ... agent setup ... # Start the agent reply as a task reply_task = asyncio.create_task(agent(msg)) # Interrupt after 1 second await asyncio.sleep(1) await agent.interrupt() result = await reply_task print(result.get_text_content())
You can override handle_interrupt in a subclass to customize the response when an interruption occurs — for example, calling the LLM to generate a context-aware acknowledgment.
As conversations grow, token counts can exceed model limits. Enable automatic compression by passing a CompressionConfig when creating the agent:
from agentscope.agent import ReActAgentfrom agentscope.token import CharTokenCounteragent = ReActAgent( name="Assistant", sys_prompt="You are a helpful assistant.", model=model, formatter=formatter, compression_config=ReActAgent.CompressionConfig( enable=True, agent_token_counter=CharTokenCounter(), trigger_threshold=10000, # compress when exceeding 10,000 tokens keep_recent=3, # keep the 3 most recent messages uncompressed ),)
When the token count exceeds trigger_threshold, the agent compresses older messages into a structured summary with these default fields:
Field
Description
task_overview
The user’s core request and success criteria
current_state
What has been completed, including files and outputs
important_discoveries
Constraints, decisions, errors, and failed approaches
next_steps
Specific actions needed to complete the task
context_to_preserve
User preferences, domain details, and promises made
Compression uses a marking mechanism — old messages are marked as compressed and excluded from future retrievals, while the summary is stored separately. Original messages are preserved.
Customizing compressionYou can control the compression behavior with summary_schema, summary_template, and compression_prompt:
from pydantic import BaseModel, Fieldclass CustomSummary(BaseModel): main_topic: str = Field(max_length=200, description="The main topic of the conversation") key_points: str = Field(max_length=400, description="Important points discussed") pending_tasks: str = Field(max_length=200, description="Tasks that remain to be done")agent = ReActAgent( name="Assistant", sys_prompt="You are a helpful assistant.", model=model, formatter=formatter, compression_config=ReActAgent.CompressionConfig( enable=True, agent_token_counter=CharTokenCounter(), trigger_threshold=10000, keep_recent=3, summary_schema=CustomSummary, compression_prompt=( "<system-hint>Summarize the conversation focusing on " "the main topic, key points, and pending tasks.</system-hint>" ), summary_template=( "<system-info>Summary:\n" "Main Topic: {main_topic}\n\n" "Key Points:\n{key_points}\n\n" "Pending Tasks:\n{pending_tasks}</system-info>" ), ),)
Use a smaller, faster model for compression by specifying compression_model and compression_formatter to reduce cost and latency.
The Plan Module enables ReActAgent to formally break down complex tasks into manageable sub-tasks and execute them systematically. Pass a PlanNotebook instance via the plan_notebook parameter to activate it. Once provided, the agent:
Is automatically equipped with plan management tool functions
Receives a hint message at the beginning of each reasoning step guiding it through the current plan
The current plan module requires subtasks to be executed sequentially. Parallel subtask execution is on the roadmap.
Key capabilities:
Creating, modifying, abandoning, and restoring plans
Switching between multiple plans
Gracefully handling interruptions by temporarily suspending the current plan
Real-time visualization and monitoring via plan change hooks
Create a plan upfront, then pass the PlanNotebook to ReActAgent:
import asyncioimport osfrom agentscope.agent import ReActAgentfrom agentscope.formatter import DashScopeChatFormatterfrom agentscope.model import DashScopeChatModelfrom agentscope.plan import PlanNotebook, SubTaskplan_notebook = PlanNotebook()async def setup_plan() -> None: await plan_notebook.create_plan( name="Research on Agent", description="Conduct a comprehensive research on the LLM-empowered agent.", expected_outcome=( "A Markdown report answering: 1. What's an agent? " "2. What's the current state of the art? " "3. What's the future trend?" ), subtasks=[ SubTask( name="Search agent-related survey papers", description=( "Search multiple sources including Google Scholar, arXiv, " "and Semantic Scholar. Must be published after 2021 with " "more than 50 citations." ), expected_outcome="A paper list in Markdown format", ), SubTask( name="Read and summarize the papers", description=( "Read the papers and summarize key points: definition, " "taxonomy, challenges, and key directions." ), expected_outcome="A summary of key points in Markdown format", ), SubTask( name="Research recent advances of large companies", description=( "Research recent advances from Google, Microsoft, OpenAI, " "Anthropic, Alibaba, and Meta via official blogs or news articles." ), expected_outcome="A recent advances summary in Markdown format", ), SubTask( name="Write a report", description="Write a report based on all previous steps.", expected_outcome=( "A Markdown report answering: 1. What's an agent? " "2. What's the current state of the art? " "3. What's the future trend?" ), ), ], ) msg = await plan_notebook.get_current_hint() print(f"{msg.name}: {msg.content}")asyncio.run(setup_plan())agent = ReActAgent( name="Friday", sys_prompt="You are a helpful assistant.", model=DashScopeChatModel( model_name="qwen-max", api_key=os.environ["DASHSCOPE_API_KEY"], ), formatter=DashScopeChatFormatter(), plan_notebook=plan_notebook,)
Pass a fresh PlanNotebook and let the agent decide when and how to plan. For complex tasks, the agent will create a plan autonomously and execute it step by step:
from agentscope.agent import UserAgentagent = ReActAgent( name="Friday", sys_prompt="You are a helpful assistant.", model=DashScopeChatModel( model_name="qwen-max", api_key=os.environ["DASHSCOPE_API_KEY"], ), formatter=DashScopeChatFormatter(), plan_notebook=PlanNotebook(),)async def interact_with_agent() -> None: user = UserAgent(name="user") msg = None while True: msg = await user(msg) if msg.get_text_content() == "exit": break msg = await agent(msg)asyncio.run(interact_with_agent())
Register a hook to react whenever the plan changes — useful for forwarding plan state to a frontend or logging system:
from agentscope.plan import PlanNotebook, Plandef on_plan_changed(self: PlanNotebook, plan: Plan) -> None: """Called whenever the plan is updated. Args: self: The PlanNotebook instance. plan: The updated plan instance. """ # Forward the plan to a frontend or logging system ...plan_notebook.register_plan_change_hook(on_plan_changed)
from typing import Anyfrom agentscope.agent import AgentBasedef my_pre_hook( self: AgentBase, kwargs: dict[str, Any],) -> dict[str, Any] | None: # modify kwargs and return, or return None to leave them unchanged return kwargs
Post-hooks receive an additional output argument:
def my_post_hook( self: AgentBase, kwargs: dict[str, Any], output: Any,) -> Any | None: # modify output and return, or return None to leave it unchanged return output
All positional and keyword arguments of the core function are passed as a single kwargs dict. When a hook returns None, the next hook receives the most recent non-None return value (or the original arguments if all previous hooks returned None).
StateModule is the foundation for state management. Any class that inherits from it can register attributes as part of its state, enabling serialization and restoration.AgentBase, MemoryBase, LongTermMemoryBase, and Toolkit all inherit from StateModule.
A session is a collection of StateModule objects (e.g., multiple agents) whose state you want to persist together.AgentScope provides JSONSession, which saves and loads session state as a JSON file named by session ID:
from agentscope.session import JSONSessionsession = JSONSession(save_dir="./sessions")
Saving a session:
await session.save_session_state( session_id="user_1", agent=agent, # keyword argument name must match when loading)
Loading a session:
await session.load_session_state( session_id="user_1", agent=agent, # same keyword argument name as used when saving)
You can pass multiple agents to save_session_state and load_session_state as keyword arguments. The keyword names must be consistent between save and load calls.
JSONSession is a concrete implementation of SessionBase. You can implement your own session class with a custom storage backend (e.g., Redis, a database) by subclassing SessionBase and implementing save_session_state and load_session_state.
The realtime agent is currently under active development. Contributions, discussions, and feedback are welcome.
RealtimeAgent is designed for real-time interactions such as voice conversations. It bridges realtime model APIs with your application via a unified event interface.
ChatRoom manages multiple RealtimeAgent instances in a shared conversation space, with automatic message broadcasting and unified lifecycle management.
