Tool Node

graphorchestrator.nodes.nodes.ToolNode

Bases: ProcessingNode

A node that represents a tool.

This node is a specialized ProcessingNode that wraps a tool method.

Source code in graphorchestrator\nodes\nodes.py

class ToolNode(ProcessingNode):
    """
    A node that represents a tool.

    This node is a specialized ProcessingNode that wraps a tool method.
    """

    def __init__(
        self,
        node_id: str,
        description: Optional[str],
        tool_method: Callable[[State], State],
    ) -> None:
        if not getattr(tool_method, "is_tool_method", False):
            raise ToolMethodNotDecorated(tool_method)
        if not (description or (tool_method.__doc__ or "").strip()):
            raise EmptyToolNodeDescriptionError(tool_method)

        super().__init__(node_id, tool_method)
        self.description = description

        GraphLogger.get().info(
            **wrap_constants(
                message="ToolNode created",
                **{
                    LC.EVENT_TYPE: "tool",
                    LC.NODE_ID: self.node_id,
                    LC.NODE_TYPE: "ToolNode",
                    LC.ACTION: "node_created",
                    LC.CUSTOM: {
                        "function": tool_method.__name__,
                        "has_description": bool(description),
                    },
                },
            )
        )

    async def execute(self, state: State) -> State:
        """
        Executes the tool method.

        Args:
            state (State): The input state for the node.

        Returns:
            State: The state after executing the tool method.
        """
        log = GraphLogger.get()

        log.info(
            **wrap_constants(
                message="ToolNode execution started",
                **{
                    LC.EVENT_TYPE: "tool",
                    LC.NODE_ID: self.node_id,
                    LC.NODE_TYPE: "ToolNode",
                    LC.ACTION: "execute_start",
                    LC.INPUT_SIZE: len(state.messages),
                },
            )
        )

        result = (
            await self.func(state)
            if asyncio.iscoroutinefunction(self.func)
            else self.func(state)
        )

        log.info(
            **wrap_constants(
                message="ToolNode execution completed",
                **{
                    LC.EVENT_TYPE: "tool",
                    LC.NODE_ID: self.node_id,
                    LC.NODE_TYPE: "ToolNode",
                    LC.ACTION: "execute_end",
                    LC.OUTPUT_SIZE: len(result.messages),
                },
            )
        )

        return result

execute(state) async

Executes the tool method.

Parameters:

Name	Type	Description	Default
state	State	The input state for the node.	required

Returns:

Name	Type	Description
State	State	The state after executing the tool method.

Source code in graphorchestrator\nodes\nodes.py

async def execute(self, state: State) -> State:
    """
    Executes the tool method.

    Args:
        state (State): The input state for the node.

    Returns:
        State: The state after executing the tool method.
    """
    log = GraphLogger.get()

    log.info(
        **wrap_constants(
            message="ToolNode execution started",
            **{
                LC.EVENT_TYPE: "tool",
                LC.NODE_ID: self.node_id,
                LC.NODE_TYPE: "ToolNode",
                LC.ACTION: "execute_start",
                LC.INPUT_SIZE: len(state.messages),
            },
        )
    )

    result = (
        await self.func(state)
        if asyncio.iscoroutinefunction(self.func)
        else self.func(state)
    )

    log.info(
        **wrap_constants(
            message="ToolNode execution completed",
            **{
                LC.EVENT_TYPE: "tool",
                LC.NODE_ID: self.node_id,
                LC.NODE_TYPE: "ToolNode",
                LC.ACTION: "execute_end",
                LC.OUTPUT_SIZE: len(result.messages),
            },
        )
    )

    return result