Refactor run_graph function to add support for running a JSON flow

This commit is contained in:
Gabriel Luiz Freitas Almeida 2024-04-01 16:18:02 -03:00
commit bb19581750
3 changed files with 93 additions and 7 deletions

View file

@ -23,7 +23,7 @@ from langflow.graph.schema import RunOutputs
from langflow.interface.custom.custom_component import CustomComponent from langflow.interface.custom.custom_component import CustomComponent
from langflow.interface.custom.directory_reader import DirectoryReader from langflow.interface.custom.directory_reader import DirectoryReader
from langflow.interface.custom.utils import build_custom_component_template 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.auth.utils import api_key_security, get_current_active_user
from langflow.services.cache.utils import save_uploaded_file from langflow.services.cache.utils import save_uploaded_file
from langflow.services.database.models.flow import Flow 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()) 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, graph=graph,
flow_id=flow_id, flow_id=flow_id,
session_id=input_request.session_id, session_id=input_request.session_id,
@ -255,7 +255,7 @@ async def experimental_run_flow(
graph_data = flow.data graph_data = flow.data
graph_data = process_tweaks(graph_data, tweaks or {}) graph_data = process_tweaks(graph_data, tweaks or {})
graph = Graph.from_payload(graph_data, flow_id=flow_id) 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, graph=graph,
flow_id=flow_id, flow_id=flow_id,
session_id=session_id, session_id=session_id,

View file

@ -1,9 +1,10 @@
import json import json
from pathlib import Path from pathlib import Path
from typing import Optional, Union from typing import List, Optional, Union
from langflow.graph import Graph 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: 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) graph = Graph.from_payload(graph_data)
return graph 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

View file

@ -124,10 +124,10 @@ class Result(BaseModel):
session_id: str session_id: str
async def run_graph( async def run_graph_internal(
graph: "Graph", graph: "Graph",
flow_id: str, flow_id: str,
stream: bool, stream: bool = False,
session_id: Optional[str] = None, session_id: Optional[str] = None,
inputs: Optional[List["InputValueRequest"]] = None, inputs: Optional[List["InputValueRequest"]] = None,
outputs: Optional[List[str]] = None, outputs: Optional[List[str]] = None,
@ -167,6 +167,58 @@ async def run_graph(
return run_outputs, session_id_str 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( def validate_input(
graph_data: Dict[str, Any], tweaks: Union["Tweaks", Dict[str, Dict[str, Any]]] graph_data: Dict[str, Any], tweaks: Union["Tweaks", Dict[str, Dict[str, Any]]]
) -> List[Dict[str, Any]]: ) -> List[Dict[str, Any]]: