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 Data
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"]
trace_type = "tool"
beta = True
def get_flow_names(self) -> List[str]:
flow_datas = self.list_flows()
return [flow_data.data["name"] for flow_data in flow_datas]
def get_flow(self, flow_name: str) -> Optional[Data]:
"""
Retrieves a flow by its name.
Args:
flow_name (str): The name of the flow to retrieve.
Returns:
Optional[Text]: The flow record if found, None otherwise.
"""
flow_datas = self.list_flows()
for flow_data in flow_datas:
if flow_data.data["name"] == flow_name:
return flow_data
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_data = self.get_flow(flow_name)
if not flow_data:
raise ValueError("Flow not found.")
graph = Graph.from_payload(flow_data.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_data.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
1条答案
按热度按时间h22fl7wq1#
要解决在渲染上的Langflow中的"Flow as Tool"组件问题,请按照以下步骤操作:
get_flow_names
方法是否正确获取并返回流程名称列表。此方法应填充流程选择下拉框的选项。update_build_config
方法正确使用可用的流程名称更新构建配置。此方法应在需要刷新流程名称时调用。以下是一个代码片段,展示了如何更新构建配置和处理流程选择:
确保
get_flow_names
方法已正确实现,并且适当调用update_build_config
方法以刷新流程名称。此外,处理在流程刷新过程中可能发生的任何异常,以防止崩溃 1(https://github.com/langflow-ai/langflow/blob/dev/src/backend/base/langflow/components/experimental/FlowTool.py)2(https://github.com/langflow-ai/langflow/blob/dev/src/backend/base/langflow/components/prototypes/FlowTool.py)。对于在Langflow 1.0之前创建的组件,您可能需要向现有流程中添加一些额外的组件,以确保兼容性:
这应该有助于稳定组件并使它们与新版本的Langflow 3(https://github.com/logspace-ai/langflow/blob/dev/docs/docs/migration/compatibility.mdx)4(https://github.com/langflow-ai/langflow/blob/dev/docs/docs/migration/compatibility.mdx)兼容。