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

.github/workflows/ci.yml

@@ -4,10 +4,18 @@ on:
   push:
     branches:
       - dev
+    paths:
+      - "tests/**"
+      - "src/backend/**"
+      - "pyproject.toml"
   pull_request:
     branches:
       - dev
       - main
+    paths:
+      - "tests/**"
+      - "src/backend/**"
+      - "pyproject.toml"
 
 jobs:
   build-and-test:

.github/workflows/lint.yml

@@ -3,7 +3,16 @@ name: lint
 on:
   push:
     branches: [main]
+    paths:
+      - "tests/**"
+      - "src/backend/**"
+      - "pyproject.toml"
+
   pull_request:
+    paths:
+      - "tests/**"
+      - "src/backend/**"
+      - "pyproject.toml"
 
 env:
   POETRY_VERSION: "1.4.0"

.github/workflows/test.yml

@@ -3,9 +3,17 @@ name: test
 on:
   push:
     branches: [main]
+    paths:
+      - "tests/**"
+      - "src/backend/**"
+      - "pyproject.toml"
+
   pull_request:
     branches: [dev]
-
+    paths:
+      - "tests/**"
+      - "src/backend/**"
+      - "pyproject.toml"
 env:
   POETRY_VERSION: "1.5.0"
 

Makefile

@@ -53,8 +53,8 @@ setup_devcontainer:
 	poetry run langflow --path src/frontend/build
 
 frontend:
-	make install_frontend
-	make run_frontend
+	@-make install_frontend || (echo "An error occurred while installing frontend dependencies. Attempting to fix." && make install_frontendc)
+	@make run_frontend
 
 frontendc:
 	make install_frontendc
@@ -65,7 +65,13 @@ install_backend:
 
 backend:
 	make install_backend
-	poetry run uvicorn --factory src.backend.langflow.main:create_app --port 7860 --reload --log-level debug
+ifeq ($(login),1)
+	@echo "Running backend without autologin";
+	poetry run langflow run --backend-only --port 7860 --host 0.0.0.0 --no-open-browser
+else
+	@echo "Running backend with autologin";
+	LANGFLOW_AUTO_LOGIN=True poetry run langflow run --backend-only --port 7860 --host 0.0.0.0 --no-open-browser
+endif
 
 build_and_run:
 	echo 'Removing dist folder'

README.md

@@ -143,7 +143,7 @@ Alternatively, click the **"Open in Cloud Shell"** button below to launch Google
 
 # 🎨 Creating Flows
 
-Creating flows with Langflow is easy. Simply drag sidebar components onto the canvas and connect them together to create your pipeline. Langflow provides a range of [LangChain components](https://langchain.readthedocs.io/en/latest/reference.html) to choose from, including LLMs, prompt serializers, agents, and chains.
+Creating flows with Langflow is easy. Simply drag sidebar components onto the canvas and connect them together to create your pipeline. Langflow provides a range of [LangChain components](https://docs.langchain.com/docs/category/components) to choose from, including LLMs, prompt serializers, agents, and chains.
 
 Explore by editing prompt parameters, link chains and agents, track an agent's thought process, and export your flow.
 

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

@@ -187,14 +187,18 @@ async def stream_build(
                     valid = False
                     update_build_status(cache_service, flow_id, BuildStatus.FAILURE)
 
-                response = {
-                    "valid": valid,
-                    "params": params,
-                    "id": vertex.id,
-                    "progress": round(i / number_of_nodes, 2),
-                }
+                vertex_id = (
+                    vertex.parent_node_id if vertex.parent_is_top_level else vertex.id
+                )
+                if vertex_id in graph.top_level_nodes:
+                    response = {
+                        "valid": valid,
+                        "params": params,
+                        "id": vertex_id,
+                        "progress": round(i / number_of_nodes, 2),
+                    }
 
-                yield str(StreamData(event="message", data=response))
+                    yield str(StreamData(event="message", data=response))
 
             langchain_object = graph.build()
             # Now we  need to check the input_keys to send them to the client

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

@@ -69,7 +69,7 @@ def get_all(
     "/process/{flow_id}",
     response_model=ProcessResponse,
 )
-async def process_flow(
+async def process(
     session: Annotated[Session, Depends(get_session)],
     flow_id: str,
     inputs: Optional[dict] = None,

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

@@ -1,28 +1,79 @@
 from loguru import logger
 from typing import TYPE_CHECKING
+from pydantic import BaseModel, Field
+from typing import List, Optional
 
 if TYPE_CHECKING:
     from langflow.graph.vertex.base import Vertex
 
 
+class SourceHandle(BaseModel):
+    baseClasses: List[str] = Field(
+        ..., description="List of base classes for the source handle."
+    )
+    dataType: str = Field(..., description="Data type for the source handle.")
+    id: str = Field(..., description="Unique identifier for the source handle.")
+
+
+class TargetHandle(BaseModel):
+    fieldName: str = Field(..., description="Field name for the target handle.")
+    id: str = Field(..., description="Unique identifier for the target handle.")
+    inputTypes: Optional[List[str]] = Field(
+        None, description="List of input types for the target handle."
+    )
+    type: str = Field(..., description="Type of the target handle.")
+
+
 class Edge:
     def __init__(self, source: "Vertex", target: "Vertex", edge: dict):
         self.source: "Vertex" = source
         self.target: "Vertex" = target
-        self.source_handle = edge.get("sourceHandle", "")
-        self.target_handle = edge.get("targetHandle", "")
-        # 'BaseLoader;BaseOutputParser|documents|PromptTemplate-zmTlD'
-        # target_param is documents
-        self.target_param = self.target_handle.split("|")[1]
-
+        if data := edge.get("data", {}):
+            self._source_handle = data.get("sourceHandle", {})
+            self._target_handle = data.get("targetHandle", {})
+            self.source_handle: SourceHandle = SourceHandle(**self._source_handle)
+            self.target_handle: TargetHandle = TargetHandle(**self._target_handle)
+            self.target_param = self.target_handle.fieldName
+            # validate handles
+            self.validate_handles()
+        else:
+            # Logging here because this is a breaking change
+            logger.error("Edge data is empty")
+            self._source_handle = edge.get("sourceHandle", "")
+            self._target_handle = edge.get("targetHandle", "")
+            # 'BaseLoader;BaseOutputParser|documents|PromptTemplate-zmTlD'
+            # target_param is documents
+            self.target_param = self._target_handle.split("|")[1]
+        # Validate in __init__ to fail fast
         self.validate_edge()
 
+    def validate_handles(self) -> None:
+        if self.target_handle.inputTypes is None:
+            self.valid_handles = (
+                self.target_handle.type in self.source_handle.baseClasses
+            )
+        else:
+            self.valid_handles = (
+                any(
+                    baseClass in self.target_handle.inputTypes
+                    for baseClass in self.source_handle.baseClasses
+                )
+                or self.target_handle.type in self.source_handle.baseClasses
+            )
+        if not self.valid_handles:
+            logger.debug(self.source_handle)
+            logger.debug(self.target_handle)
+            raise ValueError(
+                f"Edge between {self.source.vertex_type} and {self.target.vertex_type} "
+                f"has invalid handles"
+            )
+
     def __setstate__(self, state):
         self.source = state["source"]
         self.target = state["target"]
         self.target_param = state["target_param"]
-        self.source_handle = state["source_handle"]
-        self.target_handle = state["target_handle"]
+        self.source_handle = state.get("source_handle")
+        self.target_handle = state.get("target_handle")
 
     def reset(self) -> None:
         self.source._build_params()

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

@@ -2,6 +2,7 @@ from typing import Dict, Generator, List, Type, Union
 
 from langflow.graph.edge.base import Edge
 from langflow.graph.graph.constants import lazy_load_vertex_dict
+from langflow.graph.graph.utils import process_flow
 from langflow.graph.vertex.base import Vertex
 from langflow.graph.vertex.types import (
     FileToolVertex,
@@ -19,11 +20,21 @@ class Graph:
 
     def __init__(
         self,
-        nodes: List[Dict[str, Union[str, Dict[str, Union[str, List[str]]]]]],
+        nodes: List[Dict],
         edges: List[Dict[str, str]],
     ) -> None:
         self._nodes = nodes
         self._edges = edges
+        self.raw_graph_data = {"nodes": nodes, "edges": edges}
+
+        self.top_level_nodes = []
+        for node in self._nodes:
+            if node_id := node.get("id"):
+                self.top_level_nodes.append(node_id)
+
+        self._graph_data = process_flow(self.raw_graph_data)
+        self._nodes = self._graph_data["nodes"]
+        self._edges = self._graph_data["edges"]
         self._build_graph()
 
     def __setstate__(self, state):
@@ -50,6 +61,7 @@ class Graph:
             edges = payload["edges"]
             return cls(nodes, edges)
         except KeyError as exc:
+            logger.exception(exc)
             raise ValueError(
                 f"Invalid payload. Expected keys 'nodes' and 'edges'. Found {list(payload.keys())}"
             ) from exc
@@ -215,7 +227,9 @@ class Graph:
             node_lc_type: str = node_data["node"]["template"]["_type"]  # type: ignore
 
             VertexClass = self._get_vertex_class(node_type, node_lc_type)
-            nodes.append(VertexClass(node))
+            vertex = VertexClass(node)
+            vertex.set_top_level(self.top_level_nodes)
+            nodes.append(vertex)
 
         return nodes
 

src/backend/langflow/graph/graph/utils.py

@@ -0,0 +1,230 @@
+from collections import deque
+import copy
+
+
+def find_last_node(nodes, edges):
+    """
+    This function receives a flow and returns the last node.
+    """
+    return next((n for n in nodes if all(e["source"] != n["id"] for e in edges)), None)
+
+
+def add_parent_node_id(nodes, parent_node_id):
+    """
+    This function receives a list of nodes and adds a parent_node_id to each node.
+    """
+    for node in nodes:
+        node["parent_node_id"] = parent_node_id
+
+
+def ungroup_node(group_node_data, base_flow):
+    template, flow = (
+        group_node_data["node"]["template"],
+        group_node_data["node"]["flow"],
+    )
+    parent_node_id = group_node_data["id"]
+    g_nodes = flow["data"]["nodes"]
+    add_parent_node_id(g_nodes, parent_node_id)
+    g_edges = flow["data"]["edges"]
+
+    # Redirect edges to the correct proxy node
+    updated_edges = get_updated_edges(
+        base_flow, g_nodes, g_edges, group_node_data["id"]
+    )
+
+    # Update template values
+    update_template(template, g_nodes)
+
+    nodes = [
+        n for n in base_flow["nodes"] if n["id"] != group_node_data["id"]
+    ] + g_nodes
+    edges = (
+        [
+            e
+            for e in base_flow["edges"]
+            if e["target"] != group_node_data["id"]
+            and e["source"] != group_node_data["id"]
+        ]
+        + g_edges
+        + updated_edges
+    )
+
+    base_flow["nodes"] = nodes
+    base_flow["edges"] = edges
+
+    return nodes
+
+
+def process_flow(flow_object):
+    cloned_flow = copy.deepcopy(flow_object)
+    processed_nodes = set()  # To keep track of processed nodes
+
+    def process_node(node):
+        node_id = node.get("id")
+
+        # If node already processed, skip
+        if node_id in processed_nodes:
+            return
+
+        if (
+            node.get("data")
+            and node["data"].get("node")
+            and node["data"]["node"].get("flow")
+        ):
+            process_flow(node["data"]["node"]["flow"]["data"])
+            new_nodes = ungroup_node(node["data"], cloned_flow)
+            # Add new nodes to the queue for future processing
+            nodes_to_process.extend(new_nodes)
+
+        # Mark node as processed
+        processed_nodes.add(node_id)
+
+    nodes_to_process = deque(cloned_flow["nodes"])
+
+    while nodes_to_process:
+        node = nodes_to_process.popleft()
+        process_node(node)
+
+    return cloned_flow
+
+
+def update_template(template, g_nodes):
+    """
+    Updates the template of a node in a graph with the given template.
+
+    Args:
+        template (dict): The new template to update the node with.
+        g_nodes (list): The list of nodes in the graph.
+
+    Returns:
+        None
+    """
+    for _, value in template.items():
+        if not value.get("proxy"):
+            continue
+        proxy_dict = value["proxy"]
+        field, id_ = proxy_dict["field"], proxy_dict["id"]
+        node_index = next((i for i, n in enumerate(g_nodes) if n["id"] == id_), -1)
+        if node_index != -1:
+            display_name = None
+            show = g_nodes[node_index]["data"]["node"]["template"][field]["show"]
+            advanced = g_nodes[node_index]["data"]["node"]["template"][field][
+                "advanced"
+            ]
+            if "display_name" in g_nodes[node_index]["data"]["node"]["template"][field]:
+                display_name = g_nodes[node_index]["data"]["node"]["template"][field][
+                    "display_name"
+                ]
+            else:
+                display_name = g_nodes[node_index]["data"]["node"]["template"][field][
+                    "name"
+                ]
+
+            g_nodes[node_index]["data"]["node"]["template"][field] = value
+            g_nodes[node_index]["data"]["node"]["template"][field]["show"] = show
+            g_nodes[node_index]["data"]["node"]["template"][field][
+                "advanced"
+            ] = advanced
+            g_nodes[node_index]["data"]["node"]["template"][field][
+                "display_name"
+            ] = display_name
+
+
+def update_target_handle(new_edge, g_nodes, group_node_id):
+    """
+    Updates the target handle of a given edge if it is a proxy node.
+
+    Args:
+        new_edge (dict): The edge to update.
+        g_nodes (list): The list of nodes in the graph.
+        group_node_id (str): The ID of the group node.
+
+    Returns:
+        dict: The updated edge.
+    """
+    target_handle = new_edge["data"]["targetHandle"]
+    if target_handle.get("proxy"):
+        proxy_id = target_handle["proxy"]["id"]
+        if node := next((n for n in g_nodes if n["id"] == proxy_id), None):
+            set_new_target_handle(proxy_id, new_edge, target_handle, node)
+    return new_edge
+
+
+def set_new_target_handle(proxy_id, new_edge, target_handle, node):
+    """
+    Sets a new target handle for a given edge.
+
+    Args:
+        proxy_id (str): The ID of the proxy.
+        new_edge (dict): The new edge to be created.
+        target_handle (dict): The target handle of the edge.
+        node (dict): The node containing the edge.
+
+    Returns:
+        None
+    """
+    new_edge["target"] = proxy_id
+    _type = target_handle.get("type")
+    if _type is None:
+        raise KeyError("The 'type' key must be present in target_handle.")
+
+    field = target_handle["proxy"]["field"]
+    new_target_handle = {
+        "fieldName": field,
+        "type": _type,
+        "id": proxy_id,
+    }
+    if node["data"]["node"].get("flow"):
+        new_target_handle["proxy"] = {
+            "field": node["data"]["node"]["template"][field]["proxy"]["field"],
+            "id": node["data"]["node"]["template"][field]["proxy"]["id"],
+        }
+    if input_types := target_handle.get("inputTypes"):
+        new_target_handle["inputTypes"] = input_types
+    new_edge["data"]["targetHandle"] = new_target_handle
+
+
+def update_source_handle(new_edge, g_nodes, g_edges):
+    """
+    Updates the source handle of a given edge to the last node in the flow data.
+
+    Args:
+        new_edge (dict): The edge to update.
+        flow_data (dict): The flow data containing the nodes and edges.
+
+    Returns:
+        dict: The updated edge with the new source handle.
+    """
+    last_node = copy.deepcopy(find_last_node(g_nodes, g_edges))
+    new_edge["source"] = last_node["id"]
+    new_source_handle = new_edge["data"]["sourceHandle"]
+    new_source_handle["id"] = last_node["id"]
+    new_edge["data"]["sourceHandle"] = new_source_handle
+    return new_edge
+
+
+def get_updated_edges(base_flow, g_nodes, g_edges, group_node_id):
+    """
+    Given a base flow, a list of graph nodes and a group node id, returns a list of updated edges.
+    An updated edge is an edge that has its target or source handle updated based on the group node id.
+
+    Args:
+        base_flow (dict): The base flow containing a list of edges.
+        g_nodes (list): A list of graph nodes.
+        group_node_id (str): The id of the group node.
+
+    Returns:
+        list: A list of updated edges.
+    """
+    updated_edges = []
+    for edge in base_flow["edges"]:
+        new_edge = copy.deepcopy(edge)
+        if new_edge["target"] == group_node_id:
+            new_edge = update_target_handle(new_edge, g_nodes, group_node_id)
+
+        if new_edge["source"] == group_node_id:
+            new_edge = update_source_handle(new_edge, g_nodes, g_edges)
+
+        if edge["target"] == group_node_id or edge["source"] == group_node_id:
+            updated_edges.append(new_edge)
+    return updated_edges

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

@@ -38,6 +38,8 @@ class Vertex:
         self.task_id: Optional[str] = None
         self.is_task = is_task
         self.params = params or {}
+        self.parent_node_id: Optional[str] = self._data.get("parent_node_id")
+        self.parent_is_top_level = False
 
     def reset_params(self):
         for edge in self.edges:
@@ -88,6 +90,11 @@ class Vertex:
             self._built = False
         self.artifacts: Dict[str, Any] = {}
         self.task_id: Optional[str] = None
+        self.parent_node_id = state["parent_node_id"]
+        self.parent_is_top_level = state["parent_is_top_level"]
+
+    def set_top_level(self, top_level_nodes: List[str]) -> None:
+        self.parent_is_top_level = self.parent_node_id in top_level_nodes
 
     def _parse_data(self) -> None:
         self.data = self._data["data"]

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

@@ -32,7 +32,11 @@ def initialize_database():
     try:
         database_service.run_migrations()
     except CommandError as exc:
-        if "Can't locate revision identified by" not in str(exc):
+        # if "overlaps with other requested revisions" or "Can't locate revision identified by"
+        # are not in the exception, we can't handle it
+        if "overlaps with other requested revisions" not in str(
+            exc
+        ) and "Can't locate revision identified by" not in str(exc):
             raise exc
         # This means there's wrong revision in the DB
         # We need to delete the alembic_version table

src/backend/langflow/template/frontend_node/prompts.py

@@ -44,6 +44,7 @@ class PromptFrontendNode(FrontendNode):
 
         # All prompt fields should be password=False
         field.password = False
+        field.dynamic = True
 
 
 class PromptTemplateNode(FrontendNode):

src/frontend/.gitignore

@@ -22,5 +22,5 @@ npm-debug.log*
 yarn-debug.log*
 yarn-error.log*
 /test-results/
-/playwright-report/
+/playwright-report/*/
 /playwright/.cache/