Merge branch 'zustand/io/migration' into state_theories

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

@@ -114,7 +114,7 @@ async def build_vertex(
 
         vertex = graph.get_vertex(vertex_id)
         try:
-            if not vertex.pinned or not vertex._built:
+            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)
 
@@ -233,7 +233,7 @@ async def build_vertex_stream(
                     )
                     yield str(stream_data)
 
-                elif not vertex.pinned or not vertex._built:
+                elif not vertex.frozen or not vertex._built:
                     logger.debug(f"Streaming vertex {vertex_id}")
                     stream_data = StreamData(
                         event="message",

src/backend/langflow/base/data/utils.py

@@ -0,0 +1,89 @@
+from concurrent import futures
+from pathlib import Path
+from typing import List, Optional, Text
+
+from langflow.schema.schema import Record
+
+
+def is_hidden(path: Path) -> bool:
+    return path.name.startswith(".")
+
+
+def retrieve_file_paths(
+    path: str,
+    types: List[str],
+    load_hidden: bool,
+    recursive: bool,
+    depth: int,
+) -> List[str]:
+    path_obj = Path(path)
+    if not path_obj.exists() or not path_obj.is_dir():
+        raise ValueError(f"Path {path} must exist and be a directory.")
+
+    def match_types(p: Path) -> bool:
+        return any(p.suffix == f".{t}" for t in types) if types else True
+
+    def is_not_hidden(p: Path) -> bool:
+        return not is_hidden(p) or load_hidden
+
+    def walk_level(directory: Path, max_depth: int):
+        directory = directory.resolve()
+        prefix_length = len(directory.parts)
+        for p in directory.rglob("*" if recursive else "[!.]*"):
+            if len(p.parts) - prefix_length <= max_depth:
+                yield p
+
+    glob = "**/*" if recursive else "*"
+    paths = walk_level(path_obj, depth) if depth else path_obj.glob(glob)
+    file_paths = [
+        Text(p) for p in paths if p.is_file() and match_types(p) and is_not_hidden(p)
+    ]
+
+    return file_paths
+
+
+def parse_file_to_record(file_path: str, silent_errors: bool) -> Optional[Record]:
+    # Use the partition function to load the file
+    from unstructured.partition.auto import partition  # type: ignore
+
+    try:
+        elements = partition(file_path)
+    except Exception as e:
+        if not silent_errors:
+            raise ValueError(f"Error loading file {file_path}: {e}") from e
+        return None
+
+    # Create a Record
+    text = "\n\n".join([Text(el) for el in elements])
+    metadata = elements.metadata if hasattr(elements, "metadata") else {}
+    metadata["file_path"] = file_path
+    record = Record(text=text, data=metadata)
+    return record
+
+
+def get_elements(
+    file_paths: List[str],
+    silent_errors: bool,
+    max_concurrency: int,
+    use_multithreading: bool,
+) -> List[Optional[Record]]:
+    if use_multithreading:
+        records = parallel_load_records(file_paths, silent_errors, max_concurrency)
+    else:
+        records = [
+            parse_file_to_record(file_path, silent_errors) for file_path in file_paths
+        ]
+    records = list(filter(None, records))
+    return records
+
+
+def parallel_load_records(
+    file_paths: List[str], silent_errors: bool, max_concurrency: int
+) -> List[Optional[Record]]:
+    with futures.ThreadPoolExecutor(max_workers=max_concurrency) as executor:
+        loaded_files = executor.map(
+            lambda file_path: parse_file_to_record(file_path, silent_errors),
+            file_paths,
+        )
+    # loaded_files is an iterator, so we need to convert it to a list
+    return list(loaded_files)

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

@@ -0,0 +1,107 @@
+import warnings
+from typing import Optional, Union
+
+from langflow import CustomComponent
+from langflow.field_typing import Text
+from langflow.memory import add_messages
+from langflow.schema import Record
+
+
+class ChatComponent(CustomComponent):
+    display_name = "Chat Component"
+    description = "Use as base for chat components."
+
+    def build_config(self):
+        return {
+            "input_value": {
+                "input_types": ["Text"],
+                "display_name": "Message",
+                "multiline": True,
+            },
+            "sender": {
+                "options": ["Machine", "User"],
+                "display_name": "Sender Type",
+            },
+            "sender_name": {"display_name": "Sender Name"},
+            "session_id": {
+                "display_name": "Session ID",
+                "info": "If provided, the message will be stored in the memory.",
+            },
+            "return_record": {
+                "display_name": "Return Record",
+                "info": "Return the message as a record containing the sender, sender_name, and session_id.",
+            },
+        }
+
+    def store_message(
+        self,
+        message: Union[str, Text, Record],
+        session_id: Optional[str] = None,
+        sender: Optional[str] = None,
+        sender_name: Optional[str] = None,
+    ) -> list[Record]:
+        if not message:
+            warnings.warn("No message provided.")
+            return []
+
+        if not session_id or not sender or not sender_name:
+            raise ValueError(
+                "All of session_id, sender, and sender_name must be provided."
+            )
+        if isinstance(message, Record):
+            record = message
+            record.data.update(
+                {
+                    "session_id": session_id,
+                    "sender": sender,
+                    "sender_name": sender_name,
+                }
+            )
+        else:
+            record = Record(
+                text=message,
+                data={
+                    "session_id": session_id,
+                    "sender": sender,
+                    "sender_name": sender_name,
+                },
+            )
+
+        self.status = record
+        records = add_messages([record])
+        return records[0]
+
+    def build(
+        self,
+        sender: Optional[str] = "User",
+        sender_name: Optional[str] = "User",
+        input_value: Optional[str] = None,
+        session_id: Optional[str] = None,
+        return_record: Optional[bool] = False,
+    ) -> Union[Text, Record]:
+        input_value_record: Optional[Record] = None
+        if return_record:
+            if isinstance(input_value, Record):
+                # Update the data of the record
+                input_value.data["sender"] = sender
+                input_value.data["sender_name"] = sender_name
+                input_value.data["session_id"] = session_id
+            else:
+                input_value_record = Record(
+                    text=input_value,
+                    data={
+                        "sender": sender,
+                        "sender_name": sender_name,
+                        "session_id": session_id,
+                    },
+                )
+        if not input_value:
+            input_value = ""
+        if return_record and input_value_record:
+            result: Union[Text, Record] = input_value_record
+        else:
+            result = input_value
+        self.status = result
+        if session_id:
+            self.store_message(result, session_id, sender, sender_name)
+        return result

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

@@ -0,0 +1,19 @@
+from typing import Optional
+
+from langflow import CustomComponent
+from langflow.field_typing import Text
+
+
+class TextComponent(CustomComponent):
+    display_name = "Text Component"
+    description = "Used to pass text to the next component."
+
+    field_config = {
+        "input_value": {"display_name": "Value", "multiline": True},
+    }
+
+    def build(self, input_value: Optional[str] = "") -> Text:
+        self.status = input_value
+        if not input_value:
+            input_value = ""
+        return input_value

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

@@ -0,0 +1,76 @@
+from typing import Any, Dict, List, Optional
+
+from langflow import CustomComponent
+from langflow.base.data.utils import (
+    parallel_load_records,
+    parse_file_to_record,
+    retrieve_file_paths,
+)
+from langflow.schema import Record
+
+
+class DirectoryComponent(CustomComponent):
+    display_name = "Directory"
+    description = "Load files from a directory."
+
+    def build_config(self) -> Dict[str, Any]:
+        return {
+            "path": {"display_name": "Path"},
+            "types": {
+                "display_name": "Types",
+                "info": "File types to load. Leave empty to load all types.",
+            },
+            "depth": {"display_name": "Depth", "info": "Depth to search for files."},
+            "max_concurrency": {"display_name": "Max Concurrency", "advanced": True},
+            "load_hidden": {
+                "display_name": "Load Hidden",
+                "advanced": True,
+                "info": "If true, hidden files will be loaded.",
+            },
+            "recursive": {
+                "display_name": "Recursive",
+                "advanced": True,
+                "info": "If true, the search will be recursive.",
+            },
+            "silent_errors": {
+                "display_name": "Silent Errors",
+                "advanced": True,
+                "info": "If true, errors will not raise an exception.",
+            },
+            "use_multithreading": {
+                "display_name": "Use Multithreading",
+                "advanced": True,
+            },
+        }
+
+    def build(
+        self,
+        path: str,
+        types: Optional[List[str]] = None,
+        depth: int = 0,
+        max_concurrency: int = 2,
+        load_hidden: bool = False,
+        recursive: bool = True,
+        silent_errors: bool = False,
+        use_multithreading: bool = True,
+    ) -> List[Optional[Record]]:
+        if types is None:
+            types = []
+        resolved_path = self.resolve_path(path)
+        file_paths = retrieve_file_paths(
+            resolved_path, types, load_hidden, recursive, depth
+        )
+        loaded_records = []
+
+        if use_multithreading:
+            loaded_records = parallel_load_records(
+                file_paths, silent_errors, max_concurrency
+            )
+        else:
+            loaded_records = [
+                parse_file_to_record(file_path, silent_errors)
+                for file_path in file_paths
+            ]
+        loaded_records = list(filter(None, loaded_records))
+        self.status = loaded_records
+        return loaded_records

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

@@ -0,0 +1,28 @@
+from typing import Any, Dict, Optional
+
+from langflow import CustomComponent
+from langflow.base.data.utils import parse_file_to_record
+from langflow.schema import Record
+
+
+class FileComponent(CustomComponent):
+    display_name = "File"
+    description = "Load a file."
+
+    def build_config(self) -> Dict[str, Any]:
+        return {
+            "path": {"display_name": "Path"},
+            "silent_errors": {
+                "display_name": "Silent Errors",
+                "advanced": True,
+                "info": "If true, errors will not raise an exception.",
+            },
+        }
+
+    def build(
+        self,
+        path: str,
+        silent_errors: bool = False,
+    ) -> Optional[Record]:
+        resolved_path = self.resolve_path(path)
+        return parse_file_to_record(resolved_path, silent_errors)

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

@@ -0,0 +1,124 @@
+from typing import List
+
+from langflow import CustomComponent
+from langflow.schema import Record
+from langflow.utils.constants import LOADERS_INFO
+
+
+class FileLoaderComponent(CustomComponent):
+    display_name: str = "File Loader"
+    description: str = "Generic File Loader"
+    beta = True
+
+    def build_config(self):
+        loader_options = ["Automatic"] + [
+            loader_info["name"] for loader_info in LOADERS_INFO
+        ]
+
+        file_types = []
+        suffixes = []
+
+        for loader_info in LOADERS_INFO:
+            if "allowedTypes" in loader_info:
+                file_types.extend(loader_info["allowedTypes"])
+                suffixes.extend([f".{ext}" for ext in loader_info["allowedTypes"]])
+
+        return {
+            "file_path": {
+                "display_name": "File Path",
+                "required": True,
+                "field_type": "file",
+                "file_types": [
+                    "json",
+                    "txt",
+                    "csv",
+                    "jsonl",
+                    "html",
+                    "htm",
+                    "conllu",
+                    "enex",
+                    "msg",
+                    "pdf",
+                    "srt",
+                    "eml",
+                    "md",
+                    "mdx",
+                    "pptx",
+                    "docx",
+                ],
+                "suffixes": [
+                    ".json",
+                    ".txt",
+                    ".csv",
+                    ".jsonl",
+                    ".html",
+                    ".htm",
+                    ".conllu",
+                    ".enex",
+                    ".msg",
+                    ".pdf",
+                    ".srt",
+                    ".eml",
+                    ".md",
+                    ".mdx",
+                    ".pptx",
+                    ".docx",
+                ],
+                # "file_types" : file_types,
+                # "suffixes": suffixes,
+            },
+            "loader": {
+                "display_name": "Loader",
+                "is_list": True,
+                "required": True,
+                "options": loader_options,
+                "value": "Automatic",
+            },
+            "code": {"show": False},
+        }
+
+    def build(self, file_path: str, loader: str) -> List[Record]:
+        file_type = file_path.split(".")[-1]
+
+        # Map the loader to the correct loader class
+        selected_loader_info = None
+        for loader_info in LOADERS_INFO:
+            if loader_info["name"] == loader:
+                selected_loader_info = loader_info
+                break
+
+        if selected_loader_info is None and loader != "Automatic":
+            raise ValueError(f"Loader {loader} not found in the loader info list")
+
+        if loader == "Automatic":
+            # Determine the loader based on the file type
+            default_loader_info = None
+            for info in LOADERS_INFO:
+                if "defaultFor" in info and file_type in info["defaultFor"]:
+                    default_loader_info = info
+                    break
+
+            if default_loader_info is None:
+                raise ValueError(f"No default loader found for file type: {file_type}")
+
+            selected_loader_info = default_loader_info
+        if isinstance(selected_loader_info, dict):
+            loader_import: str = selected_loader_info["import"]
+        else:
+            raise ValueError(
+                f"Loader info for {loader} is not a dict\nLoader info:\n{selected_loader_info}"
+            )
+        module_name, class_name = loader_import.rsplit(".", 1)
+
+        try:
+            # Import the loader class
+            loader_module = __import__(module_name, fromlist=[class_name])
+            loader_instance = getattr(loader_module, class_name)
+        except ImportError as e:
+            raise ValueError(
+                f"Loader {loader} could not be imported\nLoader info:\n{selected_loader_info}"
+            ) from e
+
+        result = loader_instance(file_path=file_path)
+        docs = result.load()
+        return self.to_records(docs)

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

@@ -0,0 +1,26 @@
+from typing import Any, Dict, Optional
+
+from langchain_community.document_loaders.url import UnstructuredURLLoader
+
+from langflow import CustomComponent
+from langflow.schema import Record
+
+
+class URLComponent(CustomComponent):
+    display_name = "URL"
+    description = "Load a URL."
+
+    def build_config(self) -> Dict[str, Any]:
+        return {
+            "urls": {"display_name": "URL"},
+        }
+
+    async def build(
+        self,
+        urls: list[str],
+    ) -> Optional[Record]:
+
+        loader = UnstructuredURLLoader(urls=urls)
+        docs = loader.load()
+        records = self.to_records(docs)
+        return records

src/backend/langflow/components/data/UrlLoader.py

@@ -0,0 +1,49 @@
+from typing import List
+
+from langchain import document_loaders
+from langchain_core.documents import Document
+
+from langflow import CustomComponent
+
+
+class UrlLoaderComponent(CustomComponent):
+    display_name: str = "Url Loader"
+    description: str = "Generic Url Loader Component"
+
+    def build_config(self):
+        return {
+            "web_path": {
+                "display_name": "Url",
+                "required": True,
+            },
+            "loader": {
+                "display_name": "Loader",
+                "is_list": True,
+                "required": True,
+                "options": [
+                    "AZLyricsLoader",
+                    "CollegeConfidentialLoader",
+                    "GitbookLoader",
+                    "HNLoader",
+                    "IFixitLoader",
+                    "IMSDbLoader",
+                    "WebBaseLoader",
+                ],
+                "value": "WebBaseLoader",
+            },
+            "code": {"show": False},
+        }
+
+    def build(self, web_path: str, loader: str) -> List[Document]:
+        try:
+            loader_instance = getattr(document_loaders, loader)(web_path=web_path)
+        except Exception as e:
+            raise ValueError(f"No loader found for: {web_path}") from e
+        docs = loader_instance.load()
+        avg_length = sum(
+            len(doc.page_content) for doc in docs if hasattr(doc, "page_content")
+        ) / len(docs)
+        self.status = f"""{len(docs)} documents)
+        \nAvg. Document Length (characters): {int(avg_length)}
+        Documents: {docs[:3]}..."""
+        return docs

src/backend/langflow/components/documentloaders/GatherRecords.py

@@ -1,161 +0,0 @@
-from concurrent import futures
-from pathlib import Path
-from typing import Any, Dict, List, Optional, Text
-
-from langflow import CustomComponent
-from langflow.schema import Record
-
-
-class GatherRecordsComponent(CustomComponent):
-    display_name = "Gather Records"
-    description = "Gather records from a directory."
-
-    def build_config(self) -> Dict[str, Any]:
-        return {
-            "path": {"display_name": "Path"},
-            "types": {
-                "display_name": "Types",
-                "info": "File types to load. Leave empty to load all types.",
-            },
-            "depth": {"display_name": "Depth", "info": "Depth to search for files."},
-            "max_concurrency": {"display_name": "Max Concurrency", "advanced": True},
-            "load_hidden": {
-                "display_name": "Load Hidden",
-                "advanced": True,
-                "info": "If true, hidden files will be loaded.",
-            },
-            "recursive": {
-                "display_name": "Recursive",
-                "advanced": True,
-                "info": "If true, the search will be recursive.",
-            },
-            "silent_errors": {
-                "display_name": "Silent Errors",
-                "advanced": True,
-                "info": "If true, errors will not raise an exception.",
-            },
-            "use_multithreading": {
-                "display_name": "Use Multithreading",
-                "advanced": True,
-            },
-        }
-
-    def is_hidden(self, path: Path) -> bool:
-        return path.name.startswith(".")
-
-    def retrieve_file_paths(
-        self,
-        path: str,
-        types: List[str],
-        load_hidden: bool,
-        recursive: bool,
-        depth: int,
-    ) -> List[str]:
-        path_obj = Path(path)
-        if not path_obj.exists() or not path_obj.is_dir():
-            raise ValueError(f"Path {path} must exist and be a directory.")
-
-        def match_types(p: Path) -> bool:
-            return any(p.suffix == f".{t}" for t in types) if types else True
-
-        def is_not_hidden(p: Path) -> bool:
-            return not self.is_hidden(p) or load_hidden
-
-        def walk_level(directory: Path, max_depth: int):
-            directory = directory.resolve()
-            prefix_length = len(directory.parts)
-            for p in directory.rglob("*" if recursive else "[!.]*"):
-                if len(p.parts) - prefix_length <= max_depth:
-                    yield p
-
-        glob = "**/*" if recursive else "*"
-        paths = walk_level(path_obj, depth) if depth else path_obj.glob(glob)
-        file_paths = [
-            Text(p)
-            for p in paths
-            if p.is_file() and match_types(p) and is_not_hidden(p)
-        ]
-
-        return file_paths
-
-    def parse_file_to_record(
-        self, file_path: str, silent_errors: bool
-    ) -> Optional[Record]:
-        # Use the partition function to load the file
-        from unstructured.partition.auto import partition  # type: ignore
-
-        try:
-            elements = partition(file_path)
-        except Exception as e:
-            if not silent_errors:
-                raise ValueError(f"Error loading file {file_path}: {e}") from e
-            return None
-
-        # Create a Record
-        text = "\n\n".join([Text(el) for el in elements])
-        metadata = elements.metadata if hasattr(elements, "metadata") else {}
-        metadata["file_path"] = file_path
-        record = Record(text=text, data=metadata)
-        return record
-
-    def get_elements(
-        self,
-        file_paths: List[str],
-        silent_errors: bool,
-        max_concurrency: int,
-        use_multithreading: bool,
-    ) -> List[Optional[Record]]:
-        if use_multithreading:
-            records = self.parallel_load_records(
-                file_paths, silent_errors, max_concurrency
-            )
-        else:
-            records = [
-                self.parse_file_to_record(file_path, silent_errors)
-                for file_path in file_paths
-            ]
-        records = list(filter(None, records))
-        return records
-
-    def parallel_load_records(
-        self, file_paths: List[str], silent_errors: bool, max_concurrency: int
-    ) -> List[Optional[Record]]:
-        with futures.ThreadPoolExecutor(max_workers=max_concurrency) as executor:
-            loaded_files = executor.map(
-                lambda file_path: self.parse_file_to_record(file_path, silent_errors),
-                file_paths,
-            )
-        # loaded_files is an iterator, so we need to convert it to a list
-        return list(loaded_files)
-
-    def build(
-        self,
-        path: str,
-        types: Optional[List[str]] = None,
-        depth: int = 0,
-        max_concurrency: int = 2,
-        load_hidden: bool = False,
-        recursive: bool = True,
-        silent_errors: bool = False,
-        use_multithreading: bool = True,
-    ) -> List[Optional[Record]]:
-        if types is None:
-            types = []
-        resolved_path = self.resolve_path(path)
-        file_paths = self.retrieve_file_paths(
-            resolved_path, types, load_hidden, recursive, depth
-        )
-        loaded_records = []
-
-        if use_multithreading:
-            loaded_records = self.parallel_load_records(
-                file_paths, silent_errors, max_concurrency
-            )
-        else:
-            loaded_records = [
-                self.parse_file_to_record(file_path, silent_errors)
-                for file_path in file_paths
-            ]
-        loaded_records = list(filter(None, loaded_records))
-        self.status = loaded_records
-        return loaded_records

src/backend/langflow/components/inputs/ChatInput.py

@@ -0,0 +1,26 @@
+from typing import Optional, Union
+
+from langflow.base.io.chat import ChatComponent
+from langflow.field_typing import Text
+from langflow.schema import Record
+
+
+class ChatInput(ChatComponent):
+    display_name = "Chat Input"
+    description = "Used to get user input from the chat."
+
+    def build(
+        self,
+        sender: Optional[str] = "User",
+        sender_name: Optional[str] = "User",
+        input_value: Optional[str] = None,
+        session_id: Optional[str] = None,
+        return_record: Optional[bool] = False,
+    ) -> Union[Text, Record]:
+        return super().build(
+            sender=sender,
+            sender_name=sender_name,
+            input_value=input_value,
+            session_id=session_id,
+            return_record=return_record,
+        )

src/backend/langflow/components/inputs/TextInput.py

@@ -0,0 +1,12 @@
+from typing import Optional
+
+from langflow.base.io.text import TextComponent
+from langflow.field_typing import Text
+
+
+class TextInput(TextComponent):
+    display_name = "Text Input"
+    description = "Used to pass text input to the next component."
+
+    def build(self, input_value: Optional[str] = "") -> Text:
+        return super().build(input_value=input_value)

src/backend/langflow/components/langchain_utilities/BingSearchAPIWrapper.py

@@ -0,0 +1,31 @@
+from langflow import CustomComponent
+
+# Assuming `BingSearchAPIWrapper` is a class that exists in the context
+# and has the appropriate methods and attributes.
+# We need to make sure this class is importable from the context where this code will be running.
+from langchain_community.utilities.bing_search import BingSearchAPIWrapper
+
+
+class BingSearchAPIWrapperComponent(CustomComponent):
+    display_name = "BingSearchAPIWrapper"
+    description = "Wrapper for Bing Search API."
+
+    def build_config(self):
+        return {
+            "bing_search_url": {"display_name": "Bing Search URL"},
+            "bing_subscription_key": {
+                "display_name": "Bing Subscription Key",
+                "password": True,
+            },
+            "k": {"display_name": "Number of results", "advanced": True},
+            # 'k' is not included as it is not shown (show=False)
+        }
+
+    def build(
+        self,
+        bing_search_url: str,
+        bing_subscription_key: str,
+        k: int = 10,
+    ) -> BingSearchAPIWrapper:
+        # 'k' has a default value and is not shown (show=False), so it is hardcoded here
+        return BingSearchAPIWrapper(bing_search_url=bing_search_url, bing_subscription_key=bing_subscription_key, k=k)

src/backend/langflow/components/langchain_utilities/DocumentToRecord.py

@@ -0,0 +1,22 @@
+from typing import List
+
+from langchain_core.documents import Document
+
+from langflow import CustomComponent
+from langflow.schema import Record
+
+
+class DocumentToRecordComponent(CustomComponent):
+    display_name = "Documents to Records"
+    description = "Convert documents to records."
+
+    field_config = {
+        "documents": {"display_name": "Documents"},
+    }
+
+    def build(self, documents: List[Document]) -> List[Record]:
+        if isinstance(documents, Document):
+            documents = [documents]
+        records = [Record.from_document(document) for document in documents]
+        self.status = records
+        return records

src/backend/langflow/components/langchain_utilities/GoogleSearchAPIWrapper.py

@@ -0,0 +1,22 @@
+from typing import Callable, Union
+
+from langchain_community.utilities.google_search import GoogleSearchAPIWrapper
+from langflow import CustomComponent
+
+
+class GoogleSearchAPIWrapperComponent(CustomComponent):
+    display_name = "GoogleSearchAPIWrapper"
+    description = "Wrapper for Google Search API."
+
+    def build_config(self):
+        return {
+            "google_api_key": {"display_name": "Google API Key", "password": True},
+            "google_cse_id": {"display_name": "Google CSE ID", "password": True},
+        }
+
+    def build(
+        self,
+        google_api_key: str,
+        google_cse_id: str,
+    ) -> Union[GoogleSearchAPIWrapper, Callable]:
+        return GoogleSearchAPIWrapper(google_api_key=google_api_key, google_cse_id=google_cse_id)  # type: ignore