From 20810988b0bde2d3cf7874b76221c68a9f4c1a19 Mon Sep 17 00:00:00 2001 From: anovazzi1 Date: Tue, 27 Feb 2024 15:59:42 -0300 Subject: [PATCH 01/28] Add updateFlowPool function to FlowStoreType --- src/frontend/src/stores/flowStore.ts | 16 +++++++++++++++- src/frontend/src/types/zustand/flow/index.ts | 1 + 2 files changed, 16 insertions(+), 1 deletion(-) diff --git a/src/frontend/src/stores/flowStore.ts b/src/frontend/src/stores/flowStore.ts index 300378406..7f0f88a5f 100644 --- a/src/frontend/src/stores/flowStore.ts +++ b/src/frontend/src/stores/flowStore.ts @@ -19,7 +19,7 @@ import { sourceHandleType, targetHandleType, } from "../types/flow"; -import { FlowStoreType } from "../types/zustand/flow"; +import { FlowPoolObjectType, FlowStoreType } from "../types/zustand/flow"; import { buildVertices } from "../utils/buildUtils"; import { cleanEdges, @@ -59,6 +59,20 @@ const useFlowStore = create((set, get) => ({ } get().setFlowPool(newFlowPool); }, + updateFlowPool:(nodeId:string,data:FlowPoolObjectType,buildId?:string)=>{ + let newFlowPool = cloneDeep({ ...get().flowPool }); + if (!newFlowPool[nodeId]){ + return; + } + else { + let index = newFlowPool[nodeId].length-1; + if(buildId){ + index = newFlowPool[nodeId].findIndex((flow)=>flow.id===buildId); + } + newFlowPool[nodeId][index] = data; + } + get().setFlowPool(newFlowPool); + }, CleanFlowPool: () => { get().setFlowPool({}); }, diff --git a/src/frontend/src/types/zustand/flow/index.ts b/src/frontend/src/types/zustand/flow/index.ts index cf6e786e0..01dbbf009 100644 --- a/src/frontend/src/types/zustand/flow/index.ts +++ b/src/frontend/src/types/zustand/flow/index.ts @@ -92,4 +92,5 @@ export type FlowStoreType = { updateBuildStatus: (nodeId: string[], status: BuildStatus) => void; revertBuiltStatusFromBuilding: () => void; flowBuildStatus: { [key: string]: BuildStatus }; + updateFlowPool: (nodeId:string, data:FlowPoolObjectType,buildId?:string) => void; }; From 2add242a187713c3e3e802e86b0c12e7c6dc9802 Mon Sep 17 00:00:00 2001 From: anovazzi1 Date: Tue, 27 Feb 2024 16:19:37 -0300 Subject: [PATCH 02/28] Update FlowStoreType to include ChatOutputType and chatInputType --- src/frontend/src/stores/flowStore.ts | 14 +++++++++++--- src/frontend/src/types/zustand/flow/index.ts | 4 ++-- 2 files changed, 13 insertions(+), 5 deletions(-) diff --git a/src/frontend/src/stores/flowStore.ts b/src/frontend/src/stores/flowStore.ts index 7f0f88a5f..351997aaf 100644 --- a/src/frontend/src/stores/flowStore.ts +++ b/src/frontend/src/stores/flowStore.ts @@ -19,7 +19,7 @@ import { sourceHandleType, targetHandleType, } from "../types/flow"; -import { FlowPoolObjectType, FlowStoreType } from "../types/zustand/flow"; +import { ChatOutputType, FlowPoolObjectType, FlowStoreType, chatInputType } from "../types/zustand/flow"; import { buildVertices } from "../utils/buildUtils"; import { cleanEdges, @@ -59,7 +59,7 @@ const useFlowStore = create((set, get) => ({ } get().setFlowPool(newFlowPool); }, - updateFlowPool:(nodeId:string,data:FlowPoolObjectType,buildId?:string)=>{ + updateFlowPool:(nodeId:string,data:FlowPoolObjectType| ChatOutputType | chatInputType,buildId?:string)=>{ let newFlowPool = cloneDeep({ ...get().flowPool }); if (!newFlowPool[nodeId]){ return; @@ -69,7 +69,15 @@ const useFlowStore = create((set, get) => ({ if(buildId){ index = newFlowPool[nodeId].findIndex((flow)=>flow.id===buildId); } - newFlowPool[nodeId][index] = data; + //check if the data is a flowpool object + if((data as FlowPoolObjectType).data?.artifacts!==undefined){ + newFlowPool[nodeId][index] = (data as FlowPoolObjectType); + } + //update data artifact + else + { + newFlowPool[nodeId][index].data.artifacts = data; + } } get().setFlowPool(newFlowPool); }, diff --git a/src/frontend/src/types/zustand/flow/index.ts b/src/frontend/src/types/zustand/flow/index.ts index 01dbbf009..8b6093b51 100644 --- a/src/frontend/src/types/zustand/flow/index.ts +++ b/src/frontend/src/types/zustand/flow/index.ts @@ -24,7 +24,7 @@ export type FlowPoolObjectType = { timestamp: string; valid: boolean; params: any; - data: { artifacts: any; results: any | ChatOutputType | chatInputType }; + data: { artifacts: any | ChatOutputType | chatInputType; results: any | ChatOutputType | chatInputType }; duration: string; progress: number; id: string; @@ -92,5 +92,5 @@ export type FlowStoreType = { updateBuildStatus: (nodeId: string[], status: BuildStatus) => void; revertBuiltStatusFromBuilding: () => void; flowBuildStatus: { [key: string]: BuildStatus }; - updateFlowPool: (nodeId:string, data:FlowPoolObjectType,buildId?:string) => void; + updateFlowPool: (nodeId:string, data:FlowPoolObjectType | ChatOutputType | chatInputType,buildId?:string) => void; }; From 4428cb913c201f91b514e49a802866ed0e5df43a Mon Sep 17 00:00:00 2001 From: Gabriel Luiz Freitas Almeida Date: Tue, 27 Feb 2024 16:36:13 -0300 Subject: [PATCH 03/28] Update base classes to support subclassing CustomComponent --- .../custom/code_parser/code_parser.py | 70 ++++++++++++++++--- 1 file changed, 59 insertions(+), 11 deletions(-) diff --git a/src/backend/langflow/interface/custom/code_parser/code_parser.py b/src/backend/langflow/interface/custom/code_parser/code_parser.py index 7a102e33f..e54051a5c 100644 --- a/src/backend/langflow/interface/custom/code_parser/code_parser.py +++ b/src/backend/langflow/interface/custom/code_parser/code_parser.py @@ -6,6 +6,9 @@ from typing import Any, Dict, List, Type, Union from cachetools import TTLCache, cachedmethod, keys from fastapi import HTTPException +from loguru import logger + +from langflow.interface.custom.eval import eval_custom_component_code from langflow.interface.custom.schema import CallableCodeDetails, ClassCodeDetails @@ -92,7 +95,9 @@ class CodeParser: elif isinstance(node, ast.ImportFrom): for alias in node.names: if alias.asname: - self.data["imports"].append((node.module, f"{alias.name} as {alias.asname}")) + self.data["imports"].append( + (node.module, f"{alias.name} as {alias.asname}") + ) else: self.data["imports"].append((node.module, alias.name)) @@ -141,7 +146,9 @@ class CodeParser: return_type = None if node.returns: return_type_str = ast.unparse(node.returns) - eval_env = self.construct_eval_env(return_type_str, tuple(self.data["imports"])) + eval_env = self.construct_eval_env( + return_type_str, tuple(self.data["imports"]) + ) try: return_type = eval(return_type_str, eval_env) @@ -183,14 +190,22 @@ class CodeParser: num_defaults = len(node.args.defaults) num_missing_defaults = num_args - num_defaults missing_defaults = [None] * num_missing_defaults - default_values = [ast.unparse(default).strip("'") if default else None for default in node.args.defaults] + default_values = [ + ast.unparse(default).strip("'") if default else None + for default in node.args.defaults + ] # Now check all default values to see if there # are any "None" values in the middle - default_values = [None if value == "None" else value for value in default_values] + default_values = [ + None if value == "None" else value for value in default_values + ] defaults = missing_defaults + default_values - args = [self.parse_arg(arg, default) for arg, default in zip(node.args.args, defaults)] + args = [ + self.parse_arg(arg, default) + for arg, default in zip(node.args.args, defaults) + ] return args def parse_varargs(self, node: ast.FunctionDef) -> List[Dict[str, Any]]: @@ -208,11 +223,17 @@ class CodeParser: """ Parses the keyword-only arguments of a function or method node. """ - kw_defaults = [None] * (len(node.args.kwonlyargs) - len(node.args.kw_defaults)) + [ - ast.unparse(default) if default else None for default in node.args.kw_defaults + kw_defaults = [None] * ( + len(node.args.kwonlyargs) - len(node.args.kw_defaults) + ) + [ + ast.unparse(default) if default else None + for default in node.args.kw_defaults ] - args = [self.parse_arg(arg, default) for arg, default in zip(node.args.kwonlyargs, kw_defaults)] + args = [ + self.parse_arg(arg, default) + for arg, default in zip(node.args.kwonlyargs, kw_defaults) + ] return args def parse_kwargs(self, node: ast.FunctionDef) -> List[Dict[str, Any]]: @@ -268,15 +289,28 @@ class CodeParser: method = self.parse_callable_details(stmt) return (method, True) if stmt.name == "__init__" else (method, False) + def get_base_classes(self): + """ + Returns the base classes of the custom component class. + """ + try: + bases = self.execute_and_inspect_classes(self.code) + except Exception as e: + # If the code cannot be executed, return an empty list + logger.exception(e) + bases = [] + raise e + return bases + def parse_classes(self, node: ast.ClassDef) -> None: """ Extracts "classes" from the code, including inheritance and init methods. """ - + bases = self.get_base_classes() or [ast.unparse(b) for b in node.bases] class_details = ClassCodeDetails( name=node.name, doc=ast.get_docstring(node), - bases=[ast.unparse(base) for base in node.bases], + bases=bases, attributes=[], methods=[], init=None, @@ -303,11 +337,25 @@ class CodeParser: Extracts global variables from the code. """ global_var = { - "targets": [t.id if hasattr(t, "id") else ast.dump(t) for t in node.targets], + "targets": [ + t.id if hasattr(t, "id") else ast.dump(t) for t in node.targets + ], "value": ast.unparse(node.value), } self.data["global_vars"].append(global_var) + def execute_and_inspect_classes(self, code: str): + custom_component_class = eval_custom_component_code(code) + custom_component = custom_component_class() + dunder_class = custom_component.__class__ + # Get the base classes at two levels of inheritance + bases = [] + for base in dunder_class.__bases__: + bases.append(base.__name__) + for bases_base in base.__bases__: + bases.append(bases_base.__name__) + return bases + def parse_code(self) -> Dict[str, Any]: """ Runs all parsing operations and returns the resulting data. From af69ac3333a6c198ff5ba3f63b053b0213510778 Mon Sep 17 00:00:00 2001 From: Gabriel Luiz Freitas Almeida Date: Tue, 27 Feb 2024 16:36:27 -0300 Subject: [PATCH 04/28] Refactor code formatting in validate.py --- src/backend/langflow/utils/validate.py | 58 +++++++++++++++++++------- 1 file changed, 44 insertions(+), 14 deletions(-) diff --git a/src/backend/langflow/utils/validate.py b/src/backend/langflow/utils/validate.py index 21821538c..6b789c08c 100644 --- a/src/backend/langflow/utils/validate.py +++ b/src/backend/langflow/utils/validate.py @@ -45,7 +45,9 @@ def validate_code(code): # Evaluate the function definition for node in tree.body: if isinstance(node, ast.FunctionDef): - code_obj = compile(ast.Module(body=[node], type_ignores=[]), "", "exec") + code_obj = compile( + ast.Module(body=[node], type_ignores=[]), "", "exec" + ) try: exec(code_obj) except Exception as e: @@ -89,15 +91,23 @@ def execute_function(code, function_name, *args, **kwargs): exec_globals, locals(), ) - exec_globals[alias.asname or alias.name] = importlib.import_module(alias.name) + exec_globals[alias.asname or alias.name] = importlib.import_module( + alias.name + ) except ModuleNotFoundError as e: - raise ModuleNotFoundError(f"Module {alias.name} not found. Please install it and try again.") from e + raise ModuleNotFoundError( + f"Module {alias.name} not found. Please install it and try again." + ) from e function_code = next( - node for node in module.body if isinstance(node, ast.FunctionDef) and node.name == function_name + node + for node in module.body + if isinstance(node, ast.FunctionDef) and node.name == function_name ) function_code.parent = None - code_obj = compile(ast.Module(body=[function_code], type_ignores=[]), "", "exec") + code_obj = compile( + ast.Module(body=[function_code], type_ignores=[]), "", "exec" + ) try: exec(code_obj, exec_globals, locals()) except Exception as exc: @@ -124,15 +134,23 @@ def create_function(code, function_name): if isinstance(node, ast.Import): for alias in node.names: try: - exec_globals[alias.asname or alias.name] = importlib.import_module(alias.name) + exec_globals[alias.asname or alias.name] = importlib.import_module( + alias.name + ) except ModuleNotFoundError as e: - raise ModuleNotFoundError(f"Module {alias.name} not found. Please install it and try again.") from e + raise ModuleNotFoundError( + f"Module {alias.name} not found. Please install it and try again." + ) from e function_code = next( - node for node in module.body if isinstance(node, ast.FunctionDef) and node.name == function_name + node + for node in module.body + if isinstance(node, ast.FunctionDef) and node.name == function_name ) function_code.parent = None - code_obj = compile(ast.Module(body=[function_code], type_ignores=[]), "", "exec") + code_obj = compile( + ast.Module(body=[function_code], type_ignores=[]), "", "exec" + ) with contextlib.suppress(Exception): exec(code_obj, exec_globals, locals()) exec_globals[function_name] = locals()[function_name] @@ -194,9 +212,13 @@ def prepare_global_scope(code, module): if isinstance(node, ast.Import): for alias in node.names: try: - exec_globals[alias.asname or alias.name] = importlib.import_module(alias.name) + exec_globals[alias.asname or alias.name] = importlib.import_module( + alias.name + ) except ModuleNotFoundError as e: - raise ModuleNotFoundError(f"Module {alias.name} not found. Please install it and try again.") from e + raise ModuleNotFoundError( + f"Module {alias.name} not found. Please install it and try again." + ) from e elif isinstance(node, ast.ImportFrom) and node.module is not None: try: imported_module = importlib.import_module(node.module) @@ -217,7 +239,11 @@ def extract_class_code(module, class_name): :param class_name: Name of the class to extract :return: AST node of the specified class """ - class_code = next(node for node in module.body if isinstance(node, ast.ClassDef) and node.name == class_name) + class_code = next( + node + for node in module.body + if isinstance(node, ast.ClassDef) and node.name == class_name + ) class_code.parent = None return class_code @@ -230,7 +256,9 @@ def compile_class_code(class_code): :param class_code: AST node of the class :return: Compiled code object of the class """ - code_obj = compile(ast.Module(body=[class_code], type_ignores=[]), "", "exec") + code_obj = compile( + ast.Module(body=[class_code], type_ignores=[]), "", "exec" + ) return code_obj @@ -274,7 +302,9 @@ def get_default_imports(code_string): langflow_imports = list(CUSTOM_COMPONENT_SUPPORTED_TYPES.keys()) necessary_imports = find_names_in_code(code_string, langflow_imports) langflow_module = importlib.import_module("langflow.field_typing") - default_imports.update({name: getattr(langflow_module, name) for name in necessary_imports}) + default_imports.update( + {name: getattr(langflow_module, name) for name in necessary_imports} + ) return default_imports From e0a67b0410df98f51db13aea146748fa9db2ddec Mon Sep 17 00:00:00 2001 From: Gabriel Luiz Freitas Almeida Date: Tue, 27 Feb 2024 16:52:04 -0300 Subject: [PATCH 05/28] Update model invocation in OpenAIModel, VertexAiModel, CohereModel, LlamaCppModel, AnthropicModel, GoogleGenerativeAIModel, AzureOpenAIModel, CTransformersModel, OllamaModel, AmazonBedrockModel, BaiduQianfanChatModel, and HuggingFaceModel --- src/backend/langflow/components/chains/ConversationChain.py | 2 +- src/backend/langflow/components/models/AmazonBedrockModel.py | 2 +- src/backend/langflow/components/models/AnthropicModel.py | 2 +- src/backend/langflow/components/models/AzureOpenAIModel.py | 2 +- src/backend/langflow/components/models/BaiduQianfanChatModel.py | 2 +- src/backend/langflow/components/models/CTransformersModel.py | 2 +- src/backend/langflow/components/models/CohereModel.py | 2 +- .../langflow/components/models/GoogleGenerativeAIModel.py | 2 +- src/backend/langflow/components/models/HuggingFaceModel.py | 2 +- src/backend/langflow/components/models/LlamaCppModel.py | 2 +- src/backend/langflow/components/models/OllamaModel.py | 2 +- src/backend/langflow/components/models/OpenAIModel.py | 2 +- src/backend/langflow/components/models/VertexAiModel.py | 2 +- 13 files changed, 13 insertions(+), 13 deletions(-) diff --git a/src/backend/langflow/components/chains/ConversationChain.py b/src/backend/langflow/components/chains/ConversationChain.py index 7d9d28dcc..726056138 100644 --- a/src/backend/langflow/components/chains/ConversationChain.py +++ b/src/backend/langflow/components/chains/ConversationChain.py @@ -31,7 +31,7 @@ class ConversationChainComponent(CustomComponent): chain = ConversationChain(llm=llm) else: chain = ConversationChain(llm=llm, memory=memory) - result = chain.invoke(inputs) + result = chain.invoke(input_value) # result is an AIMessage which is a subclass of BaseMessage # We need to check if it is a string or a BaseMessage if hasattr(result, "content") and isinstance(result.content, str): diff --git a/src/backend/langflow/components/models/AmazonBedrockModel.py b/src/backend/langflow/components/models/AmazonBedrockModel.py index 68e404773..478bf8e9a 100644 --- a/src/backend/langflow/components/models/AmazonBedrockModel.py +++ b/src/backend/langflow/components/models/AmazonBedrockModel.py @@ -60,7 +60,7 @@ class AmazonBedrockComponent(CustomComponent): ) # type: ignore except Exception as e: raise ValueError("Could not connect to AmazonBedrock API.") from e - message = output.invoke(inputs) + message = output.invoke(input_value) result = message.content if hasattr(message, "content") else message self.status = result return result diff --git a/src/backend/langflow/components/models/AnthropicModel.py b/src/backend/langflow/components/models/AnthropicModel.py index be6e46d9a..cb8e55194 100644 --- a/src/backend/langflow/components/models/AnthropicModel.py +++ b/src/backend/langflow/components/models/AnthropicModel.py @@ -77,7 +77,7 @@ class AnthropicLLM(CustomComponent): ) except Exception as e: raise ValueError("Could not connect to Anthropic API.") from e - message = output.invoke(inputs) + message = output.invoke(input_value) result = message.content if hasattr(message, "content") else message self.status = result return result diff --git a/src/backend/langflow/components/models/AzureOpenAIModel.py b/src/backend/langflow/components/models/AzureOpenAIModel.py index be1f724bf..df8413870 100644 --- a/src/backend/langflow/components/models/AzureOpenAIModel.py +++ b/src/backend/langflow/components/models/AzureOpenAIModel.py @@ -99,7 +99,7 @@ class AzureChatOpenAIComponent(CustomComponent): ) except Exception as e: raise ValueError("Could not connect to AzureOpenAI API.") from e - message = output.invoke(inputs) + message = output.invoke(input_value) result = message.content if hasattr(message, "content") else message self.status = result return result diff --git a/src/backend/langflow/components/models/BaiduQianfanChatModel.py b/src/backend/langflow/components/models/BaiduQianfanChatModel.py index 9eadb7013..0075316a2 100644 --- a/src/backend/langflow/components/models/BaiduQianfanChatModel.py +++ b/src/backend/langflow/components/models/BaiduQianfanChatModel.py @@ -94,7 +94,7 @@ class QianfanChatEndpointComponent(CustomComponent): ) except Exception as e: raise ValueError("Could not connect to Baidu Qianfan API.") from e - message = output.invoke(inputs) + message = output.invoke(input_value) result = message.content if hasattr(message, "content") else message self.status = result return result diff --git a/src/backend/langflow/components/models/CTransformersModel.py b/src/backend/langflow/components/models/CTransformersModel.py index 60cc2eb12..fb292d9cf 100644 --- a/src/backend/langflow/components/models/CTransformersModel.py +++ b/src/backend/langflow/components/models/CTransformersModel.py @@ -42,7 +42,7 @@ class CTransformersComponent(CustomComponent): output = CTransformers( model=model, model_file=model_file, model_type=model_type, config=config ) - message = output.invoke(inputs) + message = output.invoke(input_value) result = message.content if hasattr(message, "content") else message self.status = result return result diff --git a/src/backend/langflow/components/models/CohereModel.py b/src/backend/langflow/components/models/CohereModel.py index 28b198ec1..c2a004c38 100644 --- a/src/backend/langflow/components/models/CohereModel.py +++ b/src/backend/langflow/components/models/CohereModel.py @@ -43,7 +43,7 @@ class CohereComponent(CustomComponent): max_tokens=max_tokens, temperature=temperature, ) - message = output.invoke(inputs) + message = output.invoke(input_value) result = message.content if hasattr(message, "content") else message self.status = result return result diff --git a/src/backend/langflow/components/models/GoogleGenerativeAIModel.py b/src/backend/langflow/components/models/GoogleGenerativeAIModel.py index 2ff01c4c7..3b0c758e8 100644 --- a/src/backend/langflow/components/models/GoogleGenerativeAIModel.py +++ b/src/backend/langflow/components/models/GoogleGenerativeAIModel.py @@ -73,7 +73,7 @@ class GoogleGenerativeAIComponent(CustomComponent): n=n or 1, google_api_key=SecretStr(google_api_key), ) - message = output.invoke(inputs) + message = output.invoke(input_value) result = message.content if hasattr(message, "content") else message self.status = result return result diff --git a/src/backend/langflow/components/models/HuggingFaceModel.py b/src/backend/langflow/components/models/HuggingFaceModel.py index 394938344..99fcc6ab0 100644 --- a/src/backend/langflow/components/models/HuggingFaceModel.py +++ b/src/backend/langflow/components/models/HuggingFaceModel.py @@ -45,7 +45,7 @@ class HuggingFaceEndpointsComponent(CustomComponent): except Exception as e: raise ValueError("Could not connect to HuggingFace Endpoints API.") from e output = ChatHuggingFace(llm=llm) - message = output.invoke(inputs) + message = output.invoke(input_value)alue) result = message.content if hasattr(message, "content") else message self.status = result return result diff --git a/src/backend/langflow/components/models/LlamaCppModel.py b/src/backend/langflow/components/models/LlamaCppModel.py index 53a6f8ace..5941f213b 100644 --- a/src/backend/langflow/components/models/LlamaCppModel.py +++ b/src/backend/langflow/components/models/LlamaCppModel.py @@ -135,7 +135,7 @@ class LlamaCppComponent(CustomComponent): verbose=verbose, vocab_only=vocab_only, ) - message = output.invoke(inputs) + message = output.invoke(input_value) result = message.content if hasattr(message, "content") else message self.status = result return result diff --git a/src/backend/langflow/components/models/OllamaModel.py b/src/backend/langflow/components/models/OllamaModel.py index 3dc8dacab..7ae896532 100644 --- a/src/backend/langflow/components/models/OllamaModel.py +++ b/src/backend/langflow/components/models/OllamaModel.py @@ -250,7 +250,7 @@ class ChatOllamaComponent(CustomComponent): output = ChatOllama(**llm_params) # type: ignore except Exception as e: raise ValueError("Could not initialize Ollama LLM.") from e - message = output.invoke(inputs) + message = output.invoke(input_value) result = message.content if hasattr(message, "content") else message self.status = result return result diff --git a/src/backend/langflow/components/models/OpenAIModel.py b/src/backend/langflow/components/models/OpenAIModel.py index 07ba7013c..ee6809e0b 100644 --- a/src/backend/langflow/components/models/OpenAIModel.py +++ b/src/backend/langflow/components/models/OpenAIModel.py @@ -80,7 +80,7 @@ class OpenAIModelComponent(CustomComponent): temperature=temperature, ) - message = model.invoke(inputs) + message = model.invoke(input_value) result = message.content if hasattr(message, "content") else message self.status = result return result diff --git a/src/backend/langflow/components/models/VertexAiModel.py b/src/backend/langflow/components/models/VertexAiModel.py index 81338f723..c05dc5e94 100644 --- a/src/backend/langflow/components/models/VertexAiModel.py +++ b/src/backend/langflow/components/models/VertexAiModel.py @@ -92,7 +92,7 @@ class ChatVertexAIComponent(CustomComponent): top_p=top_p, verbose=verbose, ) - message = output.invoke(inputs) + message = output.invoke(input_value) result = message.content if hasattr(message, "content") else message self.status = result return result From 0e9aa23a1c7c210064b7cc196521440992d0f076 Mon Sep 17 00:00:00 2001 From: Gabriel Luiz Freitas Almeida Date: Tue, 27 Feb 2024 16:57:51 -0300 Subject: [PATCH 06/28] Refactor ChatOutput class to inherit from ChatComponent --- src/backend/langflow/components/io/ChatOutput.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/backend/langflow/components/io/ChatOutput.py b/src/backend/langflow/components/io/ChatOutput.py index e4c5d03ee..842a083fb 100644 --- a/src/backend/langflow/components/io/ChatOutput.py +++ b/src/backend/langflow/components/io/ChatOutput.py @@ -1,11 +1,11 @@ from typing import Optional, Union -from langflow import CustomComponent from langflow.field_typing import Text +from langflow.io.schema import ChatComponent from langflow.schema import Record -class ChatOutput(CustomComponent): +class ChatOutput(ChatComponent): display_name = "Chat Output" description = "Used to send a message to the chat." From 40baf204244c6af9b4038d09d4eac1d06003cbcc Mon Sep 17 00:00:00 2001 From: Gabriel Luiz Freitas Almeida Date: Tue, 27 Feb 2024 17:51:02 -0300 Subject: [PATCH 07/28] Refactor ChatComponent's add_message method --- src/backend/langflow/io/schema.py | 39 ++++++++++++++++--------------- 1 file changed, 20 insertions(+), 19 deletions(-) diff --git a/src/backend/langflow/io/schema.py b/src/backend/langflow/io/schema.py index d83d3472e..4660b4276 100644 --- a/src/backend/langflow/io/schema.py +++ b/src/backend/langflow/io/schema.py @@ -48,27 +48,28 @@ class ChatComponent(CustomComponent): raise ValueError( "All of session_id, sender, and sender_name must be provided." ) - - if not record: - record = [] - if not session_id or not sender or not sender_name: - raise ValueError - for text in text: - record = Record( - text=text, - data={ - "session_id": session_id, - "sender": sender, - "sender_name": sender_name, - }, - ) - record.append(record) - elif isinstance(record, Record): - record = [record] + if isinstance(message, Record): + record = message + record.data.update( + { + "session_id": session_id, + "sender": sender, + "sender_name": sender_name, + } + ) + else: + record = Record( + text=message, + data={ + "session_id": session_id, + "sender": sender, + "sender_name": sender_name, + }, + ) self.status = record - record = add_messages(record) - return record + records = add_messages([record]) + return records[0] def build( self, From 4007b9728fde7402cbeecf0ee1b0ee2fcea9c19b Mon Sep 17 00:00:00 2001 From: Gabriel Luiz Freitas Almeida Date: Tue, 27 Feb 2024 17:57:29 -0300 Subject: [PATCH 08/28] Refactor typing import in types.py --- src/backend/langflow/graph/vertex/types.py | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/src/backend/langflow/graph/vertex/types.py b/src/backend/langflow/graph/vertex/types.py index 45bcb9ccd..92d4a1bd7 100644 --- a/src/backend/langflow/graph/vertex/types.py +++ b/src/backend/langflow/graph/vertex/types.py @@ -1,7 +1,6 @@ import ast import json -from typing import (AsyncIterator, Callable, Dict, Iterator, List, Optional, - Union) +from typing import AsyncIterator, Callable, Dict, Iterator, List, Optional, Union import yaml from langchain_core.messages import AIMessage @@ -396,7 +395,7 @@ class ChatVertex(StatelessVertex): if artifacts: self.artifacts = artifacts.model_dump() if isinstance(self._built_object, (AsyncIterator, Iterator)): - if self.params["as_record"]: + if self.params["return_record"]: self._built_object = Record(text=message, data=self.artifacts) else: self._built_object = message From 27fe14237768c2d0275e238f214c4608f5304b7f Mon Sep 17 00:00:00 2001 From: Gabriel Luiz Freitas Almeida Date: Tue, 27 Feb 2024 18:06:35 -0300 Subject: [PATCH 09/28] Update message parameter name in ChatVertex class --- src/backend/langflow/graph/vertex/types.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/backend/langflow/graph/vertex/types.py b/src/backend/langflow/graph/vertex/types.py index 92d4a1bd7..5c6aaecf8 100644 --- a/src/backend/langflow/graph/vertex/types.py +++ b/src/backend/langflow/graph/vertex/types.py @@ -361,7 +361,7 @@ class ChatVertex(StatelessVertex): artifacts = None sender = self.params.get("sender", None) sender_name = self.params.get("sender_name", None) - message = self.params.get("message", None) + message = self.params.get("input_value", None) stream_url = None if isinstance(self._built_object, AIMessage): artifacts = ChatOutputResponse.from_message( From 39a2801095f0165629e14851bd97d10a84b2ea6d Mon Sep 17 00:00:00 2001 From: Gabriel Luiz Freitas Almeida Date: Tue, 27 Feb 2024 18:07:12 -0300 Subject: [PATCH 10/28] Refactor message parameter in ContractEdge --- src/backend/langflow/graph/edge/base.py | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/src/backend/langflow/graph/edge/base.py b/src/backend/langflow/graph/edge/base.py index 99b24e2da..6706156a0 100644 --- a/src/backend/langflow/graph/edge/base.py +++ b/src/backend/langflow/graph/edge/base.py @@ -137,14 +137,15 @@ class ContractEdge(Edge): log_transaction(self, source, target, "success") # If the target vertex is a power component we log messages if target.vertex_type == "ChatOutput" and ( - isinstance(target.params.get("message"), str) or isinstance(target.params.get("message"), dict) + isinstance(target.params.get("input_value"), str) + or isinstance(target.params.get("input_value"), dict) ): if target.params.get("message") == "": return self.result await log_message( sender=target.params.get("sender", ""), sender_name=target.params.get("sender_name", ""), - message=target.params.get("message", {}), + message=target.params.get("input_value", {}), session_id=target.params.get("session_id", ""), artifacts=target.artifacts, ) From a9bea1872ad258ebf27366e2532d552dae6bd675 Mon Sep 17 00:00:00 2001 From: Gabriel Luiz Freitas Almeida Date: Tue, 27 Feb 2024 18:08:50 -0300 Subject: [PATCH 11/28] Update parameter name in ChatVertex stream method --- src/backend/langflow/graph/vertex/types.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/backend/langflow/graph/vertex/types.py b/src/backend/langflow/graph/vertex/types.py index 5c6aaecf8..f8d4ae1e4 100644 --- a/src/backend/langflow/graph/vertex/types.py +++ b/src/backend/langflow/graph/vertex/types.py @@ -405,7 +405,7 @@ class ChatVertex(StatelessVertex): await super()._run(*args, **kwargs) async def stream(self): - iterator = self.params.get("message", None) + iterator = self.params.get("input_value", None) if not isinstance(iterator, (AsyncIterator, Iterator)): raise ValueError("The message must be an iterator or an async iterator.") is_async = isinstance(iterator, AsyncIterator) From d6963b581257d6ba023ec5aaaccf4458c41d0f3d Mon Sep 17 00:00:00 2001 From: Gabriel Luiz Freitas Almeida Date: Tue, 27 Feb 2024 18:12:33 -0300 Subject: [PATCH 12/28] Add loguru logger to types.py --- src/backend/langflow/graph/vertex/types.py | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/src/backend/langflow/graph/vertex/types.py b/src/backend/langflow/graph/vertex/types.py index f8d4ae1e4..eef69a10d 100644 --- a/src/backend/langflow/graph/vertex/types.py +++ b/src/backend/langflow/graph/vertex/types.py @@ -4,6 +4,7 @@ from typing import AsyncIterator, Callable, Dict, Iterator, List, Optional, Unio import yaml from langchain_core.messages import AIMessage +from loguru import logger from langflow.graph.utils import UnbuiltObject, flatten_list from langflow.graph.vertex.base import StatefulVertex, StatelessVertex @@ -361,7 +362,7 @@ class ChatVertex(StatelessVertex): artifacts = None sender = self.params.get("sender", None) sender_name = self.params.get("sender_name", None) - message = self.params.get("input_value", None) + message = self.params.get(INPUT_FIELD_NAME, None) stream_url = None if isinstance(self._built_object, AIMessage): artifacts = ChatOutputResponse.from_message( @@ -405,7 +406,7 @@ class ChatVertex(StatelessVertex): await super()._run(*args, **kwargs) async def stream(self): - iterator = self.params.get("input_value", None) + iterator = self.params.get(INPUT_FIELD_NAME, None) if not isinstance(iterator, (AsyncIterator, Iterator)): raise ValueError("The message must be an iterator or an async iterator.") is_async = isinstance(iterator, AsyncIterator) @@ -431,6 +432,7 @@ class ChatVertex(StatelessVertex): sender=self.params.get("sender", ""), sender_name=self.params.get("sender_name", ""), ).model_dump() + logger.debug(f"Streamed message: {complete_message}") await log_message( sender=self.params.get("sender", ""), From 6ad4de86550ec24dd84de31c02df63347e69cf59 Mon Sep 17 00:00:00 2001 From: Gabriel Luiz Freitas Almeida Date: Tue, 27 Feb 2024 18:12:58 -0300 Subject: [PATCH 13/28] Add updated_raw_params flag and INPUT_FIELD_NAME constant --- src/backend/langflow/graph/edge/base.py | 50 +++++++++++++++++------ src/backend/langflow/graph/graph/base.py | 38 ++++------------- src/backend/langflow/graph/schema.py | 2 + src/backend/langflow/graph/vertex/base.py | 17 +++++--- 4 files changed, 60 insertions(+), 47 deletions(-) diff --git a/src/backend/langflow/graph/edge/base.py b/src/backend/langflow/graph/edge/base.py index 6706156a0..cfcd33dd1 100644 --- a/src/backend/langflow/graph/edge/base.py +++ b/src/backend/langflow/graph/edge/base.py @@ -4,6 +4,7 @@ from loguru import logger from pydantic import BaseModel, Field from langflow.graph.edge.utils import build_clean_params +from langflow.graph.schema import INPUT_FIELD_NAME from langflow.services.deps import get_monitor_service from langflow.services.monitor.utils import log_message @@ -12,7 +13,9 @@ if TYPE_CHECKING: class SourceHandle(BaseModel): - baseClasses: List[str] = Field(..., description="List of base classes for the source handle.") + baseClasses: List[str] = Field( + ..., description="List of base classes for the source handle." + ) dataType: str = Field(..., description="Data type for the source handle.") id: str = Field(..., description="Unique identifier for the source handle.") @@ -20,7 +23,9 @@ class SourceHandle(BaseModel): class TargetHandle(BaseModel): fieldName: str = Field(..., description="Field name for the target handle.") id: str = Field(..., description="Unique identifier for the target handle.") - inputTypes: Optional[List[str]] = Field(None, description="List of input types for the target handle.") + inputTypes: Optional[List[str]] = Field( + None, description="List of input types for the target handle." + ) type: str = Field(..., description="Type of the target handle.") @@ -49,16 +54,24 @@ class Edge: def validate_handles(self, source, target) -> None: if self.target_handle.inputTypes is None: - self.valid_handles = self.target_handle.type in self.source_handle.baseClasses + self.valid_handles = ( + self.target_handle.type in self.source_handle.baseClasses + ) else: self.valid_handles = ( - any(baseClass in self.target_handle.inputTypes for baseClass in self.source_handle.baseClasses) + any( + baseClass in self.target_handle.inputTypes + for baseClass in self.source_handle.baseClasses + ) or self.target_handle.type in self.source_handle.baseClasses ) if not self.valid_handles: logger.debug(self.source_handle) logger.debug(self.target_handle) - raise ValueError(f"Edge between {source.vertex_type} and {target.vertex_type} " f"has invalid handles") + raise ValueError( + f"Edge between {source.vertex_type} and {target.vertex_type} " + f"has invalid handles" + ) def __setstate__(self, state): self.source_id = state["source_id"] @@ -75,7 +88,11 @@ class Edge: # Both lists contain strings and sometimes a string contains the value we are # looking for e.g. comgin_out=["Chain"] and target_reqs=["LLMChain"] # so we need to check if any of the strings in source_types is in target_reqs - self.valid = any(output in target_req for output in self.source_types for target_req in self.target_reqs) + self.valid = any( + output in target_req + for output in self.source_types + for target_req in self.target_reqs + ) # Get what type of input the target node is expecting self.matched_type = next( @@ -86,7 +103,10 @@ class Edge: if no_matched_type: logger.debug(self.source_types) logger.debug(self.target_reqs) - raise ValueError(f"Edge between {source.vertex_type} and {target.vertex_type} " f"has no matched type") + raise ValueError( + f"Edge between {source.vertex_type} and {target.vertex_type} " + f"has no matched type" + ) def __repr__(self) -> str: return ( @@ -98,7 +118,11 @@ class Edge: return hash(self.__repr__()) def __eq__(self, __value: object) -> bool: - return self.__repr__() == __value.__repr__() if isinstance(__value, Edge) else False + return ( + self.__repr__() == __value.__repr__() + if isinstance(__value, Edge) + else False + ) class ContractEdge(Edge): @@ -137,15 +161,15 @@ class ContractEdge(Edge): log_transaction(self, source, target, "success") # If the target vertex is a power component we log messages if target.vertex_type == "ChatOutput" and ( - isinstance(target.params.get("input_value"), str) - or isinstance(target.params.get("input_value"), dict) + isinstance(target.params.get(INPUT_FIELD_NAME), str) + or isinstance(target.params.get(INPUT_FIELD_NAME), dict) ): if target.params.get("message") == "": return self.result await log_message( sender=target.params.get("sender", ""), sender_name=target.params.get("sender_name", ""), - message=target.params.get("input_value", {}), + message=target.params.get(INPUT_FIELD_NAME, {}), session_id=target.params.get("session_id", ""), artifacts=target.artifacts, ) @@ -155,7 +179,9 @@ class ContractEdge(Edge): return f"{self.source_id} -[{self.target_param}]-> {self.target_id}" -def log_transaction(edge: ContractEdge, source: "Vertex", target: "Vertex", status, error=None): +def log_transaction( + edge: ContractEdge, source: "Vertex", target: "Vertex", status, error=None +): try: monitor_service = get_monitor_service() clean_params = build_clean_params(target) diff --git a/src/backend/langflow/graph/graph/base.py b/src/backend/langflow/graph/graph/base.py index 341a4729c..0ffc9825b 100644 --- a/src/backend/langflow/graph/graph/base.py +++ b/src/backend/langflow/graph/graph/base.py @@ -33,8 +33,6 @@ class Graph: edges: List[Dict[str, str]], flow_id: Optional[str] = None, ) -> None: - self.inputs = [] - self.outputs = [] self._vertices = nodes self._edges = edges self.raw_graph_data = {"nodes": nodes, "edges": edges} @@ -77,7 +75,7 @@ class Graph: async def _run(self, inputs: Dict[str, str]) -> List["ResultData"]: """Runs the graph with the given inputs.""" - for vertex_id in self.inputs: + for vertex_id in self._is_input_vertices: vertex = self.get_vertex(vertex_id) if vertex is None: raise ValueError(f"Vertex {vertex_id} not found") @@ -89,7 +87,7 @@ class Graph: logger.exception(exc) raise ValueError(f"Error running graph: {exc}") from exc outputs = [] - for vertex_id in self.outputs: + for vertex_id in self._is_output_vertices: vertex = self.get_vertex(vertex_id) if vertex is None: raise ValueError(f"Vertex {vertex_id} not found") @@ -104,11 +102,11 @@ class Graph: # of the vertices that are inputs # if the value is a list, we need to run multiple times outputs = [] - inputs_values = inputs.get("input_value") + inputs_values = inputs.get(INPUT_FIELD_NAME) if not isinstance(inputs_values, list): inputs_values = [inputs_values] for input_value in inputs_values: - run_outputs = await self._run({"input_value": input_value}) + run_outputs = await self._run({INPUT_FIELD_NAME: input_value}) logger.debug(f"Run outputs: {run_outputs}") outputs.extend(run_outputs) return outputs @@ -317,28 +315,6 @@ class Graph: # Now that we have the vertices and edges # We need to map the vertices that are connected to # to ChatVertex instances - self._map_chat_vertices() - - def _map_chat_vertices(self) -> None: - """Maps the vertices that are connected to ChatVertex instances.""" - # For each edge, we need to check if the source or target vertex is a ChatVertex - # If it is, we need to update the other vertex `is_external` attribute - # and store the id of the ChatVertex in the attributes self.inputs and self.outputs - for edge in self.edges: - source_vertex = self.get_vertex(edge.source_id) - target_vertex = self.get_vertex(edge.target_id) - if isinstance(source_vertex, ChatVertex): - # The source vertex is a ChatVertex - # thus the target vertex is an external vertex - # and the source vertex is an input - target_vertex.has_external_input = True - self.inputs.append(source_vertex.id) - if isinstance(target_vertex, ChatVertex): - # The target vertex is a ChatVertex - # thus the source vertex is an external vertex - # and the target vertex is an output - source_vertex.has_external_output = True - self.outputs.append(target_vertex.id) def remove_vertex(self, vertex_id: str) -> None: """Removes a vertex from the graph.""" @@ -443,13 +419,15 @@ class Graph: async def _execute_tasks(self, tasks): """Executes tasks in parallel, handling exceptions for each task.""" results = [] - for task in asyncio.as_completed(tasks): + for i, task in enumerate(asyncio.as_completed(tasks)): try: result = await task results.append(result) except Exception as e: # Log the exception along with the task name for easier debugging - task_name = task.get_name() + # task_name = task.get_name() + # coroutine has not attribute get_name + task_name = tasks[i].get_name() logger.error(f"Task {task_name} failed with exception: {e}") return results diff --git a/src/backend/langflow/graph/schema.py b/src/backend/langflow/graph/schema.py index d41e0544a..028b8db9f 100644 --- a/src/backend/langflow/graph/schema.py +++ b/src/backend/langflow/graph/schema.py @@ -35,3 +35,5 @@ OUTPUT_COMPONENTS = [ InterfaceComponentTypes.ChatOutput, InterfaceComponentTypes.TextOutput, ] + +INPUT_FIELD_NAME = "input_value" diff --git a/src/backend/langflow/graph/vertex/base.py b/src/backend/langflow/graph/vertex/base.py index 3e1133491..dd308f9f1 100644 --- a/src/backend/langflow/graph/vertex/base.py +++ b/src/backend/langflow/graph/vertex/base.py @@ -44,7 +44,7 @@ class Vertex: ) -> None: # is_external means that the Vertex send or receives data from # an external source (e.g the chat) - + self.updated_raw_params = False self.id: str = data["id"] self.is_input = any( input_component_name in self.id for input_component_name in INPUT_COMPONENTS @@ -285,6 +285,10 @@ class Vertex: if self.graph is None: raise ValueError("Graph not found") + if self.updated_raw_params: + self.updated_raw_params = False + return + template_dict = { key: value for key, value in self.data["node"]["template"].items() @@ -386,10 +390,11 @@ class Vertex: Raises: ValueError: If any key in new_params is not found in self._raw_params. """ - for key in new_params: - if key not in self._raw_params: - raise ValueError(f"Key {key} not found in raw params") + # First check if the input_value in _raw_params is not a vertex + if any(isinstance(self._raw_params.get(key), Vertex) for key in new_params): + return self._raw_params.update(new_params) + self.updated_raw_params = True async def _build(self, user_id=None): """ @@ -451,6 +456,8 @@ class Vertex: await self._build_node_and_update_params(key, value, user_id) elif isinstance(value, list) and self._is_list_of_nodes(value): await self._build_list_of_nodes_and_update_params(key, value, user_id) + elif key not in self.params: + self.params[key] = value def _is_node(self, value): """ @@ -586,7 +593,7 @@ class Vertex: logger.warning(message) - def _reset(self): + def _reset(self, params_update: Optional[Dict[str, Any]] = None): self._built = False self._built_object = UnbuiltObject() self._built_result = UnbuiltResult() From 8ea252c0333f50e3c759e39d1d022f791c5af7c7 Mon Sep 17 00:00:00 2001 From: Gabriel Luiz Freitas Almeida Date: Tue, 27 Feb 2024 18:13:08 -0300 Subject: [PATCH 14/28] Add flow_id parameter to Graph.from_payload() method --- src/backend/langflow/processing/process.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/src/backend/langflow/processing/process.py b/src/backend/langflow/processing/process.py index 69e47b242..aa419b962 100644 --- a/src/backend/langflow/processing/process.py +++ b/src/backend/langflow/processing/process.py @@ -269,6 +269,7 @@ async def build_graph_and_generate_result( async def run_graph( graph: Union["Graph", dict], + flow_id: str, session_id: str, inputs: Optional[Union[dict, List[dict]]] = None, artifacts: Optional[Dict[str, Any]] = None, @@ -276,7 +277,8 @@ async def run_graph( ): """Run the graph and generate the result""" if isinstance(graph, dict): - graph = Graph.from_payload(graph) + graph = Graph.from_payload(graph, flow_id=flow_id) + outputs = await graph.run(inputs) if session_id and session_service: session_service.update_session(session_id, (graph, artifacts)) From e40395197d32bff61358e71fec9c222cab05ac0d Mon Sep 17 00:00:00 2001 From: Gabriel Luiz Freitas Almeida Date: Tue, 27 Feb 2024 18:13:23 -0300 Subject: [PATCH 15/28] Add model serialization to RunResponse schema --- src/backend/langflow/api/v1/schemas.py | 16 ++++++++++++++-- 1 file changed, 14 insertions(+), 2 deletions(-) diff --git a/src/backend/langflow/api/v1/schemas.py b/src/backend/langflow/api/v1/schemas.py index 0c33fdb36..0092efa4e 100644 --- a/src/backend/langflow/api/v1/schemas.py +++ b/src/backend/langflow/api/v1/schemas.py @@ -4,7 +4,7 @@ from pathlib import Path from typing import Any, Dict, List, Optional, Union from uuid import UUID -from pydantic import BaseModel, Field, field_validator +from pydantic import BaseModel, Field, field_validator, model_serializer from langflow.services.database.models.api_key.model import ApiKeyRead from langflow.services.database.models.base import orjson_dumps @@ -70,9 +70,21 @@ class RunResponse(BaseModel): """Run response schema.""" outputs: Optional[List[Any]] = None - status: Optional[str] = None session_id: Optional[str] = None + @model_serializer(mode="wrap") + def serialize(self, handler): + # Serialize all the outputs if they are base models + if self.outputs: + serialized_outputs = [] + for output in self.outputs: + if isinstance(output, BaseModel): + serialized_outputs.append(output.model_dump(exclude_none=True)) + else: + serialized_outputs.append(output) + self.outputs = serialized_outputs + return handler(self) + class PreloadResponse(BaseModel): """Preload response schema.""" From feac452f1c8172a313f35e07dcdbe5729d45e2ac Mon Sep 17 00:00:00 2001 From: Gabriel Luiz Freitas Almeida Date: Tue, 27 Feb 2024 18:13:29 -0300 Subject: [PATCH 16/28] Refactor run_flow_with_caching function in endpoints.py --- src/backend/langflow/api/v1/endpoints.py | 25 ++++++++++++------------ 1 file changed, 13 insertions(+), 12 deletions(-) diff --git a/src/backend/langflow/api/v1/endpoints.py b/src/backend/langflow/api/v1/endpoints.py index 2dc79e85a..fce4f9649 100644 --- a/src/backend/langflow/api/v1/endpoints.py +++ b/src/backend/langflow/api/v1/endpoints.py @@ -220,7 +220,9 @@ async def preload_flow( raise HTTPException(status_code=500, detail=str(exc)) from exc -@router.post("/run/{flow_id}", response_model=ProcessResponse) +@router.post( + "/run/{flow_id}", response_model=RunResponse, response_model_exclude_none=True +) async def run_flow_with_caching( session: Annotated[Session, Depends(get_session)], flow_id: str, @@ -235,13 +237,13 @@ async def run_flow_with_caching( session_data = await session_service.load_session(session_id) graph, artifacts = session_data if session_data else (None, None) task_result: Any = None - task_status = None if not graph: raise ValueError("Graph not found in the session") task_result = await run_graph( - graph, - session_id, - inputs, + graph=graph, + flow_id=flow_id, + session_id=session_id, + inputs=inputs, artifacts=artifacts, session_service=session_service, ) @@ -262,16 +264,15 @@ async def run_flow_with_caching( graph_data = flow.data graph_data = process_tweaks(graph_data, tweaks) task_result = await run_graph( - graph_data, - inputs, - tweaks, - session_id, + graph=graph_data, + flow_id=flow_id, + session_id=session_id, + inputs=inputs, + artifacts={}, session_service=session_service, ) - return RunResponse( - outputs=task_result, session_id=session_id, status=task_status - ) + return RunResponse(outputs=task_result, session_id=session_id) except sa.exc.StatementError as exc: # StatementError('(builtins.ValueError) badly formed hexadecimal UUID string') if "badly formed hexadecimal UUID string" in str(exc): From e4dd94021d6209aef7a4ef3ab783da7869d89eb2 Mon Sep 17 00:00:00 2001 From: Gabriel Luiz Freitas Almeida Date: Tue, 27 Feb 2024 18:13:34 -0300 Subject: [PATCH 17/28] Refactor buildInputs function and update code that calls it --- src/frontend/src/utils/utils.ts | 22 ++++++++-------------- 1 file changed, 8 insertions(+), 14 deletions(-) diff --git a/src/frontend/src/utils/utils.ts b/src/frontend/src/utils/utils.ts index 77e668c26..d6ebff378 100644 --- a/src/frontend/src/utils/utils.ts +++ b/src/frontend/src/utils/utils.ts @@ -13,7 +13,7 @@ import { tweakType, } from "../types/components"; import { FlowType, NodeType } from "../types/flow"; -import { FlowState, FlowsState } from "../types/tabs"; +import { FlowState } from "../types/tabs"; import { buildTweaks } from "./reactflowUtils"; export function classNames(...classes: Array): string { @@ -217,12 +217,8 @@ export function groupByFamily( })); } -export function buildInputs(flowState?: FlowState): string { - return flowState && - flowState.input_keys && - Object.keys(flowState.input_keys!).length > 0 - ? JSON.stringify(flowState.input_keys) - : '{"input": "message"}'; +export function buildInputs(): string { + return '{"input_value": "message"}'; } export function getRandomElement(array: T[]): T { @@ -323,7 +319,7 @@ export function getPythonApiCode( // node.data.id // } const tweaks = buildTweaks(flow); - const inputs = buildInputs(flowState); + const inputs = buildInputs(); return `import requests from typing import Optional @@ -382,12 +378,10 @@ export function getCurlCode( ): string { const flowId = flow.id; const tweaks = buildTweaks(flow); - const inputs = buildInputs(flowState); + const inputs = buildInputs(); return `curl -X POST \\ - ${window.location.protocol}//${ - window.location.host - }/api/v1/process/${flowId} \\ + ${window.location.protocol}//${window.location.host}/api/v1/run/${flowId} \\ -H 'Content-Type: application/json'\\${ !isAuth ? `\n -H 'x-api-key: '\\` : "" } @@ -410,7 +404,7 @@ export function getPythonCode( ): string { const flowName = flow.name; const tweaks = buildTweaks(flow); - const inputs = buildInputs(flowState); + const inputs = buildInputs(); return `from langflow import load_flow_from_json TWEAKS = ${ tweak && tweak.length > 0 @@ -435,7 +429,7 @@ export function getWidgetCode( ): string { const flowId = flow.id; const flowName = flow.name; - const inputs = buildInputs(flowState); + const inputs = buildInputs(); let chat_input_field = getChatInputField(flow, flowState); return ` From af9786bdef108db4ce9caab7db9395399da2e0d9 Mon Sep 17 00:00:00 2001 From: Gabriel Luiz Freitas Almeida Date: Tue, 27 Feb 2024 18:24:00 -0300 Subject: [PATCH 18/28] Add import statement for INPUT_FIELD_NAME --- src/backend/langflow/graph/vertex/types.py | 1 + 1 file changed, 1 insertion(+) diff --git a/src/backend/langflow/graph/vertex/types.py b/src/backend/langflow/graph/vertex/types.py index eef69a10d..d60b7412c 100644 --- a/src/backend/langflow/graph/vertex/types.py +++ b/src/backend/langflow/graph/vertex/types.py @@ -6,6 +6,7 @@ import yaml from langchain_core.messages import AIMessage from loguru import logger +from langflow.graph.schema import INPUT_FIELD_NAME from langflow.graph.utils import UnbuiltObject, flatten_list from langflow.graph.vertex.base import StatefulVertex, StatelessVertex from langflow.interface.utils import extract_input_variables_from_prompt From df07cf413b53defa37b2e71f55db9c3a3302d267 Mon Sep 17 00:00:00 2001 From: anovazzi1 Date: Tue, 27 Feb 2024 18:29:57 -0300 Subject: [PATCH 19/28] fix(frontend): add useRef hook to manage eventSource in ChatMessage component feat(frontend): add support for process.env.PORT environment variable in server.ts feat(frontend): add updateFlowPool function to NewChatView component feat(frontend): add buildId parameter to addDataToFlowPool function in flowStore feat(frontend): add stream_url property to ChatOutputType in flow types --- .../newChatView/chatMessage/index.tsx | 45 +++++++++--------- .../src/components/newChatView/index.tsx | 47 +++++++++---------- src/frontend/src/stores/flowStore.ts | 7 +-- src/frontend/src/types/chat/index.ts | 2 +- src/frontend/src/types/components/index.ts | 6 +-- src/frontend/src/types/zustand/flow/index.ts | 8 ++-- src/frontend/src/utils/buildUtils.ts | 9 ++-- 7 files changed, 64 insertions(+), 60 deletions(-) diff --git a/src/frontend/src/components/newChatView/chatMessage/index.tsx b/src/frontend/src/components/newChatView/chatMessage/index.tsx index 266444d8b..c2819a3a1 100644 --- a/src/frontend/src/components/newChatView/chatMessage/index.tsx +++ b/src/frontend/src/components/newChatView/chatMessage/index.tsx @@ -1,5 +1,5 @@ import Convert from "ansi-to-html"; -import { useEffect, useMemo, useState } from "react"; +import { useEffect, useMemo, useState,useRef } from "react"; import Markdown from "react-markdown"; import rehypeMathjax from "rehype-mathjax"; import remarkGfm from "remark-gfm"; @@ -12,6 +12,7 @@ import IconComponent from "../../../components/genericIconComponent"; import { chatMessagePropsType } from "../../../types/components"; import { classNames } from "../../../utils/utils"; import FileCard from "../fileComponent"; +import useFlowStore from "../../../stores/flowStore"; export default function ChatMessage({ chat, @@ -29,6 +30,9 @@ export default function ChatMessage({ const chatMessageString = chat.message ? chat.message.toString() : ""; const [chatMessage, setChatMessage] = useState(chatMessageString); const [isStreaming, setIsStreaming] = useState(false); + const eventSource = useRef(undefined); + const updateFlowPool = useFlowStore((state) => state.updateFlowPool); + // The idea now is that chat.stream_url MAY be a URL if we should stream the output of the chat // probably the message is empty when we have a stream_url @@ -36,49 +40,48 @@ export default function ChatMessage({ const streamChunks = (url: string) => { setIsStreaming(true); // Streaming starts return new Promise((resolve, reject) => { - const eventSource = new EventSource(url); - eventSource.onmessage = (event) => { + eventSource.current = new EventSource(url); + eventSource.current.onmessage = (event) => { let parsedData = JSON.parse(event.data); if (parsedData.chunk) { setChatMessage((prev) => prev + parsedData.chunk); } }; - eventSource.onerror = (event) => { + eventSource.current.onerror = (event) => { + setIsStreaming(false); + eventSource.current?.close(); + setStreamUrl(undefined); reject(new Error("Streaming failed")); - setIsStreaming(false); - eventSource.close(); }; - eventSource.addEventListener("close", (event) => { - setStreamUrl(null); // Update state to reflect the stream is closed - resolve(true); + eventSource.current.addEventListener("close", (event) => { + setStreamUrl(undefined); // Update state to reflect the stream is closed + eventSource.current?.close(); setIsStreaming(false); - eventSource.close(); + resolve(true); }); }); }; + + useEffect(() => { - if (streamUrl && chat.message === "") { + console.log(streamUrl) + if (streamUrl&& !isStreaming) { streamChunks(streamUrl) .then(() => { if (updateChat) { - updateChat(chat, chatMessage, streamUrl); + console.log("rodou") + updateChat(chat, chatMessage); } }) .catch((error) => { console.error(error); }); } - }, [streamUrl]); - - useEffect(() => { - // This effect is specifically for calling updateChat after streaming ends - if (!isStreaming && streamUrl) { - if (updateChat) { - updateChat(chat, chatMessage, streamUrl); - } + return () => { + eventSource.current?.close(); } - }, [isStreaming]); + }, [streamUrl,chatMessage]); useEffect(() => { const element = document.getElementById("last-chat-message"); diff --git a/src/frontend/src/components/newChatView/index.tsx b/src/frontend/src/components/newChatView/index.tsx index 86b8ace6a..b1b41b8fc 100644 --- a/src/frontend/src/components/newChatView/index.tsx +++ b/src/frontend/src/components/newChatView/index.tsx @@ -34,6 +34,7 @@ export default function NewChatView({ const inputIds = inputs.map((obj) => obj.id); const outputIds = outputs.map((obj) => obj.id); const outputTypes = outputs.map((obj) => obj.type); + const updateFlowPool = useFlowStore((state)=>state.updateFlowPool) useEffect(() => { if (!outputTypes.includes("ChatOutput")) { @@ -67,14 +68,12 @@ export default function NewChatView({ const { sender, message, sender_name, stream_url } = output.data .artifacts as ChatOutputType; - const componentId = output.id + index; - const is_ai = sender === "Machine" || sender === null; return { isSend: !is_ai, message: message, sender_name, - id: componentId, + componentId: output.id, stream_url: stream_url, }; } catch (e) { @@ -83,7 +82,7 @@ export default function NewChatView({ isSend: false, message: "Error parsing message", sender_name: "Error", - id: output.id + index, + componentId: output.id, }; } }); @@ -120,27 +119,25 @@ export default function NewChatView({ function updateChat( chat: ChatMessageType, message: string, - stream_url: string | null - ) { - if (message === "") return; - console.log(`updateChat: ${message}`); - console.log("chatHistory:", chatHistory); - chat.message = message; - chat.stream_url = stream_url; + stream_url?: string + ) { + if (message === "") return; + chat.message = message; + console.log(message) // chat is one of the chatHistory - setChatHistory((oldChatHistory) => { - const index = oldChatHistory.findIndex((ch) => ch.id === chat.id); - - if (index === -1) return oldChatHistory; - let newChatHistory = _.cloneDeep(oldChatHistory); - newChatHistory = [ - ...newChatHistory.slice(0, index), - chat, - ...newChatHistory.slice(index + 1), - ]; - console.log("newChatHistory:", newChatHistory); - return newChatHistory; - }); + updateFlowPool(chat.componentId,{message,sender_name:chat.sender_name??"Bot",sender:"Machine"}) + // setChatHistory((oldChatHistory) => { + // const index = oldChatHistory.findIndex((ch) => ch.id === chat.id); + // if (index === -1) return oldChatHistory; + // let newChatHistory = _.cloneDeep(oldChatHistory); + // newChatHistory = [ + // ...newChatHistory.slice(0, index), + // chat, + // ...newChatHistory.slice(index + 1), + // ]; + // console.log("newChatHistory:", newChatHistory); + // return newChatHistory; + // }); } return ( @@ -167,7 +164,7 @@ export default function NewChatView({ lockChat={lockChat} chat={chat} lastMessage={chatHistory.length - 1 === index ? true : false} - key={`${chat.id}-${index}`} + key={`${chat.componentId}-${index}`} updateChat={updateChat} /> )) diff --git a/src/frontend/src/stores/flowStore.ts b/src/frontend/src/stores/flowStore.ts index 351997aaf..f9a211e86 100644 --- a/src/frontend/src/stores/flowStore.ts +++ b/src/frontend/src/stores/flowStore.ts @@ -51,7 +51,7 @@ const useFlowStore = create((set, get) => ({ setFlowPool: (flowPool) => { set({ flowPool }); }, - addDataToFlowPool: (data: any, nodeId: string) => { + addDataToFlowPool: (data: FlowPoolObjectType, nodeId: string) => { let newFlowPool = cloneDeep({ ...get().flowPool }); if (!newFlowPool[nodeId]) newFlowPool[nodeId] = [data]; else { @@ -416,12 +416,13 @@ const useFlowStore = create((set, get) => ({ } function handleBuildUpdate( vertexBuildData: VertexBuildTypeAPI, - status: BuildStatus + status: BuildStatus, + buildId:string ) { if (vertexBuildData && vertexBuildData.inactive_vertices) { get().removeFromVerticesBuild(vertexBuildData.inactive_vertices); } - get().addDataToFlowPool(vertexBuildData, vertexBuildData.id); + get().addDataToFlowPool({...vertexBuildData,buildId}, vertexBuildData.id); useFlowStore.getState().updateBuildStatus([vertexBuildData.id], status); } await updateFlowInDatabase({ diff --git a/src/frontend/src/types/chat/index.ts b/src/frontend/src/types/chat/index.ts index 3914dd2dc..e24c6e891 100644 --- a/src/frontend/src/types/chat/index.ts +++ b/src/frontend/src/types/chat/index.ts @@ -9,7 +9,7 @@ export type ChatMessageType = { files?: Array<{ data: string; type: string; data_type: string }>; prompt?: string; chatKey?: string; - id?: string; + componentId: string; stream_url?: string | null; sender_name?: string; }; diff --git a/src/frontend/src/types/components/index.ts b/src/frontend/src/types/components/index.ts index e7767eabb..b36b5d44c 100644 --- a/src/frontend/src/types/components/index.ts +++ b/src/frontend/src/types/components/index.ts @@ -527,7 +527,7 @@ export type chatMessagePropsType = { updateChat: ( chat: ChatMessageType, message: string, - stream_url: string + stream_url?: string ) => void; }; @@ -632,9 +632,9 @@ export type validationStatusType = { id: string; data: object | any; params: string; - progress: number; + progress?: number; valid: boolean; - duration: string; + duration?: string; }; export type ApiKey = { diff --git a/src/frontend/src/types/zustand/flow/index.ts b/src/frontend/src/types/zustand/flow/index.ts index 8b6093b51..c506b034f 100644 --- a/src/frontend/src/types/zustand/flow/index.ts +++ b/src/frontend/src/types/zustand/flow/index.ts @@ -18,6 +18,7 @@ export type ChatOutputType = { message: string; sender: string; sender_name: string; + stream_url?: string; }; export type FlowPoolObjectType = { @@ -25,9 +26,10 @@ export type FlowPoolObjectType = { valid: boolean; params: any; data: { artifacts: any | ChatOutputType | chatInputType; results: any | ChatOutputType | chatInputType }; - duration: string; - progress: number; + duration?: string; + progress?: number; id: string; + buildId: string; }; export type FlowPoolType = { @@ -40,7 +42,7 @@ export type FlowStoreType = { outputs: Array<{ type: string; id: string }>; hasIO: boolean; setFlowPool: (flowPool: FlowPoolType) => void; - addDataToFlowPool: (data: any, nodeId: string) => void; + addDataToFlowPool: (data: FlowPoolObjectType, nodeId: string) => void; CleanFlowPool: () => void; isBuilding: boolean; isPending: boolean; diff --git a/src/frontend/src/utils/buildUtils.ts b/src/frontend/src/utils/buildUtils.ts index 862472ee7..3acc3d972 100644 --- a/src/frontend/src/utils/buildUtils.ts +++ b/src/frontend/src/utils/buildUtils.ts @@ -9,7 +9,7 @@ type BuildVerticesParams = { flowId: string; // Assuming FlowType is the type for your flow nodeId?: string | null; // Assuming nodeId is of type string, and it's optional onGetOrderSuccess?: () => void; - onBuildUpdate?: (data: VertexBuildTypeAPI, status: BuildStatus) => void; // Replace any with the actual type if it's not any + onBuildUpdate?: (data: VertexBuildTypeAPI, status: BuildStatus,buildId:string) => void; // Replace any with the actual type if it's not any onBuildComplete?: (allNodesValid: boolean) => void; onBuildError?: (title, list, idList: string[]) => void; onBuildStart?: (idList: string[]) => void; @@ -48,7 +48,7 @@ export async function buildVertices({ let orderResponse; try { orderResponse = await getVerticesOrder(flowId, nodeId); - } catch (error) { + } catch (error:any) { console.log(error); setErrorData({ title: "Oops! Looks like you missed something", @@ -59,6 +59,7 @@ export async function buildVertices({ } if (onGetOrderSuccess) onGetOrderSuccess(); let verticesOrder: Array> = orderResponse.data.ids; + const runId = orderResponse.data.run_id; let vertices_layers: Array> = []; let stop = false; if (validateNodes) { @@ -102,14 +103,14 @@ export async function buildVertices({ onBuildUpdate ) { // If it is, skip building and set the state to inactive - onBuildUpdate(getInactiveVertexData(id), BuildStatus.INACTIVE); + onBuildUpdate(getInactiveVertexData(id), BuildStatus.INACTIVE,runId); buildResults.push(false); continue; } await buildVertex({ flowId, id, - onBuildUpdate, + onBuildUpdate:(data: VertexBuildTypeAPI, status: BuildStatus) => {if(onBuildUpdate) onBuildUpdate(data, status,runId)}, onBuildError, verticesIds, buildResults, From 3c56b7a2b735ad85aa81d02ef881d643860dfbbd Mon Sep 17 00:00:00 2001 From: anovazzi1 Date: Tue, 27 Feb 2024 20:37:57 -0300 Subject: [PATCH 20/28] fix premature closing of websocket --- .../src/components/newChatView/chatMessage/index.tsx | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/src/frontend/src/components/newChatView/chatMessage/index.tsx b/src/frontend/src/components/newChatView/chatMessage/index.tsx index c2819a3a1..59befcb1b 100644 --- a/src/frontend/src/components/newChatView/chatMessage/index.tsx +++ b/src/frontend/src/components/newChatView/chatMessage/index.tsx @@ -78,10 +78,13 @@ export default function ChatMessage({ console.error(error); }); } + }, [streamUrl,chatMessage]); + + useEffect(()=>{ return () => { eventSource.current?.close(); } - }, [streamUrl,chatMessage]); + },[]) useEffect(() => { const element = document.getElementById("last-chat-message"); From 915de79cf24e3dd946c0b5014e98a93228674353 Mon Sep 17 00:00:00 2001 From: Gabriel Luiz Freitas Almeida Date: Tue, 27 Feb 2024 20:39:07 -0300 Subject: [PATCH 21/28] Refactor logger configuration logic --- src/backend/langflow/utils/logger.py | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/src/backend/langflow/utils/logger.py b/src/backend/langflow/utils/logger.py index 060ad9731..6755e53b1 100644 --- a/src/backend/langflow/utils/logger.py +++ b/src/backend/langflow/utils/logger.py @@ -25,7 +25,10 @@ def patching(record): def configure(log_level: Optional[str] = None, log_file: Optional[Path] = None): - if os.getenv("LANGFLOW_LOG_LEVEL") in VALID_LOG_LEVELS and log_level is None: + if ( + os.getenv("LANGFLOW_LOG_LEVEL", "").upper() in VALID_LOG_LEVELS + and log_level is None + ): log_level = os.getenv("LANGFLOW_LOG_LEVEL") if log_level is None: log_level = "INFO" From b2549d92fc6232ec26495322c381d26cea7fe1c9 Mon Sep 17 00:00:00 2001 From: Gabriel Luiz Freitas Almeida Date: Tue, 27 Feb 2024 20:39:39 -0300 Subject: [PATCH 22/28] Refactor custom component building functions --- .../langflow/interface/custom/utils.py | 67 ++++++++++++++----- 1 file changed, 50 insertions(+), 17 deletions(-) diff --git a/src/backend/langflow/interface/custom/utils.py b/src/backend/langflow/interface/custom/utils.py index d2be47333..381193116 100644 --- a/src/backend/langflow/interface/custom/utils.py +++ b/src/backend/langflow/interface/custom/utils.py @@ -27,14 +27,18 @@ from langflow.utils import validate from langflow.utils.util import get_base_classes -def add_output_types(frontend_node: CustomComponentFrontendNode, return_types: List[str]): +def add_output_types( + frontend_node: CustomComponentFrontendNode, return_types: List[str] +): """Add output types to the frontend node""" for return_type in return_types: if return_type is None: raise HTTPException( status_code=400, detail={ - "error": ("Invalid return type. Please check your code and try again."), + "error": ( + "Invalid return type. Please check your code and try again." + ), "traceback": traceback.format_exc(), }, ) @@ -63,14 +67,18 @@ def reorder_fields(frontend_node: CustomComponentFrontendNode, field_order: List frontend_node.template.fields = reordered_fields -def add_base_classes(frontend_node: CustomComponentFrontendNode, return_types: List[str]): +def add_base_classes( + frontend_node: CustomComponentFrontendNode, return_types: List[str] +): """Add base classes to the frontend node""" for return_type_instance in return_types: if return_type_instance is None: raise HTTPException( status_code=400, detail={ - "error": ("Invalid return type. Please check your code and try again."), + "error": ( + "Invalid return type. Please check your code and try again." + ), "traceback": traceback.format_exc(), }, ) @@ -145,10 +153,14 @@ def add_new_custom_field( # If options is a list, then it's a dropdown # If options is None, then it's a list of strings is_list = isinstance(field_config.get("options"), list) - field_config["is_list"] = is_list or field_config.get("is_list", False) or field_contains_list + field_config["is_list"] = ( + is_list or field_config.get("is_list", False) or field_contains_list + ) if "name" in field_config: - warnings.warn("The 'name' key in field_config is used to build the object and can't be changed.") + warnings.warn( + "The 'name' key in field_config is used to build the object and can't be changed." + ) required = field_config.pop("required", field_required) placeholder = field_config.pop("placeholder", "") @@ -179,7 +191,9 @@ def add_extra_fields(frontend_node, field_config, function_args): if "name" not in extra_field or extra_field["name"] == "self": continue - field_name, field_type, field_value, field_required = get_field_properties(extra_field) + field_name, field_type, field_value, field_required = get_field_properties( + extra_field + ) config = field_config.get(field_name, {}) frontend_node = add_new_custom_field( frontend_node, @@ -217,7 +231,9 @@ def run_build_config( raise HTTPException( status_code=400, detail={ - "error": ("Invalid type convertion. Please check your code and try again."), + "error": ( + "Invalid type convertion. Please check your code and try again." + ), "traceback": traceback.format_exc(), }, ) from exc @@ -245,7 +261,9 @@ def run_build_config( raise HTTPException( status_code=400, detail={ - "error": ("Invalid type convertion. Please check your code and try again."), + "error": ( + "Invalid type convertion. Please check your code and try again." + ), "traceback": traceback.format_exc(), }, ) from exc @@ -300,16 +318,24 @@ def build_custom_component_template( frontend_node = build_frontend_node(custom_component.template_config) logger.debug("Updated attributes") - field_config, custom_instance = run_build_config(custom_component, user_id=user_id, update_field=update_field) + field_config, custom_instance = run_build_config( + custom_component, user_id=user_id, update_field=update_field + ) logger.debug("Built field config") entrypoint_args = custom_component.get_function_entrypoint_args add_extra_fields(frontend_node, field_config, entrypoint_args) - frontend_node = add_code_field(frontend_node, custom_component.code, field_config.get("code", {})) + frontend_node = add_code_field( + frontend_node, custom_component.code, field_config.get("code", {}) + ) - add_base_classes(frontend_node, custom_component.get_function_entrypoint_return_type) - add_output_types(frontend_node, custom_component.get_function_entrypoint_return_type) + add_base_classes( + frontend_node, custom_component.get_function_entrypoint_return_type + ) + add_output_types( + frontend_node, custom_component.get_function_entrypoint_return_type + ) logger.debug("Added base classes") reorder_fields(frontend_node, custom_instance._get_field_order()) @@ -321,7 +347,9 @@ def build_custom_component_template( raise HTTPException( status_code=400, detail={ - "error": ("Invalid type convertion. Please check your code and try again."), + "error": ( + "Invalid type convertion. Please check your code and try again." + ), "traceback": traceback.format_exc(), }, ) from exc @@ -345,7 +373,9 @@ def build_custom_components(settings_service): if not settings_service.settings.COMPONENTS_PATH: return {} - logger.info(f"Building custom components from {settings_service.settings.COMPONENTS_PATH}") + logger.info( + f"Building custom components from {settings_service.settings.COMPONENTS_PATH}" + ) custom_components_from_file = {} processed_paths = set() for path in settings_service.settings.COMPONENTS_PATH: @@ -356,7 +386,9 @@ def build_custom_components(settings_service): custom_component_dict = build_custom_component_list_from_path(path_str) if custom_component_dict: category = next(iter(custom_component_dict)) - logger.info(f"Loading {len(custom_component_dict[category])} component(s) from category {category}") + logger.info( + f"Loading {len(custom_component_dict[category])} component(s) from category {category}" + ) custom_components_from_file = merge_nested_dicts_with_renaming( custom_components_from_file, custom_component_dict ) @@ -400,8 +432,9 @@ def sanitize_field_config(field_config: Dict): def build_component(component): """Build a single component.""" component_name = determine_component_name(component) + logger.debug(f"Building component: {component_name}") component_template = create_component_template(component) - logger.debug(f"Building component: {component_name, component.get('output_types')}") + return component_name, component_template From 8b7ee7225206aeda99c54f0bec6527c8d4574f6c Mon Sep 17 00:00:00 2001 From: Gabriel Luiz Freitas Almeida Date: Tue, 27 Feb 2024 20:39:59 -0300 Subject: [PATCH 23/28] Refactor ChatVertex class to update artifacts and params --- src/backend/langflow/graph/vertex/types.py | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/src/backend/langflow/graph/vertex/types.py b/src/backend/langflow/graph/vertex/types.py index d60b7412c..100390d11 100644 --- a/src/backend/langflow/graph/vertex/types.py +++ b/src/backend/langflow/graph/vertex/types.py @@ -424,15 +424,16 @@ class ChatVertex(StatelessVertex): message = message.text if hasattr(message, "text") else message yield message complete_message += message - self._built_object = Record(text=complete_message, data=self.artifacts) - self._built_result = complete_message - # Update artifacts with the message - # and remove the stream_url self.artifacts = ChatOutputResponse( message=complete_message, sender=self.params.get("sender", ""), sender_name=self.params.get("sender_name", ""), ).model_dump() + self.params[INPUT_FIELD_NAME] = complete_message + self._built_object = Record(text=complete_message, data=self.artifacts) + self._built_result = complete_message + # Update artifacts with the message + # and remove the stream_url logger.debug(f"Streamed message: {complete_message}") await log_message( @@ -443,6 +444,9 @@ class ChatVertex(StatelessVertex): artifacts=self.artifacts, ) + self._validate_built_object() + self._built = True + class RoutingVertex(StatelessVertex): def __init__(self, data: Dict, graph): From 12fe2ffbe6c8e3fb33b7e47ed1fbf9eeb662ea1b Mon Sep 17 00:00:00 2001 From: Gabriel Luiz Freitas Almeida Date: Tue, 27 Feb 2024 20:40:11 -0300 Subject: [PATCH 24/28] Update langflow.graph.schema import --- src/backend/langflow/graph/graph/base.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/backend/langflow/graph/graph/base.py b/src/backend/langflow/graph/graph/base.py index 0ffc9825b..3f5e376a5 100644 --- a/src/backend/langflow/graph/graph/base.py +++ b/src/backend/langflow/graph/graph/base.py @@ -8,7 +8,7 @@ from loguru import logger from langflow.graph.edge.base import ContractEdge from langflow.graph.graph.constants import lazy_load_vertex_dict from langflow.graph.graph.utils import process_flow -from langflow.graph.schema import InterfaceComponentTypes +from langflow.graph.schema import INPUT_FIELD_NAME, InterfaceComponentTypes from langflow.graph.vertex.base import Vertex from langflow.graph.vertex.types import ( ChatVertex, From 34d48992d576f685d9a5c5c5a9cb5239b529c54c Mon Sep 17 00:00:00 2001 From: Gabriel Luiz Freitas Almeida Date: Tue, 27 Feb 2024 20:40:24 -0300 Subject: [PATCH 25/28] Refactor chat.py to support vertex streaming --- src/backend/langflow/api/v1/chat.py | 28 ++++++++++++++++++++++------ 1 file changed, 22 insertions(+), 6 deletions(-) diff --git a/src/backend/langflow/api/v1/chat.py b/src/backend/langflow/api/v1/chat.py index 7f85bb211..e1b52aeb5 100644 --- a/src/backend/langflow/api/v1/chat.py +++ b/src/backend/langflow/api/v1/chat.py @@ -1,6 +1,6 @@ import time -from typing import Optional import uuid +from typing import TYPE_CHECKING, Optional from fastapi import ( APIRouter, @@ -35,6 +35,9 @@ from langflow.services.chat.service import ChatService from langflow.services.deps import get_chat_service, get_session from langflow.services.monitor.utils import log_vertex_build +if TYPE_CHECKING: + from langflow.graph.vertex.types import ChatVertex + router = APIRouter(tags=["Chat"]) @@ -237,27 +240,40 @@ async def build_vertex_stream( else: graph = cache.get("result") - vertex = graph.get_vertex(vertex_id) + vertex: "ChatVertex" = graph.get_vertex(vertex_id) + if not hasattr(vertex, "stream"): + raise ValueError(f"Vertex {vertex_id} does not support streaming") if not vertex.pinned or not vertex._built: + logger.debug(f"Streaming vertex {vertex_id}") stream_data = StreamData( event="message", - data={"message": "Building vertex"}, + data={"message": f"Streaming vertex {vertex_id}"}, ) yield str(stream_data) - + number_of_chunks = 0 async for chunk in vertex.stream(): stream_data = StreamData( event="message", data={"chunk": chunk}, ) + number_of_chunks += 1 yield str(stream_data) + logger.debug(f"Number of chunks: {number_of_chunks}") + elif vertex.result is not None: + stream_data = StreamData( + event="message", + data={"chunk": vertex._built_result}, + ) + yield str(stream_data) else: raise ValueError(f"No result found for vertex {vertex_id}") except Exception as exc: + logger.error(f"Error building vertex: {exc}") yield str(StreamData(event="error", data={"error": str(exc)})) - - yield str(StreamData(event="close", data={"message": "Stream closed"})) + finally: + logger.debug("Closing stream") + yield str(StreamData(event="close", data={"message": "Stream closed"})) return StreamingResponse(stream_vertex(), media_type="text/event-stream") except Exception as exc: From b42caa27ad17f4a56e4246fbfd92883d660653c0 Mon Sep 17 00:00:00 2001 From: Gabriel Luiz Freitas Almeida Date: Tue, 27 Feb 2024 20:40:49 -0300 Subject: [PATCH 26/28] Add LANGFLOW_LOG_LEVEL debug to launch.json --- .vscode/launch.json | 3 +++ 1 file changed, 3 insertions(+) diff --git a/.vscode/launch.json b/.vscode/launch.json index a8229b155..3332b67e9 100644 --- a/.vscode/launch.json +++ b/.vscode/launch.json @@ -17,6 +17,9 @@ ], "jinja": true, "justMyCode": true, + "env": { + "LANGFLOW_LOG_LEVEL": "debug" + }, "envFile": "${workspaceFolder}/.env" }, { From 42338a829b375d51d06e0ef3e719978e1180ceab Mon Sep 17 00:00:00 2001 From: anovazzi1 Date: Tue, 27 Feb 2024 20:50:22 -0300 Subject: [PATCH 27/28] Fix useRef typo and update chat message reference --- .../newChatView/chatMessage/index.tsx | 77 ++++++++++--------- .../src/components/newChatView/index.tsx | 1 - 2 files changed, 41 insertions(+), 37 deletions(-) diff --git a/src/frontend/src/components/newChatView/chatMessage/index.tsx b/src/frontend/src/components/newChatView/chatMessage/index.tsx index 59befcb1b..8b2c181f5 100644 --- a/src/frontend/src/components/newChatView/chatMessage/index.tsx +++ b/src/frontend/src/components/newChatView/chatMessage/index.tsx @@ -1,5 +1,5 @@ import Convert from "ansi-to-html"; -import { useEffect, useMemo, useState,useRef } from "react"; +import { useEffect, useMemo, useState, useRef } from "react"; import Markdown from "react-markdown"; import rehypeMathjax from "rehype-mathjax"; import remarkGfm from "remark-gfm"; @@ -32,6 +32,13 @@ export default function ChatMessage({ const [isStreaming, setIsStreaming] = useState(false); const eventSource = useRef(undefined); const updateFlowPool = useFlowStore((state) => state.updateFlowPool); + const chatMessageRef = useRef(chatMessage); + + // Sync ref with state + useEffect(() => { + chatMessageRef.current = chatMessage; + }, [chatMessage]); + // The idea now is that chat.stream_url MAY be a URL if we should stream the output of the chat @@ -63,28 +70,26 @@ export default function ChatMessage({ }; - useEffect(() => { - console.log(streamUrl) - if (streamUrl&& !isStreaming) { + console.log("chatMessage", chatMessage); + if (streamUrl && !isStreaming) { streamChunks(streamUrl) .then(() => { if (updateChat) { - console.log("rodou") - updateChat(chat, chatMessage); + updateChat(chat, chatMessageRef.current); } }) .catch((error) => { console.error(error); }); } - }, [streamUrl,chatMessage]); + }, [streamUrl, chatMessage]); - useEffect(()=>{ + useEffect(() => { return () => { eventSource.current?.close(); } - },[]) + }, []) useEffect(() => { const element = document.getElementById("last-chat-message"); @@ -213,7 +218,7 @@ dark:prose-invert" }, ]} activeTab={"0"} - setActiveTab={() => {}} + setActiveTab={() => { }} /> ) : ( @@ -270,33 +275,33 @@ dark:prose-invert" {promptOpen ? template?.split("\n")?.map((line, index) => { - const regex = /{([^}]+)}/g; - let match; - let parts: Array = []; - let lastIndex = 0; - while ((match = regex.exec(line)) !== null) { - // Push text up to the match - if (match.index !== lastIndex) { - parts.push(line.substring(lastIndex, match.index)); - } - // Push div with matched text - if (chat.message[match[1]]) { - parts.push( - - {chat.message[match[1]]} - - ); - } + const regex = /{([^}]+)}/g; + let match; + let parts: Array = []; + let lastIndex = 0; + while ((match = regex.exec(line)) !== null) { + // Push text up to the match + if (match.index !== lastIndex) { + parts.push(line.substring(lastIndex, match.index)); + } + // Push div with matched text + if (chat.message[match[1]]) { + parts.push( + + {chat.message[match[1]]} + + ); + } - // Update last index - lastIndex = regex.lastIndex; - } - // Push text after the last match - if (lastIndex !== line.length) { - parts.push(line.substring(lastIndex)); - } - return

{parts}

; - }) + // Update last index + lastIndex = regex.lastIndex; + } + // Push text after the last match + if (lastIndex !== line.length) { + parts.push(line.substring(lastIndex)); + } + return

{parts}

; + }) : chatMessage}
diff --git a/src/frontend/src/components/newChatView/index.tsx b/src/frontend/src/components/newChatView/index.tsx index b1b41b8fc..72a203cb8 100644 --- a/src/frontend/src/components/newChatView/index.tsx +++ b/src/frontend/src/components/newChatView/index.tsx @@ -123,7 +123,6 @@ export default function NewChatView({ ) { if (message === "") return; chat.message = message; - console.log(message) // chat is one of the chatHistory updateFlowPool(chat.componentId,{message,sender_name:chat.sender_name??"Bot",sender:"Machine"}) // setChatHistory((oldChatHistory) => { From e79c5771ee83f60aaafb971dc5f47bf6d2142caa Mon Sep 17 00:00:00 2001 From: anovazzi1 Date: Tue, 27 Feb 2024 20:58:03 -0300 Subject: [PATCH 28/28] Add setLockChat prop to ChatMessage component --- src/frontend/src/components/newChatView/chatMessage/index.tsx | 4 ++++ src/frontend/src/components/newChatView/index.tsx | 1 + src/frontend/src/types/components/index.ts | 1 + 3 files changed, 6 insertions(+) diff --git a/src/frontend/src/components/newChatView/chatMessage/index.tsx b/src/frontend/src/components/newChatView/chatMessage/index.tsx index 8b2c181f5..4f6d8e534 100644 --- a/src/frontend/src/components/newChatView/chatMessage/index.tsx +++ b/src/frontend/src/components/newChatView/chatMessage/index.tsx @@ -19,6 +19,7 @@ export default function ChatMessage({ lockChat, lastMessage, updateChat, + setLockChat }: chatMessagePropsType): JSX.Element { const convert = new Convert({ newline: true }); const [hidden, setHidden] = useState(true); @@ -73,14 +74,17 @@ export default function ChatMessage({ useEffect(() => { console.log("chatMessage", chatMessage); if (streamUrl && !isStreaming) { + setLockChat(true); streamChunks(streamUrl) .then(() => { + setLockChat(false); if (updateChat) { updateChat(chat, chatMessageRef.current); } }) .catch((error) => { console.error(error); + setLockChat(false); }); } }, [streamUrl, chatMessage]); diff --git a/src/frontend/src/components/newChatView/index.tsx b/src/frontend/src/components/newChatView/index.tsx index 72a203cb8..b1f888f1c 100644 --- a/src/frontend/src/components/newChatView/index.tsx +++ b/src/frontend/src/components/newChatView/index.tsx @@ -160,6 +160,7 @@ export default function NewChatView({ {chatHistory?.length > 0 ? ( chatHistory.map((chat, index) => ( void; updateChat: ( chat: ChatMessageType, message: string,