|
|
from typing import Any |
|
|
|
|
|
from langchain.tools import StructuredTool |
|
|
from langchain_community.utilities.serpapi import SerpAPIWrapper |
|
|
from langchain_core.tools import ToolException |
|
|
from loguru import logger |
|
|
from pydantic import BaseModel, Field |
|
|
|
|
|
from langflow.base.langchain_utilities.model import LCToolComponent |
|
|
from langflow.field_typing import Tool |
|
|
from langflow.inputs import DictInput, IntInput, MultilineInput, SecretStrInput |
|
|
from langflow.schema import Data |
|
|
|
|
|
|
|
|
class SerpAPISchema(BaseModel): |
|
|
"""Schema for SerpAPI search parameters.""" |
|
|
|
|
|
query: str = Field(..., description="The search query") |
|
|
params: dict[str, Any] | None = Field( |
|
|
default={ |
|
|
"engine": "google", |
|
|
"google_domain": "google.com", |
|
|
"gl": "us", |
|
|
"hl": "en", |
|
|
}, |
|
|
description="Additional search parameters", |
|
|
) |
|
|
max_results: int = Field(5, description="Maximum number of results to return") |
|
|
max_snippet_length: int = Field(100, description="Maximum length of each result snippet") |
|
|
|
|
|
|
|
|
class SerpAPIComponent(LCToolComponent): |
|
|
display_name = "Serp Search API" |
|
|
description = "Call Serp Search API with result limiting" |
|
|
name = "SerpAPI" |
|
|
icon = "SerpSearch" |
|
|
|
|
|
inputs = [ |
|
|
SecretStrInput(name="serpapi_api_key", display_name="SerpAPI API Key", required=True), |
|
|
MultilineInput( |
|
|
name="input_value", |
|
|
display_name="Input", |
|
|
), |
|
|
DictInput(name="search_params", display_name="Parameters", advanced=True, is_list=True), |
|
|
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), |
|
|
] |
|
|
|
|
|
def _build_wrapper(self, params: dict[str, Any] | None = None) -> SerpAPIWrapper: |
|
|
"""Build a SerpAPIWrapper with the provided parameters.""" |
|
|
params = params or {} |
|
|
if params: |
|
|
return SerpAPIWrapper( |
|
|
serpapi_api_key=self.serpapi_api_key, |
|
|
params=params, |
|
|
) |
|
|
return SerpAPIWrapper(serpapi_api_key=self.serpapi_api_key) |
|
|
|
|
|
def build_tool(self) -> Tool: |
|
|
wrapper = self._build_wrapper(self.search_params) |
|
|
|
|
|
def search_func( |
|
|
query: str, params: dict[str, Any] | None = None, max_results: int = 5, max_snippet_length: int = 100 |
|
|
) -> list[dict[str, Any]]: |
|
|
try: |
|
|
local_wrapper = wrapper |
|
|
if params: |
|
|
local_wrapper = self._build_wrapper(params) |
|
|
|
|
|
full_results = local_wrapper.results(query) |
|
|
organic_results = full_results.get("organic_results", [])[:max_results] |
|
|
|
|
|
limited_results = [] |
|
|
for result in organic_results: |
|
|
limited_result = { |
|
|
"title": result.get("title", "")[:max_snippet_length], |
|
|
"link": result.get("link", ""), |
|
|
"snippet": result.get("snippet", "")[:max_snippet_length], |
|
|
} |
|
|
limited_results.append(limited_result) |
|
|
|
|
|
except Exception as e: |
|
|
error_message = f"Error in SerpAPI search: {e!s}" |
|
|
logger.debug(error_message) |
|
|
raise ToolException(error_message) from e |
|
|
return limited_results |
|
|
|
|
|
tool = StructuredTool.from_function( |
|
|
name="serp_search_api", |
|
|
description="Search for recent results using SerpAPI with result limiting", |
|
|
func=search_func, |
|
|
args_schema=SerpAPISchema, |
|
|
) |
|
|
|
|
|
self.status = "SerpAPI Tool created" |
|
|
return tool |
|
|
|
|
|
def run_model(self) -> list[Data]: |
|
|
tool = self.build_tool() |
|
|
try: |
|
|
results = tool.run( |
|
|
{ |
|
|
"query": self.input_value, |
|
|
"params": self.search_params or {}, |
|
|
"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 Exception as e: |
|
|
logger.opt(exception=True).debug("Error running SerpAPI") |
|
|
self.status = f"Error: {e}" |
|
|
return [Data(data={"error": str(e)}, text=str(e))] |
|
|
|
|
|
self.status = data_list |
|
|
return data_list |
|
|
|
