agentscope.agent

The agent base class.

class AgentBase

基类：StateModule

Base class for asynchronous agents.

supported_hook_types: list[str] = ['pre_reply', 'post_reply', 'pre_print', 'post_print', 'pre_observe', 'post_observe']

Supported hook types for the agent base class.

__init__()

Initialize the agent.

返回类型:

None

id: str

The agent's unique identifier, generated using shortuuid.

async observe(msg)

Receive the given message(s) without generating a reply.

参数:

msg (Msg | list[Msg] | None) -- The message(s) to be observed.

返回类型:

None

async reply(*args, **kwargs)

The main logic of the agent, which generates a reply based on the current state and input arguments.

参数:

• args (Any)

• kwargs (Any)

返回类型:

Msg

async print(msg, last=True, speech=None)

The function to display the message.

参数:

• msg (Msg) -- The message object to be printed.

• last (bool, defaults to True) -- Whether this is the last one in streaming messages. For non-streaming message, this should always be True.

• speech (AudioBlock | list[AudioBlock] | None, optional) -- The audio content block(s) to be played along with the message.

返回类型:

None

async __call__(*args, **kwargs)

Call the reply function with the given arguments.

参数:

• args (Any)

• kwargs (Any)

返回类型:

Msg

async handle_interrupt(*args, **kwargs)

The post-processing logic when the reply is interrupted by the user or something else.

参数:

• args (Any)

• kwargs (Any)

返回类型:

Msg

async interrupt(msg=None)

Interrupt the current reply process.

参数:

msg (Msg | list[Msg] | None)

返回类型:

None

register_instance_hook(hook_type, hook_name, hook)

Register a hook to the agent instance, which only takes effect for the current instance.

参数:

• hook_type (str) -- The type of the hook, indicating where the hook is to be triggered.

• hook_name (str) -- The name of the hook. If the name is already registered, the hook will be overwritten.

• hook (Callable) -- The hook function.

返回类型:

None

remove_instance_hook(hook_type, hook_name)

Remove an instance-level hook from the agent instance.

参数:

• hook_type (AgentHookTypes) -- The type of the hook, indicating where the hook is to be triggered.

• hook_name (str) -- The name of the hook to remove.

返回类型:

None

classmethod register_class_hook(hook_type, hook_name, hook)

The universal function to register a hook to the agent class, which will take effect for all instances of the class.

参数:

• hook_type (AgentHookTypes) -- The type of the hook, indicating where the hook is to be triggered.

• hook_name (str) -- The name of the hook. If the name is already registered, the hook will be overwritten.

• hook (Callable) -- The hook function.

返回类型:

None

classmethod remove_class_hook(hook_type, hook_name)

Remove a class-level hook from the agent class.

参数:

• hook_type (AgentHookTypes) -- The type of the hook, indicating where the hook is to be triggered.

• hook_name (str) -- The name of the hook to remove.

返回类型:

None

classmethod clear_class_hooks(hook_type=None)

Clear all class-level hooks.

参数:

hook_type (AgentHookTypes, optional) -- The type of the hook to clear. If not specified, all class-level hooks will be cleared.

返回类型:

None

clear_instance_hooks(hook_type=None)

If hook_type is not specified, clear all instance-level hooks. Otherwise, clear the specified type of instance-level hooks.

参数:

hook_type (str | Literal['pre_reply', 'post_reply', 'pre_print', 'post_print', 'pre_observe', 'post_observe'] | None)

返回类型:

None

reset_subscribers(msghub_name, subscribers)

Reset the subscribers of the agent.

参数:

• msghub_name (str) -- The name of the MsgHub that manages the subscribers.

• subscribers (list[AgentBase]) -- A list of agents that will receive the reply message from this agent via their observe method.

返回类型:

None

remove_subscribers(msghub_name)

Remove the msghub subscribers by the given msg hub name.

参数:

msghub_name (str) -- The name of the MsgHub that manages the subscribers.

返回类型:

None

disable_console_output()

This function will disable the console output of the agent, e.g. in a production environment to avoid messy logs.

返回类型:

None

set_console_output_enabled(enabled)

Enable or disable the console output of the agent. E.g. in a production environment, you may want to disable the console output to avoid messy logs.

参数:

enabled (bool) -- If True, enable the console output. If False, disable the console output.

返回类型:

None

set_msg_queue_enabled(enabled, queue=None)

Enable or disable the message queue for streaming outputs.

参数:

• enabled (bool) -- If True, enable the message queue to allow streaming outputs. If False, disable the message queue.

• queue (Queue | None, optional) -- The queue instance that will be used to initialize the message queue when enable is True.

返回类型:

None

class ReActAgentBase

基类：AgentBase

The ReAct agent base class.

To support ReAct algorithm, this class extends the AgentBase class by adding two abstract interfaces: reasoning and acting, while supporting hook functions at four positions: pre-reasoning, post-reasoning, pre-acting, and post-acting by the _ReActAgentMeta metaclass.

supported_hook_types: list[str] = ['pre_reply', 'post_reply', 'pre_print', 'post_print', 'pre_observe', 'post_observe', 'pre_reasoning', 'post_reasoning', 'pre_acting', 'post_acting']

Supported hook types for the agent base class.

__init__()

Initialize the ReAct agent base class.

返回类型:

None

class ReActAgent

基类：ReActAgentBase

A ReAct agent implementation in AgentScope, which supports

• Realtime steering

• API-based (parallel) tool calling

• Hooks around reasoning, acting, reply, observe and print functions

• Structured output generation

finish_function_name: str = 'generate_response'

The name of the function used to generate structured output. Only registered when structured output model is provided in the reply call.

__init__(name, sys_prompt, model, formatter, toolkit=None, memory=None, long_term_memory=None, long_term_memory_mode='both', enable_meta_tool=False, parallel_tool_calls=False, knowledge=None, enable_rewrite_query=True, plan_notebook=None, print_hint_msg=False, max_iters=10, tts_model=None)

Initialize the ReAct agent

参数:

• name (str) -- The name of the agent.

• sys_prompt (str) -- The system prompt of the agent.

• model (ChatModelBase) -- The chat model used by the agent.

• formatter (FormatterBase) -- The formatter used to format the messages into the required format of the model API provider.

• toolkit (Toolkit | None, optional) -- A Toolkit object that contains the tool functions. If not provided, a default empty Toolkit will be created.

• memory (MemoryBase | None, optional) -- The memory used to store the dialogue history. If not provided, a default InMemoryMemory will be created, which stores messages in a list in memory.

• long_term_memory (LongTermMemoryBase | None, optional) -- The optional long-term memory, which will provide two tool functions: retrieve_from_memory and record_to_memory, and will attach the retrieved information to the system prompt before each reply.

• enable_meta_tool (bool, defaults to False) -- If True, a meta tool function reset_equipped_tools will be added to the toolkit, which allows the agent to manage its equipped tools dynamically.

• long_term_memory_mode (Literal['agent_control', 'static_control', 'both'], defaults to both) -- The mode of the long-term memory. If agent_control, two tool functions retrieve_from_memory and record_to_memory will be registered in the toolkit to allow the agent to manage the long-term memory. If static_control, retrieving and recording will happen in the beginning and end of each reply respectively.

• parallel_tool_calls (bool, defaults to False) -- When LLM generates multiple tool calls, whether to execute them in parallel.

• knowledge (KnowledgeBase | list[KnowledgeBase] | None, optional) -- The knowledge object(s) used by the agent to retrieve relevant documents at the beginning of each reply.

• enable_rewrite_query (bool, defaults to True) -- Whether ask the agent to rewrite the user input query before retrieving from the knowledge base(s), e.g. rewrite "Who am I" to "{user's name}" to get more relevant documents. Only works when the knowledge base(s) is provided.

• plan_notebook (PlanNotebook | None, optional) -- The plan notebook instance, allow the agent to finish the complex task by decomposing it into a sequence of subtasks.

• print_hint_msg (bool, defaults to False) -- Whether to print the hint messages, including the reasoning hint from the plan notebook, the retrieved information from the long-term memory and knowledge base(s).

• max_iters (int, defaults to 10) -- The maximum number of iterations of the reasoning-acting loops.

• tts_model (TTSModelBase | None optional) -- The TTS model used by the agent.

返回类型:

None

property sys_prompt: str

The dynamic system prompt of the agent.

async reply(msg=None, structured_model=None)

Generate a reply based on the current state and input arguments.

参数:

• msg (Msg | list[Msg] | None, optional) -- The input message(s) to the agent.

• structured_model (Type[BaseModel] | None, optional) -- The required structured output model. If provided, the agent is expected to generate structured output in the metadata field of the output message.

返回:

The output message generated by the agent.

返回类型:

Msg

async observe(msg)

Receive observing message(s) without generating a reply.

参数:

msg (Msg | list[Msg] | None) -- The message or messages to be observed.

返回类型:

None

async handle_interrupt(msg=None, structured_model=None)

The post-processing logic when the reply is interrupted by the user or something else.

参数:

• msg (Msg | list[Msg] | None, optional) -- The input message(s) to the agent.

• structured_model (Type[BaseModel] | None, optional) -- The required structured output model.

返回类型:

Msg

generate_response(**kwargs)

Generate required structured output by this function and return it

参数:

kwargs (Any)

返回类型:

ToolResponse

class UserInputData

基类：object

The user input data.

blocks_input: List[TextBlock | ImageBlock | AudioBlock | VideoBlock] = None

The text input from the user

structured_input: dict[str, Any] | None = None

The structured input from the user

__init__(blocks_input=None, structured_input=None)

参数:

• blocks_input (List[TextBlock | ImageBlock | AudioBlock | VideoBlock] | None)

• structured_input (dict[str, Any] | None)

返回类型:

None

class UserInputBase

基类：object

The base class used to handle the user input from different sources.

abstract async __call__(agent_id, agent_name, *args, structured_model=None, **kwargs)

The user input method, which returns the user input and the required structured data.

参数:

• agent_id (str) -- The agent identifier.

• agent_name (str) -- The agent name.

• structured_model (Type[BaseModel] | None, optional) -- A base model class that defines the structured input format.

• args (Any)

• kwargs (Any)

返回:

The user input data.

返回类型:

UserInputData

class TerminalUserInput

基类：UserInputBase

The terminal user input.

__init__(input_hint='User Input: ')

Initialize the terminal user input with a hint.

参数:

input_hint (str)

返回类型:

None

async __call__(agent_id, agent_name, *args, structured_model=None, **kwargs)

Handle the user input from the terminal.

参数:

• agent_id (str) -- The agent identifier.

• agent_name (str) -- The agent name.

• structured_model (Type[BaseModel] | None, optional) -- A base model class that defines the structured input format.

• args (Any)

• kwargs (Any)

返回:

The user input data.

返回类型:

UserInputData

class StudioUserInput

基类：UserInputBase

The class that host the user input on the AgentScope Studio.

__init__(studio_url, run_id, max_retries=3, reconnect_attempts=3, reconnection_delay=1, reconnection_delay_max=5)

Initialize the StudioUserInput object.

参数:

• studio_url (str) -- The URL of the AgentScope Studio.

• run_id (str) -- The current run identity.

• max_retries (int, defaults to 3) -- The maximum number of retries to get user input.

• reconnect_attempts (int)

• reconnection_delay (int)

• reconnection_delay_max (int)

返回类型:

None

async __call__(agent_id, agent_name, *args, structured_model=None)

Get the user input from AgentScope Studio.

参数:

• agent_id (str) -- The identity of the agent.

• agent_name (str) -- The name of the agent.

• structured_model (Type[BaseModel] | None, optional) -- The base model class of the structured input.

• args (Any)

抛出:

RuntimeError -- Failed to get user input from AgentScope Studio.

返回:

The user input.

返回类型:

UserInputData

class UserAgent

基类：AgentBase

The class for user interaction, allowing developers to handle the user input from different sources, such as web UI, cli, and other interfaces.

__init__(name)

Initialize the user agent with a name.

参数:

name (str)

返回类型:

None

async reply(msg=None, structured_model=None)

Receive input message(s) and generate a reply message from the user.

参数:

• msg (Msg | list[Msg] | None, defaults to None) -- The message(s) to be replied. If None, the agent will wait for user input.

• structured_model (Type[BaseModel] | None, defaults to None) -- A child class of pydantic.BaseModel that defines the structured output format. If provided, the user will be prompted to fill in the required fields.

返回:

The reply message generated by the user.

返回类型:

Msg

override_instance_input_method(input_method)

Override the input method of the current UserAgent instance.

参数:

input_method (UserInputBase) -- The callable input method, which should be an object of a class that inherits from UserInputBase.

返回类型:

None

classmethod override_class_input_method(input_method)

Override the input method of the current UserAgent class.

参数:

input_method (UserInputBase) -- The callable input method, which should be an object of a class that inherits from UserInputBase.

返回类型:

None

async handle_interrupt(*args, **kwargs)

The post-processing logic when the reply is interrupted by the user or something else.

参数:

• args (Any)

• kwargs (Any)

返回类型:

Msg

async observe(msg)

Observe the message(s) from the other agents or the environment.

参数:

msg (Msg | list[Msg] | None)

返回类型:

None

class A2AAgent

基类：AgentBase

An A2A agent implementation in AgentScope, which supports

• Communication with remote agents using the A2A protocol

• Bidirectional message conversion between AgentScope and A2A formats

• Task lifecycle management with streaming and polling

• Artifact handling and status tracking

备注

Due to the limitation of A2A protocol. The A2AAgent class

• Only support chatbot-scenario (a user and an assistant) interactions.

To support multi-agent interactions requires the server side to handle the name field in the A2A messages properly.

• Does not support structured output in reply() method due to the lack

of structured output support in A2A protocol.

• Stores observed messages locally and merges them with input messages

when reply() is called. Observed messages are cleared after processing.

__init__(agent_card, client_config=None, consumers=None, additional_transport_producers=None)

Initialize the A2A agent instance by the given agent card.

参数:

• agent_card (AgentCard) -- The agent card that contains the information about the remote agent, such as its URL and capabilities.

• client_config (ClientConfig | None, optional) -- The configuration for the A2A client, including transport preferences and streaming options.

• consumers (list[Consumer] | None, optional) -- The list of consumers for handling A2A client events. These intercept request/response flows for logging, metrics, and security.

• additional_transport_producers (dict[str, TransportProducer] | None, optional) -- Additional transport producers for creating A2A clients with specific transport protocols.

返回类型:

None

state_dict()

Get the state dictionary of the A2A agent.

返回:

The state dictionary containing the observed messages.

返回类型:

dict

load_state_dict(state_dict, strict=True)

Load the state dictionary into the module.

参数:

• state_dict (dict) -- The state dictionary to load.

• strict (bool, defaults to True) -- If True, raises an error if any key in the module is not found in the state_dict. If False, skips missing keys.

抛出:

KeyError -- If a required key is missing in the state_dict when strict is True.

返回类型:

None

async observe(msg)

Receive the given message(s) without generating a reply.

The observed messages are stored and will be merged with the input messages when reply is called. After reply completes, the stored messages will be cleared.

参数:

msg (Msg | list[Msg] | None) -- The message(s) to be observed. If None, no action is taken.

返回类型:

None

async reply(msg=None, **kwargs)

Send message(s) to the remote A2A agent and receive a response.

备注

This method merges any previously observed messages with the

input messages, sends them to the remote agent, and clears the observed messages after processing.

备注

The A2A protocol does not support structured output, so the structured_model parameter is not supported in this method.

参数:

• msg (Msg | list[Msg] | None, optional) -- The message(s) to send to the remote agent. Can be a single Msg, a list of Msgs, or None. If None, only observed messages will be sent. Defaults to None.

• kwargs (Any)

返回:

The response message from the remote agent. For tasks, this may be either a status update message or the final artifacts message, depending on the task state. If no messages are provided (both msg and observed messages are empty), returns a prompt message. If an error occurs during communication, returns an error message.

返回类型:

Msg

async handle_interrupt(msg=None, structured_model=None)

The post-processing logic when the reply is interrupted by the user or something else.

参数:

• msg (Msg | list[Msg] | None)

• structured_model (Type[BaseModel] | None)

返回类型:

Msg