fix: refactor DuckDuckGo search component to use new Component base class (nightly fix) (#6064)

Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>

src/backend/base/langflow/components/tools/duck_duck_go_search_run.py

@@ -1,75 +1,91 @@
-from typing import Any
-
-from langchain.tools import StructuredTool
 from langchain_community.tools import DuckDuckGoSearchRun
-from langchain_core.tools import ToolException
-from pydantic import BaseModel, Field
 
-from langflow.base.langchain_utilities.model import LCToolComponent
+from langflow.custom import Component
-from langflow.field_typing import Tool
 from langflow.inputs import IntInput, MessageTextInput
+from langflow.io import Output
 from langflow.schema import Data
+from langflow.schema.message import Message
 
 
-class DuckDuckGoSearchComponent(LCToolComponent):
+class DuckDuckGoSearchComponent(Component):
-    display_name: str = "DuckDuckGo Search"
+    """Component for performing web searches using DuckDuckGo."""
-    description: str = "Perform web searches using the DuckDuckGo search engine with result limiting"
+
-    name = "DuckDuckGoSearch"
+    display_name = "DuckDuckGo Search"
-    documentation: str = "https://python.langchain.com/docs/integrations/tools/ddg"
+    description = "Search the web using DuckDuckGo with customizable result limits"
-    icon: str = "DuckDuckGo"
+    documentation = "https://python.langchain.com/docs/integrations/tools/ddg"
+    icon = "DuckDuckGo"
+
     inputs = [
         MessageTextInput(
             name="input_value",
             display_name="Search Query",
+            required=True,
+            info="The search query to execute with DuckDuckGo",
+            tool_mode=True,
+        ),
+        IntInput(
+            name="max_results",
+            display_name="Max Results",
+            value=5,
+            required=False,
+            advanced=True,
+            info="Maximum number of search results to return",
+        ),
+        IntInput(
+            name="max_snippet_length",
+            display_name="Max Snippet Length",
+            value=100,
+            required=False,
+            advanced=True,
+            info="Maximum length of each result snippet",
         ),
-        IntInput(name="max_results", display_name="Max Results", value=5, advanced=True),
-        IntInput(name="max_snippet_length", display_name="Max Snippet Length", value=100, advanced=True),
     ]
 
-    class DuckDuckGoSearchSchema(BaseModel):
+    outputs = [
-        query: str = Field(..., description="The search query")
+        Output(display_name="Data", name="data", method="fetch_content"),
-        max_results: int = Field(5, description="Maximum number of results to return")
+        Output(display_name="Text", name="text", method="fetch_content_text"),
-        max_snippet_length: int = Field(100, description="Maximum length of each result snippet")
+    ]
 
-    def _build_wrapper(self):
+    def _build_wrapper(self) -> DuckDuckGoSearchRun:
+        """Build the DuckDuckGo search wrapper."""
         return DuckDuckGoSearchRun()
 
-    def build_tool(self) -> Tool:
+    def run_model(self) -> list[Data]:
+        return self.fetch_content()
+
+    def fetch_content(self) -> list[Data]:
+        """Execute the search and return results as Data objects."""
+        try:
             wrapper = self._build_wrapper()
 
-        def search_func(query: str, max_results: int = 5, max_snippet_length: int = 100) -> list[dict[str, Any]]:
+            full_results = wrapper.run(f"{self.input_value} (site:*)")
-            try:
+
-                full_results = wrapper.run(f"{query} (site:*)")
+            result_list = full_results.split("\n")[: self.max_results]
-                result_list = full_results.split("\n")[:max_results]
+
-                limited_results = []
+            data_results = []
             for result in result_list:
-                    limited_result = {
+                if result.strip():
-                        "snippet": result[:max_snippet_length],
+                    snippet = result[: self.max_snippet_length]
-                    }
+                    data_results.append(
-                    limited_results.append(limited_result)
+                        Data(
-            except Exception as e:
+                            text=snippet,
-                msg = f"Error in DuckDuckGo Search: {e!s}"
+                            data={
-                raise ToolException(msg) from e
+                                "content": result,
-            return limited_results
+                                "snippet": snippet,
-
+                            },
-        tool = StructuredTool.from_function(
-            name="duckduckgo_search",
-            description="Search for recent results using DuckDuckGo with result limiting",
-            func=search_func,
-            args_schema=self.DuckDuckGoSearchSchema,
                         )
-        self.status = "DuckDuckGo Search Tool created"
-        return tool
-
-    def run_model(self) -> list[Data]:
-        tool = self.build_tool()
-        results = tool.run(
-            {
-                "query": self.input_value,
-                "max_results": self.max_results,
-                "max_snippet_length": self.max_snippet_length,
-            }
                     )
-        data_list = [Data(data=result, text=result.get("snippet", "")) for result in results]
+        except (ValueError, AttributeError) as e:
-        self.status = data_list  # type: ignore[assignment]
+            error_data = [Data(text=str(e), data={"error": str(e)})]
-        return data_list
+            self.status = error_data
+            return error_data
+        else:
+            self.status = data_results
+            return data_results
+
+    def fetch_content_text(self) -> Message:
+        """Return search results as a single text message."""
+        data = self.fetch_content()
+        result_string = "\n".join(item.text for item in data)
+        self.status = result_string
+        return Message(text=result_string)