Skip to main content

agentchat.contrib.swarm_agent

initiate_swarm_chat​

def initiate_swarm_chat(
    initial_agent: "SwarmAgent",
    messages: Union[List[Dict[str, Any]], str],
    agents: List["SwarmAgent"],
    user_agent: Optional[UserProxyAgent] = None,
    max_rounds: int = 20,
    context_variables: Optional[Dict[str, Any]] = None,
    after_work: Optional[Union[AFTER_WORK, Callable]] = AFTER_WORK(
        AfterWorkOption.TERMINATE)
) -> Tuple[ChatResult, Dict[str, Any], "SwarmAgent"]

Initialize and run a swarm chat

Arguments:

  • initial_agent - The first receiving agent of the conversation.
  • messages - Initial message(s).
  • agents - List of swarm agents.
  • user_agent - Optional user proxy agent for falling back to.
  • max_rounds - Maximum number of conversation rounds.
  • context_variables - Starting context variables.
  • after_work - Method to handle conversation continuation when an agent doesn't select the next agent. If no agent is selected and no tool calls are output, we will use this method to determine the next agent. Must be a AFTER_WORK instance (which is a dataclass accepting a SwarmAgent, AfterWorkOption, A str (of the AfterWorkOption)) or a callable. AfterWorkOption:
    • TERMINATE (Default): Terminate the conversation.
    • REVERT_TO_USER : Revert to the user agent if a user agent is provided. If not provided, terminate the conversation.
    • STAY : Stay with the last speaker.
  • Callable - A custom function that takes the current agent, messages, groupchat, and context_variables as arguments and returns the next agent. The function should return None to terminate. python def custom_afterwork_func(last_speaker: SwarmAgent, messages: List[Dict[str, Any]], groupchat: GroupChat, context_variables: Optional[Dict[str, Any]]) -> Optional[SwarmAgent]:

Returns:

  • ChatResult - Conversations chat history. Dict[str, Any]: Updated Context variables.
  • SwarmAgent - Last speaker.

SwarmResult​

class SwarmResult(BaseModel)

Encapsulates the possible return values for a swarm agent function.

Arguments:

  • values str - The result values as a string.
  • agent SwarmAgent - The swarm agent instance, if applicable.
  • context_variables dict - A dictionary of context variables.

SwarmAgent​

class SwarmAgent(ConversableAgent)

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.

register_update_agent_state_before_reply​

def register_update_agent_state_before_reply(
        functions: Optional[Union[List[Callable], Callable]])

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.

Arguments:

  • functions List[Callable[[], None]] - A list of functions to be registered. Each function is called when the agent is selected and before it speaks.

register_hand_off​

def register_hand_off(hand_to: Union[List[Union[ON_CONDITION, AFTER_WORK]],
                                     ON_CONDITION, AFTER_WORK])

Register a function to hand off to another agent.

Arguments:

  • hand_to - A list of ON_CONDITIONs and an, optional, AFTER_WORK condition

    Hand off template: def transfer_to_agent_name() -> SwarmAgent: return agent_name

    1. register the function with the agent
    2. register the schema with the agent, description set to the condition

generate_swarm_tool_reply​

def generate_swarm_tool_reply(
        messages: Optional[List[Dict]] = None,
        sender: Optional[Agent] = None,
        config: Optional[OpenAIWrapper] = 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.

add_single_function​

def add_single_function(func: Callable, name=None, description="")

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

process_nested_chat_carryover​

@staticmethod
def process_nested_chat_carryover(chat: Dict[str, Any],
                                  recipient: ConversableAgent,
                                  messages: List[Dict[str, Any]],
                                  sender: ConversableAgent,
                                  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

Arguments:

  • chat - The chat dictionary containing the carryover configuration
  • recipient - The recipient agent
  • messages - The messages from the parent chat
  • sender - The sender agent
  • trim_n_messages - The number of latest messages to trim from the messages list