Add LLM cache (#541)

✨ feat(main.py): call setup_llm_caching function on app startup
The `setup_llm_caching` function is added to `utils.py` to set up LLM
caching. The function is then called on app startup in `main.py` using
the `app.on_event("startup")` method. This improves the performance of
the application by caching LLM objects.

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "langflow"
-version = "0.1.7"
+version = "0.2.0"
 description = "A Python package with a built-in web application"
 authors = ["Logspace <contact@logspace.ai>"]
 maintainers = [
@@ -23,14 +23,14 @@ langflow = "langflow.__main__:main"
 
 [tool.poetry.dependencies]
 python = ">=3.9,<3.12"
-fastapi = "^0.97.0"
+fastapi = "^0.98.0"
 uvicorn = "^0.22.0"
 beautifulsoup4 = "^4.11.2"
 google-search-results = "^2.4.1"
 google-api-python-client = "^2.79.0"
 typer = "^0.9.0"
 gunicorn = "^20.1.0"
-langchain = "^0.0.208"
+langchain = "^0.0.211"
 openai = "^0.27.8"
 types-pyyaml = "^6.0.12.8"
 pandas = "^1.5.3"
@@ -51,12 +51,13 @@ tiktoken = "~0.4.0"
 wikipedia = "^1.4.0"
 langchain-serve = { version = ">0.0.39", optional = true }
 qdrant-client = "^1.2.0"
-websockets = "^11.0.3"
+websockets = "^10.3"
 weaviate-client = "^3.21.0"
 jina = "3.15.2"
 sentence-transformers = "^2.2.2"
 ctransformers = "^0.2.2"
 cohere = "^4.6.0"
+python-multipart = "^0.0.6"
 sqlmodel = "^0.0.8"
 faiss-cpu = "^1.7.4"
 anthropic = "^0.2.10"
@@ -64,6 +65,11 @@ orjson = "^3.9.1"
 multiprocess = "^0.70.14"
 cachetools = "^5.3.1"
 types-cachetools = "^5.3.0.5"
+appdirs = "^1.4.4"
+pinecone-client = "^2.2.2"
+supabase = "^1.0.3"
+pymongo = "^4.4.0"
+certifi = "^2023.5.7"
 
 
 [tool.poetry.group.dev.dependencies]
@@ -78,6 +84,7 @@ requests = "^2.28.0"
 pytest-cov = "^4.0.0"
 pandas-stubs = "^2.0.0.230412"
 types-pillow = "^9.5.0.2"
+types-appdirs = "^1.4.3.5"
 
 
 [tool.poetry.extras]

src/backend/langflow/api/v1/chat.py

@@ -1,4 +1,3 @@
-import json
 from fastapi import (
     APIRouter,
     HTTPException,
@@ -7,7 +6,7 @@ from fastapi import (
     status,
 )
 from fastapi.responses import StreamingResponse
-from langflow.api.v1.schemas import BuiltResponse, InitResponse
+from langflow.api.v1.schemas import BuiltResponse, InitResponse, StreamData
 
 from langflow.chat.manager import ChatManager
 from langflow.graph.graph.base import Graph
@@ -71,30 +70,37 @@ async def stream_build(flow_id: str):
     """Stream the build process based on stored flow data."""
 
     async def event_stream(flow_id):
-        final_response = json.dumps({"end_of_stream": True})
+        final_response = {"end_of_stream": True}
         try:
             if flow_id not in flow_data_store:
                 error_message = "Invalid session ID"
-                yield f"data: {json.dumps({'error': error_message})}\n\n"
+                yield str(StreamData(event="error", data={"error": error_message}))
                 return
 
             graph_data = flow_data_store[flow_id].get("data")
 
             if not graph_data:
                 error_message = "No data provided"
-                yield f"data: {json.dumps({'error': error_message})}\n\n"
+                yield str(StreamData(event="error", data={"error": error_message}))
                 return
 
             logger.debug("Building langchain object")
-            graph = Graph.from_payload(graph_data)
+            try:
+                # Some error could happen when building the graph
+                graph = Graph.from_payload(graph_data)
+            except Exception as exc:
+                logger.exception(exc)
+                error_message = str(exc)
+                yield str(StreamData(event="error", data={"error": error_message}))
+                return
+
             number_of_nodes = len(graph.nodes)
             for i, vertex in enumerate(graph.generator_build(), 1):
                 try:
                     log_dict = {
                         "log": f"Building node {vertex.vertex_type}",
-                        "progress": round(i / number_of_nodes, 2),
                     }
-                    yield f"data: {json.dumps(log_dict)}\n\n"
+                    yield str(StreamData(event="log", data=log_dict))
                     vertex.build()
                     params = vertex._built_object_repr()
                     valid = True
@@ -105,21 +111,21 @@ async def stream_build(flow_id: str):
                     params = str(exc)
                     valid = False
 
-                response = json.dumps(
-                    {
-                        "valid": valid,
-                        "params": params,
-                        "id": vertex.id,
-                    }
-                )
-                yield f"data: {response}\n\n"
+                response = {
+                    "valid": valid,
+                    "params": params,
+                    "id": vertex.id,
+                    "progress": round(i / number_of_nodes, 2),
+                }
+
+                yield str(StreamData(event="message", data=response))
 
             chat_manager.set_cache(flow_id, graph.build())
         except Exception as exc:
             logger.error("Error while building the flow: %s", exc)
-            yield f"error: {json.dumps({'error': str(exc)})}\n\n"
+            yield str(StreamData(event="error", data={"error": str(exc)}))
         finally:
-            yield f"data: {final_response}\n\n"
+            yield str(StreamData(event="message", data=final_response))
 
     try:
         return StreamingResponse(event_stream(flow_id), media_type="text/event-stream")

src/backend/langflow/api/v1/endpoints.py

@@ -1,12 +1,14 @@
+from typing import Optional
+from langflow.cache.utils import save_uploaded_file
 from langflow.database.models.flow import Flow
 from langflow.processing.process import process_graph_cached, process_tweaks
 from langflow.utils.logger import logger
 
-from fastapi import APIRouter, Depends, HTTPException
+from fastapi import APIRouter, Depends, HTTPException, UploadFile
 
 from langflow.api.v1.schemas import (
-    PredictRequest,
-    PredictResponse,
+    ProcessResponse,
+    UploadFileResponse,
 )
 
 from langflow.interface.types import build_langchain_types_dict
@@ -22,14 +24,17 @@ def get_all():
     return build_langchain_types_dict()
 
 
-@router.post("/predict/{flow_id}", response_model=PredictResponse)
-async def predict_flow(
-    predict_request: PredictRequest,
+# For backwards compatibility we will keep the old endpoint
+@router.post("/predict/{flow_id}", response_model=ProcessResponse)
+@router.post("/process/{flow_id}", response_model=ProcessResponse)
+async def process_flow(
     flow_id: str,
+    inputs: Optional[dict] = None,
+    tweaks: Optional[dict] = None,
     session: Session = Depends(get_session),
 ):
     """
-    Endpoint to process a message using the flow passed in the bearer token.
+    Endpoint to process an input with a given flow_id.
     """
 
     try:
@@ -40,15 +45,14 @@ async def predict_flow(
         if flow.data is None:
             raise ValueError(f"Flow {flow_id} has no data")
         graph_data = flow.data
-        if predict_request.tweaks:
+        if tweaks:
             try:
-                graph_data = process_tweaks(graph_data, predict_request.tweaks)
+                graph_data = process_tweaks(graph_data, tweaks)
             except Exception as exc:
                 logger.error(f"Error processing tweaks: {exc}")
-        response = process_graph_cached(graph_data, predict_request.message)
-        return PredictResponse(
-            result=response.get("result", ""),
-            intermediate_steps=response.get("thought", ""),
+        response = process_graph_cached(graph_data, inputs)
+        return ProcessResponse(
+            result=response,
         )
     except Exception as e:
         # Log stack trace
@@ -56,6 +60,21 @@ async def predict_flow(
         raise HTTPException(status_code=500, detail=str(e)) from e
 
 
+@router.post("/upload/{flow_id}", response_model=UploadFileResponse, status_code=201)
+async def create_upload_file(file: UploadFile, flow_id: str):
+    # Cache file
+    try:
+        file_path = save_uploaded_file(file.file, folder_name=flow_id)
+
+        return UploadFileResponse(
+            flowId=flow_id,
+            file_path=file_path,
+        )
+    except Exception as exc:
+        logger.error(f"Error saving file: {exc}")
+        raise HTTPException(status_code=500, detail=str(exc)) from exc
+
+
 # get endpoint to return version of langflow
 @router.get("/version")
 def get_version():

src/backend/langflow/api/v1/schemas.py

@@ -1,6 +1,8 @@
+from pathlib import Path
 from typing import Any, Dict, List, Optional, Union
 from langflow.database.models.flow import FlowCreate, FlowRead
 from pydantic import BaseModel, Field, validator
+import json
 
 
 class GraphData(BaseModel):
@@ -19,34 +21,22 @@ class ExportedFlow(BaseModel):
     data: GraphData
 
 
-class PredictRequest(BaseModel):
-    """Predict request schema."""
+class InputRequest(BaseModel):
+    input: dict
 
-    message: str
+
+class TweaksRequest(BaseModel):
     tweaks: Optional[Dict[str, Dict[str, str]]] = Field(default_factory=dict)
 
-    class Config:
-        schema_extra = {
-            "example": {
-                "message": "Hello, how are you?",
-                "tweaks": {
-                    "dndnode_986363f0-4677-4035-9f38-74b94af5dd78": {
-                        "name": "A tool name",
-                        "description": "A tool description",
-                    },
-                    "dndnode_986363f0-4677-4035-9f38-74b94af57378": {
-                        "template": "A {template}",
-                    },
-                },
-            }
-        }
+
+class UpdateTemplateRequest(BaseModel):
+    template: dict
 
 
-class PredictResponse(BaseModel):
-    """Predict response schema."""
+class ProcessResponse(BaseModel):
+    """Process response schema."""
 
-    result: str
-    intermediate_steps: str = ""
+    result: dict
 
 
 class ChatMessage(BaseModel):
@@ -101,3 +91,18 @@ class InitResponse(BaseModel):
 
 class BuiltResponse(BaseModel):
     built: bool
+
+
+class UploadFileResponse(BaseModel):
+    """Upload file response schema."""
+
+    flowId: str
+    file_path: Path
+
+
+class StreamData(BaseModel):
+    event: str
+    data: dict
+
+    def __str__(self) -> str:
+        return f"event: {self.event}\ndata: {json.dumps(self.data)}\n\n"

src/backend/langflow/api/v1/validate.py

@@ -1,5 +1,3 @@
-import json
-
 from fastapi import APIRouter, HTTPException
 
 from langflow.api.v1.base import (
@@ -9,8 +7,6 @@ from langflow.api.v1.base import (
     PromptValidationResponse,
     validate_prompt,
 )
-from langflow.graph.vertex.types import VectorStoreVertex
-from langflow.graph import Graph
 from langflow.utils.logger import logger
 from langflow.utils.validate import validate_code
 
@@ -37,21 +33,3 @@ def post_validate_prompt(prompt: Prompt):
     except Exception as e:
         logger.exception(e)
         raise HTTPException(status_code=500, detail=str(e)) from e
-
-
-# validate node
-@router.post("/node/{node_id}", status_code=200)
-def post_validate_node(node_id: str, data: dict):
-    try:
-        # build graph
-        graph = Graph.from_payload(data)
-        # validate node
-        node = graph.get_node(node_id)
-        if node is None:
-            raise ValueError(f"Node {node_id} not found")
-        if not isinstance(node, VectorStoreVertex):
-            node.build()
-        return json.dumps({"valid": True, "params": str(node._built_object_repr())})
-    except Exception as e:
-        logger.exception(e)
-        return json.dumps({"valid": False, "params": str(e)})

src/backend/langflow/cache/base.py

@@ -62,9 +62,6 @@ class BaseCache(abc.ABC):
 
         Args:
             key: The key of the item to retrieve.
-
-        Returns:
-            The value associated with the key, or None if the key is not found.
         """
 
     @abc.abstractmethod

src/backend/langflow/cache/utils.py

@@ -8,15 +8,17 @@ import tempfile
 from collections import OrderedDict
 from pathlib import Path
 from typing import Any, Dict
-
+from appdirs import user_cache_dir
 
 CACHE: Dict[str, Any] = {}
 
+CACHE_DIR = user_cache_dir("langflow", "langflow")
+
 
 def create_cache_folder(func):
     def wrapper(*args, **kwargs):
         # Get the destination folder
-        cache_path = Path(tempfile.gettempdir()) / PREFIX
+        cache_path = Path(CACHE_DIR) / PREFIX
 
         # Create the destination folder if it doesn't exist
         os.makedirs(cache_path, exist_ok=True)
@@ -118,7 +120,7 @@ def save_binary_file(content: str, file_name: str, accepted_types: list[str]) ->
         raise ValueError(f"File {file_name} is not accepted")
 
     # Get the destination folder
-    cache_path = Path(tempfile.gettempdir()) / PREFIX
+    cache_path = Path(CACHE_DIR) / PREFIX
     if not content:
         raise ValueError("Please, reload the file in the loader.")
     data = content.split(",")[1]
@@ -132,3 +134,46 @@ def save_binary_file(content: str, file_name: str, accepted_types: list[str]) ->
         file.write(decoded_bytes)
 
     return file_path
+
+
+@create_cache_folder
+def save_uploaded_file(file, folder_name):
+    """
+    Save an uploaded file to the specified folder with a hash of its content as the file name.
+
+    Args:
+        file: The uploaded file object.
+        folder_name: The name of the folder to save the file in.
+
+    Returns:
+        The path to the saved file.
+    """
+    cache_path = Path(CACHE_DIR)
+    folder_path = cache_path / folder_name
+
+    # Create the folder if it doesn't exist
+    if not folder_path.exists():
+        folder_path.mkdir()
+
+    # Create a hash of the file content
+    sha256_hash = hashlib.sha256()
+    # Reset the file cursor to the beginning of the file
+    file.seek(0)
+    # Iterate over the uploaded file in small chunks to conserve memory
+    while chunk := file.read(8192):  # Read 8KB at a time (adjust as needed)
+        sha256_hash.update(chunk)
+
+    # Use the hex digest of the hash as the file name
+    hex_dig = sha256_hash.hexdigest()
+    file_name = hex_dig
+
+    # Reset the file cursor to the beginning of the file
+    file.seek(0)
+
+    # Save the file with the hash as its name
+    file_path = folder_path / file_name
+    with open(file_path, "wb") as new_file:
+        while chunk := file.read(8192):
+            new_file.write(chunk)
+
+    return file_path

src/backend/langflow/chat/manager.py

@@ -144,6 +144,8 @@ class ChatManager:
                 if isinstance(msg, FileResponse):
                     if msg.data_type == "image":
                         # Base64 encode the image
+                        if isinstance(msg.data, str):
+                            continue
                         msg.data = pil_to_base64(msg.data)
                     file_responses.append(msg)
                 if msg.type == "start":

src/backend/langflow/config.yaml

@@ -49,6 +49,7 @@ documentloaders:
   - SlackDirectoryLoader
   - NotionDirectoryLoader
   - DirectoryLoader
+  - GitLoader
 embeddings:
   - OpenAIEmbeddings
   - HuggingFaceEmbeddings
@@ -58,7 +59,7 @@ llms:
   # - AzureOpenAI
   # - AzureChatOpenAI
   - ChatOpenAI
-  - LlamaCpp 
+  - LlamaCpp
   - CTransformers
   - Cohere
   - Anthropic
@@ -130,6 +131,9 @@ vectorstores:
   - Qdrant
   - Weaviate
   - FAISS
+  - Pinecone
+  - SupabaseVectorStore
+  - MongoDBAtlasVectorSearch
 wrappers:
   - RequestsWrapper
   # - ChatPromptTemplate

src/backend/langflow/graph/graph/base.py

@@ -146,7 +146,7 @@ class Graph:
     def generator_build(self) -> Generator:
         """Builds each vertex in the graph and yields it."""
         sorted_vertices = self.topological_sort()
-        logger.info("Sorted vertices: %s", sorted_vertices)
+        logger.debug("Sorted vertices: %s", sorted_vertices)
         yield from sorted_vertices
 
     def get_node_neighbors(self, node: Vertex) -> Dict[Vertex, int]:

src/backend/langflow/graph/graph/constants.py

@@ -30,9 +30,6 @@ from langflow.interface.wrappers.base import wrapper_creator
 from typing import Dict, Type
 
 
-DIRECT_TYPES = ["str", "bool", "code", "int", "float", "Any", "prompt"]
-
-
 VERTEX_TYPE_MAP: Dict[str, Type[Vertex]] = {
     **{t: PromptVertex for t in prompt_creator.to_list()},
     **{t: AgentVertex for t in agent_creator.to_list()},

src/backend/langflow/graph/vertex/base.py

@@ -1,6 +1,5 @@
-from langflow.cache import utils as cache_utils
-from langflow.graph.vertex.constants import DIRECT_TYPES
-from langflow.interface import loading
+from langflow.utils.constants import DIRECT_TYPES
+from langflow.interface.initialize import loading
 from langflow.interface.listing import ALL_TYPES_DICT
 from langflow.utils.logger import logger
 from langflow.utils.util import sync_to_async
@@ -90,12 +89,7 @@ class Vertex:
                 # Load the type in value.get('suffixes') using
                 # what is inside value.get('content')
                 # value.get('value') is the file name
-                file_name = value.get("value")
-                content = value.get("content")
-                type_to_load = value.get("suffixes")
-                file_path = cache_utils.save_binary_file(
-                    content=content, file_name=file_name, accepted_types=type_to_load
-                )
+                file_path = value.get("file_path")
 
                 params[key] = file_path
 

src/backend/langflow/graph/vertex/constants.py

@@ -1 +1 @@
-DIRECT_TYPES = ["str", "bool", "code", "int", "float", "Any", "prompt"]
+

src/backend/langflow/graph/vertex/types.py

@@ -106,9 +106,6 @@ class VectorStoreVertex(Vertex):
     def __init__(self, data: Dict):
         super().__init__(data, base_type="vectorstores")
 
-    def _built_object_repr(self):
-        return "Vector stores can take time to build. It will build on the first query."
-
 
 class MemoryVertex(Vertex):
     def __init__(self, data: Dict):

src/backend/langflow/interface/base.py

@@ -68,7 +68,7 @@ class LangChainTypeCreator(BaseModel, ABC):
                     value=value.get("value", None),
                     suffixes=value.get("suffixes", []),
                     file_types=value.get("fileTypes", []),
-                    content=value.get("content", None),
+                    file_path=value.get("file_path", None),
                 )
                 for key, value in signature["template"].items()
                 if key != "_type"

src/backend/langflow/interface/initialize/loading.py

@@ -0,0 +1,320 @@
+import json
+from typing import Any, Callable, Dict, Sequence
+
+from langchain.agents import ZeroShotAgent
+from langchain.agents import agent as agent_module
+from langchain.agents.agent import AgentExecutor
+from langchain.agents.agent_toolkits.base import BaseToolkit
+from langchain.agents.tools import BaseTool
+from langflow.interface.initialize.vector_store import vecstore_initializer
+
+from pydantic import ValidationError
+
+from langflow.interface.custom_lists import CUSTOM_NODES
+from langflow.interface.importing.utils import get_function, import_by_type
+from langflow.interface.toolkits.base import toolkits_creator
+from langflow.interface.chains.base import chain_creator
+from langflow.interface.utils import load_file_into_dict
+from langflow.utils import validate
+
+
+def instantiate_class(node_type: str, base_type: str, params: Dict) -> Any:
+    """Instantiate class from module type and key, and params"""
+    params = convert_params_to_sets(params)
+    params = convert_kwargs(params)
+    if node_type in CUSTOM_NODES:
+        if custom_node := CUSTOM_NODES.get(node_type):
+            if hasattr(custom_node, "initialize"):
+                return custom_node.initialize(**params)
+            return custom_node(**params)
+
+    class_object = import_by_type(_type=base_type, name=node_type)
+    return instantiate_based_on_type(class_object, base_type, node_type, params)
+
+
+def convert_params_to_sets(params):
+    """Convert certain params to sets"""
+    if "allowed_special" in params:
+        params["allowed_special"] = set(params["allowed_special"])
+    if "disallowed_special" in params:
+        params["disallowed_special"] = set(params["disallowed_special"])
+    return params
+
+
+def convert_kwargs(params):
+    # if *kwargs are passed as a string, convert to dict
+    # first find any key that has kwargs in it
+    kwargs_keys = [key for key in params.keys() if "kwargs" in key]
+    for key in kwargs_keys:
+        if isinstance(params[key], str):
+            params[key] = json.loads(params[key])
+    return params
+
+
+def instantiate_based_on_type(class_object, base_type, node_type, params):
+    if base_type == "agents":
+        return instantiate_agent(class_object, 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)
+    elif base_type == "toolkits":
+        return instantiate_toolkit(node_type, class_object, params)
+    elif base_type == "embeddings":
+        return instantiate_embedding(class_object, params)
+    elif base_type == "vectorstores":
+        return instantiate_vectorstore(class_object, params)
+    elif base_type == "documentloaders":
+        return instantiate_documentloader(class_object, params)
+    elif base_type == "textsplitters":
+        return instantiate_textsplitter(class_object, params)
+    elif base_type == "utilities":
+        return instantiate_utility(node_type, class_object, params)
+    elif base_type == "chains":
+        return instantiate_chains(node_type, class_object, params)
+    else:
+        return class_object(**params)
+
+
+def instantiate_chains(node_type, class_object, params):
+    if "retriever" in params and hasattr(params["retriever"], "as_retriever"):
+        params["retriever"] = params["retriever"].as_retriever()
+    if node_type in chain_creator.from_method_nodes:
+        method = chain_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_agent(class_object, params):
+    return load_agent_executor(class_object, params)
+
+
+def instantiate_prompt(node_type, class_object, params):
+    if node_type == "ZeroShotPrompt":
+        if "tools" not in params:
+            params["tools"] = []
+        return ZeroShotAgent.create_prompt(**params)
+    return class_object(**params)
+
+
+def instantiate_tool(node_type, class_object, params):
+    if node_type == "JsonSpec":
+        params["dict_"] = load_file_into_dict(params.pop("path"))
+        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):
+            return validate.eval_function(function_string)
+        raise ValueError("Function should be a string")
+    elif node_type.lower() == "tool":
+        return class_object(**params)
+    return class_object(**params)
+
+
+def instantiate_toolkit(node_type, class_object, params):
+    loaded_toolkit = class_object(**params)
+    # Commenting this out for now to use toolkits as normal tools
+    # if toolkits_creator.has_create_function(node_type):
+    #     return load_toolkits_executor(node_type, loaded_toolkit, params)
+    if isinstance(loaded_toolkit, BaseToolkit):
+        return loaded_toolkit.get_tools()
+    return loaded_toolkit
+
+
+def instantiate_embedding(class_object, params):
+    params.pop("model", None)
+    params.pop("headers", None)
+    try:
+        return class_object(**params)
+    except ValidationError:
+        params = {
+            key: value
+            for key, value in params.items()
+            if key in class_object.__fields__
+        }
+        return class_object(**params)
+
+
+def instantiate_vectorstore(class_object, params):
+    search_kwargs = params.pop("search_kwargs", {})
+    if initializer := vecstore_initializer.get(class_object.__name__):
+        vecstore = initializer(class_object, params)
+    else:
+        if "texts" in params:
+            params["documents"] = params.pop("texts")
+        vecstore = class_object.from_documents(**params)
+
+    # ! This might not work. Need to test
+    if search_kwargs and hasattr(vecstore, "as_retriever"):
+        vecstore = vecstore.as_retriever(search_kwargs=search_kwargs)
+
+    return vecstore
+
+
+def instantiate_documentloader(class_object, params):
+    if "file_filter" in params:
+        # file_filter will be a string but we need a function
+        # that will be used to filter the files using file_filter
+        # like lambda x: x.endswith(".txt") but as we don't know
+        # anything besides the string, we will simply check if the string is
+        # in x and if it is, we will return True
+        file_filter = params.pop("file_filter", None)
+        extensions = file_filter.split(",")
+        params["file_filter"] = lambda x: any(
+            extension.strip() in x for extension in extensions
+        )
+    metadata = params.pop("metadata", None)
+    docs = class_object(**params).load()
+    if metadata:
+        if isinstance(metadata, str):
+            try:
+                metadata = json.loads(metadata)
+            except json.JSONDecodeError as exc:
+                raise ValueError(
+                    "The metadata you provided is not a valid JSON string."
+                ) from exc
+
+        for doc in docs:
+            doc.metadata = metadata
+
+    return docs
+
+
+def instantiate_textsplitter(class_object, params):
+    try:
+        documents = params.pop("documents")
+    except KeyError as e:
+        raise ValueError(
+            "The source you provided did not load correctly or was empty."
+            "Try changing the chunk_size of the Text Splitter."
+        ) from e
+    text_splitter = class_object(**params)
+    return text_splitter.split_documents(documents)
+
+
+def instantiate_utility(node_type, class_object, params):
+    if node_type == "SQLDatabase":
+        return class_object.from_uri(params.pop("uri"))
+    return class_object(**params)
+
+
+def replace_zero_shot_prompt_with_prompt_template(nodes):
+    """Replace ZeroShotPrompt with PromptTemplate"""
+    for node in nodes:
+        if node["data"]["type"] == "ZeroShotPrompt":
+            # Build Prompt Template
+            tools = [
+                tool
+                for tool in nodes
+                if tool["type"] != "chatOutputNode"
+                and "Tool" in tool["data"]["node"]["base_classes"]
+            ]
+            node["data"] = build_prompt_template(prompt=node["data"], tools=tools)
+            break
+    return nodes
+
+
+def load_agent_executor(agent_class: type[agent_module.Agent], params, **kwargs):
+    """Load agent executor from agent class, tools and chain"""
+    allowed_tools: Sequence[BaseTool] = params.get("allowed_tools", [])
+    llm_chain = params["llm_chain"]
+    # if allowed_tools is not a list or set, make it a list
+    if not isinstance(allowed_tools, (list, set)) and isinstance(
+        allowed_tools, BaseTool
+    ):
+        allowed_tools = [allowed_tools]
+    tool_names = [tool.name for tool in allowed_tools]
+    # Agent class requires an output_parser but Agent classes
+    # have a default output_parser.
+    agent = agent_class(allowed_tools=tool_names, llm_chain=llm_chain)  # type: ignore
+    return AgentExecutor.from_agent_and_tools(
+        agent=agent,
+        tools=allowed_tools,
+        **kwargs,
+    )
+
+
+def load_toolkits_executor(node_type: str, toolkit: BaseToolkit, params: dict):
+    create_function: Callable = toolkits_creator.get_create_function(node_type)
+    if llm := params.get("llm"):
+        return create_function(llm=llm, toolkit=toolkit)
+
+
+def build_prompt_template(prompt, tools):
+    """Build PromptTemplate from ZeroShotPrompt"""
+    prefix = prompt["node"]["template"]["prefix"]["value"]
+    suffix = prompt["node"]["template"]["suffix"]["value"]
+    format_instructions = prompt["node"]["template"]["format_instructions"]["value"]
+
+    tool_strings = "\n".join(
+        [
+            f"{tool['data']['node']['name']}: {tool['data']['node']['description']}"
+            for tool in tools
+        ]
+    )
+    tool_names = ", ".join([tool["data"]["node"]["name"] for tool in tools])
+    format_instructions = format_instructions.format(tool_names=tool_names)
+    value = "\n\n".join([prefix, tool_strings, format_instructions, suffix])
+
+    prompt["type"] = "PromptTemplate"
+
+    prompt["node"] = {
+        "template": {
+            "_type": "prompt",
+            "input_variables": {
+                "type": "str",
+                "required": True,
+                "placeholder": "",
+                "list": True,
+                "show": False,
+                "multiline": False,
+            },
+            "output_parser": {
+                "type": "BaseOutputParser",
+                "required": False,
+                "placeholder": "",
+                "list": False,
+                "show": False,
+                "multline": False,
+                "value": None,
+            },
+            "template": {
+                "type": "str",
+                "required": True,
+                "placeholder": "",
+                "list": False,
+                "show": True,
+                "multiline": True,
+                "value": value,
+            },
+            "template_format": {
+                "type": "str",
+                "required": False,
+                "placeholder": "",
+                "list": False,
+                "show": False,
+                "multline": False,
+                "value": "f-string",
+            },
+            "validate_template": {
+                "type": "bool",
+                "required": False,
+                "placeholder": "",
+                "list": False,
+                "show": False,
+                "multline": False,
+                "value": True,
+            },
+        },
+        "description": "Schema to represent a prompt for an LLM.",
+        "base_classes": ["BasePromptTemplate"],
+    }
+
+    return prompt

src/backend/langflow/interface/initialize/vector_store.py

@@ -0,0 +1,223 @@
+import json
+from typing import Any, Callable, Dict, Type
+from langchain.vectorstores import (
+    Pinecone,
+    Qdrant,
+    Chroma,
+    FAISS,
+    Weaviate,
+    SupabaseVectorStore,
+    MongoDBAtlasVectorSearch,
+)
+
+import os
+
+
+def docs_in_params(params: dict) -> bool:
+    """Check if params has documents OR texts and one of them is not an empty list,
+    If any of them is not an empty list, return True, else return False"""
+    return ("documents" in params and params["documents"]) or (
+        "texts" in params and params["texts"]
+    )
+
+
+def initialize_mongodb(class_object: Type[MongoDBAtlasVectorSearch], params: dict):
+    """Initialize mongodb and return the class object"""
+
+    MONGODB_ATLAS_CLUSTER_URI = params.pop("mongodb_atlas_cluster_uri")
+    if not MONGODB_ATLAS_CLUSTER_URI:
+        raise ValueError("Mongodb atlas cluster uri must be provided in the params")
+    from pymongo import MongoClient
+    import certifi
+
+    client: MongoClient = MongoClient(
+        MONGODB_ATLAS_CLUSTER_URI, tlsCAFile=certifi.where()
+    )
+    db_name = params.pop("db_name", None)
+    collection_name = params.pop("collection_name", None)
+    if not db_name or not collection_name:
+        raise ValueError("db_name and collection_name must be provided in the params")
+
+    index_name = params.pop("index_name", None)
+    if not index_name:
+        raise ValueError("index_name must be provided in the params")
+
+    collection = client[db_name][collection_name]
+    if not docs_in_params(params):
+        # __init__ requires collection, embedding and index_name
+        init_args = {
+            "collection": collection,
+            "index_name": index_name,
+            "embedding": params.get("embedding"),
+        }
+
+        return class_object(**init_args)
+
+    if "texts" in params:
+        params["documents"] = params.pop("texts")
+
+    params["collection"] = collection
+    params["index_name"] = index_name
+
+    return class_object.from_documents(**params)
+
+
+def initialize_supabase(class_object: Type[SupabaseVectorStore], params: dict):
+    """Initialize supabase and return the class object"""
+    from supabase.client import Client, create_client
+
+    if "supabase_url" not in params or "supabase_service_key" not in params:
+        raise ValueError("Supabase url and service key must be provided in the params")
+    if "texts" in params:
+        params["documents"] = params.pop("texts")
+
+    client_kwargs = {
+        "supabase_url": params.pop("supabase_url"),
+        "supabase_key": params.pop("supabase_service_key"),
+    }
+
+    supabase: Client = create_client(**client_kwargs)
+    if not docs_in_params(params):
+        params.pop("documents", None)
+        params.pop("texts", None)
+        return class_object(client=supabase, **params)
+    # If there are docs in the params, create a new index
+
+    return class_object.from_documents(client=supabase, **params)
+
+
+def initialize_weaviate(class_object: Type[Weaviate], params: dict):
+    """Initialize weaviate and return the class object"""
+    if not docs_in_params(params):
+        import weaviate  # type: ignore
+
+        client_kwargs_json = params.get("client_kwargs", "{}")
+        client_kwargs = json.loads(client_kwargs_json)
+        client_params = {
+            "url": params.get("weaviate_url"),
+        }
+        client_params.update(client_kwargs)
+        weaviate_client = weaviate.Client(**client_params)
+
+        new_params = {
+            "client": weaviate_client,
+            "index_name": params.get("index_name"),
+            "text_key": params.get("text_key"),
+        }
+        return class_object(**new_params)
+    # If there are docs in the params, create a new index
+    if "texts" in params:
+        params["documents"] = params.pop("texts")
+
+    return class_object.from_documents(**params)
+
+
+def initialize_faiss(class_object: Type[FAISS], params: dict):
+    """Initialize faiss and return the class object"""
+
+    if not docs_in_params(params):
+        return class_object.load_local
+
+    save_local = params.get("save_local")
+    faiss_index = class_object(**params)
+    if save_local:
+        faiss_index.save_local(folder_path=save_local)
+    return faiss_index
+
+
+def initialize_pinecone(class_object: Type[Pinecone], params: dict):
+    """Initialize pinecone and return the class object"""
+
+    import pinecone  # type: ignore
+
+    pinecone_api_key = params.get("pinecone_api_key")
+    pinecone_env = params.get("pinecone_env")
+
+    if pinecone_api_key is None or pinecone_env is None:
+        if os.getenv("PINECONE_API_KEY") is not None:
+            pinecone_api_key = os.getenv("PINECONE_API_KEY")
+        if os.getenv("PINECONE_ENV") is not None:
+            pinecone_env = os.getenv("PINECONE_ENV")
+
+    if pinecone_api_key is None or pinecone_env is None:
+        raise ValueError(
+            "Pinecone API key and environment must be provided in the params"
+        )
+
+    # initialize pinecone
+    pinecone.init(
+        api_key=pinecone_api_key,  # find at app.pinecone.io
+        environment=pinecone_env,  # next to api key in console
+    )
+
+    # If there are no docs in the params, return an existing index
+    # but first remove any texts or docs keys from the params
+    if not docs_in_params(params):
+        existing_index_params = {
+            "embedding": params.pop("embedding"),
+        }
+        if "index_name" in params:
+            existing_index_params["index_name"] = params.pop("index_name")
+        if "namespace" in params:
+            existing_index_params["namespace"] = params.pop("namespace")
+
+        return class_object.from_existing_index(**existing_index_params)
+    # If there are docs in the params, create a new index
+    if "texts" in params:
+        params["documents"] = params.pop("texts")
+    return class_object.from_documents(**params)
+
+
+def initialize_chroma(class_object: Type[Chroma], params: dict):
+    """Initialize a ChromaDB object from the params"""
+    persist = params.pop("persist", False)
+    if not docs_in_params(params):
+        params.pop("documents", None)
+        params.pop("texts", None)
+        params["embedding_function"] = params.pop("embedding")
+        chromadb = class_object(**params)
+    else:
+        if "texts" in params:
+            params["documents"] = params.pop("texts")
+        for doc in params["documents"]:
+            if doc.metadata is None:
+                doc.metadata = {}
+            for key, value in doc.metadata.items():
+                if value is None:
+                    doc.metadata[key] = ""
+        chromadb = class_object.from_documents(**params)
+    if persist:
+        chromadb.persist()
+    return chromadb
+
+
+def initialize_qdrant(class_object: Type[Qdrant], params: dict):
+    if not docs_in_params(params):
+        if "location" not in params and "api_key" not in params:
+            raise ValueError("Location and API key must be provided in the params")
+        from qdrant_client import QdrantClient
+
+        client_params = {
+            "location": params.pop("location"),
+            "api_key": params.pop("api_key"),
+        }
+        lc_params = {
+            "collection_name": params.pop("collection_name"),
+            "embeddings": params.pop("embedding"),
+        }
+        client = QdrantClient(**client_params)
+
+        return class_object(client=client, **lc_params)
+
+    return class_object.from_documents(**params)
+
+
+vecstore_initializer: Dict[str, Callable[[Type[Any], dict], Any]] = {
+    "Pinecone": initialize_pinecone,
+    "Chroma": initialize_chroma,
+    "Qdrant": initialize_qdrant,
+    "Weaviate": initialize_weaviate,
+    "FAISS": initialize_faiss,
+    "SupabaseVectorStore": initialize_supabase,
+    "MongoDBAtlasVectorSearch": initialize_mongodb,
+}

src/backend/langflow/interface/utils.py

@@ -7,6 +7,7 @@ import re
 import yaml
 from langchain.base_language import BaseLanguageModel
 from PIL.Image import Image
+from langflow.utils.logger import logger
 
 
 def load_file_into_dict(file_path: str) -> dict:
@@ -58,3 +59,12 @@ def try_setting_streaming_options(langchain_object, websocket):
 def extract_input_variables_from_prompt(prompt: str) -> list[str]:
     """Extract input variables from prompt."""
     return re.findall(r"{(.*?)}", prompt)
+
+
+def setup_llm_caching():
+    """Setup LLM caching."""
+    import langchain
+    from langchain.cache import SQLiteCache
+
+    logger.debug("Setting up LLM caching")
+    langchain.llm_cache = SQLiteCache()