Deep Dive

Swarms provide controllable flows between agents that are determined at the agent level. You define handoff, post-tool, and post-work transitions from one agent to another (or to end the swarm).

In this Swarm deep-dive we run through all the components of AG2's Swarm. You can learn about Swarm's high-level concepts in the Basic Concepts section.

Components#

Here are the main components that are needed to create a swarm chat:

1. Create Agents: instantiate an AssistantAgent to be part of the swarm chat.
   • API Reference: AssistantAgent
2. Register Handoffs: utilize register_hand_off to register OnCondition, OnContextCondition, and AfterWork handoffs.
   • API Reference: register_hand_off, OnCondition, OnContextCondition, AfterWork, AfterWorkOption, SwarmResult
3. Update Agent State (Optional): update an agent's state before replying.
   • API Reference: UpdateSystemMessage
4. Start Swarm Chat: initiate the swarm chat with initiate_swarm_chat or a_initiate_swarm_chat for asynchronous calls.
   • API Reference: initiate_swarm_chat, a_initiate_swarm_chat

Create Agents#

You can directly create AssistantAgent to be used in a swarm chat. Instead of registering functions one by one, you can pass in a list of functions when creating your agent, AssistantAgent(functions=[func1, ...]). These functions will be converted to schemas to be passed to the LLMs, and you don't need to worry about registering the functions for execution as the swarm handles that automatically.

Notes for creating the function calls (Caution: The notes below only apply to functions that will used in swarm chats with initiate_swarm_chat//docs/api-reference/autogen/a_initiate_swarm_chat) - You can pass back a SwarmResult object whereby you can return a value, the next agent to call, and update context variables at the same time. - For input arguments, you must define the type of the argument, otherwise, the registration will fail (e.g. arg_name: str). - If your function requires access or modification of the context variables, you must pass in context_variables: dict as one argument. This argument will not be visible to the LLM (removed when registering the function schema). But when called, the global context variables will be passed in by the swarm chat. If you are making changes to the context variables you must return it in the SwarmResult so it can be updated. - The docstring of the function will be used as the prompt. So make sure to write a clear description. - The function name will be used as the tool name.

Registering Handoffs to agents#

While you can create functions to decide who the next agent to call is, we provide a quick way to register the handoff using OnCondition. Internally, a transition function is created and added to the LLM configuration directly.

from autogen import AssistantAgent, OnCondition, register_hand_off
# llm_config = ...
agent_2 = AssistantAgent("agent_2", llm_config=llm_config)
agent_3 = AssistantAgent("agent_3", llm_config=llm_config)

# --------Option 1---------
agent_1 = AssistantAgent("agent_1", llm_config=llm_config)
# Register the handoff
register_hand_off(
    agent = agent_1,
    hand_to=[
        OnCondition(target=agent_2, condition="condition_1"),
        OnCondition(target=agent_3, condition="condition_2"),
    ]
)

# --------Option 2---------
# This is equivalent to:
def transfer_to_agent_2():
    """condition_1"""
    return agent_2

def transfer_to_agent_3():
    """condition_2"""
    return agent_3

agent_1 = AssistantAgent("agent_1", llm_config=llm_config, functions=[transfer_to_agent_2, transfer_to_agent_3])

Registering Context Variable-based Handoffs to agents#

In the previous section you saw how agent handoffs were created using OnCondition. These handoffs utilize the agent's LLM and tool calling to determine the next agent. However, if you are able to use the swarm's context variables in an expression, you can create a context variable-based handoff, using OnContextCondition and ContextExpression, that does not require an LLM.

The syntax for creating an using OnContextCondition is very similar to OnCondition, with the exception being the condition is a ContextExpression.

ContextExpression is a powerful expression evaluator that allows you to create complex expressions using context variable keys: - Variable references use ${var_name} syntax: ${logged_in}, ${attempts} - String literals can be used, e.g. ${customer_tier} == 'gold' - Numeric comparisons, e.g. ${budget} >= 50 - Supports a wide range of operators: - Logical: not, !, and, &, or, | - Comparison: >, <, >=, <=, ==, != - Parentheses can be used for grouping - Supports len(${var_name}): Gets the length of a list, string, or other collection - Examples: - not ${logged_in} and ${is_admin} or ${guest_checkout} - !${logged_in} & ${is_admin} | ${guest_checkout} - ${attempts} > 3 | ${is_admin} == True - len(${search_results}) > 5

register_hand_off(
    agent = agent_1,
    hand_to=[
        OnContextCondition(
            target=agent_2,
            condition=ContextExpression("(${account_level} > 2 and ${budget_remaining} > 0) or ${account_tier} == 'Gold' or len(${order_count}) > 10"),
            available="logged_in"),
        OnCondition(target=agent_3, condition="condition_2"), # LLM-based, evaluated after OnContextCondition's
    ]
)

Notes: - If you pass a string into the condition parameter of OnContextCondition it is assumed to be a context variable key and will be put into a ContextExpression. - OnContextCondition's are evaluated before OnCondition handoffs.

Enabling/Disabling Handoffs#

You can enable and disable OnCondition/OnContextCondition handoffs using their available parameter.

This can be a useful mechanism for ensuring handoffs are only evaluated when applicable, such as only when a user is authenticated.

The available parameter can take: | Type | Description | | --- | --- | | String | Will look up the value of the context variable with that name, which should be a bool, to determine whether it should include this condition. | | ContextExpression | Will evaluate the logical expression against the context variables (see previous section for examples) | | Callable | def my_available_func(agent: ConversableAgent, messages: list[Dict[str, Any]]) -> bool |

In the following example the handoff will only be available to agent_1 when the swarm context variable has_plan is True.

register_hand_off(
    agent = agent_1,
    hand_to=[
        OnCondition(
            target=agent_2,
            condition="Transfer to the reviewer to evaluate the plan.",
            available="has_plan"),
    ]
)

SwarmResult#

When tools are called, a SwarmResult can be returned and that can be used to specify the next agent to speak through the SwarmResult's agent parameter.

The agent property can be an agent object, an agent's name (string), an AfterWorkOption, or None. - If it is an agent object or agent name, that agent will be the next speaker. - If None it will return to the previous speaker. - If it is an AfterWorkOption, it will follow the rules noted in the previous section.

By using an AfterWorkOption you have additional flexibility, such as terminating the swarm at this point or transferring to the swarm's user agent.

Update Agent state before replying#

It can be useful to update an agent's state before they reply. For example, you can use an agent's context variables in their system message to keep it current with the state of the workflow.

When initialising an agent use the update_agent_state_before_reply parameter to register updates that run after the agent is selected, but before they reply.

update_agent_state_before_reply takes a list of any combination of the following (executing them in the provided order):

• UpdateSystemMessage provides a simple way to update the agent's system message via an f-string that substitutes the values of context variables, or a Callable that returns a string
• Callable with two parameters of type ConversableAgent for the agent and List[Dict[str Any]] for the messages, and does not return a value

Below is an example of these options.

# Creates a system message string
def create_system_prompt_function(my_agent: ConversableAgent, messages: List[Dict[]]) -> str:
 preferred_name = my_agent.get_context("preferred_name", "(name not provided)")

    # Note that the returned string will be treated like an f-string using the context variables
    return "You are a customer service representative helping a customer named "
 + preferred_name
 + " and their passport number is '{passport_number}'."

# Function to update an Agent's state
def my_callable_state_update_function(my_agent: ConversableAgent, messages: List[Dict[]]) -> None:
 agent.set_context("context_key", 43)
 agent.update_system_message("You are a customer service representative helping customer ID " + agent.get_context("context_key"))

# Create the AssistantAgent and set agent updates
customer_service = AssistantAgent(
    name="CustomerServiceRep",
    system_message="You are a customer service representative.",
    update_agent_state_before_reply=[
 UpdateSystemMessage("You are a customer service representative. Quote passport number '{passport_number}'"),
 UpdateSystemMessage(create_system_prompt_function),
 my_callable_state_update_function]
 ...
)

Initialize SwarmChat with initiate_swarm_chat / a_initiate_swarm_chat#

After a set of swarm agents is created, you can initiate a swarm chat by calling initiate_swarm_chat (or a_initiate_swarm_chat for an asynchronous version).

chat_history, context_variables, last_active_agent = initiate_swarm_chat(
    initial_agent=agent_1, # the first agent to start the chat
    agents=[agent_1, agent_2, agent_3], # a list of agents
    messages=[{"role": "user", "content": "Hello"}], # a list of messages to start the chat, you can also pass in one string
    user_agent=user_agent, # optional, if you want to use your own user agent
    context_variables={"key": "value"} # optional, initial context variables
)

How we handle the messages parameter: - Case 1: If you pass in one single message - If there is a name in that message, we will assume this message is from that agent. The name must match an agent in the swarm. - If there is no name: - 1. User agent passed in: we assume this message is from the user agent. - 2. No user agent passed in: we will create a temporary user agent just to start the chat. - Case 2: We will use the Resume GroupChat feature to resume the chat. The name fields in these messages must match the names of the agents you passed in.

AfterWork#

When the active agent's response doesn't suggest a tool call or handoff, the chat will terminate by default. However, you can register an AfterWork handoff to control what to do next. You can register these AfterWork handoffs at the agent level and also the swarm level (through the after_work parameter on initiate_swarm_chat). The agent level takes precedence over the swarm level.

The AfterWork takes a single parameter and this can be an agent, an agent's name, an AfterWorkOption, or a callable function.

The AfterWorkOption options are: - TERMINATE: Terminate the chat - STAY: Stay at the current agent - REVERT_TO_USER: Revert to the user agent. Only if a user agent is passed in when initializing. (See below for more details) - SWARM_MANAGER: Use the internal group chat's auto speaker selection method

The callable function signature is: def my_after_work_func(last_speaker: ConversableAgent, messages: List[Dict[str, Any]], groupchat: GroupChat) -> Union[AfterWorkOption, ConversableAgent, str]:

Note: there should only be one AfterWork, if your requirement is more complex, use a Callable as the parameter.

Here are examples of registering AfterWork handoffs:

# Register the handoff to an agent
register_handoff(
    agent=agent_1,
    hand_to=[
 OnCondition(...),
 AfterWork(agent_4) # Fallback to agent_4 if no OnCondition handoff is called
 ]
)

# Register the handoff to an AfterWorkOption
register_handoff(
    agent=agent_2,
    hand_to=[AfterWork(AfterWorkOption.TERMINATE)] # Terminate the chat if no handoff is suggested
)

def my_after_work_func(last_speaker: AssistantAgent, messages: List[Dict[str, Any]], groupchat: GroupChat) -> Union[AfterWorkOption, AssistantAgent, str]:
    if last_speaker.get_context("agent_1_done"):
        return agent_2
    else:
        return AfterWorkOption.TERMINATE

# Register the handoff to a function that will return an agent or AfterWorkOption
agent_3.handoff(hand_to=[AfterWork(my_after_work_func)])
register_handoff(agent_3, hand_to=[AfterWork(my_after_work_func)])

# Register the swarm level AfterWork that becomes the default for agents that don't have one specified
chat_history, context_variables, last_active_agent = initiate_swarm_chat(
 ...
    after_work=AfterWorkOption.TERMINATE # Or an agent or Callable
)

Q&As#

How are context variables updated?#

In a swarm, the context variables are shared amongst the swarm's agents. As context variables are available at the agent level, you can use the context variable getters/setters (get_context, set_context) on the agent to view and change the shared context variables. If you're working with a function that returns a SwarmResult you should update the passed-in context variables and return it in the SwarmResult to ensure the shared context is updated.

What is the difference between OnCondition and AfterWork?#

When registering an OnCondition handoff, we are creating a function schema to be passed to the LLM. The LLM will decide whether to call this function.

When registering an AfterWork handoff, we are defining the fallback mechanism when no tool calls are suggested. This is a higher level of control from the swarm chat level.

When to pass in a user agent?#

If your application requires interactions with the user, you can pass in a user agent using the user_agent parameter of initiate_swarm_chat. This means that you don't need to write an outer loop to accept user inputs and return to the swarm.