Utilities#

General utilities.

pydantic model langchain.utilities.ApifyWrapper[source]#

Wrapper around Apify.

To use, you should have the apify-client python package installed, and the environment variable APIFY_API_TOKEN set with your API key, or pass apify_api_token as a named parameter to the constructor.

field apify_client: Any = None#

field apify_client_async: Any = None#

async acall_actor(actor_id: str, run_input: Dict, dataset_mapping_function: Callable[[Dict], langchain.schema.Document], *, build: Optional[str] = None, memory_mbytes: Optional[int] = None, timeout_secs: Optional[int] = None) → langchain.document_loaders.apify_dataset.ApifyDatasetLoader[source]#

Run an Actor on the Apify platform and wait for results to be ready.

Parameters

actor_id (str) – The ID or name of the Actor on the Apify platform.
run_input (Dict) – The input object of the Actor that you’re trying to run.
dataset_mapping_function (Callable) – A function that takes a single dictionary (an Apify dataset item) and converts it to an instance of the Document class.
build (str, optional) – Optionally specifies the actor build to run. It can be either a build tag or build number.
memory_mbytes (int, optional) – Optional memory limit for the run, in megabytes.
timeout_secs (int, optional) – Optional timeout for the run, in seconds.

Returns

A loader that will fetch the records from the: Actor run’s default dataset.

Return type

ApifyDatasetLoader

call_actor(actor_id: str, run_input: Dict, dataset_mapping_function: Callable[[Dict], langchain.schema.Document], *, build: Optional[str] = None, memory_mbytes: Optional[int] = None, timeout_secs: Optional[int] = None) → langchain.document_loaders.apify_dataset.ApifyDatasetLoader[source]#

Run an Actor on the Apify platform and wait for results to be ready.

Parameters

actor_id (str) – The ID or name of the Actor on the Apify platform.
run_input (Dict) – The input object of the Actor that you’re trying to run.
dataset_mapping_function (Callable) – A function that takes a single dictionary (an Apify dataset item) and converts it to an instance of the Document class.
build (str, optional) – Optionally specifies the actor build to run. It can be either a build tag or build number.
memory_mbytes (int, optional) – Optional memory limit for the run, in megabytes.
timeout_secs (int, optional) – Optional timeout for the run, in seconds.

Returns

A loader that will fetch the records from the: Actor run’s default dataset.

Return type

ApifyDatasetLoader

pydantic model langchain.utilities.ArxivAPIWrapper[source]#

Wrapper around ArxivAPI.

To use, you should have the arxiv python package installed. https://lukasschwab.me/arxiv.py/index.html This wrapper will use the Arxiv API to conduct searches and fetch document summaries. By default, it will return the document summaries of the top-k results of an input search.

field arxiv_exceptions: Any = None#

field top_k_results: int = 3#

run(query: str) → str[source]#: Run Arxiv search and get the document meta information. See https://lukasschwab.me/arxiv.py/index.html#Search See https://lukasschwab.me/arxiv.py/index.html#Result It uses only the most informative fields of document meta information.

class langchain.utilities.BashProcess(strip_newlines: bool = False, return_err_output: bool = False)[source]#

Executes bash commands and returns the output.

run(commands: Union[str, List[str]]) → str[source]#: Run commands and return final output.

pydantic model langchain.utilities.BingSearchAPIWrapper[source]#

Wrapper for Bing Search API.

In order to set this up, follow instructions at: https://levelup.gitconnected.com/api-tutorial-how-to-use-bing-web-search-api-in-python-4165d5592a7e

field bing_search_url: str [Required]#

field bing_subscription_key: str [Required]#

field k: int = 10#

results(query: str, num_results: int) → List[Dict][source]#

Run query through BingSearch and return metadata.

Parameters

query – The query to search for.
num_results – The number of results to return.

Returns

snippet - The description of the result. title - The title of the result. link - The link to the result.

Return type

A list of dictionaries with the following keys

run(query: str) → str[source]#: Run query through BingSearch and parse result.

pydantic model langchain.utilities.GooglePlacesAPIWrapper[source]#

Wrapper around Google Places API.

To use, you should have the googlemaps python package installed,: an API key for the google maps platform, and the enviroment variable ‘’GPLACES_API_KEY’’ set with your API key , or pass ‘gplaces_api_key’ as a named parameter to the constructor.
By default, this will return the all the results on the input query.: You can use the top_k_results argument to limit the number of results.

Example

from langchain import GooglePlacesAPIWrapper
gplaceapi = GooglePlacesAPIWrapper()

field gplaces_api_key: Optional[str] = None#

field top_k_results: Optional[int] = None#

fetch_place_details(place_id: str) → Optional[str][source]#

format_place_details(place_details: Dict[str, Any]) → Optional[str][source]#

run(query: str) → str[source]#: Run Places search and get k number of places that exists that match.

pydantic model langchain.utilities.GoogleSearchAPIWrapper[source]#

Wrapper for Google Search API.

Adapted from: Instructions adapted from https://stackoverflow.com/questions/ 37083058/ programmatically-searching-google-in-python-using-custom-search

TODO: DOCS for using it 1. Install google-api-python-client - If you don’t already have a Google account, sign up. - If you have never created a Google APIs Console project, read the Managing Projects page and create a project in the Google API Console. - Install the library using pip install google-api-python-client The current version of the library is 2.70.0 at this time

2. To create an API key: - Navigate to the APIs & Services→Credentials panel in Cloud Console. - Select Create credentials, then select API key from the drop-down menu. - The API key created dialog box displays your newly created key. - You now have an API_KEY

3. Setup Custom Search Engine so you can search the entire web - Create a custom search engine in this link. - In Sites to search, add any valid URL (i.e. www.stackoverflow.com). - That’s all you have to fill up, the rest doesn’t matter. In the left-side menu, click Edit search engine → {your search engine name} → Setup Set Search the entire web to ON. Remove the URL you added from

the list of Sites to search.

Under Search engine ID you’ll find the search-engine-ID.

4. Enable the Custom Search API - Navigate to the APIs & Services→Dashboard panel in Cloud Console. - Click Enable APIs and Services. - Search for Custom Search API and click on it. - Click Enable. URL for it: https://console.cloud.google.com/apis/library/customsearch.googleapis .com

field google_api_key: Optional[str] = None#

field google_cse_id: Optional[str] = None#

field k: int = 10#

field siterestrict: bool = False#

results(query: str, num_results: int) → List[Dict][source]#

Run query through GoogleSearch and return metadata.

Parameters

query – The query to search for.
num_results – The number of results to return.

Returns

snippet - The description of the result. title - The title of the result. link - The link to the result.

Return type

A list of dictionaries with the following keys

run(query: str) → str[source]#: Run query through GoogleSearch and parse result.

pydantic model langchain.utilities.GoogleSerperAPIWrapper[source]#

Wrapper around the Serper.dev Google Search API.

You can create a free API key at https://serper.dev.

To use, you should have the environment variable SERPER_API_KEY set with your API key, or pass serper_api_key as a named parameter to the constructor.

Example

from langchain import GoogleSerperAPIWrapper
google_serper = GoogleSerperAPIWrapper()

field gl: str = 'us'#

field hl: str = 'en'#

field k: int = 10#

field serper_api_key: Optional[str] = None#

run(query: str) → str[source]#: Run query through GoogleSearch and parse result.

pydantic model langchain.utilities.OpenWeatherMapAPIWrapper[source]#

Wrapper for OpenWeatherMap API using PyOWM.

Docs for using:

Go to OpenWeatherMap and sign up for an API key
Save your API KEY into OPENWEATHERMAP_API_KEY env variable
pip install pyowm

field openweathermap_api_key: Optional[str] = None#

field owm: Any = None#

run(location: str) → str[source]#: Get the current weather information for a specified location.

pydantic model langchain.utilities.PowerBIDataset[source]#

Create PowerBI engine from dataset ID and credential or token.

Use either the credential or a supplied token to authenticate. If both are supplied the credential is used to generate a token. The impersonated_user_name is the UPN of a user to be impersonated. If the model is not RLS enabled, this will be ignored.

field aiosession: Optional[aiohttp.ClientSession] = None#

field credential: Optional[Union[ChainedTokenCredential, InteractiveCredential]] = None#

field dataset_id: str [Required]#

field group_id: Optional[str] = None#

field impersonated_user_name: Optional[str] = None#

field sample_rows_in_table_info: int = 1#

Constraints

exclusiveMinimum = 0
maximum = 10

field schemas: Dict[str, str] [Optional]#

field table_names: List[str] [Required]#

field token: Optional[str] = None#

async aget_table_info(table_names: Optional[Union[List[str], str]] = None) → str[source]#: Get information about specified tables.

async arun(command: str) → Any[source]#: Execute a DAX command and return the result asynchronously.

get_schemas() → str[source]#: Get the available schema’s.

get_table_info(table_names: Optional[Union[List[str], str]] = None) → str[source]#: Get information about specified tables.

get_table_names() → Iterable[str][source]#: Get names of tables available.

run(command: str) → Any[source]#: Execute a DAX command and return a json representing the results.

property headers: Dict[str, str]#: Get the token.

property request_url: str#: Get the request url.

property table_info: str#: Information about all tables in the database.

pydantic model langchain.utilities.PythonREPL[source]#

Simulates a standalone Python REPL.

field globals: Optional[Dict] [Optional] (alias '_globals')#

field locals: Optional[Dict] [Optional] (alias '_locals')#

run(command: str) → str[source]#: Run command with own globals/locals and returns anything printed.

pydantic model langchain.utilities.SearxSearchWrapper[source]#

Wrapper for Searx API.

To use you need to provide the searx host by passing the named parameter searx_host or exporting the environment variable SEARX_HOST.

In some situations you might want to disable SSL verification, for example if you are running searx locally. You can do this by passing the named parameter unsecure. You can also pass the host url scheme as http to disable SSL.

Example

from langchain.utilities import SearxSearchWrapper
searx = SearxSearchWrapper(searx_host="http://localhost:8888")

Example with SSL disabled:

from langchain.utilities import SearxSearchWrapper
# note the unsecure parameter is not needed if you pass the url scheme as
# http
searx = SearxSearchWrapper(searx_host="http://localhost:8888",
                                        unsecure=True)

Validators

disable_ssl_warnings » unsecure
validate_params » all fields

field aiosession: Optional[Any] = None#

field categories: Optional[List[str]] = []#

field engines: Optional[List[str]] = []#

field headers: Optional[dict] = None#

field k: int = 10#

field params: dict [Optional]#

field query_suffix: Optional[str] = ''#

field searx_host: str = ''#

field unsecure: bool = False#

async aresults(query: str, num_results: int, engines: Optional[List[str]] = None, query_suffix: Optional[str] = '', **kwargs: Any) → List[Dict][source]#

Asynchronously query with json results.

Uses aiohttp. See results for more info.

async arun(query: str, engines: Optional[List[str]] = None, query_suffix: Optional[str] = '', **kwargs: Any) → str[source]#: Asynchronously version of run.

results(query: str, num_results: int, engines: Optional[List[str]] = None, categories: Optional[List[str]] = None, query_suffix: Optional[str] = '', **kwargs: Any) → List[Dict][source]#

Run query through Searx API and returns the results with metadata.

Parameters

query – The query to search for.
query_suffix – Extra suffix appended to the query.
num_results – Limit the number of results to return.
engines – List of engines to use for the query.
categories – List of categories to use for the query.
**kwargs – extra parameters to pass to the searx API.

Returns

{

snippet: The description of the result.

title: The title of the result.

link: The link to the result.

engines: The engines used for the result.

category: Searx category of the result.

}

Return type

Dict with the following keys

run(query: str, engines: Optional[List[str]] = None, categories: Optional[List[str]] = None, query_suffix: Optional[str] = '', **kwargs: Any) → str[source]#

Run query through Searx API and parse results.

You can pass any other params to the searx query API.

Parameters

query – The query to search for.
query_suffix – Extra suffix appended to the query.
engines – List of engines to use for the query.
categories – List of categories to use for the query.
**kwargs – extra parameters to pass to the searx API.

Returns

The result of the query.

Return type

str

Raises

ValueError – If an error occured with the query.

Example

This will make a query to the qwant engine:

from langchain.utilities import SearxSearchWrapper
searx = SearxSearchWrapper(searx_host="http://my.searx.host")
searx.run("what is the weather in France ?", engine="qwant")

# the same result can be achieved using the `!` syntax of searx
# to select the engine using `query_suffix`
searx.run("what is the weather in France ?", query_suffix="!qwant")

pydantic model langchain.utilities.SerpAPIWrapper[source]#

Wrapper around SerpAPI.

To use, you should have the google-search-results python package installed, and the environment variable SERPAPI_API_KEY set with your API key, or pass serpapi_api_key as a named parameter to the constructor.

Example

from langchain import SerpAPIWrapper
serpapi = SerpAPIWrapper()

field aiosession: Optional[aiohttp.client.ClientSession] = None#

field params: dict = {'engine': 'google', 'gl': 'us', 'google_domain': 'google.com', 'hl': 'en'}#

field serpapi_api_key: Optional[str] = None#

async arun(query: str) → str[source]#: Use aiohttp to run query through SerpAPI and parse result.

get_params(query: str) → Dict[str, str][source]#: Get parameters for SerpAPI.

results(query: str) → dict[source]#: Run query through SerpAPI and return the raw result.

run(query: str) → str[source]#: Run query through SerpAPI and parse result.

pydantic model langchain.utilities.TextRequestsWrapper[source]#

Lightweight wrapper around requests library.

The main purpose of this wrapper is to always return a text output.

field aiosession: Optional[aiohttp.client.ClientSession] = None#

field headers: Optional[Dict[str, str]] = None#

async adelete(url: str, **kwargs: Any) → str[source]#: DELETE the URL and return the text asynchronously.

async aget(url: str, **kwargs: Any) → str[source]#: GET the URL and return the text asynchronously.

async apatch(url: str, data: Dict[str, Any], **kwargs: Any) → str[source]#: PATCH the URL and return the text asynchronously.

async apost(url: str, data: Dict[str, Any], **kwargs: Any) → str[source]#: POST to the URL and return the text asynchronously.

async aput(url: str, data: Dict[str, Any], **kwargs: Any) → str[source]#: PUT the URL and return the text asynchronously.

delete(url: str, **kwargs: Any) → str[source]#: DELETE the URL and return the text.

get(url: str, **kwargs: Any) → str[source]#: GET the URL and return the text.

patch(url: str, data: Dict[str, Any], **kwargs: Any) → str[source]#: PATCH the URL and return the text.

post(url: str, data: Dict[str, Any], **kwargs: Any) → str[source]#: POST to the URL and return the text.

put(url: str, data: Dict[str, Any], **kwargs: Any) → str[source]#: PUT the URL and return the text.

property requests: langchain.requests.Requests#

pydantic model langchain.utilities.WikipediaAPIWrapper[source]#

Wrapper around WikipediaAPI.

To use, you should have the wikipedia python package installed. This wrapper will use the Wikipedia API to conduct searches and fetch page summaries. By default, it will return the page summaries of the top-k results of an input search.

field lang: str = 'en'#

field top_k_results: int = 3#

fetch_formatted_page_summary(page: str) → Optional[str][source]#

run(query: str) → str[source]#: Run Wikipedia search and get page summaries.

pydantic model langchain.utilities.WolframAlphaAPIWrapper[source]#

Wrapper for Wolfram Alpha.

Docs for using:

Go to wolfram alpha and sign up for a developer account
Create an app and get your APP ID
Save your APP ID into WOLFRAM_ALPHA_APPID env variable
pip install wolframalpha

field wolfram_alpha_appid: Optional[str] = None#

run(query: str) → str[source]#: Run query through WolframAlpha and parse result.