agentscope.agents.agent module

Base class for Agent

class agentscope.agents.agent.DistConf(host: str = 'localhost', port: int | None = None, max_pool_size: int = 8192, max_timeout_seconds: int = 7200, local_mode: bool = True, lazy_launch: bool = False)[source]

Bases: dict

Distribution configuration for agents.

__init__(host: str = 'localhost', port: int | None = None, max_pool_size: int = 8192, max_timeout_seconds: int = 7200, local_mode: bool = True, lazy_launch: bool = False)[source]

Init the distributed configuration.

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) – Max number of task results that the server can accommodate.

  • max_timeout_seconds (int, defaults to 7200) – Timeout for task results.

  • local_mode (bool, defaults to True) – Whether the started rpc server only listens to local requests.

  • lazy_launch (bool, defaults to False) – Only launch the server when the agent is called.

class agentscope.agents.agent.AgentBase(*args: tuple, **kwargs: dict)[source]

Bases: Operator

Base class for all agents.

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

__init__(name: str, sys_prompt: str | None = None, model_config_name: str | None = None, use_memory: bool = True, to_dist: DistConf | bool | None = False) → None[source]

Initialize an agent from the given arguments.

Parameters:

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

  • sys_prompt (Optional[str]) – The system prompt of the agent, which can be passed by args or hard-coded in the agent.

  • model_config_name (str, defaults to None) – The name of the model config, which is used to load model from configuration.

  • use_memory (bool, defaults to True) – Whether the agent has memory.

  • to_dist (Optional[Union[DistConf, bool]], default to False) –

    The configurations passed to to_dist() method. Used in _AgentMeta, when this parameter is provided, the agent will automatically be converted into its distributed version. Below are some examples:

    # run as a sub process
agent = XXXAgent(
    # ... other parameters
    to_dist=True,
)

# connect to an existing agent server
agent = XXXAgent(
    # ... other parameters
    to_dist=DistConf(
        host="<ip of your server>",
        port=<port of your server>,
        # other parameters
    ),
)

    See Tutorial for detail.

classmethod generate_agent_id() → str[source]

Generate the agent_id of this agent instance

classmethod get_agent_class(agent_class_name: str) → Type[AgentBase][source]

Get the agent class based on the specific agent class name.

Parameters:

agent_class_name (str) – the name of the agent class.

Raises:

ValueError – Agent class name not exits.

Returns:

the AgentBase subclass.

Return type:

Type[AgentBase]

classmethod register_agent_class(agent_class: Type[AgentBase]) → None[source]

Register the agent class into the registry.

Parameters:

agent_class (Type[AgentBase]) – the agent class to be registered.

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

Define the actions taken by this agent.

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

Note

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

speak(content: str | Msg | Generator[Tuple[bool, str], 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 –

    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.

  • Msg – 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.

  • Generator[Tuple[bool – 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.

  • str] – 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.

  • 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.

  • 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.

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.

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.

clear_audience() → None[source]

Remove the audience of this agent.

rm_audience(audience: Sequence[AgentBase] | AgentBase) → None[source]

Remove the given audience from the Sequence

property agent_id: str

The unique id of this agent.

Returns:

agent_id

Return type:

str

to_dist(host: str = 'localhost', port: int | None = None, max_pool_size: int = 8192, max_timeout_seconds: int = 7200, local_mode: bool = True, lazy_launch: bool = False, launch_server: bool | None = None) → AgentBase[source]

Convert current agent instance into a 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_timeout_seconds (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.

  • 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.

  • lazy_launch (bool, defaults to False) – Only takes effect when host and port are not filled in. If True, launch the agent server when the agent is called, otherwise, launch the agent server immediately.

  • launch_server (bool, defaults to None) – This field has been deprecated and will be removed in future releases.

Returns:

the wrapped agent instance with distributed functionality

Return type:

AgentBase