Tools
This Langflow feature is currently in public preview. Development is ongoing, and the features and functionality are subject to change. Langflow, and the use of such, is subject to the DataStax Preview Terms. |
Tool components are used to interact with external services, APIs, and tools. They can be used to search the web, query databases, and perform other tasks.
Bing Search API
This component allows you to call the Bing Search API.
Parameters
Name | Type | Description |
---|---|---|
bing_subscription_key |
SecretString |
Bing API subscription key |
input_value |
String |
Search query input |
bing_search_url |
String |
Custom Bing Search URL (optional) |
k |
Integer |
Number of search results to return |
Name | Type | Description |
---|---|---|
results |
List[Data] |
List of search results |
tool |
Tool |
Bing Search tool for use in LangChain |
Component code
BingSearchAPI.py
from typing import List, cast
from langchain_community.tools.bing_search import BingSearchResults
from langchain_community.utilities import BingSearchAPIWrapper
from langflow.base.langchain_utilities.model import LCToolComponent
from langflow.field_typing import Tool
from langflow.inputs import IntInput, MessageTextInput, MultilineInput, SecretStrInput
from langflow.schema import Data
class BingSearchAPIComponent(LCToolComponent):
display_name = "Bing Search API"
description = "Call the Bing Search API."
name = "BingSearchAPI"
inputs = [
SecretStrInput(name="bing_subscription_key", display_name="Bing Subscription Key"),
MultilineInput(
name="input_value",
display_name="Input",
),
MessageTextInput(name="bing_search_url", display_name="Bing Search URL", advanced=True),
IntInput(name="k", display_name="Number of results", value=4, required=True),
]
def run_model(self) -> List[Data]:
if self.bing_search_url:
wrapper = BingSearchAPIWrapper(
bing_search_url=self.bing_search_url, bing_subscription_key=self.bing_subscription_key
)
else:
wrapper = BingSearchAPIWrapper(bing_subscription_key=self.bing_subscription_key) # type: ignore
results = wrapper.results(query=self.input_value, num_results=self.k)
data = [Data(data=result, text=result["snippet"]) for result in results]
self.status = data
return data
def build_tool(self) -> Tool:
if self.bing_search_url:
wrapper = BingSearchAPIWrapper(
bing_search_url=self.bing_search_url, bing_subscription_key=self.bing_subscription_key
)
else:
wrapper = BingSearchAPIWrapper(bing_subscription_key=self.bing_subscription_key) # type: ignore
return cast(Tool, BingSearchResults(api_wrapper=wrapper, num_results=self.k))
Glean Search API
This component allows you to call the Glean Search API.
Parameters
Name | Type | Description |
---|---|---|
glean_api_url |
String |
URL of the Glean API |
glean_access_token |
SecretString |
Access token for Glean API authentication |
query |
String |
Search query input |
page_size |
Integer |
Number of results per page (default: 10) |
request_options |
Dict |
Additional options for the API request (optional) |
Name | Type | Description |
---|---|---|
results |
List[Data] |
List of search results |
tool |
Tool |
Glean Search tool for use in LangChain |
Component code
GleanSearchAPI.py
import httpx
import json
from typing import Any, Dict, Optional, Union
from urllib.parse import urljoin
from langchain_core.pydantic_v1 import BaseModel
from langflow.base.langchain_utilities.model import LCToolComponent
from langflow.inputs import SecretStrInput, StrInput, NestedDictInput, IntInput
from langflow.field_typing import Tool
from langflow.schema import Data
class GleanSearchAPIComponent(LCToolComponent):
display_name = "Glean Search API"
description = "Call Glean Search API"
name = "GleanAPI"
inputs = [
StrInput(
name="glean_api_url",
display_name="Glean API URL",
required=True,
),
SecretStrInput(name="glean_access_token", display_name="Glean Access Token", required=True),
StrInput(name="query", display_name="Query", required=True),
IntInput(name="page_size", display_name="Page Size", value=10),
NestedDictInput(name="request_options", display_name="Request Options", required=False),
]
def build_tool(self) -> Tool:
wrapper = self._build_wrapper()
return Tool(name="glean_search_api", description="Search with the Glean API", func=wrapper.run)
def run_model(self) -> Union[Data, list[Data]]:
wrapper = self._build_wrapper()
results = wrapper.results(
query=self.query,
page_size=self.page_size,
request_options=self.request_options,
)
list_results = results.get("results", [])
# Build the data
data = []
for result in list_results:
data.append(Data(data=result))
self.status = data
return data
def _build_wrapper(self):
class GleanAPIWrapper(BaseModel):
"""
Wrapper around Glean API.
"""
glean_api_url: str
glean_access_token: str
act_as: str = "langflow-component@datastax.com" # TODO: Detect this
def _prepare_request(
self,
query: str,
page_size: int = 10,
request_options: Optional[Dict[str, Any]] = None,
) -> dict:
# Ensure there's a trailing slash
url = self.glean_api_url
if not url.endswith("/"):
url += "/"
return {
"url": urljoin(url, "search"),
"headers": {
"Authorization": f"Bearer {self.glean_access_token}",
"X-Scio-ActAs": self.act_as,
},
"payload": {
"query": query,
"pageSize": page_size,
"requestOptions": request_options,
},
}
def run(self, query: str, **kwargs: Any) -> str:
results = self.results(query, **kwargs)
return self._result_as_string(results)
def results(self, query: str, **kwargs: Any) -> dict:
results = self._search_api_results(query, **kwargs)
return results
def _search_api_results(self, query: str, **kwargs: Any) -> Dict[str, Any]:
request_details = self._prepare_request(query, **kwargs)
response = httpx.post(
request_details["url"],
json=request_details["payload"],
headers=request_details["headers"],
)
response.raise_for_status()
return response.json()
@staticmethod
def _result_as_string(result: dict) -> str:
return json.dumps(result, indent=4)
return GleanAPIWrapper(glean_api_url=self.glean_api_url, glean_access_token=self.glean_access_token)
Google search API
This component allows you to call the Google Search API.
Parameters
Name | Type | Description |
---|---|---|
google_api_key |
SecretString |
Google API key for authentication |
google_cse_id |
SecretString |
Google Custom Search Engine ID |
input_value |
String |
Search query input |
k |
Integer |
Number of search results to return |
Name | Type | Description |
---|---|---|
results |
List[Data] |
List of search results |
tool |
Tool |
Google Search tool for use in LangChain |
Component code
GoogleSearchAPI.py
from typing import Union
from langchain_core.tools import Tool
from langflow.base.langchain_utilities.model import LCToolComponent
from langflow.inputs import SecretStrInput, MultilineInput, IntInput
from langflow.schema import Data
class GoogleSearchAPIComponent(LCToolComponent):
display_name = "Google Search API"
description = "Call Google Search API."
name = "GoogleSearchAPI"
inputs = [
SecretStrInput(name="google_api_key", display_name="Google API Key", required=True),
SecretStrInput(name="google_cse_id", display_name="Google CSE ID", required=True),
MultilineInput(
name="input_value",
display_name="Input",
),
IntInput(name="k", display_name="Number of results", value=4, required=True),
]
def run_model(self) -> Union[Data, list[Data]]:
wrapper = self._build_wrapper()
results = wrapper.results(query=self.input_value, num_results=self.k)
data = [Data(data=result, text=result["snippet"]) for result in results]
self.status = data
return data
def build_tool(self) -> Tool:
wrapper = self._build_wrapper()
return Tool(
name="google_search",
description="Search Google for recent results.",
func=wrapper.run,
)
def _build_wrapper(self):
try:
from langchain_google_community import GoogleSearchAPIWrapper # type: ignore
except ImportError:
raise ImportError("Please install langchain-google-community to use GoogleSearchAPIWrapper.")
return GoogleSearchAPIWrapper(google_api_key=self.google_api_key, google_cse_id=self.google_cse_id, k=self.k)
Google serper API
This component allows you to call the Serper.dev Google Search API.
Parameters
Name | Type | Description |
---|---|---|
serper_api_key |
SecretString |
API key for Serper.dev authentication |
input_value |
String |
Search query input |
k |
Integer |
Number of search results to return |
Name | Type | Description |
---|---|---|
results |
List[Data] |
List of search results |
tool |
Tool |
Google Serper search tool for use in LangChain |
Component code
GoogleSerperAPI.py
from typing import Union
from langchain_community.utilities.google_serper import GoogleSerperAPIWrapper
from langflow.base.langchain_utilities.model import LCToolComponent
from langflow.inputs import SecretStrInput, MultilineInput, IntInput
from langflow.schema import Data
from langflow.field_typing import Tool
class GoogleSerperAPIComponent(LCToolComponent):
display_name = "Google Serper API"
description = "Call the Serper.dev Google Search API."
name = "GoogleSerperAPI"
inputs = [
SecretStrInput(name="serper_api_key", display_name="Serper API Key", required=True),
MultilineInput(
name="input_value",
display_name="Input",
),
IntInput(name="k", display_name="Number of results", value=4, required=True),
]
def run_model(self) -> Union[Data, list[Data]]:
wrapper = self._build_wrapper()
results = wrapper.results(query=self.input_value)
list_results = results.get("organic", [])
data = [Data(data=result, text=result["snippet"]) for result in list_results]
self.status = data
return data
def build_tool(self) -> Tool:
wrapper = self._build_wrapper()
return Tool(
name="google_search",
description="Search Google for recent results.",
func=wrapper.run,
)
def _build_wrapper(self):
return GoogleSerperAPIWrapper(serper_api_key=self.serper_api_key, k=self.k)
Python code structured tool
This component creates a structured tool from Python code using a dataclass.
The component dynamically updates its configuration based on the provided Python code, allowing for custom function arguments and descriptions.
Parameters
Name | Type | Description |
---|---|---|
tool_code |
String |
Python code for the tool’s dataclass |
tool_name |
String |
Name of the tool |
tool_description |
String |
Description of the tool |
return_direct |
Boolean |
Whether to return the function output directly |
tool_function |
String |
Selected function for the tool |
global_variables |
Dict |
Global variables or data for the tool |
Name | Type | Description |
---|---|---|
result_tool |
Tool |
Structured tool created from the Python code |
Component code
PythonCodeStructuredTool.py
import ast
import json
from typing import Any
from langchain.agents import Tool
from langflow.base.langchain_utilities.model import LCToolComponent
from langflow.inputs.inputs import MultilineInput, MessageTextInput, BoolInput, DropdownInput, HandleInput, FieldTypes
from langchain_core.tools import StructuredTool
from pydantic.v1 import Field, create_model
from pydantic.v1.fields import Undefined
from langflow.io import Output
from langflow.schema import Data
from langflow.schema.dotdict import dotdict
class PythonCodeStructuredTool(LCToolComponent):
DEFAULT_KEYS = [
"code",
"_type",
"text_key",
"tool_code",
"tool_name",
"tool_description",
"return_direct",
"tool_function",
"global_variables",
"_classes",
"_functions",
]
display_name = "Python Code Structured Tool"
description = "structuredtool dataclass code to tool"
documentation = "https://python.langchain.com/docs/modules/tools/custom_tools/#structuredtool-dataclass"
name = "PythonCodeStructuredTool"
icon = "🐍"
field_order = ["name", "description", "tool_code", "return_direct", "tool_function"]
inputs = [
MultilineInput(
name="tool_code",
display_name="Tool Code",
info="Enter the dataclass code.",
placeholder="def my_function(args):\n pass",
required=True,
real_time_refresh=True,
refresh_button=True,
),
MessageTextInput(name="tool_name", display_name="Tool Name", info="Enter the name of the tool.", required=True),
MessageTextInput(
name="tool_description",
display_name="Description",
info="Enter the description of the tool.",
required=True,
),
BoolInput(
name="return_direct",
display_name="Return Directly",
info="Should the tool return the function output directly?",
),
DropdownInput(
name="tool_function",
display_name="Tool Function",
info="Select the function for additional expressions.",
options=[],
required=True,
real_time_refresh=True,
refresh_button=True,
),
HandleInput(
name="global_variables",
display_name="Global Variables",
info="Enter the global variables or Create Data Component.",
input_types=["Data"],
field_type=FieldTypes.DICT,
is_list=True,
),
MessageTextInput(name="_classes", display_name="Classes", advanced=True),
MessageTextInput(name="_functions", display_name="Functions", advanced=True),
]
outputs = [
Output(display_name="Tool", name="result_tool", method="build_tool"),
]
def update_build_config(self, build_config: dotdict, field_value: Any, field_name: str | None = None) -> dotdict:
if field_name is None:
return build_config
if field_name != "tool_code" and field_name != "tool_function":
return build_config
try:
named_functions = {}
[classes, functions] = self._parse_code(build_config["tool_code"]["value"])
existing_fields = {}
if len(build_config) > len(self.DEFAULT_KEYS):
for key in build_config.copy():
if key not in self.DEFAULT_KEYS:
existing_fields[key] = build_config.pop(key)
names = []
for func in functions:
named_functions[func["name"]] = func
names.append(func["name"])
for arg in func["args"]:
field_name = f"{func['name']}|{arg['name']}"
if field_name in existing_fields:
build_config[field_name] = existing_fields[field_name]
continue
field = MessageTextInput(
display_name=f"{arg['name']}: Description",
name=field_name,
info=f"Enter the description for {arg['name']}",
required=True,
)
build_config[field_name] = field.to_dict()
build_config["_functions"]["value"] = json.dumps(named_functions)
build_config["_classes"]["value"] = json.dumps(classes)
build_config["tool_function"]["options"] = names
except Exception as e:
self.status = f"Failed to extract names: {str(e)}"
build_config["tool_function"]["options"] = ["Failed to parse", str(e)]
return build_config
async def build_tool(self) -> Tool:
_local_namespace = {} # type: ignore
modules = self._find_imports(self.tool_code)
import_code = ""
for module in modules["imports"]:
import_code += f"global {module}\nimport {module}\n"
for from_module in modules["from_imports"]:
for alias in from_module.names:
import_code += f"global {alias.name}\n"
import_code += (
f"from {from_module.module} import {', '.join([alias.name for alias in from_module.names])}\n"
)
exec(import_code, globals())
exec(self.tool_code, globals(), _local_namespace)
class PythonCodeToolFunc:
params: dict = {}
def run(**kwargs):
for key in kwargs:
if key not in PythonCodeToolFunc.params:
PythonCodeToolFunc.params[key] = kwargs[key]
return _local_namespace[self.tool_function](**PythonCodeToolFunc.params)
_globals = globals()
_local = {} # type: ignore
_local[self.tool_function] = PythonCodeToolFunc
_globals.update(_local)
if isinstance(self.global_variables, list):
for data in self.global_variables:
if isinstance(data, Data):
_globals.update(data.data)
elif isinstance(self.global_variables, dict):
_globals.update(self.global_variables)
classes = json.loads(self._attributes["_classes"])
for class_dict in classes:
exec("\n".join(class_dict["code"]), _globals)
named_functions = json.loads(self._attributes["_functions"])
schema_fields = {}
for attr in self._attributes:
if attr in self.DEFAULT_KEYS:
continue
func_name = attr.split("|")[0]
field_name = attr.split("|")[1]
func_arg = self._find_arg(named_functions, func_name, field_name)
if func_arg is None:
raise Exception(f"Failed to find arg: {field_name}")
field_annotation = func_arg["annotation"]
field_description = self._get_value(self._attributes[attr], str)
if field_annotation:
exec(f"temp_annotation_type = {field_annotation}", _globals)
schema_annotation = _globals["temp_annotation_type"]
else:
schema_annotation = Any
schema_fields[field_name] = (
schema_annotation,
Field(
default=func_arg["default"] if "default" in func_arg else Undefined, description=field_description
),
)
if "temp_annotation_type" in _globals:
_globals.pop("temp_annotation_type")
PythonCodeToolSchema = None
if schema_fields:
PythonCodeToolSchema = create_model("PythonCodeToolSchema", **schema_fields) # type: ignore
tool = StructuredTool.from_function(
func=_local[self.tool_function].run,
args_schema=PythonCodeToolSchema,
name=self.tool_name,
description=self.tool_description,
return_direct=self.return_direct,
)
return tool # type: ignore
def post_code_processing(self, new_frontend_node: dict, current_frontend_node: dict):
"""
This function is called after the code validation is done.
"""
frontend_node = super().post_code_processing(new_frontend_node, current_frontend_node)
frontend_node["template"] = self.update_build_config(
frontend_node["template"], frontend_node["template"]["tool_code"]["value"], "tool_code"
)
frontend_node = super().post_code_processing(new_frontend_node, current_frontend_node)
for key in frontend_node["template"]:
if key in self.DEFAULT_KEYS:
continue
frontend_node["template"] = self.update_build_config(
frontend_node["template"], frontend_node["template"][key]["value"], key
)
frontend_node = super().post_code_processing(new_frontend_node, current_frontend_node)
return frontend_node
def _parse_code(self, code: str) -> tuple[list[dict], list[dict]]:
parsed_code = ast.parse(code)
lines = code.split("\n")
classes = []
functions = []
for node in parsed_code.body:
if isinstance(node, ast.ClassDef):
class_lines = lines[node.lineno - 1 : node.end_lineno]
class_lines[-1] = class_lines[-1][: node.end_col_offset]
class_lines[0] = class_lines[0][node.col_offset :]
classes.append(
{
"name": node.name,
"code": class_lines,
}
)
continue
if not isinstance(node, ast.FunctionDef):
continue
func = {"name": node.name, "args": []}
for arg in node.args.args:
if arg.lineno != arg.end_lineno:
raise Exception("Multiline arguments are not supported")
func_arg = {
"name": arg.arg,
"annotation": None,
}
for default in node.args.defaults:
if (
arg.lineno > default.lineno
or arg.col_offset > default.col_offset
or (
arg.end_lineno is not None
and default.end_lineno is not None
and arg.end_lineno < default.end_lineno
)
or (
arg.end_col_offset is not None
and default.end_col_offset is not None
and arg.end_col_offset < default.end_col_offset
)
):
continue
if isinstance(default, ast.Name):
func_arg["default"] = default.id
elif isinstance(default, ast.Constant):
func_arg["default"] = default.value
if arg.annotation:
annotation_line = lines[arg.annotation.lineno - 1]
annotation_line = annotation_line[: arg.annotation.end_col_offset]
annotation_line = annotation_line[arg.annotation.col_offset :]
func_arg["annotation"] = annotation_line
if isinstance(func_arg["annotation"], str) and func_arg["annotation"].count("=") > 0:
func_arg["annotation"] = "=".join(func_arg["annotation"].split("=")[:-1]).strip()
if isinstance(func["args"], list):
func["args"].append(func_arg)
functions.append(func)
return classes, functions
def _find_imports(self, code: str) -> dotdict:
imports = []
from_imports = []
parsed_code = ast.parse(code)
for node in parsed_code.body:
if isinstance(node, ast.Import):
for alias in node.names:
imports.append(alias.name)
elif isinstance(node, ast.ImportFrom):
from_imports.append(node)
return dotdict({"imports": imports, "from_imports": from_imports})
def _get_value(self, value: Any, annotation: Any) -> Any:
return value if isinstance(value, annotation) else value["value"]
def _find_arg(self, named_functions: dict, func_name: str, arg_name: str) -> dict | None:
for arg in named_functions[func_name]["args"]:
if arg["name"] == arg_name:
return arg
return None
Python REPL Tool
This component creates a Python REPL (Read-Eval-Print Loop) tool for executing Python code.
Parameters
Name | Type | Description |
---|---|---|
name |
String |
The name of the tool (default: "python_repl") |
description |
String |
A description of the tool’s functionality |
global_imports |
List[String] |
List of modules to import globally (default: ["math"]) |
Name | Type | Description |
---|---|---|
tool |
Tool |
Python REPL tool for use in LangChain |
Component code
PythonREPLTool.py
import importlib
from langchain_experimental.utilities import PythonREPL
from langflow.base.tools.base import build_status_from_tool
from langflow.custom import CustomComponent
from langchain_core.tools import Tool
class PythonREPLToolComponent(CustomComponent):
display_name = "Python REPL Tool"
description = "A tool for running Python code in a REPL environment."
name = "PythonREPLTool"
def build_config(self):
return {
"name": {"display_name": "Name", "info": "The name of the tool."},
"description": {"display_name": "Description", "info": "A description of the tool."},
"global_imports": {
"display_name": "Global Imports",
"info": "A list of modules to import globally, e.g. ['math', 'numpy'].",
},
}
def get_globals(self, globals: list[str]) -> dict:
"""
Retrieves the global variables from the specified modules.
Args:
globals (list[str]): A list of module names.
Returns:
dict: A dictionary containing the global variables from the specified modules.
"""
global_dict = {}
for module in globals:
try:
imported_module = importlib.import_module(module)
global_dict[imported_module.__name__] = imported_module
except ImportError:
raise ImportError(f"Could not import module {module}")
return global_dict
def build(
self,
name: str = "python_repl",
description: str = "A Python shell. Use this to execute python commands. Input should be a valid python command. If you want to see the output of a value, you should print it out with `print(...)`.",
global_imports: list[str] = ["math"],
) -> Tool:
"""
Builds a Python REPL tool.
Args:
name (str, optional): The name of the tool. Defaults to "python_repl".
description (str, optional): The description of the tool. Defaults to "A Python shell. Use this to execute python commands. Input should be a valid python command. If you want to see the output of a value, you should print it out with `print(...)`. ".
global_imports (list[str], optional): A list of global imports to be available in the Python REPL. Defaults to ["math"].
Returns:
Tool: The built Python REPL tool.
"""
_globals = self.get_globals(global_imports)
python_repl = PythonREPL(_globals=_globals)
tool = Tool(
name=name,
description=description,
func=python_repl.run,
)
self.status = build_status_from_tool(tool)
return tool
Retriever Tool
This component creates a tool for interacting with a retriever in LangChain.
Parameters
Name | Type | Description |
---|---|---|
retriever |
BaseRetriever |
The retriever to interact with |
name |
String |
The name of the tool |
description |
String |
A description of the tool’s functionality |
Name | Type | Description |
---|---|---|
tool |
Tool |
Retriever tool for use in LangChain |
Component code
RetrieverTool.py
from langchain_core.tools import create_retriever_tool
from langflow.custom import CustomComponent
from langflow.field_typing import BaseRetriever, Tool
class RetrieverToolComponent(CustomComponent):
display_name = "RetrieverTool"
description = "Tool for interacting with retriever"
name = "RetrieverTool"
def build_config(self):
return {
"retriever": {
"display_name": "Retriever",
"info": "Retriever to interact with",
"type": BaseRetriever,
"input_types": ["Retriever"],
},
"name": {"display_name": "Name", "info": "Name of the tool"},
"description": {"display_name": "Description", "info": "Description of the tool"},
}
def build(
self,
retriever: BaseRetriever,
name: str,
description: str,
) -> Tool:
return create_retriever_tool(
retriever=retriever,
name=name,
description=description,
)
SearXNG Search Tool
This component creates a tool for searching using SearXNG, a metasearch engine.
Parameters
Name | Type | Description |
---|---|---|
url |
String |
The URL of the SearXNG instance |
max_results |
Integer |
Maximum number of results to return |
categories |
List[String] |
Categories to search in |
language |
String |
Language for the search results |
Name | Type | Description |
---|---|---|
result_tool |
Tool |
SearXNG search tool for use in LangChain |
Component code
SearXNGTool.py
from typing import Any
import requests
import json
from pydantic.v1 import Field, create_model
from langchain.agents import Tool
from langchain_core.tools import StructuredTool
from langflow.base.langchain_utilities.model import LCToolComponent
from langflow.inputs import MessageTextInput, MultiselectInput, DropdownInput, IntInput
from langflow.schema.dotdict import dotdict
from langflow.io import Output
class SearXNGToolComponent(LCToolComponent):
search_headers: dict = {}
display_name = "SearXNG Search Tool"
description = "A component that searches for tools using SearXNG."
name = "SearXNGTool"
inputs = [
MessageTextInput(
name="url",
display_name="URL",
value="http://localhost",
required=True,
refresh_button=True,
),
IntInput(
name="max_results",
display_name="Max Results",
value=10,
required=True,
),
MultiselectInput(
name="categories",
display_name="Categories",
options=[],
value=[],
),
DropdownInput(
name="language",
display_name="Language",
options=[],
),
]
outputs = [
Output(display_name="Tool", name="result_tool", method="build_tool"),
]
def update_build_config(self, build_config: dotdict, field_value: Any, field_name: str | None = None) -> dotdict:
if field_name is None:
return build_config
if field_name != "url":
return build_config
try:
url = f"{field_value}/config"
response = requests.get(url=url, headers=self.search_headers.copy())
data = None
if response.headers.get("Content-Encoding") == "zstd":
data = json.loads(response.content)
else:
data = response.json()
build_config["categories"]["options"] = data["categories"].copy()
for selected_category in build_config["categories"]["value"]:
if selected_category not in build_config["categories"]["options"]:
build_config["categories"]["value"].remove(selected_category)
languages = []
for language in data["locales"]:
languages.append(language)
build_config["language"]["options"] = languages.copy()
except Exception as e:
self.status = f"Failed to extract names: {str(e)}"
build_config["categories"]["options"] = ["Failed to parse", str(e)]
return build_config
def build_tool(self) -> Tool:
class SearxSearch:
_url: str = ""
_categories: list[str] = []
_language: str = ""
_headers: dict = {}
_max_results: int = 10
@staticmethod
def search(query: str, categories: list[str] = []) -> list:
if not SearxSearch._categories and not categories:
raise ValueError("No categories provided.")
all_categories = SearxSearch._categories + list(set(categories) - set(SearxSearch._categories))
try:
url = f"{SearxSearch._url}/"
headers = SearxSearch._headers.copy()
response = requests.get(
url=url,
headers=headers,
params={
"q": query,
"categories": ",".join(all_categories),
"language": SearxSearch._language,
"format": "json",
},
).json()
results = []
num_results = min(SearxSearch._max_results, len(response["results"]))
for i in range(num_results):
results.append(response["results"][i])
return results
except Exception as e:
return [f"Failed to search: {str(e)}"]
SearxSearch._url = self.url
SearxSearch._categories = self.categories.copy()
SearxSearch._language = self.language
SearxSearch._headers = self.search_headers.copy()
SearxSearch._max_results = self.max_results
_globals = globals()
_local = {}
_local["SearxSearch"] = SearxSearch
_globals.update(_local)
schema_fields = {
"query": (str, Field(..., description="The query to search for.")),
"categories": (list[str], Field(default=[], description="The categories to search in.")),
}
SearxSearchSchema = create_model("SearxSearchSchema", **schema_fields) # type: ignore
tool = StructuredTool.from_function(
func=_local["SearxSearch"].search,
args_schema=SearxSearchSchema,
name="searxng_search_tool",
description="A tool that searches for tools using SearXNG.\nThe available categories are: "
+ ", ".join(self.categories),
)
return tool
Search API
This component calls the searchapi.io
API. It can be used to search the web for information.
For more information, see the SearchAPI documentation.
Parameters
Name | Display Name | Info |
---|---|---|
engine |
Engine |
The search engine to use (default: "google") |
api_key |
SearchAPI API Key |
The API key for authenticating with SearchAPI |
input_value |
Input |
The search query or input for the API call |
search_params |
Search parameters |
Additional parameters for customizing the search |
Name | Display Name | Info |
---|---|---|
data |
Search Results |
List of Data objects containing search results |
tool |
Search API Tool |
A Tool object for use in LangChain workflows |
Component code
SearchAPI.py
from typing import Union
from langchain_community.utilities.searchapi import SearchApiAPIWrapper
from langflow.base.langchain_utilities.model import LCToolComponent
from langflow.inputs import SecretStrInput, MultilineInput, DictInput, MessageTextInput
from langflow.schema import Data
from langflow.field_typing import Tool
class SearchAPIComponent(LCToolComponent):
display_name: str = "Search API"
description: str = "Call the searchapi.io API"
name = "SearchAPI"
documentation: str = "https://www.searchapi.io/docs/google"
inputs = [
MessageTextInput(name="engine", display_name="Engine", value="google"),
SecretStrInput(name="api_key", display_name="SearchAPI API Key", required=True),
MultilineInput(
name="input_value",
display_name="Input",
),
DictInput(name="search_params", display_name="Search parameters", advanced=True, is_list=True),
]
def run_model(self) -> Union[Data, list[Data]]:
wrapper = self._build_wrapper()
results = wrapper.results(query=self.input_value, **(self.search_params or {}))
list_results = results.get("organic_results", [])
data = [Data(data=result, text=result["snippet"]) for result in list_results]
self.status = data
return data
def build_tool(self) -> Tool:
wrapper = self._build_wrapper()
return Tool(
name="search_api",
description="Search for recent results.",
func=lambda x: wrapper.run(query=x, **(self.search_params or {})),
)
def _build_wrapper(self):
return SearchApiAPIWrapper(engine=self.engine, searchapi_api_key=self.api_key)
Serp Search API
This component creates a tool for searching using the Serp API.
Parameters
Name | Type | Description |
---|---|---|
serpapi_api_key |
SecretString |
API key for Serp API authentication |
input_value |
String |
Search query input |
search_params |
Dict |
Additional search parameters (optional) |
Name | Type | Description |
---|---|---|
results |
List[Data] |
List of search results |
tool |
Tool |
Serp API search tool for use in LangChain |
Component code
SerpAPI.py
from langchain_community.utilities.serpapi import SerpAPIWrapper
from langflow.base.langchain_utilities.model import LCToolComponent
from langflow.inputs import SecretStrInput, DictInput, MultilineInput
from langflow.schema import Data
from langflow.field_typing import Tool
class SerpAPIComponent(LCToolComponent):
display_name = "Serp Search API"
description = "Call Serp Search API"
name = "SerpAPI"
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),
]
def run_model(self) -> list[Data]:
wrapper = self._build_wrapper()
results = wrapper.results(self.input_value)
list_results = results.get("organic_results", [])
data = [Data(data=result, text=result["snippet"]) for result in list_results]
self.status = data
return data
def build_tool(self) -> Tool:
wrapper = self._build_wrapper()
return Tool(name="search_api", description="Search for recent results.", func=wrapper.run)
def _build_wrapper(self) -> SerpAPIWrapper:
if self.search_params:
return SerpAPIWrapper( # type: ignore
serpapi_api_key=self.serpapi_api_key,
params=self.search_params,
)
return SerpAPIWrapper( # type: ignore
serpapi_api_key=self.serpapi_api_key
)
Wikipedia API
This component creates a tool for searching and retrieving information from Wikipedia.
Parameters
Name | Type | Description |
---|---|---|
input_value |
String |
Search query input |
lang |
String |
Language code for Wikipedia (default: "en") |
k |
Integer |
Number of results to return |
load_all_available_meta |
Boolean |
Whether to load all available metadata (advanced) |
doc_content_chars_max |
Integer |
Maximum number of characters for document content (advanced) |
Name | Type | Description |
---|---|---|
results |
List[Data] |
List of Wikipedia search results |
tool |
Tool |
Wikipedia search tool for use in LangChain |
Component code
WikipediaAPI.py
from typing import cast
from langchain_community.tools import WikipediaQueryRun
from langchain_community.utilities.wikipedia import WikipediaAPIWrapper
from langflow.base.langchain_utilities.model import LCToolComponent
from langflow.field_typing import Tool
from langflow.inputs import BoolInput, IntInput, MessageTextInput, MultilineInput
from langflow.schema import Data
class WikipediaAPIComponent(LCToolComponent):
display_name = "Wikipedia API"
description = "Call Wikipedia API."
name = "WikipediaAPI"
inputs = [
MultilineInput(
name="input_value",
display_name="Input",
),
MessageTextInput(name="lang", display_name="Language", value="en"),
IntInput(name="k", display_name="Number of results", value=4, required=True),
BoolInput(name="load_all_available_meta", display_name="Load all available meta", value=False, advanced=True),
IntInput(
name="doc_content_chars_max", display_name="Document content characters max", value=4000, advanced=True
),
]
def run_model(self) -> list[Data]:
wrapper = self._build_wrapper()
docs = wrapper.load(self.input_value)
data = [Data.from_document(doc) for doc in docs]
self.status = data
return data
def build_tool(self) -> Tool:
wrapper = self._build_wrapper()
return cast(Tool, WikipediaQueryRun(api_wrapper=wrapper))
def _build_wrapper(self) -> WikipediaAPIWrapper:
return WikipediaAPIWrapper( # type: ignore
top_k_results=self.k,
lang=self.lang,
load_all_available_meta=self.load_all_available_meta,
doc_content_chars_max=self.doc_content_chars_max,
)
Wolfram Alpha API
This component creates a tool for querying the Wolfram Alpha API.
Parameters
Name | Type | Description |
---|---|---|
input_value |
String |
Query input for Wolfram Alpha |
app_id |
SecretString |
Wolfram Alpha API App ID |
Name | Type | Description |
---|---|---|
results |
List[Data] |
List containing the Wolfram Alpha API response |
tool |
Tool |
Wolfram Alpha API tool for use in LangChain |
Component code
WolframAlphaAPI.py
from langchain_community.utilities.wolfram_alpha import WolframAlphaAPIWrapper
from langflow.base.langchain_utilities.model import LCToolComponent
from langflow.field_typing import Tool
from langflow.inputs import MultilineInput, SecretStrInput
from langflow.schema import Data
class WolframAlphaAPIComponent(LCToolComponent):
display_name = "WolframAlphaAPI"
description = "Call Wolfram Alpha API."
name = "WolframAlphaAPI"
inputs = [
MultilineInput(
name="input_value",
display_name="Input",
),
SecretStrInput(name="app_id", display_name="App ID", required=True),
]
def run_model(self) -> list[Data]:
wrapper = self._build_wrapper()
result_str = wrapper.run(self.input_value)
data = [Data(text=result_str)]
self.status = data
return data
def build_tool(self) -> Tool:
wrapper = self._build_wrapper()
return Tool(name="wolfram_alpha_api", description="Answers mathematical questions.", func=wrapper.run)
def _build_wrapper(self) -> WolframAlphaAPIWrapper:
return WolframAlphaAPIWrapper(wolfram_alpha_appid=self.app_id) # type: ignore
Yahoo Finance News Tool
This component creates a tool for retrieving news from Yahoo Finance.
Parameters
This component does not have any input parameters.
Name | Type | Description |
---|---|---|
tool |
Tool |
Yahoo Finance News tool for use in LangChain |
Component code
YfinanceTool.py
from typing import cast
from langchain_community.tools.yahoo_finance_news import YahooFinanceNewsTool
from langflow.custom import Component
from langflow.field_typing import Tool
from langflow.io import Output
class YfinanceToolComponent(Component):
display_name = "Yahoo Finance News Tool"
description = "Tool for interacting with Yahoo Finance News."
name = "YFinanceTool"
outputs = [
Output(display_name="Tool", name="tool", method="build_tool"),
]
def build_tool(self) -> Tool:
return cast(Tool, YahooFinanceNewsTool())