jyapayne/langflow
1
0
Fork 0
Code Issues Pull requests Projects Releases 1 Packages Wiki Activity Actions

Merge branch 'dev' into python_custom_node_component

Browse source
This commit is contained in:
gustavoschaedler 2023-07-12 16:34:21 +01:00
commit a6fe4091ac
78 changed files with 1415 additions and 1669 deletions
Download patch file Download diff file Expand all files Collapse all files

14
src/backend/langflow/api/v1/base.py
View file

@ -60,12 +60,26 @@ INVALID_CHARACTERS = {
"}",
}
INVALID_NAMES = {
"input_variables",
"output_parser",
"partial_variables",
"template",
"template_format",
"validate_template",
}
def validate_prompt(template: str):
input_variables = extract_input_variables_from_prompt(template)
# Check if there are invalid characters in the input_variables
input_variables = check_input_variables(input_variables)
if any(var in INVALID_NAMES for var in input_variables):
raise ValueError(
f"Invalid input variables. None of the variables can be named {', '.join(input_variables)}. "
)
try:
PromptTemplate(template=template, input_variables=input_variables)
except Exception as exc:

29
src/backend/langflow/api/v1/callback.py
View file

@ -10,6 +10,7 @@ from fastapi import WebSocket
from langchain.schema import AgentAction, LLMResult, AgentFinish
from langflow.utils.logger import logger
# https://github.com/hwchase17/chat-langchain/blob/master/callback.py
@ -62,12 +63,36 @@ class AsyncStreamingLLMCallbackHandler(AsyncCallbackHandler):
async def on_tool_end(self, output: str, **kwargs: Any) -> Any:
"""Run when tool ends running."""
observation_prefix = kwargs.get("observation_prefix", "Tool output: ")
split_output = output.split()
first_word = split_output[0]
rest_of_output = split_output[1:]
# Create a formatted message.
intermediate_steps = f"{observation_prefix}{first_word}"
# Create a ChatResponse instance.
resp = ChatResponse(
message="",
type="stream",
intermediate_steps=f"Tool output: {output}",
intermediate_steps=intermediate_steps,
)
await self.websocket.send_json(resp.dict())
rest_of_resps = [
ChatResponse(
message="",
type="stream",
intermediate_steps=f"{word}",
)
for word in rest_of_output
]
resps = [resp] + rest_of_resps
# Try to send the response, handle potential errors.
try:
# This is to emulate the stream of tokens
for resp in resps:
await self.websocket.send_json(resp.dict())
except Exception as e:
logger.error(e)
async def on_tool_error(
self, error: Union[Exception, KeyboardInterrupt], **kwargs: Any

9
src/backend/langflow/api/v1/chat.py
View file

@ -115,13 +115,6 @@ async def stream_build(flow_id: str):
number_of_nodes = len(graph.nodes)
flow_data_store[flow_id]["status"] = BuildStatus.IN_PROGRESS
# To deal with the ZeroShotAgent case
# we need to build the root node first
# and then the rest of the graph
# This is a big problem because certain nodes require
# params that are not connected to it.
# We should consider connecting the tools to the ZeroShotPrompt
graph.build()
for i, vertex in enumerate(graph.generator_build(), 1):
try:
@ -133,7 +126,7 @@ async def stream_build(flow_id: str):
params = vertex._built_object_repr()
valid = True
logger.debug(
f"Building node {params[:50]}{'...' if len(params) > 50 else ''}"
f"Building node {str(params)[:50]}{'...' if len(str(params)) > 50 else ''}"
)
if vertex.artifacts:
# The artifacts will be prompt variables

10
src/backend/langflow/chat/manager.py
View file

@ -104,8 +104,14 @@ class ChatManager:
async def close_connection(self, client_id: str, code: int, reason: str):
if websocket := self.active_connections[client_id]:
await websocket.close(code=code, reason=reason)
self.disconnect(client_id)
try:
await websocket.close(code=code, reason=reason)
self.disconnect(client_id)
except RuntimeError as exc:
# This is to catch the following error:
# Unexpected ASGI message 'websocket.close', after sending 'websocket.close'
if "after sending" in str(exc):
logger.error(exc)
async def process_message(
self, client_id: str, payload: Dict, langchain_object: Any

5
src/backend/langflow/config.yaml
View file

@ -277,9 +277,14 @@ vectorstores:
documentation: "https://python.langchain.com/docs/modules/data_connection/vectorstores/integrations/supabase"
MongoDBAtlasVectorSearch:
documentation: "https://python.langchain.com/docs/modules/data_connection/vectorstores/integrations/mongodb_atlas"
# Requires docarray >=0.32.0 but langchain-serve requires jina 3.15.2 which doesn't support docarray >=0.32.0
# DocArrayInMemorySearch:
# documentation: "https://python.langchain.com/docs/modules/data_connection/vectorstores/integrations/docarray_in_memory"
wrappers:
RequestsWrapper:
documentation: ""
SQLDatabase:
documentation: ""
output_parsers:
StructuredOutputParser:
documentation: "https://python.langchain.com/docs/modules/model_io/output_parsers/structured"

7
src/backend/langflow/graph/graph/base.py
View file

@ -214,3 +214,10 @@ class Graph:
if node_type in node_types:
children.append(node)
return children
def __repr__(self):
node_ids = [node.id for node in self.nodes]
edges_repr = "\n".join(
[f"{edge.source.id} --> {edge.target.id}" for edge in self.edges]
)
return f"Graph:\nNodes: {node_ids}\nConnections:\n{edges_repr}"

152
src/backend/langflow/graph/vertex/base.py
View file

@ -123,89 +123,119 @@ class Vertex:
self.params = params
def _build(self):
# The params dict is used to build the module
# it contains values and keys that point to nodes which
# have their own params dict
# When build is called, we iterate through the params dict
# and if the value is a node, we call build on that node
# and use the output of that build as the value for the param
# if the value is not a node, then we use the value as the param
# and continue
# Another aspect is that the node_type is the class that we need to import
# and instantiate with these built params
"""
Initiate the build process.
"""
logger.debug(f"Building {self.vertex_type}")
# Build each node in the params dict
self._build_each_node_in_params_dict()
self._get_and_instantiate_class()
self._validate_built_object()
self._built = True
def _build_each_node_in_params_dict(self):
"""
Iterates over each node in the params dictionary and builds it.
"""
for key, value in self.params.copy().items():
# Check if Node or list of Nodes and not self
# to avoid recursion
if isinstance(value, Vertex):
if self._is_node(value):
if value == self:
del self.params[key]
continue
result = value.build()
# If the key is "func", then we need to use the run method
if key == "func":
if not isinstance(result, types.FunctionType):
# func can be
# PythonFunction(code='\ndef upper_case(text: str) -> str:\n return text.upper()\n')
# so we need to check if there is an attribute called run
if hasattr(result, "run"):
result = result.run # type: ignore
elif hasattr(result, "get_function"):
result = result.get_function() # type: ignore
elif inspect.iscoroutinefunction(result):
self.params["coroutine"] = result
else:
# turn result which is a function into a coroutine
# so that it can be awaited
self.params["coroutine"] = sync_to_async(result)
if isinstance(result, list):
# If the result is a list, then we need to extend the list
# with the result but first check if the key exists
# if it doesn't, then we need to create a new list
if isinstance(self.params[key], list):
self.params[key].extend(result)
self._build_node_and_update_params(key, value)
elif isinstance(value, list) and self._is_list_of_nodes(value):
self._build_list_of_nodes_and_update_params(key, value)
self.params[key] = result
elif isinstance(value, list) and all(
isinstance(node, Vertex) for node in value
):
self.params[key] = []
for node in value:
built = node.build()
if isinstance(built, list):
self.params[key].extend(built)
else:
self.params[key].append(built)
def _is_node(self, value):
"""
Checks if the provided value is an instance of Vertex.
"""
return isinstance(value, Vertex)
# Get the class from LANGCHAIN_TYPES_DICT
# and instantiate it with the params
# and return the instance
def _is_list_of_nodes(self, value):
"""
Checks if the provided value is a list of Vertex instances.
"""
return all(self._is_node(node) for node in value)
def _build_node_and_update_params(self, key, node):
"""
Builds a given node and updates the params dictionary accordingly.
"""
result = node.build()
self._handle_func(key, result)
if isinstance(result, list):
self._extend_params_list_with_result(key, result)
self.params[key] = result
def _build_list_of_nodes_and_update_params(self, key, nodes):
"""
Iterates over a list of nodes, builds each and updates the params dictionary.
"""
self.params[key] = []
for node in nodes:
built = node.build()
if isinstance(built, list):
self.params[key].extend(built)
else:
self.params[key].append(built)
def _handle_func(self, key, result):
"""
Handles 'func' key by checking if the result is a function and setting it as coroutine.
"""
if key == "func":
if not isinstance(result, types.FunctionType):
if hasattr(result, "run"):
result = result.run # type: ignore
elif hasattr(result, "get_function"):
result = result.get_function() # type: ignore
elif inspect.iscoroutinefunction(result):
self.params["coroutine"] = result
else:
self.params["coroutine"] = sync_to_async(result)
def _extend_params_list_with_result(self, key, result):
"""
Extends a list in the params dictionary with the given result if it exists.
"""
if isinstance(self.params[key], list):
self.params[key].extend(result)
def _get_and_instantiate_class(self):
"""
Gets the class from a dictionary and instantiates it with the params.
"""
if self.base_type is None:
raise ValueError(f"Base type for node {self.vertex_type} not found")
try:
if self.base_type is None:
raise ValueError(f"Base type for node {self.vertex_type} not found")
result = loading.instantiate_class(
node_type=self.vertex_type,
base_type=self.base_type,
params=self.params,
)
# Result could be the _built_object or
# (_built_object, dict) tuple
if isinstance(result, tuple):
self._built_object, self.artifacts = result
else:
self._built_object = result
self._update_built_object_and_artifacts(result)
except Exception as exc:
raise ValueError(
f"Error building node {self.vertex_type}: {str(exc)}"
) from exc
def _update_built_object_and_artifacts(self, result):
"""
Updates the built object and its artifacts.
"""
if isinstance(result, tuple):
self._built_object, self.artifacts = result
else:
self._built_object = result
def _validate_built_object(self):
"""
Checks if the built object is None and raises a ValueError if so.
"""
if self._built_object is None:
raise ValueError(f"Node type {self.vertex_type} not found")
self._built = True
def build(self, force: bool = False) -> Any:
if not self._built or force:
self._build()

39
src/backend/langflow/graph/vertex/types.py
View file

@ -1,3 +1,4 @@
import ast
from typing import Any, Dict, List, Optional, Union
from langflow.graph.vertex.base import Vertex
@ -79,7 +80,7 @@ class WrapperVertex(Vertex):
def build(self, force: bool = False) -> Any:
if not self._built or force:
if "headers" in self.params:
self.params["headers"] = eval(self.params["headers"])
self.params["headers"] = ast.literal_eval(self.params["headers"])
self._build()
return self._built_object
@ -93,7 +94,11 @@ class DocumentLoaderVertex(Vertex):
# show how many documents are in the list?
if self._built_object:
avg_length = sum(len(doc.page_content) for doc in self._built_object) / len(
self._built_object
)
return f"""{self.vertex_type}({len(self._built_object)} documents)
\nAvg. Document Length (characters): {avg_length}
Documents: {self._built_object[:3]}..."""
return f"{self.vertex_type}()"
@ -125,8 +130,13 @@ class TextSplitterVertex(Vertex):
def _built_object_repr(self):
# This built_object is a list of documents. Maybe we should
# show how many documents are in the list?
if self._built_object:
avg_length = sum(len(doc.page_content) for doc in self._built_object) / len(
self._built_object
)
return f"""{self.vertex_type}({len(self._built_object)} documents)
\nAvg. Document Length (characters): {avg_length}
\nDocuments: {self._built_object[:3]}..."""
return f"{self.vertex_type}()"
@ -202,13 +212,28 @@ class PromptVertex(Vertex):
return self._built_object
def _built_object_repr(self):
if self.artifacts and hasattr(self._built_object, "format"):
# We'll build the prompt with the artifacts
# to show the user what the prompt looks like
# with the variables filled in
return self._built_object.format(**self.artifacts)
else:
if (
not self.artifacts
or self._built_object is None
or not hasattr(self._built_object, "format")
):
return super()._built_object_repr()
# We'll build the prompt with the artifacts
# to show the user what the prompt looks like
# with the variables filled in
artifacts = self.artifacts.copy()
# Remove the handle_keys from the artifacts
# so the prompt format doesn't break
artifacts.pop("handle_keys", None)
try:
template = self._built_object.format(**artifacts)
return (
template
if isinstance(template, str)
else f"{self.vertex_type}({template})"
)
except KeyError:
return str(self._built_object)
class OutputParserVertex(Vertex):

6
src/backend/langflow/interface/agents/custom.py
View file

@ -157,7 +157,7 @@ class VectorStoreAgent(CustomAgentExecutor):
llm_chain=llm_chain, allowed_tools=tool_names, **kwargs # type: ignore
)
return AgentExecutor.from_agent_and_tools(
agent=agent, tools=tools, verbose=True
agent=agent, tools=tools, verbose=True, handle_parsing_errors=True
)
def run(self, *args, **kwargs):
@ -232,6 +232,7 @@ class SQLAgent(CustomAgentExecutor):
verbose=True,
max_iterations=15,
early_stopping_method="force",
handle_parsing_errors=True,
)
def run(self, *args, **kwargs):
@ -276,7 +277,7 @@ class VectorStoreRouterAgent(CustomAgentExecutor):
llm_chain=llm_chain, allowed_tools=tool_names, **kwargs # type: ignore
)
return AgentExecutor.from_agent_and_tools(
agent=agent, tools=tools, verbose=True
agent=agent, tools=tools, verbose=True, handle_parsing_errors=True
)
def run(self, *args, **kwargs):
@ -308,6 +309,7 @@ class InitializeAgent(CustomAgentExecutor):
agent=agent, # type: ignore
memory=memory,
return_intermediate_steps=True,
handle_parsing_errors=True,
)
def __init__(self, *args, **kwargs):

1
src/backend/langflow/interface/chains/base.py
View file

@ -23,6 +23,7 @@ class ChainCreator(LangChainTypeCreator):
from_method_nodes = {
"ConversationalRetrievalChain": "from_llm",
"LLMCheckerChain": "from_llm",
"SQLDatabaseChain": "from_llm",
}
@property

7
src/backend/langflow/interface/importing/utils.py
View file

@ -11,6 +11,7 @@ from langchain.chat_models.base import BaseChatModel
from langchain.tools import BaseTool
from langflow.interface.custom.custom import CustomComponent
from langflow.utils import validate
from langflow.interface.wrappers.base import wrapper_creator
def import_module(module_path: str) -> Any:
@ -103,7 +104,11 @@ def import_prompt(prompt: str) -> Type[PromptTemplate]:
def import_wrapper(wrapper: str) -> Any:
"""Import wrapper from wrapper name"""
return import_module(f"from langchain.requests import {wrapper}")
if (
isinstance(wrapper_creator.type_dict, dict)
and wrapper in wrapper_creator.type_dict
):
return wrapper_creator.type_dict.get(wrapper)
def import_toolkit(toolkit: str) -> Any:

75
src/backend/langflow/interface/initialize/loading.py
View file

@ -1,3 +1,4 @@
import contextlib
import json
from typing import Any, Callable, Dict, List, Sequence, Type
@ -24,6 +25,7 @@ from langflow.interface.toolkits.base import toolkits_creator
from langflow.interface.chains.base import chain_creator
from langflow.interface.output_parsers.base import output_parser_creator
from langflow.interface.retrievers.base import retriever_creator
from langflow.interface.wrappers.base import wrapper_creator
from langflow.interface.utils import load_file_into_dict
from langflow.utils import validate
from langchain.chains.base import Chain
@ -70,7 +72,11 @@ def instantiate_based_on_type(class_object, base_type, node_type, params):
elif base_type == "prompts":
return instantiate_prompt(node_type, class_object, params)
elif base_type == "tools":
return instantiate_tool(node_type, class_object, params)
tool = instantiate_tool(node_type, class_object, params)
if hasattr(tool, "name") and isinstance(tool, BaseTool):
# tool name shouldn't contain spaces
tool.name = tool.name.replace(" ", "_")
return tool
elif base_type == "toolkits":
return instantiate_toolkit(node_type, class_object, params)
elif base_type == "embeddings":
@ -95,6 +101,8 @@ def instantiate_based_on_type(class_object, base_type, node_type, params):
return instantiate_memory(node_type, class_object, params)
elif base_type == "custom_components":
return instantiate_custom_component(node_type, class_object, params)
elif base_type == "wrappers":
return instantiate_wrapper(node_type, class_object, params)
else:
return class_object(**params)
@ -104,6 +112,15 @@ def instantiate_custom_component(node_type, class_object, params):
return class_object().build(**params)
def instantiate_wrapper(node_type, class_object, params):
if node_type in wrapper_creator.from_method_nodes:
method = wrapper_creator.from_method_nodes[node_type]
if class_method := getattr(class_object, method, None):
return class_method(**params)
raise ValueError(f"Method {method} not found in {class_object}")
return class_object(**params)
def instantiate_output_parser(node_type, class_object, params):
if node_type in output_parser_creator.from_method_nodes:
method = output_parser_creator.from_method_nodes[node_type]
@ -119,12 +136,21 @@ def instantiate_llm(node_type, class_object, params: Dict):
# False if condition is True
if node_type == "VertexAI":
return initialize_vertexai(class_object=class_object, params=params)
# max_tokens sometimes is a string and should be an int
if "max_tokens" in params:
if isinstance(params["max_tokens"], str) and params["max_tokens"].isdigit():
params["max_tokens"] = int(params["max_tokens"])
elif not isinstance(params.get("max_tokens"), int):
params.pop("max_tokens", None)
return class_object(**params)
def instantiate_memory(node_type, class_object, params):
# process input_key and output_key to remove them if
# they are empty strings
if node_type == "ConversationEntityMemory":
params.pop("memory_key", None)
for key in ["input_key", "output_key"]:
if key in params and (params[key] == "" or not params[key]):
params.pop(key)
@ -178,8 +204,7 @@ def instantiate_agent(node_type, class_object: Type[agent_module.Agent], params:
agent = class_method(**params)
tools = params.get("tools", [])
return AgentExecutor.from_agent_and_tools(
agent=agent,
tools=tools,
agent=agent, tools=tools, handle_parsing_errors=True
)
return load_agent_executor(class_object, params)
@ -189,7 +214,7 @@ def instantiate_prompt(node_type, class_object, params: Dict):
if "tools" not in params:
params["tools"] = []
return ZeroShotAgent.create_prompt(**params)
if "MessagePromptTemplate" in node_type:
elif "MessagePromptTemplate" in node_type:
# Then we only need the template
from_template_params = {
"template": params.pop("prompt", params.pop("template", ""))
@ -197,12 +222,12 @@ def instantiate_prompt(node_type, class_object, params: Dict):
if not from_template_params.get("template"):
raise ValueError("Prompt template is required")
return class_object.from_template(**from_template_params)
prompt = class_object.from_template(**from_template_params)
if node_type == "ChatPromptTemplate":
return class_object.from_messages(**params)
prompt = class_object(**params)
elif node_type == "ChatPromptTemplate":
prompt = class_object.from_messages(**params)
else:
prompt = class_object(**params)
format_kwargs: Dict[str, Any] = {}
for input_variable in prompt.input_variables:
@ -214,18 +239,23 @@ def instantiate_prompt(node_type, class_object, params: Dict):
variable, "get_format_instructions"
):
format_kwargs[input_variable] = variable.get_format_instructions()
# check if is a list of Document
elif isinstance(variable, List) and all(
isinstance(item, Document) for item in variable
):
# Format document to contain page_content and metadata
# as one string separated by a newline
format_kwargs[input_variable] = "\n".join(
[
f"Document:{item.page_content}\nMetadata:{item.metadata}"
for item in variable
]
)
if len(variable) > 1:
content = "\n".join(
[item.page_content for item in variable if item.page_content]
)
else:
content = variable[0].page_content
# content could be a json list of strings
with contextlib.suppress(json.JSONDecodeError):
content = json.loads(content)
if isinstance(content, list):
content = ",".join([str(item) for item in content])
format_kwargs[input_variable] = content
# handle_keys will be a list but it does not exist yet
# so we need to create it
@ -247,12 +277,14 @@ def instantiate_prompt(node_type, class_object, params: Dict):
def instantiate_tool(node_type, class_object: Type[BaseTool], params: Dict):
if node_type == "JsonSpec":
params["dict_"] = load_file_into_dict(params.pop("path"))
if file_dict := load_file_into_dict(params.pop("path")):
params["dict_"] = file_dict
else:
raise ValueError("Invalid file")
return class_object(**params)
elif node_type == "PythonFunctionTool":
params["func"] = get_function(params.get("code"))
return class_object(**params)
# For backward compatibility
elif node_type == "PythonFunction":
function_string = params["code"]
if isinstance(function_string, str):
@ -354,6 +386,12 @@ def instantiate_textsplitter(
"separator_type" in params and params["separator_type"] == "Text"
) or "separator_type" not in params:
params.pop("separator_type", None)
# separators might come in as an escaped string like \\n
# so we need to convert it to a string
if "separators" in params:
params["separators"] = (
params["separators"].encode().decode("unicode-escape")
)
text_splitter = class_object(**params)
else:
from langchain.text_splitter import Language
@ -406,6 +444,7 @@ def load_agent_executor(agent_class: type[agent_module.Agent], params, **kwargs)
return AgentExecutor.from_agent_and_tools(
agent=agent,
tools=allowed_tools,
handle_parsing_errors=True,
# memory=memory,
**kwargs,
)

20
src/backend/langflow/interface/utils.py
View file

@ -16,17 +16,15 @@ def load_file_into_dict(file_path: str) -> dict:
if not os.path.exists(file_path):
raise FileNotFoundError(f"File not found: {file_path}")
file_extension = os.path.splitext(file_path)[1].lower()
if file_extension == ".json":
with open(file_path, "r") as json_file:
data = json.load(json_file)
elif file_extension in [".yaml", ".yml"]:
with open(file_path, "r") as yaml_file:
data = yaml.safe_load(yaml_file)
else:
raise ValueError("Unsupported file type. Please provide a JSON or YAML file.")
# Files names are UUID, so we can't find the extension
with open(file_path, "r") as file:
try:
data = json.load(file)
except json.JSONDecodeError:
file.seek(0)
data = yaml.safe_load(file)
except ValueError as exc:
raise ValueError("Invalid file type. Expected .json or .yaml.") from exc
return data

17
src/backend/langflow/interface/wrappers/base.py
View file

@ -1,25 +1,36 @@
from typing import Dict, List, Optional
from langchain import requests
from langchain import requests, sql_database
from langflow.interface.base import LangChainTypeCreator
from langflow.utils.logger import logger
from langflow.utils.util import build_template_from_class
from langflow.utils.util import build_template_from_class, build_template_from_method
class WrapperCreator(LangChainTypeCreator):
type_name: str = "wrappers"
from_method_nodes = {"SQLDatabase": "from_uri"}
@property
def type_to_loader_dict(self) -> Dict:
if self.type_dict is None:
self.type_dict = {
wrapper.__name__: wrapper for wrapper in [requests.TextRequestsWrapper]
wrapper.__name__: wrapper
for wrapper in [requests.TextRequestsWrapper, sql_database.SQLDatabase]
}
return self.type_dict
def get_signature(self, name: str) -> Optional[Dict]:
try:
if name in self.from_method_nodes:
return build_template_from_method(
name,
type_to_cls_dict=self.type_to_loader_dict,
add_function=True,
method_name=self.from_method_nodes[name],
)
return build_template_from_class(name, self.type_to_loader_dict)
except ValueError as exc:
raise ValueError("Wrapper not found") from exc

11
src/backend/langflow/processing/base.py
View file

@ -19,8 +19,11 @@ async def get_result_and_steps(langchain_object, inputs: Union[dict, str], **kwa
# Deactivating until we have a frontend solution
# to display intermediate steps
langchain_object.return_intermediate_steps = True
try:
fix_memory_inputs(langchain_object)
except Exception as exc:
logger.error(exc)
fix_memory_inputs(langchain_object)
try:
async_callbacks = [AsyncStreamingLLMCallbackHandler(**kwargs)]
output = await langchain_object.acall(inputs, callbacks=async_callbacks)
@ -39,7 +42,11 @@ async def get_result_and_steps(langchain_object, inputs: Union[dict, str], **kwa
if isinstance(output, dict)
else output
)
thought = format_actions(intermediate_steps) if intermediate_steps else ""
try:
thought = format_actions(intermediate_steps) if intermediate_steps else ""
except Exception as exc:
logger.exception(exc)
thought = ""
except Exception as exc:
logger.exception(exc)
raise ValueError(f"Error: {str(exc)}") from exc

24
src/backend/langflow/processing/process.py
View file

@ -22,7 +22,10 @@ def fix_memory_inputs(langchain_object):
if not hasattr(langchain_object, "memory") or langchain_object.memory is None:
return
try:
if langchain_object.memory.memory_key in langchain_object.input_variables:
if (
hasattr(langchain_object.memory, "memory_key")
and langchain_object.memory.memory_key in langchain_object.input_variables
):
return
except AttributeError:
input_variables = (
@ -92,6 +95,10 @@ def process_graph_cached(data_graph: Dict[str, Any], inputs: Optional[dict] = No
logger.debug("Loaded LangChain object")
if inputs is None:
inputs = {}
# Add artifacts to inputs
# artifacts can be documents loaded when building
# the flow
for (
key,
value,
@ -113,8 +120,7 @@ def process_graph_cached(data_graph: Dict[str, Any], inputs: Optional[dict] = No
result = get_result_and_thought(langchain_object, inputs)
logger.debug("Generated result and thought")
elif isinstance(langchain_object, VectorStore):
class_name = langchain_object.__class__.__name__
result = {"message": f"Processed {class_name} successfully"}
result = langchain_object.search(**inputs)
else:
raise ValueError(
f"Unknown langchain_object type: {type(langchain_object).__name__}"
@ -123,23 +129,23 @@ def process_graph_cached(data_graph: Dict[str, Any], inputs: Optional[dict] = No
def load_flow_from_json(
input: Union[Path, str, dict], tweaks: Optional[dict] = None, build=True
flow: Union[Path, str, dict], tweaks: Optional[dict] = None, build=True
):
"""
Load flow from a JSON file or a JSON object.
:param input: JSON file path or JSON object
:param flow: JSON file path or JSON object
:param tweaks: Optional tweaks to be processed
:param build: If True, build the graph, otherwise return the graph object
:return: Langchain object or Graph object depending on the build parameter
"""
# If input is a file path, load JSON from the file
if isinstance(input, (str, Path)):
with open(input, "r", encoding="utf-8") as f:
if isinstance(flow, (str, Path)):
with open(flow, "r", encoding="utf-8") as f:
flow_graph = json.load(f)
# If input is a dictionary, assume it's a JSON object
elif isinstance(input, dict):
flow_graph = input
elif isinstance(flow, dict):
flow_graph = flow
else:
raise TypeError(
"Input must be either a file path (str) or a JSON object (dict)"

2
src/backend/langflow/template/frontend_node/chains.py
View file

@ -81,7 +81,7 @@ class ChainFrontendNode(FrontendNode):
field.advanced = False
if field.name == "verbose":
field.required = False
field.show = True
field.show = False
field.advanced = True
if field.name == "llm":
field.required = True

4
src/backend/langflow/template/frontend_node/constants.py
View file

@ -58,3 +58,7 @@ The base URL of the OpenAI API. Defaults to https://api.openai.com/v1.
You can change this to use other APIs like JinaChat, LocalAI and Prem.
"""
INPUT_KEY_INFO = """The variable to be used as Chat Input when more than one variable is available."""
OUTPUT_KEY_INFO = """The variable to be used as Chat Output (e.g. answer in a ConversationalRetrievalChain)"""

14
src/backend/langflow/template/frontend_node/memories.py
View file

@ -2,6 +2,7 @@ from typing import Optional
from langflow.template.field.base import TemplateField
from langflow.template.frontend_node.base import FrontendNode
from langflow.template.frontend_node.constants import INPUT_KEY_INFO, OUTPUT_KEY_INFO
from langflow.template.template.base import Template
from langchain.memory.chat_message_histories.postgres import DEFAULT_CONNECTION_STRING
from langchain.memory.chat_message_histories.mongodb import (
@ -70,11 +71,15 @@ class MemoryFrontendNode(FrontendNode):
field.required = False
field.show = True
field.advanced = False
if field.name in ["input_key", "output_key"]:
if field.name in {"input_key", "output_key"}:
field.required = False
field.show = True
field.advanced = False
field.value = ""
field.info = (
INPUT_KEY_INFO if field.name == "input_key" else OUTPUT_KEY_INFO
)
if field.name == "memory_key":
field.value = "chat_history"
if field.name == "chat_memory":
@ -84,9 +89,10 @@ class MemoryFrontendNode(FrontendNode):
if field.name == "url":
field.show = True
if field.name == "entity_store":
field.show = True
if name == "SQLiteEntityStore":
field.show = True
field.show = False
if name == "ConversationEntityMemory" and field.name == "memory_key":
field.show = False
field.required = False
class PostgresChatMessageHistoryFrontendNode(MemoryFrontendNode):

2
src/backend/langflow/template/frontend_node/textsplitters.py
View file

@ -45,7 +45,7 @@ class TextSplittersFrontendNode(FrontendNode):
field_type="str",
required=True,
show=True,
value=".",
value="\\n",
name=name,
display_name="Separator",
)

2
src/backend/langflow/template/frontend_node/tools.py
View file

@ -112,7 +112,7 @@ class PythonFunctionToolNode(FrontendNode):
],
)
description: str = "Python function to be executed."
base_classes: list[str] = ["Tool"]
base_classes: list[str] = ["BaseTool", "Tool"]
def to_dict(self):
return super().to_dict()