SwarmAgent

SwarmAgent(
    name: str,
    system_message: str | None = 'You are a helpful AI Assistant.',
    llm_config: dict | Literal[False] | None = None,
    functions: list[typing.Callable] | Callable = None,
    is_termination_msg: Callable[[dict], bool] | None = None,
    max_consecutive_auto_reply: int | None = None,
    human_input_mode: Literal['ALWAYS', 'NEVER', 'TERMINATE'] = 'NEVER',
    description: str | None = None,
    code_execution_config=False,
    update_agent_state_before_reply: list[Callable | autogen.UPDATE_SYSTEM_MESSAGE] | Callable | autogen.UPDATE_SYSTEM_MESSAGE | None = None,
    **kwargs
)

Swarm agent for participating in a swarm.

SwarmAgent is a subclass of ConversableAgent.

Additional args: functions (List[Callable]): A list of functions to register with the agent. update_agent_state_before_reply (List[Callable]): A list of functions, including UPDATE_SYSTEM_MESSAGEs, called to update the agent before it replies.

Parameters:
NameDescription
namename of the agent.

Type: str
system_messagesystem message for the ChatCompletion inference.

Type: str | None

Default: ‘You are a helpful AI Assistant.‘
llm_configllm inference configuration.

Please refer to OpenAIWrapper.create for available options.

When using OpenAI or Azure OpenAI endpoints, please specify a non-empty ‘model’ either in llm_config or in each config of ‘config_list’ in llm_config.

To disable llm-based auto reply, set to False.

When set to None, will use self.DEFAULT_CONFIG, which defaults to False.

Type: dict | Literal[False] | None

Default: None
functionsType: list[typing.Callable] | Callable

Default: None
is_termination_msga function that takes a message in the form of a dictionary and returns a boolean value indicating if this received message is a termination message.

The dict can contain the following keys: “content”, “role”, “name”, “function_call”.

Type: Callable[[dict], bool] | None

Default: None
max_consecutive_auto_replythe maximum number of consecutive auto replies.

default to None (no limit provided, class attribute MAX_CONSECUTIVE_AUTO_REPLY will be used as the limit in this case).

When set to 0, no auto reply will be generated.

Type: int | None

Default: None
human_input_modewhether to ask for human inputs every time a message is received.

Possible values are “ALWAYS”, “TERMINATE”, “NEVER”.

(1) When “ALWAYS”, the agent prompts for human input every time a message is received.

Under this mode, the conversation stops when the human input is “exit”, or when is_termination_msg is True and there is no human input.

(2) When “TERMINATE”, the agent only prompts for human input only when a termination message is received or the number of auto reply reaches the max_consecutive_auto_reply.

(3) When “NEVER”, the agent will never prompt for human input.

Under this mode, the conversation stops when the number of auto reply reaches the max_consecutive_auto_reply or when is_termination_msg is True.

Type: Literal['ALWAYS', 'NEVER', 'TERMINATE']

Default: ‘NEVER’
descriptiona short description of the agent.

This description is used by other agents (e.g.

the GroupChatManager) to decide when to call upon this agent.

(Default: system_message)

Type: str | None

Default: None
code_execution_config=False
update_agent_state_before_replyType: list[Callable | autogen.UPDATE_SYSTEM_MESSAGE] | Callable | autogen.UPDATE_SYSTEM_MESSAGE | None

Default: None
**kwargs

Static Methods

process_nested_chat_carryover

process_nested_chat_carryover(
    chat: dict[str, typing.Any],
    recipient: autogen.ConversableAgent,
    messages: list[dict[str, typing.Any]],
    sender: autogen.ConversableAgent,
    config: Any,
    trim_n_messages: int = 0
) -> None

Process carryover messages for a nested chat (typically for the first chat of a swarm)

The carryover_config key is a dictionary containing: “summary_method”: The method to use to summarise the messages, can be “all”, “last_msg”, “reflection_with_llm” or a Callable “summary_args”: Optional arguments for the summary method

Supported carryover ‘summary_methods’ are: “all” - all messages will be incorporated “last_msg” - the last message will be incorporated “reflection_with_llm” - an llm will summarise all the messages and the summary will be incorporated as a single message Callable - a callable with the signature: my_method(agent: ConversableAgent, messages: List[Dict[str, Any]]) -> str

Parameters:
NameDescription
chatThe chat dictionary containing the carryover configuration

Type: dict[str, typing.Any]
recipientThe recipient agent

Type: autogen.ConversableAgent
messagesThe messages from the parent chat

Type: list[dict[str, typing.Any]]
senderThe sender agent

Type: autogen.ConversableAgent
configType: Any
trim_n_messagesThe number of latest messages to trim from the messages list

Type: int

Default: 0

Instance Methods

add_functions

add_functions(self, func_list: list[typing.Callable]) -> 
Parameters:
NameDescription
func_listType: list[typing.Callable]

add_single_function

add_single_function(
    self,
    func: Callable,
    name=None,
    description=''
) -> 

Add a single function to the agent, removing context variables for LLM use

Parameters:
NameDescription
funcType: Callable
name=None
description=''

generate_swarm_tool_reply

generate_swarm_tool_reply(
    self,
    messages: list[dict] | None = None,
    sender: autogen.Agent | None = None,
    config: autogen.OpenAIWrapper | None = None
) -> tuple[bool, dict]

Pre-processes and generates tool call replies.

This function:

  1. Adds context_variables back to the tool call for the function, if necessary.
  2. Generates the tool calls reply.
  3. Updates context_variables and next_agent based on the tool call response.
Parameters:
NameDescription
messagesType: list[dict] | None

Default: None
senderType: autogen.Agent | None

Default: None
configType: autogen.OpenAIWrapper | None

Default: None

register_hand_off

register_hand_off(self, hand_to: list[autogen.ON_CONDITION | autogen.AFTER_WORK] | autogen.ON_CONDITION | autogen.AFTER_WORK) -> 

Register a function to hand off to another agent.

Parameters:
NameDescription
hand_toA list of ON_CONDITIONs and an, optional, AFTER_WORK condition

Type: list[autogen.ON_CONDITION | autogen.AFTER_WORK] | autogen.ON_CONDITION | autogen.AFTER_WORK

register_update_agent_state_before_reply

register_update_agent_state_before_reply(self, functions: list[typing.Callable] | Callable | None) -> 

Register functions that will be called when the agent is selected and before it speaks. You can add your own validation or precondition functions here.

Parameters:
NameDescription
functionsA list of functions to be registered.

Each function is called when the agent is selected and before it speaks.

Type: list[typing.Callable] | Callable | None