agentscope.agents

Import all agent related modules in the package.

class AgentBase(*args: tuple, **kwargs: dict)[source]

Bases: object

Base class for all agents.

All agents should inherit from this class and implement the reply function.

classmethod clear_all_class_hooks() → None[source]: Clear all hooks from the agent class

clear_all_obj_hooks() → None[source]: Clear all hooks from the agent object

clear_audience() → None[source]: Remove the audience of this agent.

classmethod clear_class_hooks(hook_type: Literal['pre_reply', 'post_reply', 'pre_speak', 'post_speak', 'pre_observe', 'post_observe']) → None[source]

Clear the specified hooks from the agent class.

Parameters:

(`Literal["pre_reply" (hook_type) –

”post_speak”, “pre_observe”, “post_observe”]`):
The name of the hooks attribute, which should be one of “pre_reply”, “post_reply”, “pre_speak”, “post_speak”, “pre_observe”, and “post_observe”.
"post_reply" –

“post_speak”, “pre_observe”, “post_observe”]`):
The name of the hooks attribute, which should be one of “pre_reply”, “post_reply”, “pre_speak”, “post_speak”, “pre_observe”, and “post_observe”.
"pre_speak" –

“post_speak”, “pre_observe”, “post_observe”]`):
The name of the hooks attribute, which should be one of “pre_reply”, “post_reply”, “pre_speak”, “post_speak”, “pre_observe”, and “post_observe”.

:param :

“post_speak”, “pre_observe”, “post_observe”]`):: The name of the hooks attribute, which should be one of “pre_reply”, “post_reply”, “pre_speak”, “post_speak”, “pre_observe”, and “post_observe”.

clear_hooks(hook_type: Literal['pre_reply', 'post_reply', 'pre_speak', 'post_speak', 'pre_observe', 'post_observe']) → None[source]

Clear the specified hooks from the agent object.

Parameters:

(`Literal["pre_reply" (hook_type) –

”post_speak”, “pre_observe”, “post_observe”]`):
The type of the hook, which should be one of “pre_reply”, “post_reply”, “pre_speak”, “post_speak”, “pre_observe”, and “post_observe”.
"post_reply" –

“post_speak”, “pre_observe”, “post_observe”]`):
The type of the hook, which should be one of “pre_reply”, “post_reply”, “pre_speak”, “post_speak”, “pre_observe”, and “post_observe”.
"pre_speak" –

“post_speak”, “pre_observe”, “post_observe”]`):
The type of the hook, which should be one of “pre_reply”, “post_reply”, “pre_speak”, “post_speak”, “pre_observe”, and “post_observe”.

:param :

“post_speak”, “pre_observe”, “post_observe”]`):: The type of the hook, which should be one of “pre_reply”, “post_reply”, “pre_speak”, “post_speak”, “pre_observe”, and “post_observe”.

classmethod generate_agent_id() → str[source]: Generate the agent_id of this agent instance

observe(x: Msg | Sequence[Msg]) → None[source]

Observe the input, store it in memory without response to it.

Parameters:: x (Union[Msg, Sequence[Msg]]) – The input message to be recorded in memory.

classmethod register_class_hook(hook_type: Literal['pre_reply', 'post_reply', 'pre_speak', 'post_speak', 'pre_observe', 'post_observe'], hook_name: str, hook: Callable) → None[source]

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

Parameters:

hook_type (str) – The type of the hook, which should be one of “pre_reply”, “post_reply”, “pre_speak”, “post_speak”, “pre_observe”, and “post_observe”.
hook_name (str) – The name of the hook. If the name is already registered, the hook will be overwritten.
hook (Callable) – The hook function.

register_hook(hook_type: Literal['pre_reply', 'post_reply', 'pre_speak', 'post_speak', 'pre_observe', 'post_observe'], hook_name: str, hook: Callable) → None[source]

The universal function to register a hook to the agent.

Parameters:

(`Literal["pre_reply" (hook_type) –

”post_speak”, “pre_observe”, “post_observe”]`):
The type of the hook, which should be one of “pre_reply”, “post_reply”, “pre_speak”, “post_speak”, “pre_observe”, and “post_observe”.
"post_reply" –

“post_speak”, “pre_observe”, “post_observe”]`):
The type of the hook, which should be one of “pre_reply”, “post_reply”, “pre_speak”, “post_speak”, “pre_observe”, and “post_observe”.
"pre_speak" –

“post_speak”, “pre_observe”, “post_observe”]`):
The type of the hook, which should be one of “pre_reply”, “post_reply”, “pre_speak”, “post_speak”, “pre_observe”, and “post_observe”.

:param :

“post_speak”, “pre_observe”, “post_observe”]`):: The type of the hook, which should be one of “pre_reply”, “post_reply”, “pre_speak”, “post_speak”, “pre_observe”, and “post_observe”.

Parameters:

hook_name (str) – The name of the hook. If the name is already registered, the hook will be overwritten.
hook (Callable) – The hook function.

classmethod remove_class_hook(hook_type: Literal['pre_reply', 'post_reply', 'pre_speak', 'post_speak', 'pre_observe', 'post_observe'], hook_name: str) → None[source]

The universal function to remove a hook from the agent class.

Parameters:

(`Literal["pre_reply" (hook_type) –

”post_speak”, “pre_observe”, “post_observe”]`):
The type of the hook, which should be one of “pre_reply”, “post_reply”, “pre_speak”, “post_speak”, “pre_observe”, and “post_observe”.
"post_reply" –

“post_speak”, “pre_observe”, “post_observe”]`):
The type of the hook, which should be one of “pre_reply”, “post_reply”, “pre_speak”, “post_speak”, “pre_observe”, and “post_observe”.
"pre_speak" –

“post_speak”, “pre_observe”, “post_observe”]`):
The type of the hook, which should be one of “pre_reply”, “post_reply”, “pre_speak”, “post_speak”, “pre_observe”, and “post_observe”.

:param :

“post_speak”, “pre_observe”, “post_observe”]`):: The type of the hook, which should be one of “pre_reply”, “post_reply”, “pre_speak”, “post_speak”, “pre_observe”, and “post_observe”.

Parameters:: hook_name (str) – The name of the hook to be removed.

remove_hook(hook_type: Literal['pre_reply', 'post_reply', 'pre_speak', 'post_speak', 'pre_observe', 'post_observe'], hook_name: str) → None[source]

The universal function to remove a hook from the agent object.

Parameters:

(`Literal["pre_reply" (hook_type) –

”post_speak”, “pre_observe”, “post_observe”]`):
The type of the hook, which should be one of “pre_reply”, “post_reply”, “pre_speak”, “post_speak”, “pre_observe”, and “post_observe”.
"post_reply" –

“post_speak”, “pre_observe”, “post_observe”]`):
The type of the hook, which should be one of “pre_reply”, “post_reply”, “pre_speak”, “post_speak”, “pre_observe”, and “post_observe”.
"pre_speak" –

“post_speak”, “pre_observe”, “post_observe”]`):
The type of the hook, which should be one of “pre_reply”, “post_reply”, “pre_speak”, “post_speak”, “pre_observe”, and “post_observe”.

:param :

“post_speak”, “pre_observe”, “post_observe”]`):: The type of the hook, which should be one of “pre_reply”, “post_reply”, “pre_speak”, “post_speak”, “pre_observe”, and “post_observe”.

Parameters:: hook_name (str) – The name of the hook to be removed.

reply(x: Msg | list[Msg] | None = None) → Msg[source]

Define the actions taken by this agent.

Parameters:: x (Optional[Union[Msg, list[Msg]]], defaults to None) – The input message(s) to the agent, which also can be omitted if the agent doesn’t need any input.
Returns:: The output message generated by the agent.
Return type:: Msg

Note

Given that some agents are in an adversarial environment, their input doesn’t include the thoughts of other agents.

reset_audience(audience: Sequence[AgentBase]) → None[source]

Set the audience of this agent, which means if this agent generates a response, it will be passed to all audiences.

Parameters:: audience (Sequence[AgentBase]) – The audience of this agent, which will be notified when this agent generates a response message.

rm_audience(audience: Sequence[AgentBase] | AgentBase) → None[source]: Remove the given audience from the Sequence

speak(content: str | Msg | Generator[Tuple[bool, str], None, None] | None, tool_calls: list[ToolUseBlock] | None = None) → None[source]

Speak out the message generated by the agent. If a string is given, a Msg object will be created with the string as the content.

Parameters:

content –
(Union[str, Msg, Generator[Tuple[bool, str], None, None], None]):

The content of the message to be spoken out. If a string is given, a Msg object will be created with the agent’s name, role as “assistant”, and the given string as the content. If the content is a Generator, the agent will speak out the message chunk by chunk.
tool_calls (Optional[list[ToolUseBlock]], defaults to None) – The tool calls generated by the agent. This parameter is only used when the content is a string or a generator (TODO). When the content is a Msg object, the tool calls should be included in the content field of the Msg object.

to_dist(host: str = 'localhost', port: int | None = None, max_pool_size: int = 8192, max_expire_time: int = 7200, max_timeout_seconds: int = 5, local_mode: bool = True, retry_strategy: ~agentscope.rpc.retry_strategy.RetryBase = <agentscope.rpc.retry_strategy.RetryFixedTimes object>) → Any

Convert current object into its distributed version.

Parameters:

host (str, defaults to “localhost”) – Hostname of the rpc agent server.
port (int, defaults to None) – Port of the rpc agent server.
max_pool_size (int, defaults to 8192) – Only takes effect when host and port are not filled in. The max number of agent reply messages that the started agent server can accommodate. Note that the oldest message will be deleted after exceeding the pool size.
max_expire_time (int, defaults to 7200) – Only takes effect when host and port are not filled in. Maximum time for reply messages to be cached in the launched agent server. Note that expired messages will be deleted.
max_timeout_seconds (int, defaults to 5) – Max timeout seconds for the rpc call.
local_mode (bool, defaults to True) – Only takes effect when host and port are not filled in. Whether the started agent server only listens to local requests.
retry_strategy (RetryBase, defaults to _DEFAULT_RETRY_STRATEGY) – The retry strategy for the async rpc call.

Returns:

the wrapped agent instance with distributed functionality

Return type:

RpcObject

property agent_id: str

The unique id of this agent.

Returns:: agent_id
Return type:: str

class DialogAgent(*args: tuple, **kwargs: dict)[source]

Bases: AgentBase

A simple agent used to perform a dialogue. Your can set its role by sys_prompt.

reply(x: Msg | Sequence[Msg] | None = None) → Msg[source]

Reply function of the agent. Processes the input data, generates a prompt using the current dialogue memory and system prompt, and invokes the language model to produce a response. The response is then formatted and added to the dialogue memory.

Parameters:: x (Optional[Union[Msg, Sequence[Msg]]], defaults to None) – The input message(s) to the agent, which also can be omitted if the agent doesn’t need any input.
Returns:: The output message generated by the agent.
Return type:: Msg

to_dist(host: str = 'localhost', port: int | None = None, max_pool_size: int = 8192, max_expire_time: int = 7200, max_timeout_seconds: int = 5, local_mode: bool = True, retry_strategy: ~agentscope.rpc.retry_strategy.RetryBase = <agentscope.rpc.retry_strategy.RetryFixedTimes object>) → Any

Convert current object into its distributed version.

Parameters:

host (str, defaults to “localhost”) – Hostname of the rpc agent server.
port (int, defaults to None) – Port of the rpc agent server.
max_pool_size (int, defaults to 8192) – Only takes effect when host and port are not filled in. The max number of agent reply messages that the started agent server can accommodate. Note that the oldest message will be deleted after exceeding the pool size.
max_expire_time (int, defaults to 7200) – Only takes effect when host and port are not filled in. Maximum time for reply messages to be cached in the launched agent server. Note that expired messages will be deleted.
max_timeout_seconds (int, defaults to 5) – Max timeout seconds for the rpc call.
local_mode (bool, defaults to True) – Only takes effect when host and port are not filled in. Whether the started agent server only listens to local requests.
retry_strategy (RetryBase, defaults to _DEFAULT_RETRY_STRATEGY) – The retry strategy for the async rpc call.

Returns:

the wrapped agent instance with distributed functionality

Return type:

RpcObject

class DictDialogAgent(*args: tuple, **kwargs: dict)[source]

Bases: AgentBase

An agent that generates response in a dict format, where user can specify the required fields in the response via specifying the parser

About parser, please refer to our [tutorial](https://doc.agentscope.io/build_tutorial/structured_output.html)

For usage example, please refer to the example of werewolf in examples/game_werewolf

reply(x: Msg | Sequence[Msg] | None = None) → Msg[source]

Reply function of the agent. Processes the input data, generates a prompt using the current dialogue memory and system prompt, and invokes the language model to produce a response. The response is then formatted and added to the dialogue memory.

Parameters:: x (Optional[Union[Msg, Sequence[Msg]]], defaults to None) – The input message(s) to the agent, which also can be omitted if the agent doesn’t need any input.
Returns:: The output message generated by the agent.
Return type:: Msg
Raises:: json.decoder.JSONDecodeError – If the response from the language model is not valid JSON, it defaults to treating the response as plain text.

set_parser(parser: ParserBase) → None[source]: Set response parser, which will provide 1) format instruction; 2) response parsing; 3) filtering fields when returning message, storing message in memory. So developers only need to change the parser, and the agent will work as expected.

to_dist(host: str = 'localhost', port: int | None = None, max_pool_size: int = 8192, max_expire_time: int = 7200, max_timeout_seconds: int = 5, local_mode: bool = True, retry_strategy: ~agentscope.rpc.retry_strategy.RetryBase = <agentscope.rpc.retry_strategy.RetryFixedTimes object>) → Any

Convert current object into its distributed version.

Parameters:

host (str, defaults to “localhost”) – Hostname of the rpc agent server.
port (int, defaults to None) – Port of the rpc agent server.
max_pool_size (int, defaults to 8192) – Only takes effect when host and port are not filled in. The max number of agent reply messages that the started agent server can accommodate. Note that the oldest message will be deleted after exceeding the pool size.
max_expire_time (int, defaults to 7200) – Only takes effect when host and port are not filled in. Maximum time for reply messages to be cached in the launched agent server. Note that expired messages will be deleted.
max_timeout_seconds (int, defaults to 5) – Max timeout seconds for the rpc call.
local_mode (bool, defaults to True) – Only takes effect when host and port are not filled in. Whether the started agent server only listens to local requests.
retry_strategy (RetryBase, defaults to _DEFAULT_RETRY_STRATEGY) – The retry strategy for the async rpc call.

Returns:

the wrapped agent instance with distributed functionality

Return type:

RpcObject

class LlamaIndexAgent(*args: tuple, **kwargs: dict)[source]

Bases: AgentBase

A LlamaIndex agent build on LlamaIndex.

reply(x: Msg | Sequence[Msg] | None = None) → Msg[source]

Reply function of the RAG agent. Processes the input data, 1) use the input data to retrieve with RAG function; 2) generates a prompt using the current memory and system prompt; 3) invokes the language model to produce a response. The response is then formatted and added to the dialogue memory.

Parameters:: x (Optional[Union[Msg, Sequence[Msg]]], defaults to None) – The input message(s) to the agent, which also can be omitted if the agent doesn’t need any input.
Returns:: The output message generated by the agent.
Return type:: Msg

to_dist(host: str = 'localhost', port: int | None = None, max_pool_size: int = 8192, max_expire_time: int = 7200, max_timeout_seconds: int = 5, local_mode: bool = True, retry_strategy: ~agentscope.rpc.retry_strategy.RetryBase = <agentscope.rpc.retry_strategy.RetryFixedTimes object>) → Any

Convert current object into its distributed version.

Parameters:

host (str, defaults to “localhost”) – Hostname of the rpc agent server.
port (int, defaults to None) – Port of the rpc agent server.
max_pool_size (int, defaults to 8192) – Only takes effect when host and port are not filled in. The max number of agent reply messages that the started agent server can accommodate. Note that the oldest message will be deleted after exceeding the pool size.
max_expire_time (int, defaults to 7200) – Only takes effect when host and port are not filled in. Maximum time for reply messages to be cached in the launched agent server. Note that expired messages will be deleted.
max_timeout_seconds (int, defaults to 5) – Max timeout seconds for the rpc call.
local_mode (bool, defaults to True) – Only takes effect when host and port are not filled in. Whether the started agent server only listens to local requests.
retry_strategy (RetryBase, defaults to _DEFAULT_RETRY_STRATEGY) – The retry strategy for the async rpc call.

Returns:

the wrapped agent instance with distributed functionality

Return type:

RpcObject

class ReActAgent(*args: tuple, **kwargs: dict)[source]

Bases: AgentBase

An agent class that implements the ReAct algorithm. More details refer to https://arxiv.org/abs/2210.03629.

This is an example implementation of ReAct algorithm in AgentScope. We follow the idea within the paper, but the detailed prompt engineering maybe different. Developers are encouraged to modify the prompt to fit their own needs.

Note

1. We use the “thought” field in the response to support Chain-of-Thought, which means the tool functions cannot use “thought” as their argument name. 2. The function name “finish” is also a reserved name when using this agent, which will be used to end the reasoning-acting loop.

static finish(response: str) → ServiceResponse[source]

Finish reasoning and generate a response to the user.

Note

The function won’t be executed, actually.

Parameters:: response (str) – The response to the user.

reply(x: Msg | Sequence[Msg] | None = None) → Msg[source]: The reply method of the agent.

to_dist(host: str = 'localhost', port: int | None = None, max_pool_size: int = 8192, max_expire_time: int = 7200, max_timeout_seconds: int = 5, local_mode: bool = True, retry_strategy: ~agentscope.rpc.retry_strategy.RetryBase = <agentscope.rpc.retry_strategy.RetryFixedTimes object>) → Any

Convert current object into its distributed version.

Parameters:

host (str, defaults to “localhost”) – Hostname of the rpc agent server.
port (int, defaults to None) – Port of the rpc agent server.
max_pool_size (int, defaults to 8192) – Only takes effect when host and port are not filled in. The max number of agent reply messages that the started agent server can accommodate. Note that the oldest message will be deleted after exceeding the pool size.
max_expire_time (int, defaults to 7200) – Only takes effect when host and port are not filled in. Maximum time for reply messages to be cached in the launched agent server. Note that expired messages will be deleted.
max_timeout_seconds (int, defaults to 5) – Max timeout seconds for the rpc call.
local_mode (bool, defaults to True) – Only takes effect when host and port are not filled in. Whether the started agent server only listens to local requests.
retry_strategy (RetryBase, defaults to _DEFAULT_RETRY_STRATEGY) – The retry strategy for the async rpc call.

Returns:

the wrapped agent instance with distributed functionality

Return type:

RpcObject

class ReActAgentV2(*args: tuple, **kwargs: dict)[source]

Bases: AgentBase

The ReAct Agent V2, which use the tools API of the LLM service providers rather than assembling the prompt manually.

The currently supported LLM providers include: - Anthropic - DashScope - OpenAI

generate_response(response: str, **kwargs: Any) → ServiceResponse[source]

Generate a response. You must call this function to interact with others (e.g., users).

Parameters:: response (str) – The response to the user.

reply(x: Msg | list[Msg] | None = None, structured_model: Type[BaseModel] | None = None) → Msg[source]

The reply method of the agent.

Parameters:

x (Optional[Union[Msg, list[Msg]]]) – The input message(s) to the agent.
(Optional[Type[BaseModel]] (structured_model) –

defaults to None):
A Pydantic model class that defines the structured output. If provided, the schema of the structured output will be merged with the generate_response function, and the argument exit_reply_without_tool_calls will be invalid. The agent must call the generate_response function to end the reply with a structured output.

:param :

defaults to None):: A Pydantic model class that defines the structured output. If provided, the schema of the structured output will be merged with the generate_response function, and the argument exit_reply_without_tool_calls will be invalid. The agent must call the generate_response function to end the reply with a structured output.

Returns:: The response message from the agent.
Return type:: Msg

to_dist(host: str = 'localhost', port: int | None = None, max_pool_size: int = 8192, max_expire_time: int = 7200, max_timeout_seconds: int = 5, local_mode: bool = True, retry_strategy: ~agentscope.rpc.retry_strategy.RetryBase = <agentscope.rpc.retry_strategy.RetryFixedTimes object>) → Any

Convert current object into its distributed version.

Parameters:

host (str, defaults to “localhost”) – Hostname of the rpc agent server.
port (int, defaults to None) – Port of the rpc agent server.
max_pool_size (int, defaults to 8192) – Only takes effect when host and port are not filled in. The max number of agent reply messages that the started agent server can accommodate. Note that the oldest message will be deleted after exceeding the pool size.
max_expire_time (int, defaults to 7200) – Only takes effect when host and port are not filled in. Maximum time for reply messages to be cached in the launched agent server. Note that expired messages will be deleted.
max_timeout_seconds (int, defaults to 5) – Max timeout seconds for the rpc call.
local_mode (bool, defaults to True) – Only takes effect when host and port are not filled in. Whether the started agent server only listens to local requests.
retry_strategy (RetryBase, defaults to _DEFAULT_RETRY_STRATEGY) – The retry strategy for the async rpc call.

Returns:

the wrapped agent instance with distributed functionality

Return type:

RpcObject

class StudioUserInput(studio_url: str, run_id: str, max_retries: int = 3, reconnect_attempts: int = 3, reconnection_delay: int = 1, reconnection_delay_max: int = 5)[source]

Bases: UserInputBase

The agentscope studio user input.

class TerminalUserInput(input_prefix: str = 'User Input: ')[source]

Bases: UserInputBase

The terminal user input.

class UserAgent(*args: tuple, **kwargs: dict)[source]

Bases: AgentBase

User agent class

classmethod override_class_input_method(input_method: UserInputBase) → None[source]

Override the input method of the current UserAgent instance.

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

override_input_method(input_method: UserInputBase) → None[source]

Override the input method of the current UserAgent instance.

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

reply(x: Msg | list[Msg] | None = None, structured_output: Type[BaseModel] | None = None) → Msg[source]

Receive input message(s) and generate a reply message from the user. The input source can be set by calling override_input_method and override_class_input_method methods.

For example, if you want to receive user input from your own source, you can create a class that inherits from UserInputBase and override the input method of the UserAgent class.

Parameters:

x (Optional[Union[Msg, list[Msg]]], defaults to None) – The input message(s) to the agent, which also can be omitted if the agent doesn’t need any input.
(Optional[Type[BaseModel]] (structured_output) –

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 output will be placed in the metadata field of the output message.
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 output will be placed in the metadata field of the output message.

Returns:

The output message generated by the agent.

Return type:

Msg

to_dist(host: str = 'localhost', port: int | None = None, max_pool_size: int = 8192, max_expire_time: int = 7200, max_timeout_seconds: int = 5, local_mode: bool = True, retry_strategy: ~agentscope.rpc.retry_strategy.RetryBase = <agentscope.rpc.retry_strategy.RetryFixedTimes object>) → Any

Convert current object into its distributed version.

Parameters:

host (str, defaults to “localhost”) – Hostname of the rpc agent server.
port (int, defaults to None) – Port of the rpc agent server.
max_pool_size (int, defaults to 8192) – Only takes effect when host and port are not filled in. The max number of agent reply messages that the started agent server can accommodate. Note that the oldest message will be deleted after exceeding the pool size.
max_expire_time (int, defaults to 7200) – Only takes effect when host and port are not filled in. Maximum time for reply messages to be cached in the launched agent server. Note that expired messages will be deleted.
max_timeout_seconds (int, defaults to 5) – Max timeout seconds for the rpc call.
local_mode (bool, defaults to True) – Only takes effect when host and port are not filled in. Whether the started agent server only listens to local requests.
retry_strategy (RetryBase, defaults to _DEFAULT_RETRY_STRATEGY) – The retry strategy for the async rpc call.

Returns:

the wrapped agent instance with distributed functionality

Return type:

RpcObject

class UserInputBase[source]

Bases: ABC

The base class for user input.

Bases: BaseModel

The user input data.

blocks_input: list[ToolUseBlock | ToolResultBlock | TextBlock | ImageBlock | AudioBlock | VideoBlock | FileBlock] | None: The text input from the user.

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

structured_input: dict[str, Any] | None: The structured input from the user.