Merge remote-tracking branch 'origin/zustand/io/migration' into globalVariables

src/backend/base/langflow/__main__.py

@@ -1,23 +1,29 @@
 import platform
 import socket
 import sys
+import time
+import webbrowser
 from pathlib import Path
 from typing import Optional
 
+import httpx
 import typer
 from dotenv import load_dotenv
-from langflow.main import setup_app
-from langflow.services.database.utils import session_getter
-from langflow.services.deps import get_db_service, get_settings_service
-from langflow.services.utils import initialize_services, initialize_settings_service
-from langflow.utils.logger import configure, logger
-from multiprocess import cpu_count  # type: ignore
+from multiprocess import (
+    Process,  # type: ignore
+    cpu_count,  # type: ignore
+)
 from rich import box
 from rich import print as rprint
 from rich.console import Console
 from rich.panel import Panel
 from rich.table import Table
-from sqlmodel import select
+
+from langflow.main import setup_app
+from langflow.services.database.utils import session_getter
+from langflow.services.deps import get_db_service, get_settings_service
+from langflow.services.utils import initialize_services, initialize_settings_service
+from langflow.utils.logger import configure, logger
 
 console = Console()
 
@@ -70,7 +76,7 @@ def update_settings(
     dev: bool = False,
     remove_api_keys: bool = False,
     components_path: Optional[Path] = None,
-    store: bool = False,
+    store: bool = True,
 ):
     """Update the settings from a config file."""
 
@@ -94,33 +100,6 @@ def update_settings(
         settings_service.settings.update_settings(STORE=False)
 
 
-def version_callback(value: bool):
-    """
-    Show the version and exit.
-    """
-    from langflow import __version__
-
-    if value:
-        typer.echo(f"Langflow Version: {__version__}")
-        raise typer.Exit()
-
-
-@app.callback()
-def main_entry_point(
-    version: bool = typer.Option(
-        None,
-        "--version",
-        callback=version_callback,
-        is_eager=True,
-        help="Show the version and exit.",
-    ),
-):
-    """
-    Main entry point for the Langflow CLI.
-    """
-    pass
-
-
 @app.command()
 def run(
     host: str = typer.Option("127.0.0.1", help="Host to bind the server to.", envvar="LANGFLOW_HOST"),
@@ -212,12 +191,23 @@ def run(
         run_on_windows(host, port, log_level, options, app)
     else:
         # Run using gunicorn on Linux
-        run_on_mac_or_linux(host, port, log_level, options, app)
+        run_on_mac_or_linux(host, port, log_level, options, app, open_browser)
 
 
-def run_on_mac_or_linux(host, port, log_level, options, app):
+def run_on_mac_or_linux(host, port, log_level, options, app, open_browser=True):
+    webapp_process = Process(target=run_langflow, args=(host, port, log_level, options, app))
+    webapp_process.start()
+    status_code = 0
+    while status_code != 200:
+        try:
+            status_code = httpx.get(f"http://{host}:{port}/health").status_code
+
+        except Exception:
+            time.sleep(1)
+
     print_banner(host, port)
-    run_langflow(host, port, log_level, options, app)
+    if open_browser:
+        webbrowser.open(f"http://{host}:{port}")
 
 
 def run_on_windows(host, port, log_level, options, app):
@@ -292,11 +282,10 @@ def run_langflow(host, port, log_level, options, app):
     Run Langflow server on localhost
     """
     try:
-        if platform.system() in ["Windows", "Darwin"]:
+        if platform.system() in ["Windows"]:
             # Run using uvicorn on MacOS and Windows
             # Windows doesn't support gunicorn
             # MacOS requires an env variable to be set to use gunicorn
-
             import uvicorn
 
             uvicorn.run(
@@ -310,8 +299,7 @@ def run_langflow(host, port, log_level, options, app):
 
             LangflowApplication(app, options).run()
     except KeyboardInterrupt:
-        logger.info("Shutting down server")
-        sys.exit(0)
+        pass
     except Exception as e:
         logger.exception(e)
         sys.exit(1)
@@ -336,7 +324,7 @@ def superuser(
             # Verify that the superuser was created
             from langflow.services.database.models.user.model import User
 
-            user: User = session.exec(select(User).where(User.username == username)).first()
+            user: User = session.query(User).filter(User.username == username).first()
             if user is None or not user.is_superuser:
                 typer.echo("Superuser creation failed.")
                 return
@@ -348,23 +336,11 @@ def superuser(
 
 
 @app.command()
-def migration(
-    test: bool = typer.Option(True, help="Run migrations in test mode."),
-    fix: bool = typer.Option(
-        False,
-        help="Fix migrations. This is a destructive operation, and should only be used if you know what you are doing.",
-    ),
-):
+def migration(test: bool = typer.Option(True, help="Run migrations in test mode.")):
     """
     Run or test migrations.
     """
-    if fix:
-        if not typer.confirm(
-            "This will delete all data necessary to fix migrations. Are you sure you want to continue?"
-        ):
-            raise typer.Abort()
-
-    initialize_services(fix_migration=fix)
+    initialize_services()
     db_service = get_db_service()
     if not test:
         db_service.run_migrations()

src/backend/base/langflow/alembic/env.py

@@ -66,16 +66,10 @@ def run_migrations_online() -> None:
     try:
         from langflow.services.database.factory import DatabaseServiceFactory
         from langflow.services.deps import get_db_service
-        from langflow.services.manager import (
-            initialize_settings_service,
-            service_manager,
-        )
-        from langflow.services.schema import ServiceType
+        from langflow.services.manager import initialize_settings_service, service_manager
 
         initialize_settings_service()
-        service_manager.register_factory(
-            DatabaseServiceFactory(), [ServiceType.SETTINGS_SERVICE]
-        )
+        service_manager.register_factory(DatabaseServiceFactory())
         connectable = get_db_service().engine
     except Exception as e:
         logger.error(f"Error getting database engine: {e}")
@@ -89,9 +83,7 @@ def run_migrations_online() -> None:
             poolclass=pool.NullPool,
         )
     with connectable.connect() as connection:
-        context.configure(
-            connection=connection, target_metadata=target_metadata, render_as_batch=True
-        )
+        context.configure(connection=connection, target_metadata=target_metadata, render_as_batch=True)
 
         with context.begin_transaction():
             context.run_migrations()

src/backend/base/langflow/alembic/helpers/flow.py

@@ -0,0 +1,199 @@
+from typing import TYPE_CHECKING, Any, Callable, Coroutine, List, Optional, Tuple, Union
+
+from pydantic.v1 import BaseModel, Field, create_model
+from sqlmodel import select
+
+from langflow.schema.schema import INPUT_FIELD_NAME, Record
+from langflow.services.database.models.flow.model import Flow
+from langflow.services.deps import session_scope
+
+if TYPE_CHECKING:
+    from langflow.graph.graph.base import Graph
+    from langflow.graph.vertex.base import Vertex
+
+INPUT_TYPE_MAP = {
+    "ChatInput": {"type_hint": "Optional[str]", "default": '""'},
+    "TextInput": {"type_hint": "Optional[str]", "default": '""'},
+    "JSONInput": {"type_hint": "Optional[dict]", "default": "{}"},
+}
+
+
+def list_flows(*, user_id: Optional[str] = None) -> List[Record]:
+    if not user_id:
+        raise ValueError("Session is invalid")
+    try:
+        with session_scope() as session:
+            flows = session.exec(
+                select(Flow).where(Flow.user_id == user_id).where(Flow.is_component == False)  # noqa
+            ).all()
+
+            flows_records = [flow.to_record() for flow in flows]
+            return flows_records
+    except Exception as e:
+        raise ValueError(f"Error listing flows: {e}")
+
+
+async def load_flow(
+    user_id: str, flow_id: Optional[str] = None, flow_name: Optional[str] = None, tweaks: Optional[dict] = None
+) -> "Graph":
+    from langflow.graph.graph.base import Graph
+    from langflow.processing.process import process_tweaks
+
+    if not flow_id and not flow_name:
+        raise ValueError("Flow ID or Flow Name is required")
+    if not flow_id and flow_name:
+        flow_id = find_flow(flow_name, user_id)
+        if not flow_id:
+            raise ValueError(f"Flow {flow_name} not found")
+
+    with session_scope() as session:
+        graph_data = flow.data if (flow := session.get(Flow, flow_id)) else None
+    if not graph_data:
+        raise ValueError(f"Flow {flow_id} not found")
+    if tweaks:
+        graph_data = process_tweaks(graph_data=graph_data, tweaks=tweaks)
+    graph = Graph.from_payload(graph_data, flow_id=flow_id)
+    return graph
+
+
+def find_flow(flow_name: str, user_id: str) -> Optional[str]:
+    with session_scope() as session:
+        flow = session.exec(select(Flow).where(Flow.name == flow_name).where(Flow.user_id == user_id)).first()
+        return flow.id if flow else None
+
+
+async def run_flow(
+    inputs: Union[dict, List[dict]] = None,
+    tweaks: Optional[dict] = None,
+    flow_id: Optional[str] = None,
+    flow_name: Optional[str] = None,
+    user_id: Optional[str] = None,
+) -> Any:
+    graph = await load_flow(user_id, flow_id, flow_name, tweaks)
+
+    if inputs is None:
+        inputs = []
+    inputs_list = []
+    inputs_components = []
+    types = []
+    for input_dict in inputs:
+        inputs_list.append({INPUT_FIELD_NAME: input_dict.get("input_value")})
+        inputs_components.append(input_dict.get("components", []))
+        types.append(input_dict.get("type", []))
+
+    return await graph.arun(inputs_list, inputs_components=inputs_components, types=types)
+
+
+def generate_function_for_flow(inputs: List["Vertex"], flow_id: str) -> Coroutine:
+    """
+    Generate a dynamic flow function based on the given inputs and flow ID.
+
+    Args:
+        inputs (List[Vertex]): The list of input vertices for the flow.
+        flow_id (str): The ID of the flow.
+
+    Returns:
+        Coroutine: The dynamic flow function.
+
+    Raises:
+        None
+
+    Example:
+        inputs = [vertex1, vertex2]
+        flow_id = "my_flow"
+        function = generate_function_for_flow(inputs, flow_id)
+        result = function(input1, input2)
+    """
+    # Prepare function arguments with type hints and default values
+    args = [
+        f"{input_.display_name.lower().replace(' ', '_')}: {INPUT_TYPE_MAP[input_.base_name]['type_hint']} = {INPUT_TYPE_MAP[input_.base_name]['default']}"
+        for input_ in inputs
+    ]
+
+    # Maintain original argument names for constructing the tweaks dictionary
+    original_arg_names = [input_.display_name for input_ in inputs]
+
+    # Prepare a Pythonic, valid function argument string
+    func_args = ", ".join(args)
+
+    # Map original argument names to their corresponding Pythonic variable names in the function
+    arg_mappings = ", ".join(
+        f'"{original_name}": {name}'
+        for original_name, name in zip(original_arg_names, [arg.split(":")[0] for arg in args])
+    )
+
+    func_body = f"""
+from typing import Optional
+async def flow_function({func_args}):
+    tweaks = {{ {arg_mappings} }}
+    from langflow.helpers.flow import run_flow
+    from langchain_core.tools import ToolException
+    try:
+        return await run_flow(
+            tweaks={{key: {{'input_value': value}} for key, value in tweaks.items()}},
+            flow_id="{flow_id}",
+        )
+    except Exception as e:
+        raise ToolException(f'Error running flow: ' + e)
+"""
+
+    compiled_func = compile(func_body, "<string>", "exec")
+    local_scope = {}
+    exec(compiled_func, globals(), local_scope)
+    return local_scope["flow_function"]
+
+
+def build_function_and_schema(flow_record: Record, graph: "Graph") -> Tuple[Callable, BaseModel]:
+    """
+    Builds a dynamic function and schema for a given flow.
+
+    Args:
+        flow_record (Record): The flow record containing information about the flow.
+        graph (Graph): The graph representing the flow.
+
+    Returns:
+        Tuple[Callable, BaseModel]: A tuple containing the dynamic function and the schema.
+    """
+    flow_id = flow_record.id
+    inputs = get_flow_inputs(graph)
+    dynamic_flow_function = generate_function_for_flow(inputs, flow_id)
+    schema = build_schema_from_inputs(flow_record.name, inputs)
+    return dynamic_flow_function, schema
+
+
+def get_flow_inputs(graph: "Graph") -> List["Vertex"]:
+    """
+    Retrieves the flow inputs from the given graph.
+
+    Args:
+        graph (Graph): The graph object representing the flow.
+
+    Returns:
+        List[Record]: A list of input records, where each record contains the ID, name, and description of the input vertex.
+    """
+    inputs = []
+    for vertex in graph.vertices:
+        if vertex.is_input:
+            inputs.append(vertex)
+    return inputs
+
+
+def build_schema_from_inputs(name: str, inputs: List[tuple[str, str, str]]) -> BaseModel:
+    """
+    Builds a schema from the given inputs.
+
+    Args:
+        name (str): The name of the schema.
+        inputs (List[tuple[str, str, str]]): A list of tuples representing the inputs.
+            Each tuple contains three elements: the input name, the input type, and the input description.
+
+    Returns:
+        BaseModel: The schema model.
+
+    """
+    fields = {}
+    for input_ in inputs:
+        field_name = input_.display_name.lower().replace(" ", "_")
+        description = input_.description
+        fields[field_name] = (str, Field(default="", description=description))
+    return create_model(name, **fields)

src/backend/base/langflow/alembic/versions/006b3990db50_add_unique_constraints.py

@@ -26,20 +26,13 @@ def upgrade() -> None:
     flow_constraints = inspector.get_unique_constraints("flow")
     user_constraints = inspector.get_unique_constraints("user")
     try:
-        if not any(
-            constraint["name"] == "uq_apikey_id" for constraint in api_key_constraints
-        ):
+        if not any(constraint["name"] == "uq_apikey_id" for constraint in api_key_constraints):
             with op.batch_alter_table("apikey", schema=None) as batch_op:
-
                 batch_op.create_unique_constraint("uq_apikey_id", ["id"])
-        if not any(
-            constraint["name"] == "uq_flow_id" for constraint in flow_constraints
-        ):
+        if not any(constraint["name"] == "uq_flow_id" for constraint in flow_constraints):
             with op.batch_alter_table("flow", schema=None) as batch_op:
                 batch_op.create_unique_constraint("uq_flow_id", ["id"])
-        if not any(
-            constraint["name"] == "uq_user_id" for constraint in user_constraints
-        ):
+        if not any(constraint["name"] == "uq_user_id" for constraint in user_constraints):
             with op.batch_alter_table("user", schema=None) as batch_op:
                 batch_op.create_unique_constraint("uq_user_id", ["id"])
     except Exception as e:
@@ -57,16 +50,13 @@ def downgrade() -> None:
     flow_constraints = inspector.get_unique_constraints("flow")
     user_constraints = inspector.get_unique_constraints("user")
     try:
-        if any(
-            constraint["name"] == "uq_apikey_id" for constraint in api_key_constraints
-        ):
+        if any(constraint["name"] == "uq_apikey_id" for constraint in api_key_constraints):
             with op.batch_alter_table("user", schema=None) as batch_op:
                 batch_op.drop_constraint("uq_user_id", type_="unique")
         if any(constraint["name"] == "uq_flow_id" for constraint in flow_constraints):
             with op.batch_alter_table("flow", schema=None) as batch_op:
                 batch_op.drop_constraint("uq_flow_id", type_="unique")
         if any(constraint["name"] == "uq_user_id" for constraint in user_constraints):
-
             with op.batch_alter_table("apikey", schema=None) as batch_op:
                 batch_op.drop_constraint("uq_apikey_id", type_="unique")
     except Exception as e:

src/backend/base/langflow/alembic/versions/1a110b568907_replace_credential_table_with_variable.py

@@ -0,0 +1,65 @@
+"""Replace Credential table with Variable
+
+Revision ID: 1a110b568907
+Revises: 63b9c451fd30
+Create Date: 2024-03-25 09:40:02.743453
+
+"""
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+import sqlmodel
+from alembic import op
+from sqlalchemy.engine.reflection import Inspector
+
+# revision identifiers, used by Alembic.
+revision: str = "1a110b568907"
+down_revision: Union[str, None] = "63b9c451fd30"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    conn = op.get_bind()
+    inspector = Inspector.from_engine(conn)  # type: ignore
+    table_names = inspector.get_table_names()
+    # ### commands auto generated by Alembic - please adjust! ###
+    if "variable" not in table_names:
+        op.create_table(
+            "variable",
+            sa.Column("name", sqlmodel.sql.sqltypes.AutoString(), nullable=True),
+            sa.Column("value", sqlmodel.sql.sqltypes.AutoString(), nullable=True),
+            sa.Column("type", sqlmodel.sql.sqltypes.AutoString(), nullable=True),
+            sa.Column("id", sqlmodel.sql.sqltypes.GUID(), nullable=False),
+            sa.Column("created_at", sa.DateTime(), nullable=False),
+            sa.Column("updated_at", sa.DateTime(), nullable=True),
+            sa.Column("user_id", sqlmodel.sql.sqltypes.GUID(), nullable=False),
+            sa.ForeignKeyConstraint(["user_id"], ["user.id"], name="fk_variable_user_id"),
+            sa.PrimaryKeyConstraint("id"),
+        )
+    if "credential" in table_names:
+        op.drop_table("credential")
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    conn = op.get_bind()
+    inspector = Inspector.from_engine(conn)  # type: ignore
+    table_names = inspector.get_table_names()
+    # ### commands auto generated by Alembic - please adjust! ###
+    if "credential" not in table_names:
+        op.create_table(
+            "credential",
+            sa.Column("name", sa.VARCHAR(), nullable=True),
+            sa.Column("value", sa.VARCHAR(), nullable=True),
+            sa.Column("provider", sa.VARCHAR(), nullable=True),
+            sa.Column("user_id", sa.CHAR(length=32), nullable=False),
+            sa.Column("id", sa.CHAR(length=32), nullable=False),
+            sa.Column("created_at", sa.DATETIME(), nullable=False),
+            sa.Column("updated_at", sa.DATETIME(), nullable=True),
+            sa.ForeignKeyConstraint(["user_id"], ["user.id"], name="fk_credential_user_id"),
+            sa.PrimaryKeyConstraint("id"),
+        )
+    if "variable" in table_names:
+        op.drop_table("variable")
+    # ### end Alembic commands ###

src/backend/base/langflow/alembic/versions/1ef9c4f3765d_.py

@@ -24,10 +24,8 @@ def upgrade() -> None:
     # ### commands auto generated by Alembic - please adjust! ###
     try:
         with op.batch_alter_table("apikey", schema=None) as batch_op:
-            batch_op.alter_column(
-                "name", existing_type=sqlmodel.sql.sqltypes.AutoString(), nullable=True
-            )
-    except Exception as e:
+            batch_op.alter_column("name", existing_type=sqlmodel.sql.sqltypes.AutoString(), nullable=True)
+    except Exception:
         pass
     # ### end Alembic commands ###
 
@@ -37,6 +35,6 @@ def downgrade() -> None:
     try:
         with op.batch_alter_table("apikey", schema=None) as batch_op:
             batch_op.alter_column("name", existing_type=sa.VARCHAR(), nullable=False)
-    except Exception as e:
+    except Exception:
         pass
     # ### end Alembic commands ###

src/backend/base/langflow/alembic/versions/260dbcc8b680_adds_tables.py

@@ -31,23 +31,16 @@ def upgrade() -> None:
     # and other related indices
     if "flowstyle" in existing_tables:
         op.drop_table("flowstyle")
-        if "ix_flowstyle_flow_id" in [
-            index["name"] for index in inspector.get_indexes("flowstyle")
-        ]:
-            op.drop_index(
-                "ix_flowstyle_flow_id", table_name="flowstyle", if_exists=True
-            )
+        if "ix_flowstyle_flow_id" in [index["name"] for index in inspector.get_indexes("flowstyle")]:
+            op.drop_index("ix_flowstyle_flow_id", table_name="flowstyle", if_exists=True)
 
     existing_indices_flow = []
     existing_fks_flow = []
     if "flow" in existing_tables:
-        existing_indices_flow = [
-            index["name"] for index in inspector.get_indexes("flow")
-        ]
+        existing_indices_flow = [index["name"] for index in inspector.get_indexes("flow")]
         # Existing foreign keys for the 'flow' table, if it exists
         existing_fks_flow = [
-            fk["referred_table"] + "." + fk["referred_columns"][0]
-            for fk in inspector.get_foreign_keys("flow")
+            fk["referred_table"] + "." + fk["referred_columns"][0] for fk in inspector.get_foreign_keys("flow")
         ]
         # Now check if the columns user_id exists in the 'flow' table
         # If it does not exist, we need to create the foreign key
@@ -67,9 +60,7 @@ def upgrade() -> None:
             sa.UniqueConstraint("id", name="uq_user_id"),
         )
         with op.batch_alter_table("user", schema=None) as batch_op:
-            batch_op.create_index(
-                batch_op.f("ix_user_username"), ["username"], unique=True
-            )
+            batch_op.create_index(batch_op.f("ix_user_username"), ["username"], unique=True)
 
     if "apikey" not in existing_tables:
         op.create_table(
@@ -82,20 +73,14 @@ def upgrade() -> None:
             sa.Column("id", sqlmodel.sql.sqltypes.GUID(), nullable=False),
             sa.Column("api_key", sqlmodel.sql.sqltypes.AutoString(), nullable=False),
             sa.Column("user_id", sqlmodel.sql.sqltypes.GUID(), nullable=False),
-            sa.ForeignKeyConstraint(
-                ["user_id"], ["user.id"], name="fk_apikey_user_id_user"
-            ),
+            sa.ForeignKeyConstraint(["user_id"], ["user.id"], name="fk_apikey_user_id_user"),
             sa.PrimaryKeyConstraint("id", name="pk_apikey"),
             sa.UniqueConstraint("id", name="uq_apikey_id"),
         )
         with op.batch_alter_table("apikey", schema=None) as batch_op:
-            batch_op.create_index(
-                batch_op.f("ix_apikey_api_key"), ["api_key"], unique=True
-            )
+            batch_op.create_index(batch_op.f("ix_apikey_api_key"), ["api_key"], unique=True)
             batch_op.create_index(batch_op.f("ix_apikey_name"), ["name"], unique=False)
-            batch_op.create_index(
-                batch_op.f("ix_apikey_user_id"), ["user_id"], unique=False
-            )
+            batch_op.create_index(batch_op.f("ix_apikey_user_id"), ["user_id"], unique=False)
     if "flow" not in existing_tables:
         op.create_table(
             "flow",
@@ -104,9 +89,7 @@ def upgrade() -> None:
             sa.Column("description", sqlmodel.sql.sqltypes.AutoString(), nullable=True),
             sa.Column("id", sqlmodel.sql.sqltypes.GUID(), nullable=False),
             sa.Column("user_id", sqlmodel.sql.sqltypes.GUID(), nullable=False),
-            sa.ForeignKeyConstraint(
-                ["user_id"], ["user.id"], name="fk_flow_user_id_user"
-            ),
+            sa.ForeignKeyConstraint(["user_id"], ["user.id"], name="fk_flow_user_id_user"),
             sa.PrimaryKeyConstraint("id", name="pk_flow"),
             sa.UniqueConstraint("id", name="uq_flow_id"),
         )
@@ -129,16 +112,12 @@ def upgrade() -> None:
         if "user.id" not in existing_fks_flow:
             batch_op.create_foreign_key("fk_flow_user_id", "user", ["user_id"], ["id"])
         if "ix_flow_description" not in existing_indices_flow:
-            batch_op.create_index(
-                batch_op.f("ix_flow_description"), ["description"], unique=False
-            )
+            batch_op.create_index(batch_op.f("ix_flow_description"), ["description"], unique=False)
         if "ix_flow_name" not in existing_indices_flow:
             batch_op.create_index(batch_op.f("ix_flow_name"), ["name"], unique=False)
     with op.batch_alter_table("flow", schema=None) as batch_op:
         if "ix_flow_user_id" not in existing_indices_flow:
-            batch_op.create_index(
-                batch_op.f("ix_flow_user_id"), ["user_id"], unique=False
-            )
+            batch_op.create_index(batch_op.f("ix_flow_user_id"), ["user_id"], unique=False)
 
     # ### end Alembic commands ###
 
@@ -169,10 +148,4 @@ def downgrade() -> None:
             batch_op.drop_index(batch_op.f("ix_user_username"), if_exists=True)
 
         op.drop_table("user")
-
-    if "flowstyle" in existing_tables:
-        op.drop_table("flowstyle")
-
-    if "component" in existing_tables:
-        op.drop_table("component")
     # ### end Alembic commands ###

src/backend/base/langflow/alembic/versions/2ac71eb9c3ae_adds_credential_table.py

@@ -31,9 +31,7 @@ def upgrade() -> None:
                 "credential",
                 sa.Column("name", sqlmodel.sql.sqltypes.AutoString(), nullable=True),
                 sa.Column("value", sqlmodel.sql.sqltypes.AutoString(), nullable=True),
-                sa.Column(
-                    "provider", sqlmodel.sql.sqltypes.AutoString(), nullable=True
-                ),
+                sa.Column("provider", sqlmodel.sql.sqltypes.AutoString(), nullable=True),
                 sa.Column("user_id", sqlmodel.sql.sqltypes.GUID(), nullable=False),
                 sa.Column("id", sqlmodel.sql.sqltypes.GUID(), nullable=False),
                 sa.Column("created_at", sa.DateTime(), nullable=False),

src/backend/base/langflow/alembic/versions/63b9c451fd30_add_icon_and_icon_bg_color_to_flow.py

@@ -23,20 +23,14 @@ depends_on: Union[str, Sequence[str], None] = None
 def upgrade() -> None:
     conn = op.get_bind()
     inspector = Inspector.from_engine(conn)  # type: ignore
-    table_names = inspector.get_table_names()
+    table_names = inspector.get_table_names()  # noqa
     column_names = [column["name"] for column in inspector.get_columns("flow")]
     # ### commands auto generated by Alembic - please adjust! ###
     with op.batch_alter_table("flow", schema=None) as batch_op:
         if "icon" not in column_names:
-            batch_op.add_column(
-                sa.Column("icon", sqlmodel.sql.sqltypes.AutoString(), nullable=True)
-            )
+            batch_op.add_column(sa.Column("icon", sqlmodel.sql.sqltypes.AutoString(), nullable=True))
         if "icon_bg_color" not in column_names:
-            batch_op.add_column(
-                sa.Column(
-                    "icon_bg_color", sqlmodel.sql.sqltypes.AutoString(), nullable=True
-                )
-            )
+            batch_op.add_column(sa.Column("icon_bg_color", sqlmodel.sql.sqltypes.AutoString(), nullable=True))
 
     # ### end Alembic commands ###
 
@@ -44,7 +38,7 @@ def upgrade() -> None:
 def downgrade() -> None:
     conn = op.get_bind()
     inspector = Inspector.from_engine(conn)  # type: ignore
-    table_names = inspector.get_table_names()
+    table_names = inspector.get_table_names()  # noqa
     column_names = [column["name"] for column in inspector.get_columns("flow")]
     # ### commands auto generated by Alembic - please adjust! ###
     with op.batch_alter_table("flow", schema=None) as batch_op:

src/backend/base/langflow/alembic/versions/7843803a87b5_store_updates.py

@@ -29,18 +29,14 @@ def upgrade() -> None:
     try:
         if "is_component" not in flow_columns:
             with op.batch_alter_table("flow", schema=None) as batch_op:
-                batch_op.add_column(
-                    sa.Column("is_component", sa.Boolean(), nullable=True)
-                )
-    except Exception as e:
+                batch_op.add_column(sa.Column("is_component", sa.Boolean(), nullable=True))
+    except Exception:
         pass
     try:
         if "store_api_key" not in user_columns:
             with op.batch_alter_table("user", schema=None) as batch_op:
-                batch_op.add_column(
-                    sa.Column("store_api_key", sqlmodel.AutoString(), nullable=True)
-                )
-    except Exception as e:
+                batch_op.add_column(sa.Column("store_api_key", sqlmodel.AutoString(), nullable=True))
+    except Exception:
         pass
     # ### end Alembic commands ###
 

src/backend/base/langflow/alembic/versions/7d2162acc8b2_adds_updated_at_and_folder_cols.py

@@ -30,9 +30,7 @@ def upgrade() -> None:
     try:
         if "name" in api_key_columns:
             with op.batch_alter_table("apikey", schema=None) as batch_op:
-                batch_op.alter_column(
-                    "name", existing_type=sa.VARCHAR(), nullable=False
-                )
+                batch_op.alter_column("name", existing_type=sa.VARCHAR(), nullable=False)
     except Exception as e:
         print(e)
 
@@ -40,15 +38,9 @@ def upgrade() -> None:
     try:
         with op.batch_alter_table("flow", schema=None) as batch_op:
             if "updated_at" not in flow_columns:
-                batch_op.add_column(
-                    sa.Column("updated_at", sa.DateTime(), nullable=True)
-                )
+                batch_op.add_column(sa.Column("updated_at", sa.DateTime(), nullable=True))
             if "folder" not in flow_columns:
-                batch_op.add_column(
-                    sa.Column(
-                        "folder", sqlmodel.sql.sqltypes.AutoString(), nullable=True
-                    )
-                )
+                batch_op.add_column(sa.Column("folder", sqlmodel.sql.sqltypes.AutoString(), nullable=True))
     except Exception as e:
         print(e)
 
@@ -68,7 +60,6 @@ def downgrade() -> None:
         pass
 
     try:
-
         with op.batch_alter_table("apikey", schema=None) as batch_op:
             batch_op.alter_column("name", existing_type=sa.VARCHAR(), nullable=True)
     except Exception as e:

src/backend/base/langflow/alembic/versions/b2fa308044b5_add_unique_constraints.py

@@ -32,33 +32,19 @@ def upgrade() -> None:
         with op.batch_alter_table("flow", schema=None) as batch_op:
             flow_columns = [column["name"] for column in inspector.get_columns("flow")]
             if "is_component" not in flow_columns:
-                batch_op.add_column(
-                    sa.Column("is_component", sa.Boolean(), nullable=True)
-                )
+                batch_op.add_column(sa.Column("is_component", sa.Boolean(), nullable=True))
             if "updated_at" not in flow_columns:
-                batch_op.add_column(
-                    sa.Column("updated_at", sa.DateTime(), nullable=True)
-                )
+                batch_op.add_column(sa.Column("updated_at", sa.DateTime(), nullable=True))
             if "folder" not in flow_columns:
-                batch_op.add_column(
-                    sa.Column(
-                        "folder", sqlmodel.sql.sqltypes.AutoString(), nullable=True
-                    )
-                )
+                batch_op.add_column(sa.Column("folder", sqlmodel.sql.sqltypes.AutoString(), nullable=True))
             if "user_id" not in flow_columns:
-                batch_op.add_column(
-                    sa.Column("user_id", sqlmodel.sql.sqltypes.GUID(), nullable=True)
-                )
+                batch_op.add_column(sa.Column("user_id", sqlmodel.sql.sqltypes.GUID(), nullable=True))
             indices = inspector.get_indexes("flow")
             indices_names = [index["name"] for index in indices]
             if "ix_flow_user_id" not in indices_names:
-                batch_op.create_index(
-                    batch_op.f("ix_flow_user_id"), ["user_id"], unique=False
-                )
+                batch_op.create_index(batch_op.f("ix_flow_user_id"), ["user_id"], unique=False)
             if "fk_flow_user_id_user" not in indices_names:
-                batch_op.create_foreign_key(
-                    "fk_flow_user_id_user", "user", ["user_id"], ["id"]
-                )
+                batch_op.create_foreign_key("fk_flow_user_id_user", "user", ["user_id"], ["id"])
 
     except Exception:
         pass

src/backend/base/langflow/alembic/versions/bc2f01c40e4a_new_fixes.py

@@ -33,21 +33,13 @@ def upgrade() -> None:
         if "updated_at" not in flow_columns:
             batch_op.add_column(sa.Column("updated_at", sa.DateTime(), nullable=True))
         if "folder" not in flow_columns:
-            batch_op.add_column(
-                sa.Column("folder", sqlmodel.sql.sqltypes.AutoString(), nullable=True)
-            )
+            batch_op.add_column(sa.Column("folder", sqlmodel.sql.sqltypes.AutoString(), nullable=True))
         if "user_id" not in flow_columns:
-            batch_op.add_column(
-                sa.Column("user_id", sqlmodel.sql.sqltypes.GUID(), nullable=True)
-            )
+            batch_op.add_column(sa.Column("user_id", sqlmodel.sql.sqltypes.GUID(), nullable=True))
         if "ix_flow_user_id" not in flow_indexes:
-            batch_op.create_index(
-                batch_op.f("ix_flow_user_id"), ["user_id"], unique=False
-            )
+            batch_op.create_index(batch_op.f("ix_flow_user_id"), ["user_id"], unique=False)
         if "flow_user_id_fkey" not in flow_fks:
-            batch_op.create_foreign_key(
-                "flow_user_id_fkey", "user", ["user_id"], ["id"]
-            )
+            batch_op.create_foreign_key("flow_user_id_fkey", "user", ["user_id"], ["id"])
 
 
 def downgrade() -> None:

src/backend/base/langflow/alembic/versions/eb5866d51fd2_change_columns_to_be_nullable.py

@@ -19,7 +19,7 @@ depends_on: Union[str, Sequence[str], None] = None
 
 def upgrade() -> None:
     # ### commands auto generated by Alembic - please adjust! ###
-    connection = op.get_bind()
+    connection = op.get_bind()  # noqa
 
     pass
     # ### end Alembic commands ###

src/backend/base/langflow/alembic/versions/f5ee9749d1a6_user_id_can_be_null_in_flow.py

@@ -22,9 +22,7 @@ def upgrade() -> None:
     # ### commands auto generated by Alembic - please adjust! ###
     try:
         with op.batch_alter_table("flow", schema=None) as batch_op:
-            batch_op.alter_column(
-                "user_id", existing_type=sa.CHAR(length=32), nullable=True
-            )
+            batch_op.alter_column("user_id", existing_type=sa.CHAR(length=32), nullable=True)
     except Exception as e:
         print(e)
         pass
@@ -36,9 +34,7 @@ def downgrade() -> None:
     # ### commands auto generated by Alembic - please adjust! ###
     try:
         with op.batch_alter_table("flow", schema=None) as batch_op:
-            batch_op.alter_column(
-                "user_id", existing_type=sa.CHAR(length=32), nullable=False
-            )
+            batch_op.alter_column("user_id", existing_type=sa.CHAR(length=32), nullable=False)
     except Exception as e:
         print(e)
         pass

src/backend/base/langflow/alembic/versions/fd531f8868b1_fix_credential_table.py

@@ -31,9 +31,7 @@ def upgrade() -> None:
     try:
         if "credential" in tables and "fk_credential_user_id" not in foreign_keys_names:
             with op.batch_alter_table("credential", schema=None) as batch_op:
-                batch_op.create_foreign_key(
-                    "fk_credential_user_id", "user", ["user_id"], ["id"]
-                )
+                batch_op.create_foreign_key("fk_credential_user_id", "user", ["user_id"], ["id"])
     except Exception as e:
         print(e)
         pass

src/backend/base/langflow/api/v1/api_key.py

@@ -8,20 +8,10 @@ from langflow.api.v1.schemas import ApiKeyCreateRequest, ApiKeysResponse
 from langflow.services.auth import utils as auth_utils
 
 # Assuming you have these methods in your service layer
-from langflow.services.database.models.api_key.crud import (
-    create_api_key,
-    delete_api_key,
-    get_api_keys,
-)
-from langflow.services.database.models.api_key.model import (
-    ApiKeyCreate,
-    UnmaskedApiKeyRead,
-)
+from langflow.services.database.models.api_key.crud import create_api_key, delete_api_key, get_api_keys
+from langflow.services.database.models.api_key.model import ApiKeyCreate, UnmaskedApiKeyRead
 from langflow.services.database.models.user.model import User
-from langflow.services.deps import (
-    get_session,
-    get_settings_service,
-)
+from langflow.services.deps import get_session, get_settings_service
 
 if TYPE_CHECKING:
     pass

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

@@ -1,18 +1,13 @@
 import time
 import uuid
+from functools import partial
 from typing import TYPE_CHECKING, Annotated, Optional
 
 from fastapi import APIRouter, BackgroundTasks, Body, Depends, HTTPException
 from fastapi.responses import StreamingResponse
 from loguru import logger
 
-from langflow.api.utils import (
-    build_and_cache_graph,
-    format_elapsed_time,
-    format_exception_message,
-    get_next_runnable_vertices,
-    get_top_level_vertices,
-)
+from langflow.api.utils import build_and_cache_graph, format_elapsed_time, format_exception_message
 from langflow.api.v1.schemas import (
     InputValueRequest,
     ResultDataResponse,
@@ -56,7 +51,22 @@ async def get_vertices(
     chat_service: "ChatService" = Depends(get_chat_service),
     session=Depends(get_session),
 ):
-    """Check the flow_id is in the flow_data_store."""
+    """
+    Retrieve the vertices order for a given flow.
+
+    Args:
+        flow_id (str): The ID of the flow.
+        stop_component_id (str, optional): The ID of the stop component. Defaults to None.
+        start_component_id (str, optional): The ID of the start component. Defaults to None.
+        chat_service (ChatService, optional): The chat service dependency. Defaults to Depends(get_chat_service).
+        session (Session, optional): The session dependency. Defaults to Depends(get_session).
+
+    Returns:
+        VerticesOrderResponse: The response containing the ordered vertex IDs and the run ID.
+
+    Raises:
+        HTTPException: If there is an error checking the build status.
+    """
     try:
         # First, we need to check if the flow_id is in the cache
         graph = None
@@ -65,19 +75,25 @@ async def get_vertices(
         graph = await build_and_cache_graph(flow_id, session, chat_service, graph)
         if stop_component_id or start_component_id:
             try:
-                vertices = graph.sort_vertices(stop_component_id, start_component_id)
+                first_layer = graph.sort_vertices(stop_component_id, start_component_id)
             except Exception as exc:
                 logger.error(exc)
-                vertices = graph.sort_vertices()
+                first_layer = graph.sort_vertices()
         else:
-            vertices = graph.sort_vertices()
+            first_layer = graph.sort_vertices()
+        # When we send vertices to the frontend
+        # we need to remove them from the predecessors
+        # so they are not considered for building again
+        # which duplicates the results
+        for vertex_id in first_layer:
+            graph.remove_from_predecessors(vertex_id)
 
         # Now vertices is a list of lists
         # We need to get the id of each vertex
         # and return the same structure but only with the ids
         run_id = uuid.uuid4()
         graph.set_run_id(run_id)
-        return VerticesOrderResponse(ids=vertices, run_id=run_id)
+        return VerticesOrderResponse(ids=first_layer, run_id=run_id, vertices_to_run=list(graph.vertices_to_run))
 
     except Exception as exc:
         logger.error(f"Error checking build status: {exc}")
@@ -94,7 +110,23 @@ async def build_vertex(
     chat_service: "ChatService" = Depends(get_chat_service),
     current_user=Depends(get_current_active_user),
 ):
-    """Build a vertex instead of the entire graph."""
+    """Build a vertex instead of the entire graph.
+
+    Args:
+        flow_id (str): The ID of the flow.
+        vertex_id (str): The ID of the vertex to build.
+        background_tasks (BackgroundTasks): The background tasks object for logging.
+        inputs (Optional[InputValueRequest], optional): The input values for the vertex. Defaults to None.
+        chat_service (ChatService, optional): The chat service dependency. Defaults to Depends(get_chat_service).
+        current_user (Any, optional): The current user dependency. Defaults to Depends(get_current_active_user).
+
+    Returns:
+        VertexBuildResponse: The response containing the built vertex information.
+
+    Raises:
+        HTTPException: If there is an error building the vertex.
+
+    """
 
     start_time = time.perf_counter()
     next_runnable_vertices = []
@@ -110,23 +142,25 @@ async def build_vertex(
             graph = cache.get("result")
         result_data_response = ResultDataResponse(results={})
         duration = ""
-
         vertex = graph.get_vertex(vertex_id)
         try:
-            if not vertex.frozen or not vertex._built:
-                inputs_dict = inputs.model_dump() if inputs else {}
-                await vertex.build(user_id=current_user.id, inputs=inputs_dict)
-
-            if vertex.result is not None:
-                params = vertex._built_object_repr()
-                valid = True
-                result_dict = vertex.result
-                artifacts = vertex.artifacts
-            else:
-                raise ValueError(f"No result found for vertex {vertex_id}")
-
-            next_runnable_vertices = await get_next_runnable_vertices(graph, vertex, vertex_id, chat_service, flow_id)
-            top_level_vertices = get_top_level_vertices(graph, next_runnable_vertices)
+            lock = chat_service._cache_locks[flow_id]
+            set_cache_coro = partial(chat_service.set_cache, flow_id=flow_id)
+            (
+                next_runnable_vertices,
+                top_level_vertices,
+                result_dict,
+                params,
+                valid,
+                artifacts,
+                vertex,
+            ) = await graph.build_vertex(
+                lock=lock,
+                set_cache_coro=set_cache_coro,
+                vertex_id=vertex_id,
+                user_id=current_user.id,
+                inputs_dict=inputs.model_dump() if inputs else {},
+            )
             result_data_response = ResultDataResponse(**result_dict.model_dump())
 
         except Exception as exc:
@@ -185,9 +219,6 @@ async def build_vertex(
         raise HTTPException(status_code=500, detail=str(exc)) from exc
 
 
-# Now onto an endpoint that is an SSE endpoint
-# it will receive a component_id and a flow_id
-#
 @router.get("/build/{flow_id}/{vertex_id}/stream", response_class=StreamingResponse)
 async def build_vertex_stream(
     flow_id: str,
@@ -196,7 +227,31 @@ async def build_vertex_stream(
     chat_service: "ChatService" = Depends(get_chat_service),
     session_service: "SessionService" = Depends(get_session_service),
 ):
-    """Build a vertex instead of the entire graph."""
+    """Build a vertex instead of the entire graph.
+
+    This function is responsible for building a single vertex instead of the entire graph.
+    It takes the `flow_id` and `vertex_id` as required parameters, and an optional `session_id`.
+    It also depends on the `ChatService` and `SessionService` services.
+
+    If `session_id` is not provided, it retrieves the graph from the cache using the `chat_service`.
+    If `session_id` is provided, it loads the session data using the `session_service`.
+
+    Once the graph is obtained, it retrieves the specified vertex using the `vertex_id`.
+    If the vertex does not support streaming, an error is raised.
+    If the vertex has a built result, it sends the result as a chunk.
+    If the vertex is not frozen or not built, it streams the vertex data.
+    If the vertex has a result, it sends the result as a chunk.
+    If none of the above conditions are met, an error is raised.
+
+    If any exception occurs during the process, an error message is sent.
+    Finally, the stream is closed.
+
+    Returns:
+        A `StreamingResponse` object with the streamed vertex data in text/event-stream format.
+
+    Raises:
+        HTTPException: If an error occurs while building the vertex.
+    """
     try:
 
         async def stream_vertex():

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

@@ -17,6 +17,7 @@ from langflow.api.v1.schemas import (
     UpdateCustomComponentRequest,
     UploadFileResponse,
 )
+from langflow.graph.graph.base import Graph
 from langflow.graph.schema import RunOutputs
 from langflow.interface.custom.custom_component import CustomComponent
 from langflow.interface.custom.directory_reader import DirectoryReader
@@ -53,7 +54,7 @@ def get_all(
 async def run_flow_with_caching(
     session: Annotated[Session, Depends(get_session)],
     flow_id: str,
-    inputs: Optional[List[InputValueRequest]] = [],
+    inputs: Optional[List[InputValueRequest]] = [InputValueRequest(components=[], input_value="")],
     outputs: Optional[List[str]] = [],
     tweaks: Annotated[Optional[Tweaks], Body(embed=True)] = None,  # noqa: F821
     stream: Annotated[bool, Body(embed=True)] = False,  # noqa: F821
@@ -102,23 +103,13 @@ async def run_flow_with_caching(
         if outputs is None:
             outputs = []
 
+        task_result: List[RunOutputs] = []
+        artifacts = {}
         if session_id:
             session_data = await session_service.load_session(session_id, flow_id=flow_id)
             graph, artifacts = session_data if session_data else (None, None)
-            task_result: List[RunOutputs] = []
-            if not graph:
-                raise ValueError("Graph not found in the session")
-            task_result, session_id = await run_graph(
-                graph=graph,
-                flow_id=flow_id,
-                session_id=session_id,
-                inputs=inputs,
-                outputs=outputs,
-                artifacts=artifacts,
-                session_service=session_service,
-                stream=stream,
-            )
-
+            if graph is None:
+                raise ValueError(f"Session {session_id} not found")
         else:
             # Get the flow that matches the flow_id and belongs to the user
             # flow = session.query(Flow).filter(Flow.id == flow_id).filter(Flow.user_id == api_key_user.id).first()
@@ -130,28 +121,38 @@ async def run_flow_with_caching(
                 raise ValueError(f"Flow {flow_id} has no data")
             graph_data = flow.data
             graph_data = process_tweaks(graph_data, tweaks or {})
-            task_result, session_id = await run_graph(
-                graph=graph_data,
-                flow_id=flow_id,
-                session_id=session_id,
-                inputs=inputs,
-                outputs=outputs,
-                artifacts={},
-                session_service=session_service,
-                stream=stream,
-            )
+            graph = Graph.from_payload(graph_data, flow_id=flow_id)
+        task_result, session_id = await run_graph(
+            graph=graph,
+            flow_id=flow_id,
+            session_id=session_id,
+            inputs=inputs,
+            outputs=outputs,
+            artifacts=artifacts,
+            session_service=session_service,
+            stream=stream,
+        )
 
         return RunResponse(outputs=task_result, session_id=session_id)
     except sa.exc.StatementError as exc:
         # StatementError('(builtins.ValueError) badly formed hexadecimal UUID string')
         if "badly formed hexadecimal UUID string" in str(exc):
+            logger.error(f"Flow ID {flow_id} is not a valid UUID")
             # This means the Flow ID is not a valid UUID which means it can't find the flow
             raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc
     except ValueError as exc:
         if f"Flow {flow_id} not found" in str(exc):
+            logger.error(f"Flow {flow_id} not found")
+            raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc
+        elif f"Session {session_id} not found" in str(exc):
+            logger.error(f"Session {session_id} not found")
             raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc
         else:
+            logger.exception(exc)
             raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(exc)) from exc
+    except Exception as exc:
+        logger.exception(exc)
+        raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(exc)) from exc
 
 
 @router.post(
@@ -238,7 +239,7 @@ async def create_upload_file(
 # get endpoint to return version of langflow
 @router.get("/version")
 def get_version():
-    from langflow import __version__
+    from langflow.version import __version__
 
     return {"version": __version__}
 
@@ -309,4 +310,5 @@ async def custom_component_update(
 
         return component_node
     except Exception as exc:
+        logger.exception(exc)
         raise HTTPException(status_code=400, detail=str(exc)) from exc

src/backend/base/langflow/api/v1/files.py

@@ -4,6 +4,8 @@ from io import BytesIO
 
 from fastapi import APIRouter, Depends, HTTPException, UploadFile
 from fastapi.responses import StreamingResponse
+
+
 from langflow.api.v1.schemas import UploadFileResponse
 from langflow.services.auth.utils import get_current_active_user
 from langflow.services.database.models.flow import Flow

src/backend/base/langflow/api/v1/flows.py

@@ -12,12 +12,7 @@ from langflow.api.utils import remove_api_keys, validate_is_component
 from langflow.api.v1.schemas import FlowListCreate, FlowListRead
 from langflow.initial_setup.setup import STARTER_FOLDER_NAME
 from langflow.services.auth.utils import get_current_active_user
-from langflow.services.database.models.flow import (
-    Flow,
-    FlowCreate,
-    FlowRead,
-    FlowUpdate,
-)
+from langflow.services.database.models.flow import Flow, FlowCreate, FlowRead, FlowUpdate
 from langflow.services.database.models.user.model import User
 from langflow.services.deps import get_session, get_settings_service
 from langflow.services.settings.service import SettingsService

src/backend/base/langflow/api/v1/login.py

@@ -10,6 +10,7 @@ from langflow.services.auth.utils import (
     create_user_tokens,
 )
 from langflow.services.deps import get_session, get_settings_service
+from langflow.services.settings.manager import SettingsService
 
 router = APIRouter(tags=["Login"])
 
@@ -41,7 +42,7 @@ async def login_to_get_access_token(
             httponly=auth_settings.REFRESH_HTTPONLY,
             samesite=auth_settings.REFRESH_SAME_SITE,
             secure=auth_settings.REFRESH_SECURE,
-            expires=auth_settings.REFRESH_TOKEN_EXPIRE_MINUTES * 60,
+            expires=auth_settings.REFRESH_TOKEN_EXPIRE_SECONDS,
         )
         response.set_cookie(
             "access_token_lf",
@@ -49,7 +50,7 @@ async def login_to_get_access_token(
             httponly=auth_settings.ACCESS_HTTPONLY,
             samesite=auth_settings.ACCESS_SAME_SITE,
             secure=auth_settings.ACCESS_SECURE,
-            expires=auth_settings.ACCESS_TOKEN_EXPIRE_MINUTES * 60,
+            expires=auth_settings.ACCESS_TOKEN_EXPIRE_SECONDS,
         )
         return tokens
     else:
@@ -75,7 +76,7 @@ async def auto_login(
             httponly=auth_settings.ACCESS_HTTPONLY,
             samesite=auth_settings.ACCESS_SAME_SITE,
             secure=auth_settings.ACCESS_SECURE,
-            expires=auth_settings.ACCESS_TOKEN_EXPIRE_MINUTES * 60,
+            expires=None,  # Set to None to make it a session cookie
         )
         return tokens
 
@@ -89,7 +90,9 @@ async def auto_login(
 
 
 @router.post("/refresh")
-async def refresh_token(request: Request, response: Response, settings_service=Depends(get_settings_service)):
+async def refresh_token(
+    request: Request, response: Response, settings_service: "SettingsService" = Depends(get_settings_service)
+):
     auth_settings = settings_service.auth_settings
 
     token = request.cookies.get("refresh_token_lf")
@@ -102,7 +105,7 @@ async def refresh_token(request: Request, response: Response, settings_service=D
             httponly=auth_settings.REFRESH_HTTPONLY,
             samesite=auth_settings.REFRESH_SAME_SITE,
             secure=auth_settings.REFRESH_SECURE,
-            expires=auth_settings.REFRESH_TOKEN_EXPIRE_MINUTES * 60,
+            expires=auth_settings.REFRESH_TOKEN_EXPIRE_SECONDS,
         )
         response.set_cookie(
             "access_token_lf",
@@ -110,7 +113,7 @@ async def refresh_token(request: Request, response: Response, settings_service=D
             httponly=auth_settings.ACCESS_HTTPONLY,
             samesite=auth_settings.ACCESS_SAME_SITE,
             secure=auth_settings.ACCESS_SECURE,
-            expires=auth_settings.ACCESS_TOKEN_EXPIRE_MINUTES * 60,
+            expires=auth_settings.ACCESS_TOKEN_EXPIRE_SECONDS,
         )
         return tokens
     else:

src/backend/base/langflow/api/v1/monitor.py

@@ -1,6 +1,8 @@
 from typing import Optional
 
 from fastapi import APIRouter, Depends, HTTPException, Query
+
+
 from langflow.services.deps import get_monitor_service
 from langflow.services.monitor.schema import VertexBuildMapModel
 from langflow.services.monitor.service import MonitorService

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

@@ -1,17 +1,10 @@
 from datetime import datetime
 from enum import Enum
 from pathlib import Path
-from typing import Any, Dict, List, Optional, Union
+from typing import Any, Dict, List, Literal, Optional, Union
 from uuid import UUID
 
-from pydantic import (
-    BaseModel,
-    ConfigDict,
-    Field,
-    RootModel,
-    field_validator,
-    model_serializer,
-)
+from pydantic import BaseModel, ConfigDict, Field, RootModel, field_validator, model_serializer
 
 from langflow.graph.schema import RunOutputs
 from langflow.schema import dotdict
@@ -61,18 +54,19 @@ class RunResponse(BaseModel):
     outputs: Optional[List[RunOutputs]] = []
     session_id: Optional[str] = None
 
-    @model_serializer(mode="wrap")
-    def serialize(self, handler):
+    @model_serializer(mode="plain")
+    def serialize(self):
         # Serialize all the outputs if they are base models
+        serialized = {"session_id": self.session_id, "outputs": []}
         if self.outputs:
             serialized_outputs = []
             for output in self.outputs:
-                if isinstance(output, BaseModel):
+                if isinstance(output, BaseModel) and not isinstance(output, RunOutputs):
                     serialized_outputs.append(output.model_dump(exclude_none=True))
                 else:
                     serialized_outputs.append(output)
-            self.outputs = serialized_outputs
-        return handler(self)
+            serialized["outputs"] = serialized_outputs
+        return serialized
 
 
 class PreloadResponse(BaseModel):
@@ -234,6 +228,7 @@ class ApiKeyCreateRequest(BaseModel):
 class VerticesOrderResponse(BaseModel):
     ids: List[str]
     run_id: UUID
+    vertices_to_run: List[str]
 
 
 class ResultDataResponse(BaseModel):
@@ -264,10 +259,14 @@ class VerticesBuiltResponse(BaseModel):
 class InputValueRequest(BaseModel):
     components: Optional[List[str]] = []
     input_value: Optional[str] = None
+    type: Optional[Literal["chat", "text", "json", "any"]] = Field(
+        "any",
+        description="Defines on which components the input value should be applied. 'any' applies to all input components.",
+    )
 
     # add an example
-    model_config = {
-        "json_schema_extra": {
+    model_config = ConfigDict(
+        json_schema_extra={
             "examples": [
                 {
                     "components": ["components_id", "Component Name"],
@@ -275,9 +274,12 @@ class InputValueRequest(BaseModel):
                 },
                 {"components": ["Component Name"], "input_value": "input_value"},
                 {"input_value": "input_value"},
+                {"type": "chat", "input_value": "input_value"},
+                {"type": "json", "input_value": '{"key": "value"}'},
             ]
-        }
-    }
+        },
+        extra="forbid",
+    )
 
 
 class Tweaks(RootModel):

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

@@ -3,12 +3,7 @@ from collections import defaultdict
 from fastapi import APIRouter, HTTPException
 from loguru import logger
 
-from langflow.api.v1.base import (
-    Code,
-    CodeValidationResponse,
-    PromptValidationResponse,
-    ValidatePromptRequest,
-)
+from langflow.api.v1.base import Code, CodeValidationResponse, PromptValidationResponse, ValidatePromptRequest
 from langflow.base.prompts.utils import (
     add_new_variables_to_template,
     get_old_custom_fields,

src/backend/base/langflow/base/agents/agent.py

@@ -0,0 +1,69 @@
+from typing import List, Union
+
+from langchain.agents import AgentExecutor, BaseMultiActionAgent, BaseSingleActionAgent
+from langflow.field_typing import BaseMemory, Text, Tool
+from langflow.interface.custom.custom_component import CustomComponent
+
+
+class LCAgentComponent(CustomComponent):
+    def build_config(self):
+        return {
+            "lc": {
+                "display_name": "LangChain",
+                "info": "The LangChain to interact with.",
+            },
+            "handle_parsing_errors": {
+                "display_name": "Handle Parsing Errors",
+                "info": "If True, the agent will handle parsing errors. If False, the agent will raise an error.",
+                "advanced": True,
+            },
+            "output_key": {
+                "display_name": "Output Key",
+                "info": "The key to use to get the output from the agent.",
+                "advanced": True,
+            },
+            "memory": {
+                "display_name": "Memory",
+                "info": "Memory to use for the agent.",
+            },
+            "tools": {
+                "display_name": "Tools",
+                "info": "Tools the agent can use.",
+            },
+            "input_value": {
+                "display_name": "Input",
+                "info": "Input text to pass to the agent.",
+            },
+        }
+
+    async def run_agent(
+        self,
+        agent: Union[BaseSingleActionAgent, BaseMultiActionAgent, AgentExecutor],
+        inputs: str,
+        input_variables: list[str],
+        tools: List[Tool],
+        memory: BaseMemory = None,
+        handle_parsing_errors: bool = True,
+        output_key: str = "output",
+    ) -> Text:
+        if isinstance(agent, AgentExecutor):
+            runnable = agent
+        else:
+            runnable = AgentExecutor.from_agent_and_tools(
+                agent=agent, tools=tools, verbose=True, memory=memory, handle_parsing_errors=handle_parsing_errors
+            )
+        input_dict = {"input": inputs}
+        for var in input_variables:
+            if var not in ["agent_scratchpad", "input"]:
+                input_dict[var] = ""
+        result = await runnable.ainvoke(input_dict)
+        self.status = result
+        if output_key in result:
+            return result.get(output_key)
+        elif "output" not in result:
+            if output_key != "output":
+                raise ValueError(f"Output key not found in result. Tried '{output_key}' and 'output'.")
+            else:
+                raise ValueError("Output key not found in result. Tried 'output'.")
+
+        return result.get("output")

src/backend/base/langflow/base/io/chat.py

@@ -1,8 +1,8 @@
 import warnings
 from typing import Optional, Union
 
-from langflow import CustomComponent
 from langflow.field_typing import Text
+from langflow.interface.custom.custom_component import CustomComponent
 from langflow.memory import add_messages
 from langflow.schema import Record
 

src/backend/base/langflow/base/io/text.py

@@ -1,7 +1,7 @@
 from typing import Optional
 
-from langflow import CustomComponent
 from langflow.field_typing import Text
+from langflow.interface.custom.custom_component import CustomComponent
 
 
 class TextComponent(CustomComponent):

src/backend/base/langflow/base/models/model.py

@@ -0,0 +1,48 @@
+from typing import Optional
+
+from langchain_core.language_models.chat_models import BaseChatModel
+from langchain_core.language_models.llms import LLM
+from langchain_core.messages import HumanMessage, SystemMessage
+
+from langflow.interface.custom.custom_component import CustomComponent
+
+
+class LCModelComponent(CustomComponent):
+    display_name: str = "Model Name"
+    description: str = "Model Description"
+
+    def get_result(self, runnable: LLM, stream: bool, input_value: str):
+        """
+        Retrieves the result from the output of a Runnable object.
+
+        Args:
+            output (Runnable): The output object to retrieve the result from.
+            stream (bool): Indicates whether to use streaming or invocation mode.
+            input_value (str): The input value to pass to the output object.
+
+        Returns:
+            The result obtained from the output object.
+        """
+        if stream:
+            result = runnable.stream(input_value)
+        else:
+            message = runnable.invoke(input_value)
+            result = message.content if hasattr(message, "content") else message
+            self.status = result
+        return result
+
+    def get_chat_result(
+        self, runnable: BaseChatModel, stream: bool, input_value: str, system_message: Optional[str] = None
+    ):
+        messages = []
+        if system_message:
+            messages.append(SystemMessage(system_message))
+        if input_value:
+            messages.append(HumanMessage(input_value))
+        if stream:
+            result = runnable.stream(messages)
+        else:
+            message = runnable.invoke(messages)
+            result = message.content
+            self.status = result
+        return result

src/backend/base/langflow/components/agents/AgentInitializer.py

@@ -1,8 +1,9 @@
 from typing import Callable, List, Optional, Union
 
 from langchain.agents import AgentExecutor, AgentType, initialize_agent, types
-from langflow import CustomComponent
+
 from langflow.field_typing import BaseChatMemory, BaseLanguageModel, Tool
+from langflow.interface.custom.custom_component import CustomComponent
 
 
 class AgentInitializerComponent(CustomComponent):

src/backend/base/langflow/components/agents/CSVAgent.py

@@ -0,0 +1,34 @@
+from langchain_experimental.agents.agent_toolkits.csv.base import create_csv_agent
+
+from langflow.custom import CustomComponent
+from langflow.field_typing import AgentExecutor, BaseLanguageModel
+
+
+class CSVAgentComponent(CustomComponent):
+    display_name = "CSVAgent"
+    description = "Construct a CSV agent from a CSV and tools."
+    documentation = "https://python.langchain.com/docs/modules/agents/toolkits/csv"
+
+    def build_config(self):
+        return {
+            "llm": {"display_name": "LLM", "type": BaseLanguageModel},
+            "path": {"display_name": "Path", "field_type": "file", "suffixes": [".csv"], "file_types": [".csv"]},
+            "handle_parsing_errors": {"display_name": "Handle Parse Errors", "advanced": True},
+            "agent_type": {
+                "display_name": "Agent Type",
+                "options": ["zero-shot-react-description", "openai-functions", "openai-tools"],
+                "advanced": True,
+            },
+        }
+
+    def build(
+        self, llm: BaseLanguageModel, path: str, handle_parsing_errors: bool = True, agent_type: str = "openai-tools"
+    ) -> AgentExecutor:
+        # Instantiate and return the CSV agent class with the provided llm and path
+        return create_csv_agent(
+            llm=llm,
+            path=path,
+            agent_type=agent_type,
+            verbose=True,
+            agent_executor_kwargs=dict(handle_parsing_errors=handle_parsing_errors),
+        )

src/backend/base/langflow/components/agents/JsonAgent.py

@@ -1,9 +1,10 @@
-from langflow import CustomComponent
 from langchain.agents import AgentExecutor, create_json_agent
+from langchain_community.agent_toolkits.json.toolkit import JsonToolkit
+
 from langflow.field_typing import (
     BaseLanguageModel,
 )
-from langchain_community.agent_toolkits.json.toolkit import JsonToolkit
+from langflow.interface.custom.custom_component import CustomComponent
 
 
 class JsonAgentComponent(CustomComponent):

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

@@ -9,8 +9,9 @@ from langchain.prompts.chat import MessagesPlaceholder
 from langchain.schema.memory import BaseMemory
 from langchain.tools import Tool
 from langchain_community.chat_models import ChatOpenAI
-from langflow import CustomComponent
+
 from langflow.field_typing.range_spec import RangeSpec
+from langflow.interface.custom.custom_component import CustomComponent
 
 
 class ConversationalAgent(CustomComponent):

src/backend/base/langflow/components/agents/SQLAgent.py

@@ -1,10 +1,12 @@
-from langflow import CustomComponent
-from typing import Union, Callable
+from typing import Callable, Union
+
 from langchain.agents import AgentExecutor
-from langflow.field_typing import BaseLanguageModel
-from langchain_community.agent_toolkits.sql.base import create_sql_agent
 from langchain.sql_database import SQLDatabase
 from langchain_community.agent_toolkits import SQLDatabaseToolkit
+from langchain_community.agent_toolkits.sql.base import create_sql_agent
+
+from langflow.field_typing import BaseLanguageModel
+from langflow.interface.custom.custom_component import CustomComponent
 
 
 class SQLAgentComponent(CustomComponent):

src/backend/base/langflow/components/agents/VectorStoreAgent.py

@@ -1,8 +1,10 @@
-from langflow import CustomComponent
+from typing import Callable, Union
+
 from langchain.agents import AgentExecutor, create_vectorstore_agent
 from langchain.agents.agent_toolkits.vectorstore.toolkit import VectorStoreToolkit
-from typing import Union, Callable
+
 from langflow.field_typing import BaseLanguageModel
+from langflow.interface.custom.custom_component import CustomComponent
 
 
 class VectorStoreAgentComponent(CustomComponent):

src/backend/base/langflow/components/agents/VectorStoreRouterAgent.py

@@ -1,9 +1,11 @@
-from langflow import CustomComponent
-from langchain_core.language_models.base import BaseLanguageModel
-from langchain.agents.agent_toolkits.vectorstore.toolkit import VectorStoreRouterToolkit
-from langchain.agents import create_vectorstore_router_agent
 from typing import Callable
 
+from langchain.agents import create_vectorstore_router_agent
+from langchain.agents.agent_toolkits.vectorstore.toolkit import VectorStoreRouterToolkit
+from langchain_core.language_models.base import BaseLanguageModel
+
+from langflow.interface.custom.custom_component import CustomComponent
+
 
 class VectorStoreRouterAgentComponent(CustomComponent):
     display_name = "VectorStoreRouterAgent"

src/backend/base/langflow/components/agents/XMLAgent.py

@@ -0,0 +1,89 @@
+from typing import List
+
+from langchain.agents import create_xml_agent
+from langchain_core.prompts import PromptTemplate
+
+from langflow.base.agents.agent import LCAgentComponent
+from langflow.field_typing import BaseLLM, BaseMemory, Text, Tool
+
+
+class XMLAgentComponent(LCAgentComponent):
+    display_name = "XMLAgent"
+    description = "Construct an XML agent from an LLM and tools."
+
+    def build_config(self):
+        return {
+            "llm": {"display_name": "LLM"},
+            "tools": {"display_name": "Tools"},
+            "prompt": {
+                "display_name": "Prompt",
+                "multiline": True,
+                "info": "This prompt must contain 'tools' and 'agent_scratchpad' keys.",
+                "value": """You are a helpful assistant. Help the user answer any questions.
+
+            You have access to the following tools:
+
+            {tools}
+
+            In order to use a tool, you can use <tool></tool> and <tool_input></tool_input> tags. You will then get back a response in the form <observation></observation>
+            For example, if you have a tool called 'search' that could run a google search, in order to search for the weather in SF you would respond:
+
+            <tool>search</tool><tool_input>weather in SF</tool_input>
+            <observation>64 degrees</observation>
+
+            When you are done, respond with a final answer between <final_answer></final_answer>. For example:
+
+            <final_answer>The weather in SF is 64 degrees</final_answer>
+
+            Begin!
+
+            Previous Conversation:
+            {chat_history}
+
+            Question: {input}
+            {agent_scratchpad}""",
+            },
+            "tool_template": {
+                "display_name": "Tool Template",
+                "info": "Template for rendering tools in the prompt. Tools have 'name' and 'description' keys.",
+                "advanced": True,
+            },
+            "handle_parsing_errors": {
+                "display_name": "Handle Parsing Errors",
+                "info": "If True, the agent will handle parsing errors. If False, the agent will raise an error.",
+                "advanced": True,
+            },
+            "memory": {
+                "display_name": "Memory",
+                "info": "Memory to use for the agent.",
+            },
+            "input_value": {
+                "display_name": "Inputs",
+                "info": "Input text to pass to the agent.",
+            },
+        }
+
+    async def build(
+        self,
+        input_value: str,
+        llm: BaseLLM,
+        tools: List[Tool],
+        prompt: str,
+        memory: BaseMemory = None,
+        tool_template: str = "{name}: {description}",
+        handle_parsing_errors: bool = True,
+    ) -> Text:
+        if "input" not in prompt:
+            raise ValueError("Prompt must contain 'input' key.")
+
+        def render_tool_description(tools):
+            return "\n".join(
+                [tool_template.format(name=tool.name, description=tool.description, args=tool.args) for tool in tools]
+            )
+
+        prompt_template = PromptTemplate.from_template(prompt)
+        input_variables = prompt_template.input_variables
+        agent = create_xml_agent(llm, tools, prompt_template, tools_renderer=render_tool_description)
+        result = await self.run_agent(agent, input_value, input_variables, tools, memory, handle_parsing_errors)
+        self.status = result
+        return result

src/backend/base/langflow/components/chains/ConversationChain.py

@@ -2,8 +2,8 @@ from typing import Optional
 
 from langchain.chains import ConversationChain
 
-from langflow import CustomComponent
 from langflow.field_typing import BaseLanguageModel, BaseMemory, Text
+from langflow.interface.custom.custom_component import CustomComponent
 
 
 class ConversationChainComponent(CustomComponent):
@@ -18,7 +18,10 @@ class ConversationChainComponent(CustomComponent):
                 "display_name": "Memory",
                 "info": "Memory to load context from. If none is provided, a ConversationBufferMemory will be used.",
             },
-            "code": {"show": False},
+            "input_value": {
+                "display_name": "Input Value",
+                "info": "The input value to pass to the chain.",
+            },
         }
 
     def build(

src/backend/base/langflow/components/chains/LLMChain.py

@@ -2,13 +2,8 @@ from typing import Optional
 
 from langchain.chains import LLMChain
 
-from langflow import CustomComponent
-from langflow.field_typing import (
-    BaseLanguageModel,
-    BaseMemory,
-    BasePromptTemplate,
-    Text,
-)
+from langflow.field_typing import BaseLanguageModel, BaseMemory, BasePromptTemplate, Text
+from langflow.interface.custom.custom_component import CustomComponent
 
 
 class LLMChainComponent(CustomComponent):
@@ -20,7 +15,6 @@ class LLMChainComponent(CustomComponent):
             "prompt": {"display_name": "Prompt"},
             "llm": {"display_name": "LLM"},
             "memory": {"display_name": "Memory"},
-            "code": {"show": False},
         }
 
     def build(

src/backend/base/langflow/components/chains/LLMCheckerChain.py

@@ -1,7 +1,7 @@
 from langchain.chains import LLMCheckerChain
 
-from langflow import CustomComponent
 from langflow.field_typing import BaseLanguageModel, Text
+from langflow.interface.custom.custom_component import CustomComponent
 
 
 class LLMCheckerChainComponent(CustomComponent):
@@ -12,6 +12,10 @@ class LLMCheckerChainComponent(CustomComponent):
     def build_config(self):
         return {
             "llm": {"display_name": "LLM"},
+            "input_value": {
+                "display_name": "Input Value",
+                "info": "The input value to pass to the chain.",
+            },
         }
 
     def build(

src/backend/base/langflow/components/chains/LLMMathChain.py

@@ -2,8 +2,8 @@ from typing import Optional
 
 from langchain.chains import LLMChain, LLMMathChain
 
-from langflow import CustomComponent
 from langflow.field_typing import BaseLanguageModel, BaseMemory, Text
+from langflow.interface.custom.custom_component import CustomComponent
 
 
 class LLMMathChainComponent(CustomComponent):
@@ -18,6 +18,10 @@ class LLMMathChainComponent(CustomComponent):
             "memory": {"display_name": "Memory"},
             "input_key": {"display_name": "Input Key"},
             "output_key": {"display_name": "Output Key"},
+            "input_value": {
+                "display_name": "Input Value",
+                "info": "The input value to pass to the chain.",
+            },
         }
 
     def build(

src/backend/base/langflow/components/chains/RetrievalQA.py

@@ -4,8 +4,8 @@ from langchain.chains.combine_documents.base import BaseCombineDocumentsChain
 from langchain.chains.retrieval_qa.base import RetrievalQA
 from langchain_core.documents import Document
 
-from langflow import CustomComponent
 from langflow.field_typing import BaseMemory, BaseRetriever, Text
+from langflow.interface.custom.custom_component import CustomComponent
 
 
 class RetrievalQAComponent(CustomComponent):

src/backend/base/langflow/components/chains/RetrievalQAWithSourcesChain.py

@@ -3,8 +3,8 @@ from typing import Optional
 from langchain.chains import RetrievalQAWithSourcesChain
 from langchain_core.documents import Document
 
-from langflow import CustomComponent
 from langflow.field_typing import BaseLanguageModel, BaseMemory, BaseRetriever, Text
+from langflow.interface.custom.custom_component import CustomComponent
 
 
 class RetrievalQAWithSourcesChainComponent(CustomComponent):
@@ -22,6 +22,10 @@ class RetrievalQAWithSourcesChainComponent(CustomComponent):
             "memory": {"display_name": "Memory"},
             "return_source_documents": {"display_name": "Return Source Documents"},
             "retriever": {"display_name": "Retriever"},
+            "input_value": {
+                "display_name": "Input Value",
+                "info": "The input value to pass to the chain.",
+            },
         }
 
     def build(

src/backend/base/langflow/components/chains/SQLGenerator.py

@@ -5,8 +5,8 @@ from langchain_community.utilities.sql_database import SQLDatabase
 from langchain_core.prompts import PromptTemplate
 from langchain_core.runnables import Runnable
 
-from langflow import CustomComponent
 from langflow.field_typing import BaseLanguageModel, Text
+from langflow.interface.custom.custom_component import CustomComponent
 
 
 class SQLGeneratorComponent(CustomComponent):
@@ -25,6 +25,10 @@ class SQLGeneratorComponent(CustomComponent):
                 "display_name": "Top K",
                 "info": "The number of results per select statement to return. If 0, no limit.",
             },
+            "input_value": {
+                "display_name": "Input Value",
+                "info": "The input value to pass to the chain.",
+            },
         }
 
     def build(

src/backend/base/langflow/components/data/APIRequest.py

@@ -4,7 +4,7 @@ from typing import List, Optional
 
 import httpx
 
-from langflow import CustomComponent
+from langflow.interface.custom.custom_component import CustomComponent
 from langflow.schema import Record
 
 

src/backend/base/langflow/components/data/Directory.py

@@ -1,11 +1,7 @@
 from typing import Any, Dict, List, Optional
 
-from langflow import CustomComponent
-from langflow.base.data.utils import (
-    parallel_load_records,
-    parse_text_file_to_record,
-    retrieve_file_paths,
-)
+from langflow.base.data.utils import parallel_load_records, parse_text_file_to_record, retrieve_file_paths
+from langflow.interface.custom.custom_component import CustomComponent
 from langflow.schema import Record
 
 

src/backend/base/langflow/components/data/File.py

@@ -1,7 +1,7 @@
 from typing import Any, Dict, List
 
-from langflow import CustomComponent
 from langflow.base.data.utils import TEXT_FILE_TYPES, parse_text_file_to_record
+from langflow.interface.custom.custom_component import CustomComponent
 from langflow.schema import Record
 
 

src/backend/base/langflow/components/data/FileLoader.py

@@ -1,6 +1,6 @@
 from typing import List
 
-from langflow import CustomComponent
+from langflow.interface.custom.custom_component import CustomComponent
 from langflow.schema import Record
 from langflow.utils.constants import LOADERS_INFO
 

src/backend/base/langflow/components/data/URL.py

@@ -2,7 +2,7 @@ from typing import Any, Dict
 
 from langchain_community.document_loaders.web_base import WebBaseLoader
 
-from langflow import CustomComponent
+from langflow.interface.custom.custom_component import CustomComponent
 from langflow.schema import Record
 
 

src/backend/base/langflow/components/embeddings/AmazonBedrockEmbeddings.py

@@ -3,8 +3,7 @@ from typing import Optional
 from langchain.embeddings.base import Embeddings
 from langchain_community.embeddings import BedrockEmbeddings
 
-
-from langflow import CustomComponent
+from langflow.interface.custom.custom_component import CustomComponent
 
 
 class AmazonBedrockEmeddingsComponent(CustomComponent):

src/backend/base/langflow/components/embeddings/AzureOpenAIEmbeddings.py

@@ -1,7 +1,7 @@
 from langchain.embeddings.base import Embeddings
 from langchain_community.embeddings import AzureOpenAIEmbeddings
 
-from langflow import CustomComponent
+from langflow.interface.custom.custom_component import CustomComponent
 
 
 class AzureOpenAIEmbeddingsComponent(CustomComponent):

src/backend/base/langflow/components/embeddings/CohereEmbeddings.py

@@ -1,7 +1,8 @@
 from typing import Optional
 
 from langchain_community.embeddings.cohere import CohereEmbeddings
-from langflow import CustomComponent
+
+from langflow.interface.custom.custom_component import CustomComponent
 
 
 class CohereEmbeddingsComponent(CustomComponent):

src/backend/base/langflow/components/embeddings/HuggingFaceEmbeddings.py

@@ -1,7 +1,9 @@
-from langflow import CustomComponent
-from typing import Optional, Dict
+from typing import Dict, Optional
+
 from langchain_community.embeddings.huggingface import HuggingFaceEmbeddings
 
+from langflow.interface.custom.custom_component import CustomComponent
+
 
 class HuggingFaceEmbeddingsComponent(CustomComponent):
     display_name = "HuggingFaceEmbeddings"

src/backend/base/langflow/components/embeddings/HuggingFaceInferenceAPIEmbeddings.py

@@ -1,9 +1,10 @@
 from typing import Dict, Optional
 
 from langchain_community.embeddings.huggingface import HuggingFaceInferenceAPIEmbeddings
-from langflow import CustomComponent
 from pydantic.v1.types import SecretStr
 
+from langflow.interface.custom.custom_component import CustomComponent
+
 
 class HuggingFaceInferenceAPIEmbeddingsComponent(CustomComponent):
     display_name = "HuggingFaceInferenceAPIEmbeddings"

src/backend/base/langflow/components/embeddings/OllamaEmbeddings.py

@@ -1,9 +1,10 @@
 from typing import Optional
 
-from langflow import CustomComponent
 from langchain.embeddings.base import Embeddings
 from langchain_community.embeddings import OllamaEmbeddings
 
+from langflow.interface.custom.custom_component import CustomComponent
+
 
 class OllamaEmbeddingsComponent(CustomComponent):
     """

src/backend/base/langflow/components/embeddings/OpenAIEmbeddings.py

@@ -1,10 +1,11 @@
 from typing import Any, Callable, Dict, List, Optional, Union
 
 from langchain_openai.embeddings.base import OpenAIEmbeddings
-from langflow import CustomComponent
-from langflow.field_typing import NestedDict
 from pydantic.v1.types import SecretStr
 
+from langflow.field_typing import NestedDict
+from langflow.interface.custom.custom_component import CustomComponent
+
 
 class OpenAIEmbeddingsComponent(CustomComponent):
     display_name = "OpenAIEmbeddings"

src/backend/base/langflow/components/embeddings/VertexAIEmbeddings.py

@@ -1,6 +1,8 @@
-from langflow import CustomComponent
+from typing import List, Optional
+
 from langchain_community.embeddings import VertexAIEmbeddings
-from typing import Optional, List
+
+from langflow.interface.custom.custom_component import CustomComponent
 
 
 class VertexAIEmbeddingsComponent(CustomComponent):

src/backend/base/langflow/components/experimental/ClearMessageHistory.py

@@ -1,4 +1,4 @@
-from langflow import CustomComponent
+from langflow.interface.custom.custom_component import CustomComponent
 from langflow.memory import delete_messages, get_messages
 
 

src/backend/base/langflow/components/experimental/ExtractDataFromRecord.py

@@ -1,4 +1,4 @@
-from langflow import CustomComponent
+from langflow.interface.custom.custom_component import CustomComponent
 from langflow.schema import Record
 
 

src/backend/base/langflow/components/experimental/FlowTool.py

@@ -0,0 +1,85 @@
+from typing import Any, List, Optional, Text
+
+from langchain_core.tools import StructuredTool
+from loguru import logger
+
+from langflow.custom import CustomComponent
+from langflow.field_typing import Tool
+from langflow.graph.graph.base import Graph
+from langflow.helpers.flow import build_function_and_schema
+from langflow.schema.dotdict import dotdict
+
+
+class FlowToolComponent(CustomComponent):
+    display_name = "Flow as Tool"
+    description = "Construct a Tool from a function that runs the loaded Flow."
+    field_order = ["flow_name", "name", "description", "return_direct"]
+
+    def get_flow_names(self) -> List[str]:
+        flow_records = self.list_flows()
+        return [flow_record.data["name"] for flow_record in flow_records]
+
+    def get_flow(self, flow_name: str) -> Optional[Text]:
+        """
+        Retrieves a flow by its name.
+
+        Args:
+            flow_name (str): The name of the flow to retrieve.
+
+        Returns:
+            Optional[Text]: The flow record if found, None otherwise.
+        """
+        flow_records = self.list_flows()
+        for flow_record in flow_records:
+            if flow_record.data["name"] == flow_name:
+                return flow_record
+        return None
+
+    def update_build_config(self, build_config: dotdict, field_value: Any, field_name: str | None = None):
+        logger.debug(f"Updating build config with field value {field_value} and field name {field_name}")
+        if field_name == "flow_name":
+            build_config["flow_name"]["options"] = self.get_flow_names()
+
+        return build_config
+
+    def build_config(self):
+        return {
+            "flow_name": {
+                "display_name": "Flow Name",
+                "info": "The name of the flow to run.",
+                "options": [],
+                "real_time_refresh": True,
+                "refresh_button": True,
+            },
+            "name": {
+                "display_name": "Name",
+                "description": "The name of the tool.",
+            },
+            "description": {
+                "display_name": "Description",
+                "description": "The description of the tool.",
+            },
+            "return_direct": {
+                "display_name": "Return Direct",
+                "description": "Return the result directly from the Tool.",
+                "advanced": True,
+            },
+        }
+
+    async def build(self, flow_name: str, name: str, description: str, return_direct: bool = False) -> Tool:
+        flow_record = self.get_flow(flow_name)
+        if not flow_record:
+            raise ValueError("Flow not found.")
+        graph = Graph.from_payload(flow_record.data["data"])
+        dynamic_flow_function, schema = build_function_and_schema(flow_record, graph)
+        tool = StructuredTool.from_function(
+            coroutine=dynamic_flow_function,
+            name=name,
+            description=description,
+            return_direct=return_direct,
+            args_schema=schema,
+        )
+        description_repr = repr(tool.description).strip("'")
+        args_str = "\n".join([f"- {arg_name}: {arg_data['description']}" for arg_name, arg_data in tool.args.items()])
+        self.status = f"{description_repr}\nArguments:\n{args_str}"
+        return tool

src/backend/base/langflow/components/experimental/ListFlows.py

@@ -1,6 +1,6 @@
 from typing import List
 
-from langflow import CustomComponent
+from langflow.interface.custom.custom_component import CustomComponent
 from langflow.schema import Record
 
 

src/backend/base/langflow/components/experimental/Listen.py

@@ -1,10 +1,10 @@
-from langflow import CustomComponent
+from langflow.interface.custom.custom_component import CustomComponent
 from langflow.schema import Record
 
 
-class GetNotifiedComponent(CustomComponent):
-    display_name = "Get Notified"
-    description = "A component to get notified by Notify component."
+class ListenComponent(CustomComponent):
+    display_name = "Listen"
+    description = "A component to listen for a notification."
     beta: bool = True
 
     def build_config(self):

src/backend/base/langflow/components/experimental/MergeRecords.py

@@ -1,4 +1,4 @@
-from langflow import CustomComponent
+from langflow.interface.custom.custom_component import CustomComponent
 from langflow.schema import Record
 
 

src/backend/base/langflow/components/experimental/Notify.py

@@ -1,6 +1,6 @@
 from typing import Optional
 
-from langflow import CustomComponent
+from langflow.interface.custom.custom_component import CustomComponent
 from langflow.schema import Record
 
 

src/backend/base/langflow/components/experimental/RunFlow.py

@@ -1,9 +1,9 @@
-from typing import List, Optional
+from typing import Any, List, Optional
 
-from langflow import CustomComponent
+from langflow.custom import CustomComponent
 from langflow.field_typing import NestedDict, Text
 from langflow.graph.schema import ResultData
-from langflow.schema import Record
+from langflow.schema import Record, dotdict
 
 
 class RunFlowComponent(CustomComponent):
@@ -15,6 +15,12 @@ class RunFlowComponent(CustomComponent):
         flow_records = self.list_flows()
         return [flow_record.data["name"] for flow_record in flow_records]
 
+    def update_build_config(self, build_config: dotdict, field_value: Any, field_name: str | None = None):
+        if field_name == "flow_name":
+            build_config["flow_name"]["options"] = self.get_flow_names()
+
+        return build_config
+
     def build_config(self):
         return {
             "input_value": {
@@ -24,7 +30,8 @@ class RunFlowComponent(CustomComponent):
             "flow_name": {
                 "display_name": "Flow Name",
                 "info": "The name of the flow to run.",
-                "options": self.get_flow_names,
+                "options": [],
+                "refresh_button": True,
             },
             "tweaks": {
                 "display_name": "Tweaks",
@@ -45,7 +52,7 @@ class RunFlowComponent(CustomComponent):
 
     async def build(self, input_value: Text, flow_name: str, tweaks: NestedDict) -> List[Record]:
         results: List[Optional[ResultData]] = await self.run_flow(
-            input_value=input_value, flow_name=flow_name, tweaks=tweaks
+            inputs={"input_value": input_value}, flow_name=flow_name, tweaks=tweaks
         )
         if isinstance(results, list):
             records = []

src/backend/base/langflow/components/experimental/RunnableExecutor.py

@@ -1,7 +1,7 @@
 from langchain_core.runnables import Runnable
 
-from langflow import CustomComponent
 from langflow.field_typing import Text
+from langflow.interface.custom.custom_component import CustomComponent
 
 
 class RunnableExecComponent(CustomComponent):

src/backend/base/langflow/components/experimental/SQLExecutor.py

@@ -1,8 +1,8 @@
 from langchain_community.tools.sql_database.tool import QuerySQLDataBaseTool
 from langchain_experimental.sql.base import SQLDatabase
 
-from langflow import CustomComponent
 from langflow.field_typing import Text
+from langflow.interface.custom.custom_component import CustomComponent
 
 
 class SQLExecutorComponent(CustomComponent):

src/backend/base/langflow/components/experimental/SubFlow.py

@@ -0,0 +1,122 @@
+from typing import Any, List, Optional, Text, Tuple
+
+from loguru import logger
+
+from langflow.custom import CustomComponent
+from langflow.graph.graph.base import Graph
+from langflow.graph.schema import ResultData, RunOutputs
+from langflow.schema import Record
+from langflow.schema.dotdict import dotdict
+from langflow.template.field.base import TemplateField
+
+
+class SubFlowComponent(CustomComponent):
+    display_name = "SubFlow"
+    description = "Dynamically Generates a Component from a Flow. The output is a list of records with keys 'result' and 'message'."
+    beta: bool = True
+    field_order = ["flow_name"]
+
+    def get_flow_names(self) -> List[str]:
+        flow_records = self.list_flows()
+        return [flow_record.data["name"] for flow_record in flow_records]
+
+    def get_flow(self, flow_name: str) -> Optional[Text]:
+        flow_records = self.list_flows()
+        for flow_record in flow_records:
+            if flow_record.data["name"] == flow_name:
+                return flow_record
+        return None
+
+    def update_build_config(self, build_config: dotdict, field_value: Any, field_name: str | None = None):
+        logger.debug(f"Updating build config with field value {field_value} and field name {field_name}")
+        if field_name == "flow_name":
+            build_config["flow_name"]["options"] = self.get_flow_names()
+        # Clean up the build config
+        for key in list(build_config.keys()):
+            if key not in self.field_order + ["code", "_type"]:
+                del build_config[key]
+        if field_value is not None and field_name == "flow_name":
+            try:
+                flow_record = self.get_flow(field_value)
+                if not flow_record:
+                    raise ValueError(f"Flow {field_value} not found.")
+                graph = Graph.from_payload(flow_record.data["data"])
+                # Get all inputs from the graph
+                inputs = self.get_flow_inputs(graph)
+                # Add inputs to the build config
+                build_config = self.add_inputs_to_build_config(inputs, build_config)
+            except Exception as e:
+                logger.error(f"Error getting flow {field_value}: {str(e)}")
+
+        return build_config
+
+    def get_flow_inputs(self, graph: Graph) -> List[Record]:
+        inputs = []
+        for vertex in graph.vertices:
+            if vertex.is_input:
+                inputs.append((vertex.id, vertex.display_name, vertex.description))
+        logger.debug(inputs)
+        return inputs
+
+    def add_inputs_to_build_config(self, inputs: List[Tuple], build_config: dotdict):
+        new_fields: list[TemplateField] = []
+        for input_id, input_display_name, input_description in inputs:
+            field = TemplateField(
+                display_name=input_display_name,
+                name=input_id,
+                info=input_description,
+                field_type="str",
+                default=None,
+            )
+            new_fields.append(field)
+        logger.debug(new_fields)
+        for field in new_fields:
+            build_config[field.name] = field.to_dict()
+        return build_config
+
+    def build_config(self):
+        return {
+            "input_value": {
+                "display_name": "Input Value",
+                "multiline": True,
+            },
+            "flow_name": {
+                "display_name": "Flow Name",
+                "info": "The name of the flow to run.",
+                "options": [],
+                "real_time_refresh": True,
+                "refresh_button": True,
+            },
+            "tweaks": {
+                "display_name": "Tweaks",
+                "info": "Tweaks to apply to the flow.",
+            },
+        }
+
+    def build_records_from_result_data(self, result_data: ResultData) -> List[Record]:
+        messages = result_data.messages
+        if not messages:
+            return []
+        records = []
+        for message in messages:
+            message_dict = message if isinstance(message, dict) else message.model_dump()
+            record = Record(data={"result": result_data.model_dump(), "message": message_dict.get("message", "")})
+            records.append(record)
+        return records
+
+    async def build(self, flow_name: str, **kwargs) -> List[Record]:
+        tweaks = {key: {"input_value": value} for key, value in kwargs.items()}
+        run_outputs: List[Optional[RunOutputs]] = await self.run_flow(
+            tweaks=tweaks,
+            flow_name=flow_name,
+        )
+        run_output = run_outputs[0]
+
+        records = []
+        for output in run_output.outputs:
+            if output:
+                records.extend(self.build_records_from_result_data(output))
+
+        self.status = records
+        logger.debug(records)
+        return records

src/backend/base/langflow/components/experimental/__init__.py

@@ -1,6 +1,6 @@
 from .ClearMessageHistory import ClearMessageHistoryComponent
 from .ExtractDataFromRecord import ExtractKeyFromRecordComponent
-from .GetNotified import GetNotifiedComponent
+from .Listen import GetNotifiedComponent
 from .ListFlows import ListFlowsComponent
 from .MergeRecords import MergeRecordsComponent
 from .Notify import NotifyComponent