def register_hand_off(
agent: ConversableAgent,
hand_to: Union[list[Union[OnCondition, AfterWork]], OnCondition, AfterWork],
):
"""Register a function to hand off to another agent.
Args:
agent: The agent to register the hand off with.
hand_to: A list of OnCondition's and an, optional, AfterWork condition
Hand off template:
def transfer_to_agent_name() -> ConversableAgent:
return agent_name
1. register the function with the agent
2. register the schema with the agent, description set to the condition
"""
# If the agent hasn't been established as a swarm agent, do so first
if not hasattr(agent, "_swarm_is_established"):
_establish_swarm_agent(agent)
# Ensure that hand_to is a list or OnCondition or AfterWork
if not isinstance(hand_to, (list, OnCondition, AfterWork)):
raise ValueError("hand_to must be a list of OnCondition or AfterWork")
if isinstance(hand_to, (OnCondition, AfterWork)):
hand_to = [hand_to]
for transit in hand_to:
if isinstance(transit, AfterWork):
assert isinstance(transit.agent, (AfterWorkOption, ConversableAgent, str, Callable)), (
"Invalid After Work value"
)
agent._swarm_after_work = transit
agent._swarm_after_work_selection_msg = transit.next_agent_selection_msg
elif isinstance(transit, OnCondition):
if isinstance(transit.target, ConversableAgent):
# Transition to agent
# Create closure with current loop transit value
# to ensure the condition matches the one in the loop
def make_transfer_function(current_transit: OnCondition):
def transfer_to_agent() -> ConversableAgent:
return current_transit.target
return transfer_to_agent
transfer_func = make_transfer_function(transit)
# Store function to add/remove later based on it being 'available'
# Function names are made unique and allow multiple OnCondition's to the same agent
base_func_name = f"transfer_{agent.name}_to_{transit.target.name}"
func_name = base_func_name
count = 2
while func_name in agent._swarm_conditional_functions:
func_name = f"{base_func_name}_{count}"
count += 1
# Store function to add/remove later based on it being 'available'
agent._swarm_conditional_functions[func_name] = (transfer_func, transit)
elif isinstance(transit.target, dict):
# Transition to a nested chat
# We will store them here and establish them in the initiate_swarm_chat
agent._swarm_nested_chat_handoffs.append({
"nested_chats": transit.target,
"condition": transit.condition,
"available": transit.available,
})
else:
raise ValueError("Invalid hand off condition, must be either OnCondition or AfterWork")