autogen.a_initiate_swarm_chat

a_initiate_swarm_chat

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: dict[str, Any] | None = None,
    after_work: AfterWorkOption | Callable[..., Any] | None = AfterWork(agent=<AfterWorkOption.TERMINATE: 'TERMINATE'>, next_agent_selection_msg=None),
    exclude_transit_message: bool = True
) -> tuple[ChatResult, dict[str, Any], ConversableAgent]

Initialize and run a swarm chat asynchronously

Parameters:

Name	Description
initial_agent	The first receiving agent of the conversation. Type: ConversableAgent
messages	Initial message(s). Type: list[dict[str, typing.Any]] | str
agents	List of swarm agents. Type: list[ConversableAgent]
user_agent	Optional user proxy agent for falling back to. Type: UserProxyAgent | None 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: dict[str, typing.Any] | None Default: None
max_rounds	Maximum number of conversation rounds. Type: int Default: 20
context_variables	Starting context variables. Type: dict[str, typing.Any] | None 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 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). def custom_afterwork_func(last_speaker: ConversableAgent, messages: list[Dict[str, Any]], groupchat: GroupChat) -> Union[AfterWorkOption, ConversableAgent, str]: Type: AfterWorkOption | Callable[…, Any] | None = AfterWork(agent Default: <AfterWorkOption.TERMINATE
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 cumstomized hook to process_all_messages_before_reply to remove that. Type: bool Default: True

Returns:

Type	Description
tuple[ChatResult, dict[str, typing.Any], ConversableAgent]	ChatResult: Conversations chat history. Dict[str, Any]: Updated Context variables. ConversableAgent: Last speaker.