Merge remote-tracking branch 'origin/dev' into authentication

src/backend/langflow/__main__.py

@@ -1,6 +1,7 @@
 import sys
 import time
 import httpx
+from langflow.services.manager import initialize_settings_manager
 from langflow.services.utils import get_settings_manager
 from langflow.utils.util import get_number_of_workers
 from multiprocess import Process  # type: ignore
@@ -30,6 +31,7 @@ def update_settings(
     """Update the settings from a config file."""
 
     # Check for database_url in the environment variables
+    initialize_settings_manager()
     settings_manager = get_settings_manager()
     if config:
         logger.debug(f"Loading settings from {config}")

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

@@ -1,5 +1,5 @@
 from http import HTTPStatus
-from typing import Annotated, Optional
+from typing import Annotated, Optional, Union
 
 from langflow.services.cache.utils import save_uploaded_file
 from langflow.services.database.models.flow import Flow
@@ -44,15 +44,21 @@ def get_all():
         logger.info(
             f"Building custom components from {settings_manager.settings.COMPONENTS_PATH}"
         )
-        custom_component_dicts = [
-            build_langchain_custom_component_list_from_path(str(path))
-            for path in settings_manager.settings.COMPONENTS_PATH
-        ]
+
+        custom_component_dicts = []
+        processed_paths = []
+        for path in settings_manager.settings.COMPONENTS_PATH:
+            if str(path) in processed_paths:
+                continue
+            custom_component_dict = build_langchain_custom_component_list_from_path(
+                str(path)
+            )
+            custom_component_dicts.append(custom_component_dict)
+            processed_paths.append(str(path))
+
         logger.info(f"Loading {len(custom_component_dicts)} category(ies)")
         for custom_component_dict in custom_component_dicts:
             # custom_component_dict is a dict of dicts
-            if not custom_component_dict:
-                continue
             category = list(custom_component_dict.keys())[0]
             logger.info(
                 f"Loading {len(custom_component_dict[category])} component(s) from category {category}"
@@ -75,6 +81,7 @@ async def process_flow(
     inputs: Optional[dict] = None,
     tweaks: Optional[dict] = None,
     clear_cache: Annotated[bool, Body(embed=True)] = False,  # noqa: F821
+    session_id: Annotated[Union[None, str], Body(embed=True)] = None,  # noqa: F821
     session: Session = Depends(get_session),
 ):
     """
@@ -94,10 +101,10 @@ async def process_flow(
                 graph_data = process_tweaks(graph_data, tweaks)
             except Exception as exc:
                 logger.error(f"Error processing tweaks: {exc}")
-        response = process_graph_cached(graph_data, inputs, clear_cache)
-        return ProcessResponse(
-            result=response,
+        response, session_id = process_graph_cached(
+            graph_data, inputs, clear_cache, session_id
         )
+        return ProcessResponse(result=response, session_id=session_id)
     except Exception as e:
         # Log stack trace
         logger.exception(e)

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

@@ -47,6 +47,7 @@ class ProcessResponse(BaseModel):
     """Process response schema."""
 
     result: dict
+    session_id: Optional[str] = None
 
 
 class ChatMessage(BaseModel):

src/backend/langflow/components/agents/OpenAIConversationalAgent.py

@@ -0,0 +1,82 @@
+from langflow import CustomComponent
+from typing import Optional
+from langchain.prompts import SystemMessagePromptTemplate
+from langchain.tools import Tool
+from langchain.schema.memory import BaseMemory
+from langchain.chat_models import ChatOpenAI
+
+from langchain.agents.agent import AgentExecutor
+from langchain.agents.openai_functions_agent.base import OpenAIFunctionsAgent
+from langchain.memory.token_buffer import ConversationTokenBufferMemory
+from langchain.prompts.chat import MessagesPlaceholder
+from langchain.agents.agent_toolkits.conversational_retrieval.openai_functions import (
+    _get_default_system_message,
+)
+
+
+class ConversationalAgent(CustomComponent):
+    display_name: str = "OpenAI Conversational Agent"
+    description: str = "Conversational Agent that can use OpenAI's function calling API"
+
+    def build_config(self):
+        openai_function_models = [
+            "gpt-3.5-turbo-0613",
+            "gpt-3.5-turbo-16k-0613",
+            "gpt-4-0613",
+            "gpt-4-32k-0613",
+        ]
+        return {
+            "tools": {"is_list": True, "display_name": "Tools"},
+            "memory": {"display_name": "Memory"},
+            "system_message": {"display_name": "System Message"},
+            "max_token_limit": {"display_name": "Max Token Limit"},
+            "model_name": {
+                "display_name": "Model Name",
+                "options": openai_function_models,
+                "value": openai_function_models[0],
+            },
+            "code": {"show": False},
+        }
+
+    def build(
+        self,
+        model_name: str,
+        openai_api_key: str,
+        openai_api_base: str,
+        tools: Tool,
+        memory: Optional[BaseMemory] = None,
+        system_message: Optional[SystemMessagePromptTemplate] = None,
+        max_token_limit: int = 2000,
+    ) -> AgentExecutor:
+        llm = ChatOpenAI(
+            model=model_name,
+            openai_api_key=openai_api_key,
+            openai_api_base=openai_api_base,
+        )
+        if not memory:
+            memory_key = "chat_history"
+            memory = ConversationTokenBufferMemory(
+                memory_key=memory_key,
+                return_messages=True,
+                output_key="output",
+                llm=llm,
+                max_token_limit=max_token_limit,
+            )
+        else:
+            memory_key = memory.memory_key  # type: ignore
+
+        _system_message = system_message or _get_default_system_message()
+        prompt = OpenAIFunctionsAgent.create_prompt(
+            system_message=_system_message,  # type: ignore
+            extra_prompt_messages=[MessagesPlaceholder(variable_name=memory_key)],
+        )
+        agent = OpenAIFunctionsAgent(
+            llm=llm, tools=tools, prompt=prompt  # type: ignore
+        )
+        return AgentExecutor(
+            agent=agent,
+            tools=tools,  # type: ignore
+            memory=memory,
+            verbose=True,
+            return_intermediate_steps=True,
+        )

src/backend/langflow/graph/utils.py

@@ -3,6 +3,10 @@ from typing import Any, Union
 from langflow.interface.utils import extract_input_variables_from_prompt
 
 
+class UnbuiltObject:
+    pass
+
+
 def validate_prompt(prompt: str):
     """Validate prompt."""
     if extract_input_variables_from_prompt(prompt):

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

@@ -1,4 +1,5 @@
 import ast
+from langflow.graph.utils import UnbuiltObject
 from langflow.interface.initialize import loading
 from langflow.interface.listing import lazy_load_dict
 from langflow.utils.constants import DIRECT_TYPES
@@ -22,7 +23,7 @@ class Vertex:
         self.edges: List["Edge"] = []
         self.base_type: Optional[str] = base_type
         self._parse_data()
-        self._built_object = None
+        self._built_object = UnbuiltObject()
         self._built = False
         self.artifacts: Dict[str, Any] = {}
 
@@ -245,8 +246,14 @@ class Vertex:
         """
         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")
+        if isinstance(self._built_object, UnbuiltObject):
+            raise ValueError(f"{self.vertex_type}: {self._built_object_repr()}")
+        elif self._built_object is None:
+            message = f"{self.vertex_type} returned None."
+            if self.base_type == "custom_components":
+                message += " Make sure your build method returns a component."
+
+            raise ValueError(message)
 
     def build(self, force: bool = False) -> Any:
         if not self._built or force:

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

@@ -226,7 +226,12 @@ class PromptVertex(Vertex):
         # so the prompt format doesn't break
         artifacts.pop("handle_keys", None)
         try:
-            template = self._built_object.template
+            if not hasattr(self._built_object, "template") and hasattr(
+                self._built_object, "prompt"
+            ):
+                template = self._built_object.prompt.template
+            else:
+                template = self._built_object.template
             for key, value in artifacts.items():
                 if value:
                     replace_key = "{" + key + "}"

src/backend/langflow/interface/custom/constants.py

@@ -8,10 +8,13 @@ from langchain.text_splitter import TextSplitter
 from langchain.tools import Tool
 from langchain.vectorstores.base import VectorStore
 from langchain.schema import BaseOutputParser
-
+from langchain.schema.memory import BaseMemory
+from langchain.memory.chat_memory import BaseChatMemory
+from langchain.agents.agent import AgentExecutor
 
 LANGCHAIN_BASE_TYPES = {
     "Chain": Chain,
+    "AgentExecutor": AgentExecutor,
     "Tool": Tool,
     "BaseLLM": BaseLLM,
     "PromptTemplate": PromptTemplate,
@@ -22,6 +25,8 @@ LANGCHAIN_BASE_TYPES = {
     "Embeddings": Embeddings,
     "BaseRetriever": BaseRetriever,
     "BaseOutputParser": BaseOutputParser,
+    "BaseMemory": BaseMemory,
+    "BaseChatMemory": BaseChatMemory,
 }
 
 # Langchain base types plus Python base types

src/backend/langflow/interface/custom/custom_component.py

@@ -51,8 +51,8 @@ class CustomComponent(Component, extra=Extra.allow):
 
         for type_hint in TYPE_HINT_LIST:
             if reader._is_type_hint_used_in_args(
-                "Optional", code
-            ) and not reader._is_type_hint_imported("Optional", code):
+                type_hint, code
+            ) and not reader._is_type_hint_imported(type_hint, code):
                 error_detail = {
                     "error": "Type hint Error",
                     "traceback": f"Type hint '{type_hint}' is used but not imported in the code.",

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

@@ -51,7 +51,9 @@ def handle_partial_variables(prompt, format_kwargs: Dict):
     }
     # Remove handle_keys otherwise LangChain raises an error
     partial_variables.pop("handle_keys", None)
-    return prompt.partial(**partial_variables)
+    if partial_variables and hasattr(prompt, "partial"):
+        return prompt.partial(**partial_variables)
+    return prompt
 
 
 def handle_variable(params: Dict, input_variable: str, format_kwargs: Dict):

src/backend/langflow/interface/run.py

@@ -1,3 +1,4 @@
+from typing import Any, Dict, Tuple
 from langflow.services.cache.utils import memoize_dict
 from langflow.graph import Graph
 from langflow.utils.logger import logger
@@ -15,7 +16,7 @@ def build_langchain_object_with_caching(data_graph):
 
 
 @memoize_dict(maxsize=10)
-def build_sorted_vertices_with_caching(data_graph):
+def build_sorted_vertices_with_caching(data_graph) -> Tuple[Any, Dict]:
     """
     Build langchain object from data_graph.
     """

src/backend/langflow/processing/base.py

@@ -5,6 +5,7 @@ from langflow.api.v1.callback import (
 )
 from langflow.processing.process import fix_memory_inputs, format_actions
 from langflow.utils.logger import logger
+from langchain.agents.agent import AgentExecutor
 
 
 async def get_result_and_steps(langchain_object, inputs: Union[dict, str], **kwargs):
@@ -20,7 +21,8 @@ async def get_result_and_steps(langchain_object, inputs: Union[dict, str], **kwa
             # to display intermediate steps
             langchain_object.return_intermediate_steps = True
         try:
-            fix_memory_inputs(langchain_object)
+            if not isinstance(langchain_object, AgentExecutor):
+                fix_memory_inputs(langchain_object)
         except Exception as exc:
             logger.error(f"Error fixing memory inputs: {exc}")
 

src/backend/langflow/processing/process.py

@@ -85,39 +85,55 @@ def get_input_str_if_only_one_input(inputs: dict) -> Optional[str]:
     return list(inputs.values())[0] if len(inputs) == 1 else None
 
 
-def process_graph_cached(
-    data_graph: Dict[str, Any], inputs: Optional[dict] = None, clear_cache=False
-):
-    """
-    Process graph by extracting input variables and replacing ZeroShotPrompt
-    with PromptTemplate,then run the graph and return the result and thought.
-    """
-    # Load langchain object
+def get_build_result(data_graph, session_id):
+    # If session_id is provided, load the langchain_object from the session
+    # using build_sorted_vertices_with_caching.get_result_by_session_id
+    # if it returns something different than None, return it
+    # otherwise, build the graph and return the result
+    if session_id:
+        logger.debug(f"Loading LangChain object from session {session_id}")
+        result = build_sorted_vertices_with_caching.get_result_by_session_id(session_id)
+        if result is not None:
+            logger.debug("Loaded LangChain object")
+            return result
+
+    logger.debug("Building langchain object")
+    return build_sorted_vertices_with_caching(data_graph)
+
+
+def clear_caches_if_needed(clear_cache: bool):
     if clear_cache:
         build_sorted_vertices_with_caching.clear_cache()
         logger.debug("Cleared cache")
-    langchain_object, artifacts = build_sorted_vertices_with_caching(data_graph)
-    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,
-    ) in artifacts.items():
-        if key not in inputs or not inputs[key]:
-            inputs[key] = value
+
+def load_langchain_object(
+    data_graph: Dict[str, Any], session_id: str
+) -> Tuple[Union[Chain, VectorStore], Dict[str, Any], str]:
+    langchain_object, artifacts = get_build_result(data_graph, session_id)
+    session_id = build_sorted_vertices_with_caching.hash
+    logger.debug("Loaded LangChain object")
 
     if langchain_object is None:
-        # Raise user facing error
         raise ValueError(
             "There was an error loading the langchain_object. Please, check all the nodes and try again."
         )
 
-    # Generate result and thought
+    return langchain_object, artifacts, session_id
+
+
+def process_inputs(inputs: Optional[dict], artifacts: Dict[str, Any]) -> dict:
+    if inputs is None:
+        inputs = {}
+
+    for key, value in artifacts.items():
+        if key not in inputs or not inputs[key]:
+            inputs[key] = value
+
+    return inputs
+
+
+def generate_result(langchain_object: Union[Chain, VectorStore], inputs: dict):
     if isinstance(langchain_object, Chain):
         if inputs is None:
             raise ValueError("Inputs must be provided for a Chain")
@@ -130,9 +146,28 @@ def process_graph_cached(
         raise ValueError(
             f"Unknown langchain_object type: {type(langchain_object).__name__}"
         )
+
     return result
 
 
+def process_graph_cached(
+    data_graph: Dict[str, Any],
+    inputs: Optional[dict] = None,
+    clear_cache=False,
+    session_id=None,
+) -> Tuple[Any, str]:
+    clear_caches_if_needed(clear_cache)
+    # If session_id is provided, load the langchain_object from the session
+    # else build the graph and return the result and the new session_id
+    langchain_object, artifacts, session_id = load_langchain_object(
+        data_graph, session_id
+    )
+    processed_inputs = process_inputs(inputs, artifacts)
+    result = generate_result(langchain_object, processed_inputs)
+
+    return result, session_id
+
+
 def load_flow_from_json(
     flow: Union[Path, str, dict], tweaks: Optional[dict] = None, build=True
 ):

src/backend/langflow/services/cache/utils.py

@@ -30,6 +30,7 @@ def create_cache_folder(func):
 
 def memoize_dict(maxsize=128):
     cache = OrderedDict()
+    hash_to_key = {}  # Mapping from hash to cache key
 
     def decorator(func):
         @functools.wraps(func)
@@ -39,16 +40,29 @@ def memoize_dict(maxsize=128):
             if key not in cache:
                 result = func(*args, **kwargs)
                 cache[key] = result
+                hash_to_key[hashed] = key  # Store the mapping
                 if len(cache) > maxsize:
-                    cache.popitem(last=False)
+                    oldest_key = next(iter(cache))
+                    oldest_hash = oldest_key[1]
+                    del cache[oldest_key]
+                    del hash_to_key[oldest_hash]
             else:
                 result = cache[key]
+
+            wrapper.session_id = hashed  # Store hash in the wrapper
             return result
 
         def clear_cache():
             cache.clear()
+            hash_to_key.clear()
+
+        def get_result_by_session_id(session_id):
+            key = hash_to_key.get(session_id)
+            return cache.get(key) if key is not None else None
 
         wrapper.clear_cache = clear_cache  # type: ignore
+        wrapper.get_result_by_session_id = get_result_by_session_id  # type: ignore
+        wrapper.hash = None
         wrapper.cache = cache  # type: ignore
         return wrapper
 

src/backend/langflow/services/manager.py

@@ -18,7 +18,8 @@ class ServiceManager:
         """
         Registers a new factory.
         """
-        self.factories[service_factory.service_class.name] = service_factory
+        if service_factory.service_class.name not in self.factories:
+            self.factories[service_factory.service_class.name] = service_factory
 
     def get(self, service_name: ServiceType):
         """
@@ -85,3 +86,12 @@ def initialize_services():
     service_manager.register_factory(database_factory.DatabaseManagerFactory())
     service_manager.register_factory(cache_factory.CacheManagerFactory())
     service_manager.register_factory(chat_factory.ChatManagerFactory())
+
+
+def initialize_settings_manager():
+    """
+    Initialize the settings manager.
+    """
+    from langflow.services.settings import factory as settings_factory
+
+    service_manager.register_factory(settings_factory.SettingsManagerFactory())

src/backend/langflow/services/settings/base.py

@@ -1,6 +1,7 @@
 import contextlib
 import json
 import os
+from shutil import copy2
 import secrets
 from typing import Optional, List
 from pathlib import Path
@@ -9,7 +10,8 @@ import yaml
 from pydantic import BaseSettings, root_validator, validator
 from langflow.utils.logger import logger
 
-BASE_COMPONENTS_PATH = str(Path(__file__).parent / "components")
+# BASE_COMPONENTS_PATH = str(Path(__file__).parent / "components")
+BASE_COMPONENTS_PATH = str(Path(__file__).parent.parent.parent / "components")
 
 
 class Settings(BaseSettings):
@@ -30,6 +32,9 @@ class Settings(BaseSettings):
     OUTPUT_PARSERS: dict = {}
     CUSTOM_COMPONENTS: dict = {}
 
+    # Define the default LANGFLOW_DIR
+    CONFIG_DIR: Optional[str] = None
+
     DEV: bool = False
     DATABASE_URL: Optional[str] = None
     CACHE: str = "InMemoryCache"
@@ -54,8 +59,31 @@ class Settings(BaseSettings):
     FIRST_SUPERUSER: str = "langflow"
     FIRST_SUPERUSER_PASSWORD: str = "langflow"
 
+    @validator("CONFIG_DIR", pre=True, allow_reuse=True)
+    def set_langflow_dir(cls, value):
+        if not value:
+            import appdirs
+
+            # Define the app name and author
+            app_name = "langflow"
+            app_author = "logspace"
+
+            # Get the cache directory for the application
+            cache_dir = appdirs.user_cache_dir(app_name, app_author)
+
+            # Create a .langflow directory inside the cache directory
+            value = Path(cache_dir)
+            value.mkdir(parents=True, exist_ok=True)
+
+        if isinstance(value, str):
+            value = Path(value)
+        if not value.exists():
+            value.mkdir(parents=True, exist_ok=True)
+
+        return str(value)
+
     @validator("DATABASE_URL", pre=True)
-    def set_database_url(cls, value):
+    def set_database_url(cls, value, values):
         if not value:
             logger.debug(
                 "No database_url provided, trying LANGFLOW_DATABASE_URL env variable"
@@ -65,7 +93,28 @@ class Settings(BaseSettings):
                 logger.debug("Using LANGFLOW_DATABASE_URL env variable.")
             else:
                 logger.debug("No DATABASE_URL env variable, using sqlite database")
-                value = "sqlite:///./langflow.db"
+                # Originally, we used sqlite:///./langflow.db
+                # so we need to migrate to the new format
+                # if there is a database in that location
+                if not values["CONFIG_DIR"]:
+                    raise ValueError(
+                        "CONFIG_DIR not set, please set it or provide a DATABASE_URL"
+                    )
+
+                new_path = f"{values['CONFIG_DIR']}/langflow.db"
+                if Path("./langflow.db").exists():
+                    if Path(new_path).exists():
+                        logger.debug(f"Database already exists at {new_path}, using it")
+                    else:
+                        try:
+                            logger.debug("Copying existing database to new location")
+                            copy2("./langflow.db", new_path)
+                            logger.debug(f"Copied existing database to {new_path}")
+                        except Exception:
+                            logger.error("Failed to copy database, using default path")
+                            new_path = "./langflow.db"
+
+                value = f"sqlite:///{new_path}"
 
         return value
 
@@ -148,12 +197,17 @@ class Settings(BaseSettings):
                     value = json.loads(str(value))
                 if isinstance(value, list):
                     for item in value:
+                        if isinstance(item, Path):
+                            item = str(item)
                         if item not in getattr(self, key):
                             getattr(self, key).append(item)
                     logger.debug(f"Extended {key}")
                 else:
-                    getattr(self, key).append(value)
-                    logger.debug(f"Appended {key}")
+                    if isinstance(value, Path):
+                        value = str(value)
+                    if value not in getattr(self, key):
+                        getattr(self, key).append(value)
+                        logger.debug(f"Appended {key}")
 
             else:
                 setattr(self, key, value)