a_initiate_swarm_chat - AG2

a_initiate_swarm_chat

autogen.a_initiate_swarm_chat async

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

PARAMETER DESCRIPTION
initial_agent The first receiving agent of the conversation.
TYPE: ConversableAgent
messages Initial message(s).
TYPE: Union[list[dict[str, Any]], str]
agents List of swarm agents.
TYPE: list[ConversableAgent]
user_agent Optional user proxy agent for falling back to.
TYPE: Optional[UserProxyAgent] DEFAULT: None
swarm_manager_args Optional group chat manager arguments used to establish the swarm's groupchat manager, required when AfterWorkOption.SWARM_MANAGER is used.
TYPE: Optional[dict[str, Any]] DEFAULT: None
max_rounds Maximum number of conversation rounds.
TYPE: int DEFAULT: 20
context_variables Starting context variables.
TYPE: Optional[dict[str, Any]] DEFAULT: None
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 an AfterWork instance or a callable.
TYPE: Optional[Union[AfterWorkOption, Callable[[ConversableAgent, list[dict[str, Any]], GroupChat], Union[AfterWorkOption, ConversableAgent, str]]]] DEFAULT: TERMINATE
exclude_transit_message Indicates whether all registered handoff function call and response messages should be removed from the message list before calling an LLM.
TYPE: bool DEFAULT: True

Returns: ChatResult: Conversations chat history. dict[str, Any]: Updated Context variables. ConversableAgent: Last speaker.

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

@export_module("autogen")
async def a_initiate_swarm_chat(
    initial_agent: ConversableAgent,
    messages: Union[list[dict[str, Any]], str],
    agents: list[ConversableAgent],
    user_agent: Optional[UserProxyAgent] = None,
    swarm_manager_args: Optional[dict[str, Any]] = None,
    max_rounds: int = 20,
    context_variables: Optional[dict[str, Any]] = None,
    after_work: Optional[
        Union[
            AfterWorkOption,
            Callable[
                [ConversableAgent, list[dict[str, Any]], GroupChat], Union[AfterWorkOption, ConversableAgent, str]
            ],
        ]
    ] = AfterWorkOption.TERMINATE,
    exclude_transit_message: bool = True,
) -> tuple[ChatResult, dict[str, Any], ConversableAgent]:
    """Initialize and run a swarm chat asynchronously
    ... (documentation continues here) ...
    """
    context_variables = context_variables or {}
    ... (function logic continues here) ...
    return chat_result, context_variables if context_variables != {} else None, manager.last_speaker