Documentation to use Flow as Tool

[sepa85]

I'm trying to use the experimental Flow as Tool component (class FlowToolComponent). I have the Flow A with an agent with tools, connected to the experimental Flow as Tool. This tool uses the Flow B, which have a text input and a text output.

With these configuration, I get the error:
Error building Component Tool Calling Agent: 'message'

If I use chat input and chat output in Flow B, I get the same error:
Error building Component Tool Calling Agent: 'message'

If I use the experimental "Message" component as Flow B output, I got no error, but the response from the tool is empty. I used Machine as sender type and tool as sender name.

How must I build the Flow used as tool? Which are the correct input and output component type? Is there documentation on Flow as Tool usage, maybe with examples?

[dosubot]

Hello @sepa85! I'm here to help you with any questions or issues you have. I'm a bot that can assist you with bug fixes, answer questions, and guide you on how to become a contributor. How can I assist you today?

To build the FlowTool used as a tool, you need to define a class that inherits from BaseTool and implement the required methods and properties. The correct input and output component types are Vertex for inputs and a formatted string for outputs.

Here is the relevant code snippet:

from typing import Any, List, Optional, Type
from langchain.tools import BaseTool
from langflow.graph.vertex.base import Vertex
from langflow.helpers.flow import build_schema_from_inputs, get_arg_names, get_flow_inputs, run_flow
from langflow.base.flow_processing.utils import build_records_from_result_data, format_flow_output_records
from langchain_core.tools import ToolException
from pydantic.v1 import BaseModel

class FlowTool(BaseTool):
    name: str
    description: str
    graph: Optional[Graph] = None
    flow_id: Optional[str] = None
    user_id: Optional[str] = None
    inputs: List["Vertex"] = []
    get_final_results_only: bool = True

    @property
    def args(self) -> dict:
        schema = self.get_input_schema()
        return schema.schema()["properties"]

    def get_input_schema(self, config: Optional[RunnableConfig] = None) -> Type[BaseModel]:
        """The tool's input schema."""
        if self.args_schema is not None:
            return self.args_schema
        elif self.graph is not None:
            return build_schema_from_inputs(self.name, get_flow_inputs(self.graph))
        else:
            raise ToolException("No input schema available.")

    def _run(
        self,
        *args: Any,
        **kwargs: Any,
    ) -> str:
        """Use the tool."""
        args_names = get_arg_names(self.inputs)
        if len(args_names) == len(args):
            kwargs = {arg["arg_name"]: arg_value for arg, arg_value in zip(args_names, args)}
        elif len(args_names) != len(args) and len(args) != 0:
            raise ToolException(
                "Number of arguments does not match the number of inputs. Pass keyword arguments instead."
            )
        tweaks = {arg["component_name"]: kwargs[arg["arg_name"]] for arg in args_names}

        run_outputs = syncify(run_flow, raise_sync_error=False)(
            tweaks={key: {"input_value": value} for key, value in tweaks.items()},
            flow_id=self.flow_id,
            user_id=self.user_id,
        )
        if not run_outputs:
            return "No output"
        run_output = run_outputs[0]

        records = []
        if run_output is not None:
            for output in run_output.outputs:
                if output:
                    records.extend(
                        build_records_from_result_data(output, get_final_results_only=self.get_final_results_only)
                    )
        return format_flow_output_records(records)

Additionally, the FlowToolComponent class helps in constructing the FlowTool by providing methods to retrieve flow names, get flow details, and build the tool configuration. Here is the relevant code snippet:

from typing import Any, List, Optional
from loguru import logger
from langflow.base.tools.flow_tool import FlowTool
from langflow.custom import CustomComponent
from langflow.field_typing import Tool
from langflow.graph.graph.base import Graph
from langflow.helpers.flow import get_flow_inputs
from langflow.schema import Record
from langflow.schema.dotdict import dotdict

class FlowToolComponent(CustomComponent):
    display_name = "Flow as Tool"
    description = "Construct a Tool from a function that runs the loaded Flow."
    field_order = ["flow_name", "name", "description", "return_direct"]

    def get_flow_names(self) -> List[str]:
        flow_records = self.list_flows()
        return [flow_record.data["name"] for flow_record in flow_records]

    def get_flow(self, flow_name: str) -> Optional[Record]:
        flow_records = self.list_flows()
        for flow_record in flow_records:
            if flow_record.data["name"] == flow_name:
                return flow_record
        return None

    def update_build_config(self, build_config: dotdict, field_value: Any, field_name: str | None = None):
        logger.debug(f"Updating build config with field value {field_value} and field name {field_name}")
        if field_name == "flow_name":
            build_config["flow_name"]["options"] = self.get_flow_names()

        return build_config

    def build_config(self):
        return {
            "flow_name": {
                "display_name": "Flow Name",
                "info": "The name of the flow to run.",
                "options": [],
                "real_time_refresh": True,
                "refresh_button": True,
            },
            "name": {
                "display_name": "Name",
                "description": "The name of the tool.",
            },
            "description": {
                "display_name": "Description",
                "description": "The description of the tool.",
            },
            "return_direct": {
                "display_name": "Return Direct",
                "description": "Return the result directly from the Tool.",
                "advanced": True,
            },
        }

    async def build(self, flow_name: str, name: str, description: str, return_direct: bool = False) -> Tool:
        FlowTool.update_forward_refs()
        flow_record = self.get_flow(flow_name)
        if not flow_record:
            raise ValueError("Flow not found.")
        graph = Graph.from_payload(flow_record.data["data"])
        inputs = get_flow_inputs(graph)
        tool = FlowTool(
            name=name,
            description=description,
            graph=graph,
            return_direct=return_direct,
            inputs=inputs,
            flow_id=str(flow_record.id),
            user_id=str(self._user_id),
        )
        description_repr = repr(tool.description).strip("'")
        args_str = "\n".join([f"- {arg_name}: {arg_data['description']}" for arg_name, arg_data in tool.args.items()])
        self.status = f"{description_repr}\nArguments:\n{args_str}"
        return tool  # type: ignore

In summary:

• Input Component Type: Vertex
• Output Component Type: Formatted string (result of format_flow_output_records(records)) ^[1][2][3].

To continue talking to Dosu, mention @dosu.

[sepa85]

What do you mean with Vertex, @dosu? The only components with Vertex in langflow are related to Google vertex AI. I just need to pass a string as input for Flow B, and get the string from the output of the Flow B.

[dosubot]

To correctly configure the input and output components for Flow B in Langflow to pass a string as input and get a string as output when using the experimental Flow as Tool component, you need to ensure that the flow is set up with the appropriate input and output types. Here is how you can do it:

1. Define the Flow with String Input and Output:
   Ensure that Flow B is designed to accept a string as input and produce a string as output. This involves setting up the nodes in the flow to handle string data types.

2. Configure the Flow as Tool Component:
   When building the tool, you need to specify the flow name and ensure that the inputs and outputs are correctly mapped. The build method in the FlowToolComponent class handles this configuration.

Here is an example of how you might configure the Flow as Tool component:

async def build(self, flow_name: str, name: str, description: str, return_direct: bool = False) -> Tool:
    FlowTool.update_forward_refs()
    flow_record = self.get_flow(flow_name)
    if not flow_record:
        raise ValueError("Flow not found.")
    graph = Graph.from_payload(flow_record.data["data"])
    inputs = get_flow_inputs(graph)
    
    # Ensure the inputs are set to accept a string
    if 'input_string' not in inputs:
        raise ValueError("Flow does not accept a string input named 'input_string'.")
    
    tool = FlowTool(
        name=name,
        description=description,
        graph=graph,
        return_direct=return_direct,
        inputs=inputs,
        flow_id=str(flow_record.id),
        user_id=str(self._user_id),
    )
    description_repr = repr(tool.description).strip("'")
    args_str = "\n".join([f"- {arg_name}: {arg_data['description']}" for arg_name, arg_data in tool.args.items()])
    self.status = f"{description_repr}\nArguments:\n{args_str}"
    return tool  # type: ignore

In this example, the inputs dictionary should contain an entry for input_string to ensure that the flow is configured to accept a string input. The FlowTool is then created with these inputs, and the tool is returned.

This setup ensures that Flow B is correctly configured to accept a string as input and produce a string as output when used with the Flow as Tool component ^[1].

To continue talking to Dosu, mention @dosu.