a_initiate_swarm_chat

autogen.a_initiate_swarm_chat

async def a_initiate_swarm_chat(initial_agent, messages, agents, user_agent=None, swarm_manager_args=None, max_rounds=20, context_variables=None, after_work=TERMINATE, exclude_transit_message=True)

Initialize and run a swarm chat asynchronously

Source code in autogen/agentchat/contrib/swarm_agent.py

@export_module("autogen")
async def a_initiate_swarm_chat(
    initial_agent: ConversableAgent,
    messages: list[dict[str, Any]] | str,
    agents: list[ConversableAgent],
    user_agent: UserProxyAgent | None = None,
    swarm_manager_args: dict[str, Any] | None = None,
    max_rounds: int = 20,
    context_variables: ContextVariables | None = None,
    after_work: AfterWorkOption
    | Callable[[ConversableAgent, list[dict[str, Any]], GroupChat], AfterWorkOption | ConversableAgent | str]
    | None = AfterWorkOption.TERMINATE,
    exclude_transit_message: bool = True,
) -> tuple[ChatResult, ContextVariables, ConversableAgent]:
    """Initialize and run a swarm chat asynchronously
    Args:
        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.
        swarm_manager_args: Optional group chat manager arguments used to establish the swarm's groupchat manager, required when AfterWorkOption.SWARM_MANAGER is used.
        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 AfterWork instance (which is a dataclass accepting a ConversableAgent, 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, and groupchat as arguments and returns an AfterWorkOption or a ConversableAgent (by reference or string name).
                ```python
                def custom_afterwork_func(last_speaker: ConversableAgent, messages: list[dict[str, Any]], groupchat: GroupChat) -> Union[AfterWorkOption, ConversableAgent, str]:
                ```
        exclude_transit_message:  all registered handoff function call and responses messages will be removed from message list before calling an LLM.
            Note: only with transition functions added with `register_handoff` will be removed. If you pass in a function to manage workflow, it will not be removed. You may register a customized hook to `process_all_messages_before_reply` to remove that.
    Returns:
        ChatResult:     Conversations chat history.
        ContextVariables: Updated Context variables.
        ConversableAgent:     Last speaker.
    """
    context_variables = context_variables or ContextVariables()
    tool_execution, nested_chat_agents = _prepare_swarm_agents(
        initial_agent, agents, context_variables, exclude_transit_message
    )
    processed_messages, last_agent, swarm_agent_names, temp_user_list = _process_initial_messages(
        messages, user_agent, agents, nested_chat_agents
    )
    # Create transition function (has enclosed state for initial agent)
    swarm_transition = create_swarm_transition(
        initial_agent=initial_agent,
        tool_execution=tool_execution,
        swarm_agent_names=swarm_agent_names,
        user_agent=user_agent,
        swarm_after_work=after_work,
    )
    groupchat = GroupChat(
        agents=[tool_execution] + agents + nested_chat_agents + ([user_agent] if user_agent else temp_user_list),
        messages=[],
        max_round=max_rounds,
        speaker_selection_method=swarm_transition,
    )
    manager = _create_swarm_manager(groupchat, swarm_manager_args, agents)
    # Point all ConversableAgent's context variables to this function's context_variables
    _setup_context_variables(tool_execution, agents, manager, context_variables)
    # Link all agents with the GroupChatManager to allow access to the group chat
    # and other agents, particularly the tool executor for setting _swarm_next_agent
    _link_agents_to_swarm_manager(groupchat.agents, manager)
    if len(processed_messages) > 1:
        last_agent, last_message = await manager.a_resume(messages=processed_messages)
        clear_history = False
    else:
        last_message = processed_messages[0]
        clear_history = True
    if last_agent is None:
        raise ValueError("No agent selected to start the conversation")
    chat_result = await last_agent.a_initiate_chat(  # type: ignore[attr-defined]
        manager,
        message=last_message,
        clear_history=clear_history,
    )
    _cleanup_temp_user_messages(chat_result)
    return chat_result, context_variables, manager.last_speaker  # type: ignore[return-value]