Skip to content

FunctionTarget

autogen.agentchat.group.targets.function_target.FunctionTarget #

FunctionTarget(incoming_fn, *, extra_args=None, **kwargs)

Bases: TransitionTarget

Transition target that invokes a tool function with (prev_output, context).

The function must return a FunctionTargetResult object that includes the next target to transition to.

Source code in autogen/agentchat/group/targets/function_target.py
def __init__(
    self,
    incoming_fn: Callable[..., FunctionTargetResult],
    *,
    extra_args: dict[str, Any] | None = None,
    **kwargs: Any,
) -> None:
    if callable(incoming_fn):
        extra_args = extra_args or {}

        validate_fn_sig(incoming_fn, extra_args)

        super().__init__(fn_name=incoming_fn.__name__, fn=incoming_fn, extra_args=extra_args, **kwargs)
    else:
        raise ValueError(
            "FunctionTarget must be initialized with a callable function as the first argument or 'fn' keyword argument."
        )

fn_name class-attribute instance-attribute #

fn_name = Field(...)

fn class-attribute instance-attribute #

fn = Field(..., repr=False)

extra_args class-attribute instance-attribute #

extra_args = Field(default_factory=dict)

can_resolve_for_speaker_selection #

can_resolve_for_speaker_selection()
Source code in autogen/agentchat/group/targets/function_target.py
def can_resolve_for_speaker_selection(self) -> bool:
    return False

resolve #

resolve(groupchat, current_agent, user_agent)

Invoke the function, update context variables (optional), broadcast messages (optional), and return the next target to transition to.

Source code in autogen/agentchat/group/targets/function_target.py
def resolve(
    self,
    groupchat: GroupChat,
    current_agent: ConversableAgent,
    user_agent: ConversableAgent | None,
) -> SpeakerSelectionResult:
    """Invoke the function, update context variables (optional), broadcast messages (optional), and return the next target to transition to."""
    last_message = (
        groupchat.messages[-1]["content"] if groupchat.messages and "content" in groupchat.messages[-1] else ""
    )

    # Run the function to get the FunctionTargetResult
    function_target_result = self.fn(
        last_message,
        current_agent.context_variables,
        **self.extra_args,
    )

    if not isinstance(function_target_result, FunctionTargetResult):
        raise ValueError("FunctionTarget function must return a FunctionTargetResult object.")

    if function_target_result.context_variables:
        # Update the group's Context Variables if the function returned any
        current_agent.context_variables.update(function_target_result.context_variables.to_dict())

    if function_target_result.messages:
        # If we have messages, we need to broadcast them to the appropriate agent based on the target
        broadcast(
            function_target_result.messages,
            groupchat,
            current_agent,
            self.fn_name,
            function_target_result.target,
            user_agent,
        )

    # Resolve and return the next target
    return function_target_result.target.resolve(groupchat, current_agent, user_agent)

display_name #

display_name()
Source code in autogen/agentchat/group/targets/function_target.py
def display_name(self) -> str:
    return self.fn_name

normalized_name #

normalized_name()
Source code in autogen/agentchat/group/targets/function_target.py
def normalized_name(self) -> str:
    return self.fn_name.replace(" ", "_")

needs_agent_wrapper #

needs_agent_wrapper()
Source code in autogen/agentchat/group/targets/function_target.py
def needs_agent_wrapper(self) -> bool:
    return False

create_wrapper_agent #

create_wrapper_agent(parent_agent, index)
Source code in autogen/agentchat/group/targets/function_target.py
def create_wrapper_agent(self, parent_agent: ConversableAgent, index: int) -> ConversableAgent:
    raise NotImplementedError("FunctionTarget is executed inline and needs no wrapper")

activate_target #

activate_target(groupchat)

Activate the target in the groupchat, setting the next target for GroupToolExecutor.

The Tool Executor's next target attribute will be picked up on the next iteration when _determine_next_agent is called

Source code in autogen/agentchat/group/targets/transition_target.py
def activate_target(self, groupchat: "GroupChat") -> None:
    """Activate the target in the groupchat, setting the next target for GroupToolExecutor.

    The Tool Executor's next target attribute will be picked up on the next iteration when _determine_next_agent is called
    """
    for agent in groupchat.agents:  # type: ignore[attr-defined]
        # get the GroupToolExecutor agent
        if type(agent).__name__ == "GroupToolExecutor":
            agent.set_next_target(self)  # type: ignore[attr-defined]
            return