@export_module("autogen")
def run_swarm(
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,
) -> RunResponseProtocol:
iostream = ThreadIOStream()
response = RunResponse(iostream, agents) # type: ignore[arg-type]
def stream_run(
iostream: ThreadIOStream = iostream,
response: RunResponse = response,
) -> None:
with IOStream.set_default(iostream):
try:
chat_result, returned_context_variables, last_speaker = initiate_swarm_chat(
initial_agent=initial_agent,
messages=messages,
agents=agents,
user_agent=user_agent,
swarm_manager_args=swarm_manager_args,
max_rounds=max_rounds,
context_variables=context_variables,
after_work=after_work,
exclude_transit_message=exclude_transit_message,
)
IOStream.get_default().send(
RunCompletionEvent( # type: ignore[call-arg]
history=chat_result.chat_history,
summary=chat_result.summary,
cost=chat_result.cost,
last_speaker=last_speaker.name,
context_variables=returned_context_variables,
)
)
except Exception as e:
response.iostream.send(ErrorEvent(error=e)) # type: ignore[call-arg]
threading.Thread(
target=stream_run,
).start()
return response