diff --git a/src/backend/base/langflow/api/v1/endpoints.py b/src/backend/base/langflow/api/v1/endpoints.py index bc6604089..cdcf40443 100644 --- a/src/backend/base/langflow/api/v1/endpoints.py +++ b/src/backend/base/langflow/api/v1/endpoints.py @@ -23,7 +23,7 @@ from langflow.graph.schema import RunOutputs from langflow.interface.custom.custom_component import CustomComponent from langflow.interface.custom.directory_reader import DirectoryReader from langflow.interface.custom.utils import build_custom_component_template -from langflow.processing.process import process_tweaks, run_graph +from langflow.processing.process import process_tweaks, run_graph_internal from langflow.services.auth.utils import api_key_security, get_current_active_user from langflow.services.cache.utils import save_uploaded_file from langflow.services.database.models.flow import Flow @@ -149,7 +149,7 @@ async def simplified_run_flow( and (input_request.output_type == "any" or input_request.output_type in vertex.id.lower()) ) ] - task_result, session_id = await run_graph( + task_result, session_id = await run_graph_internal( graph=graph, flow_id=flow_id, session_id=input_request.session_id, @@ -255,7 +255,7 @@ async def experimental_run_flow( graph_data = flow.data graph_data = process_tweaks(graph_data, tweaks or {}) graph = Graph.from_payload(graph_data, flow_id=flow_id) - task_result, session_id = await run_graph( + task_result, session_id = await run_graph_internal( graph=graph, flow_id=flow_id, session_id=session_id, diff --git a/src/backend/base/langflow/processing/load.py b/src/backend/base/langflow/processing/load.py index 562a37d08..43d9ecc78 100644 --- a/src/backend/base/langflow/processing/load.py +++ b/src/backend/base/langflow/processing/load.py @@ -1,9 +1,10 @@ import json from pathlib import Path -from typing import Optional, Union +from typing import List, Optional, Union from langflow.graph import Graph -from langflow.processing.process import process_tweaks +from langflow.graph.schema import RunOutputs +from langflow.processing.process import process_tweaks, run_graph def load_flow_from_json(flow: Union[Path, str, dict], tweaks: Optional[dict] = None) -> Graph: @@ -31,3 +32,36 @@ def load_flow_from_json(flow: Union[Path, str, dict], tweaks: Optional[dict] = N graph = Graph.from_payload(graph_data) return graph + + +def run_flow_from_json( + flow: Union[Path, str, dict], + input_value: str, + tweaks: Optional[dict] = None, + input_type: str = "chat", + output_type: str = "chat", + output_component: Optional[str] = None, +) -> List[RunOutputs]: + """ + Runs a JSON flow by loading it from a file or dictionary and executing it with the given input value. + + Args: + flow (Union[Path, str, dict]): The path to the JSON file, or the JSON dictionary representing the flow. + input_value (str): The input value to be processed by the flow. + tweaks (Optional[dict], optional): Optional tweaks to be applied to the flow. Defaults to None. + input_type (str, optional): The type of the input value. Defaults to "chat". + output_type (str, optional): The type of the output value. Defaults to "chat". + output_component (Optional[str], optional): The specific output component to retrieve. Defaults to None. + + Returns: + None: The result of running the flow. + """ + graph = load_flow_from_json(flow, tweaks) + result = run_graph( + graph=graph, + input_value=input_value, + input_type=input_type, + output_type=output_type, + output_component=output_component, + ) + return result diff --git a/src/backend/base/langflow/processing/process.py b/src/backend/base/langflow/processing/process.py index c7e496dd7..c05163933 100644 --- a/src/backend/base/langflow/processing/process.py +++ b/src/backend/base/langflow/processing/process.py @@ -124,10 +124,10 @@ class Result(BaseModel): session_id: str -async def run_graph( +async def run_graph_internal( graph: "Graph", flow_id: str, - stream: bool, + stream: bool = False, session_id: Optional[str] = None, inputs: Optional[List["InputValueRequest"]] = None, outputs: Optional[List[str]] = None, @@ -167,6 +167,58 @@ async def run_graph( return run_outputs, session_id_str +def run_graph( + graph: "Graph", + input_value: str, + input_type: str, + output_type: str, + output_component: Optional[str] = None, +) -> List[RunOutputs]: + """ + Runs the given Langflow Graph with the specified input and returns the outputs. + + Args: + graph (Graph): The graph to be executed. + input_value (str): The input value to be passed to the graph. + input_type (str): The type of the input value. + output_type (str): The type of the desired output. + output_component (Optional[str], optional): The specific output component to retrieve. Defaults to None. + + Returns: + List[RunOutputs]: A list of RunOutputs objects representing the outputs of the graph. + + """ + inputs = [InputValueRequest(components=[], input_value=input_value, type=input_type)] + if output_component: + outputs = [output_component] + else: + outputs = [ + vertex.id + for vertex in graph.vertices + if output_type == "debug" + or (vertex.is_output and (output_type == "any" or output_type in vertex.id.lower())) + ] + components = [] + inputs_list = [] + types = [] + for input_value_request in inputs: + if input_value_request.input_value is None: + logger.warning("InputValueRequest input_value cannot be None, defaulting to an empty string.") + input_value_request.input_value = "" + components.append(input_value_request.components or []) + inputs_list.append({INPUT_FIELD_NAME: input_value_request.input_value}) + types.append(input_value_request.type) + run_outputs = graph.run( + inputs_list, + components, + types, + outputs or [], + stream=False, + session_id="", + ) + return run_outputs + + def validate_input( graph_data: Dict[str, Any], tweaks: Union["Tweaks", Dict[str, Dict[str, Any]]] ) -> List[Dict[str, Any]]: