def register_hand_off(
agent: ConversableAgent,
hand_to: Union[list[Union[OnCondition, OnContextCondition, AfterWork]], OnCondition, OnContextCondition, AfterWork],
) -> None:
"""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, OnContextCondition, AfterWork)):
raise ValueError("hand_to must be a list of OnCondition, OnContextCondition, or AfterWork")
if isinstance(hand_to, (OnCondition, OnContextCondition, AfterWork)):
hand_to = [hand_to]
for transit in hand_to:
if isinstance(transit, AfterWork):
if not (isinstance(transit.agent, (AfterWorkOption, ConversableAgent, str)) or callable(transit.agent)):
raise ValueError(f"Invalid AfterWork agent: {transit.agent}")
agent._swarm_after_work = transit # type: ignore[attr-defined]
agent._swarm_after_work_selection_msg = transit.next_agent_selection_msg # type: ignore[attr-defined]
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) -> Callable[[], ConversableAgent]:
def transfer_to_agent() -> ConversableAgent:
return current_transit.target # type: ignore[return-value]
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: # type: ignore[attr-defined]
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) # type: ignore[attr-defined]
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({ # type: ignore[attr-defined]
"nested_chats": transit.target,
"condition": transit.condition,
"available": transit.available,
})
elif isinstance(transit, OnContextCondition):
agent._swarm_oncontextconditions.append(transit) # type: ignore[attr-defined]
else:
raise ValueError("Invalid hand off condition, must be either OnCondition or AfterWork")