Add experimental components for listing flows, getting notified, and executing runnables

src/backend/langflow/components/experimental/GetNotified.py

@@ -0,0 +1,20 @@
+from langflow import CustomComponent
+from langflow.schema import Record
+
+
+class GetNotifiedComponent(CustomComponent):
+    display_name = "Get Notified"
+    description = "A component to get notified by Notify component."
+
+    def build_config(self):
+        return {
+            "name": {
+                "display_name": "Name",
+                "info": "The name of the notification to listen for.",
+            },
+        }
+
+    def build(self, name: str) -> Record:
+        state = self.get_state(name)
+        self.status = state
+        return state

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

@@ -0,0 +1,19 @@
+from typing import List
+
+from langflow import CustomComponent
+from langflow.schema import Record
+
+
+class ListFlowsComponent(CustomComponent):
+    display_name = "List Flows"
+    description = "A component to list all available flows."
+
+    def build_config(self):
+        return {}
+
+    def build(
+        self,
+    ) -> List[Record]:
+        flows = self.list_flows()
+        self.status = flows
+        return flows

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

@@ -4,38 +4,38 @@ from langflow import CustomComponent
 from langflow.schema import Record
 
 
-class SharedState(CustomComponent):
-    display_name = "Shared State"
-    description = "A component to share state between components."
+class NotifyComponent(CustomComponent):
+    display_name = "Notify"
+    description = "A component to generate a notification to Get Notified component."
 
     def build_config(self):
         return {
-            "name": {"display_name": "Name", "info": "The name of the state."},
+            "name": {"display_name": "Name", "info": "The name of the notification."},
             "record": {"display_name": "Record", "info": "The record to store."},
             "append": {
                 "display_name": "Append",
-                "info": "If True, the record will be appended to the state.",
+                "info": "If True, the record will be appended to the notification.",
             },
         }
 
     def build(
         self, name: str, record: Optional[Record] = None, append: bool = False
     ) -> Record:
+        if record and not isinstance(record, Record):
+            if isinstance(record, str):
+                record = Record(text=record)
+            elif isinstance(record, dict):
+                record = Record(data=record)
+            else:
+                record = Record(text=str(record))
+        elif not record:
+            record = Record(text="")
         if record:
             if append:
                 self.append_state(name, record)
             else:
                 self.update_state(name, record)
-
-        state = self.get_state(name)
-        if state and not isinstance(state, Record):
-            if isinstance(state, str):
-                state = Record(text=state)
-            elif isinstance(state, dict):
-                state = Record(data=state)
-            else:
-                state = Record(text=str(state))
-        elif not state:
-            state = Record(text="")
-        self.status = state
-        return state
+        else:
+            self.status = "No record provided."
+        self.status = record
+        return record

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

@@ -0,0 +1,58 @@
+from typing import List, Optional
+
+from langflow import CustomComponent
+from langflow.field_typing import NestedDict, Text
+from langflow.graph.schema import ResultData
+from langflow.schema import Record
+
+
+class RunFlowComponent(CustomComponent):
+    display_name = "Run Flow"
+    description = "A component to run a flow."
+
+    def get_flow_names(self) -> List[str]:
+        flow_records = self.list_flows()
+        return [flow_record.data["name"] for flow_record in flow_records]
+
+    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": self.get_flow_names,
+            },
+            "tweaks": {
+                "display_name": "Tweaks",
+                "info": "Tweaks to apply to the flow.",
+            },
+        }
+
+    def build_records_from_result_data(self, result_data: ResultData) -> Record:
+        messages = result_data.messages
+        records = []
+        for message in messages:
+            record = Record(text=message.get("text", ""), data={"result": result_data})
+            records.append(record)
+        return records
+
+    async def build(
+        self, input_value: Text, flow_name: str, tweaks: NestedDict
+    ) -> Record:
+
+        results: List[Optional[ResultData]] = await self.run_flow(
+            input_value=input_value, flow_name=flow_name, tweaks=tweaks
+        )
+        if isinstance(results, list):
+            records = []
+            for result in results:
+                if result:
+                    records.extend(self.build_records_from_result_data(result))
+        else:
+            records = self.build_records_from_result_data(results)
+
+        self.status = records
+        return records

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

@@ -0,0 +1,42 @@
+from langchain_core.runnables import Runnable
+
+from langflow import CustomComponent
+from langflow.field_typing import Text
+
+
+class RunnableExecComponent(CustomComponent):
+    documentation: str = "http://docs.langflow.org/components/custom"
+    display_name = "Runnable Executor"
+    beta = True
+
+    def build_config(self):
+        return {
+            "input_key": {
+                "display_name": "Input Key",
+                "info": "The key to use for the input.",
+            },
+            "input_value": {
+                "display_name": "Inputs",
+                "info": "The inputs to pass to the runnable.",
+            },
+            "runnable": {
+                "display_name": "Runnable",
+                "info": "The runnable to execute.",
+            },
+            "output_key": {
+                "display_name": "Output Key",
+                "info": "The key to use for the output.",
+            },
+        }
+
+    def build(
+        self,
+        input_key: str,
+        input_value: Text,
+        runnable: Runnable,
+        output_key: str = "output",
+    ) -> Text:
+        result = runnable.invoke({input_key: input_value})
+        result = result.get(output_key)
+        self.status = result
+        return result

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

@@ -0,0 +1,68 @@
+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
+
+
+class SQLExecutorComponent(CustomComponent):
+    display_name = "SQL Executor"
+    description = "Execute SQL query."
+
+    def build_config(self):
+        return {
+            "database_url": {
+                "display_name": "Database URL",
+                "info": "The URL of the database.",
+            },
+            "include_columns": {
+                "display_name": "Include Columns",
+                "info": "Include columns in the result.",
+            },
+            "passthrough": {
+                "display_name": "Passthrough",
+                "info": "If an error occurs, return the query instead of raising an exception.",
+            },
+            "add_error": {
+                "display_name": "Add Error",
+                "info": "Add the error to the result.",
+            },
+        }
+
+    def clean_up_uri(self, uri: str) -> str:
+        if uri.startswith("postgresql://"):
+            uri = uri.replace("postgresql://", "postgres://")
+        return uri.strip()
+
+    def build(
+        self,
+        query: str,
+        database_url: str,
+        include_columns: bool = False,
+        passthrough: bool = False,
+        add_error: bool = False,
+    ) -> Text:
+        error = None
+        try:
+            database = SQLDatabase.from_uri(database_url)
+        except Exception as e:
+            raise ValueError(f"An error occurred while connecting to the database: {e}")
+        try:
+            tool = QuerySQLDataBaseTool(db=database)
+            result = tool.run(query, include_columns=include_columns)
+            self.status = result
+        except Exception as e:
+            result = Text(e)
+            self.status = result
+            if not passthrough:
+                raise e
+            error = repr(e)
+
+        if add_error and error is not None:
+            result = f"{result}\n\nError: {error}\n\nQuery: {query}"
+        elif error is not None:
+            # Then we won't add the error to the result
+            # but since we are in passthrough mode, we will return the query
+            result = query
+
+        return result

src/backend/langflow/components/helpers/ShouldRunNext.py

@@ -1,49 +0,0 @@
-# Implement ShouldRunNext component
-from typing import Text
-from langchain_core.prompts import PromptTemplate
-
-from langflow import CustomComponent
-from langflow.field_typing import BaseLanguageModel, Prompt
-
-
-class ShouldRunNext(CustomComponent):
-    display_name = "Should Run Next"
-    description = "Decides whether to run the next component."
-
-    def build_config(self):
-        return {
-            "prompt": {
-                "display_name": "Prompt",
-                "info": "The prompt to use for the decision. It should generate a boolean response (True or False).",
-            },
-            "llm": {
-                "display_name": "LLM",
-                "info": "The language model to use for the decision.",
-            },
-        }
-
-    def build(self, template: Prompt, llm: BaseLanguageModel, **kwargs) -> dict:
-        # This is a simple component that always returns True
-        prompt_template = PromptTemplate.from_template(Text(template))
-
-        attributes_to_check = ["text", "page_content"]
-        for key, value in kwargs.items():
-            for attribute in attributes_to_check:
-                if hasattr(value, attribute):
-                    kwargs[key] = getattr(value, attribute)
-
-        chain = prompt_template | llm
-        result = chain.invoke(kwargs)
-        if hasattr(result, "content") and isinstance(result.content, str):
-            result = result.content
-        elif isinstance(result, str):
-            result = result
-        else:
-            result = result.get("response")
-
-        if result.lower() not in ["true", "false"]:
-            raise ValueError("The prompt should generate a boolean response (True or False).")
-        # The string should be the words true or false
-        # if not raise an error
-        bool_result = result.lower() == "true"
-        return {"condition": bool_result, "result": kwargs}