In this blog, we will learn to build personalized custom tools for agents. In the previous blog, I discussed building a Langchain agent with tools, this blog is a continuation of the previous one. In case you have missed the previous blog you can find it here.
Having a custom tool gives us the flexibility to fine-tune the task as per our choice. Today we will build an Agent who will research a topic, write a summary on the topic, and save it to a file. The Agent will use custom tools to perform the 3 tasks:
Search the topic using Google APIs.
Scrape website content and summarise it according to the user's question.
Store the summarised result in the file.
Building a Custom Google Search Tool
For performing a Google search on a user topic we will be using Custom Search JSON API. For using this API we would need an API key from Google API service and an API key from Custom Search JSON API.
In our custom tool, we will pass a user query as an argument and that query will be used in the Google search. Google will respond to us with the search result which will contain the search items. Search items are JSON object that contains information like page titles, page links, metadata, etc. Following is the screenshot of the Google Search response:
From the API response, we will extract the page URLs, below is the code of our Custom Google Search Tool.
import json
import requests
GOOGLE_API_KEY = "<Your Google API Key>"
GOOGLE_CSE_ID = "<Your Custom Google Search Key>"
def get_research_urls(query):
search_url = build_googlesearch_url(query, GOOGLE_CSE_ID, 10, GOOGLE_API_KEY)
search_response = google_search_response(search_url)
research_urls = []
for item in search_response['items']:
research_urls.append(item['link'])
return research_urls
def build_googlesearch_url(q, cx, num, key):
q = q.replace(" ", "+")
base_url = "https://customsearch.googleapis.com/customsearch/v1"
url_params = f"?q={q}&cx={cx}&num={num}&key={key}&alt=json"
full_url = base_url + url_params
return full_url
def google_search_response(url):
try:
response = requests.get(url)
response.raise_for_status() # Raise an exception for 4xx and 5xx status codes
json_data = response.json()
return json_data
except requests.exceptions.RequestException as e:
print(f"Error fetching data: {e}")
return None
except ValueError as ve:
print(f"Error parsing JSON data: {ve}")
return None
We have defined 3 functions here, the main function is get_research_urls which internally calls build_googlesearch_url and google_search_response. As the name suggests, the build_googlesearch_url function will create the Google search URL. The google_search_response function will call the search endpoint and fetch the response. And finally, the get_research_urls function will extract the URLs from the Google search response.
Custom Scraper and Summarizer Tool
Post getting the list of URLs from the search results the next step is to scrape those websites and summarise them. For scraping the website we are using Selenium. For extracting data from the raw HTML response we have used BeautifulSoup.
Our next task is to summarise the page content to extract the user-relevant content by using OpenAI. The text size in the website can be huge and might not be possible to summarise the entire text in one go as there is a limitation from the OpenAI end to receive only 4000 tokens in the request.
To overcome this obstacle we are using map reduce documents chain. Map-reduce is a data processing technique that involves dividing the data into smaller chunks. It starts by invoking the Large Language Model (LLM) with an initial prompt for each data chunk. After processing the initial prompts, the LLM is invoked again, but this time with a modified prompt that incorporates the responses from the first round of prompts.
Let's have a look at our Scraper and Summarizer Tool code:
from langchain import OpenAI
from langchain.chains.summarize import load_summarize_chain
from langchain.docstore.document import Document
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.prompts import PromptTemplate
from langchain.document_loaders import SeleniumURLLoader
from bs4 import BeautifulSoup
from pydantic import BaseModel, Field
from langchain.tools import BaseTool
from typing import Type
def scrape_content(url, question):
urls = [url]
loader = SeleniumURLLoader(urls=urls)
data = loader.load()
if data is not None and len(data) > 0:
soup = BeautifulSoup(data[0].page_content, "html.parser")
text = soup.get_text()
if len(text) > 1000:
summary = summarise_content(text, question)
return summary
else:
return text
return ''
def summarise_content(content, ques):
llm = ChatOpenAI(temperature=0)
text_splitter = RecursiveCharacterTextSplitter(
separators=["\n\n", "\n"], chunk_size=3500, chunk_overlap=300)
docs = text_splitter.create_documents([content])
reduce_template_string = """I want you to act as a text summarizer to help me create a concise summary
of the text I provide. The summary can be up to 10 sentences in length, expressing the key points and
concepts written in the original text without adding your interpretations. The summary should be focused on
answering the given question and the other details which are not relevant to the given question can be ignored.
My first request is to summarize this text –
{text}
Question: {question}
Answer:
"""
reduce_template = PromptTemplate(
template=reduce_template_string, input_variables=["text", "question"])
chain = load_summarize_chain(
llm=llm,
chain_type='map_reduce',
map_prompt=reduce_template,
combine_prompt=reduce_template,
verbose=True
)
return chain.run(input_documents=docs, question=ques)
class ScraperInput(BaseModel):
"""Inputs for scrape_content function"""
url: str = Field(description="The url or link of the website to be scraped")
question: str = Field(description="The question or the query that users give to the agent")
class ScraperTool(BaseTool):
name = "website_scraper_tool"
description = "useful when you need to get data from a website url, passing both url and question to the function; DO NOT make up any url, the url should only be from the search results"
args_schema: Type[BaseModel] = ScraperInput
def _run(self, url: str, question: str):
return scrape_content(url, question)
def _arun(self, url: str):
raise NotImplementedError("error here")
Our main function is scrape_content which internally calls summarise_content function to summarise the content. The scrape_content function currently accepts two parameters url, the URL of the webpage which needs to be scrapped. The second parameter is question, it is the user-asked query based on which the content should be summarised. The idea here is to summarise only those portions of the text that are relevant to the user's query, the remaining can be ignored as it is irrelevant to the user.
If we directly try to use the function scrape_content as our Agent tool, it will throw an exception since each tool is allowed to have only one parameter. If we have a requirement of sending two arguments to our tool then we have to take the Tool class approach. In the above code, we have defined a new class called ScraperTool and it inherits from the BaseTool class. We have defined a separate class for our function parameter called ScraperInput which inherits BaseModel, and the class contains only 2 attributes url and a question. The ScraperTool class has a run method that calls our scrape_content function. This is the format we should follow when one pass more that one arguments to our tools. We also have the attribute fields in our ScraperTool class which are name, description and args_schema.
Custom File Appender Tool
The tool is pretty straightforward, we will use this tool for writing our summarised content in the file. Following is the code for our tool:
FILE_NAME = "summary_content.txt"
def append_text_to_file(text):
try:
# Open the file in 'a' mode (append mode) which creates the file if it doesn't exist
with open(FILE_NAME, 'a') as file:
file.write(text + '\n') # Append the text to the file
print(f"Text appended to '{FILE_NAME}' successfully.")
except IOError as e:
print(f"Error: {e}")
We only have one function append_text_to_file and it accepts only one parameter. It then writes the content in a file.
Using Tools in Agent
So far we have defined all our custom tools and now it's time to add them to our agent. Following is the code of our agent.
from langchain.schema import SystemMessage
from langchain.chat_models import ChatOpenAI
from langchain.agents import initialize_agent
from langchain.agents import AgentType
from langchain.tools import Tool
system_message = SystemMessage(
content="""You are a world class researcher, who can do detailed research on any topic.
Please make sure you complete the objective above with the following rules:
1/ Your job is to first breakdown the topic into relevant questions for understanding the topic in detail. You should have at max only 3 questions not more than that, in case there are more than 3 questions consider only the first 3 questions and ignore the others.
2/ You should use the search tool to get more information about the topic. You are allowed to use the search tool only 3 times in this process.
3/ Aggregate all the resources or urls that you can get on this topic.
4/ Use the provide scrape tool to scrape through your aggregrated list of urls one by one.
5/ Write the content you got from the scrape tool into a file. Use the write_to_file tool for writing to the file.
Your task is complete once you have written all the scraped content to the file.
"""
)
tools = [
Tool(
name="Search",
func=get_research_urls,
description="useful for when you need to answer questions about current events, data. This will return list of useful urls which should be scraped to get additional information."
),
ScraperTool(),
Tool(
name="write_to_file",
func=append_text_to_file,
description="useful when you need to write any text in the file."
),
]
agent_kwargs = {
"system_message": system_message,
}
llm = ChatOpenAI(temperature=0, model="gpt-3.5-turbo", verbose=True)
agent = initialize_agent(
tools,
llm,
agent=AgentType.OPENAI_FUNCTIONS,
verbose=True,
agent_kwargs=agent_kwargs,
)
In our tools array, we have defined the list of tools. The search and write to file tools have only one parameter so are able to define them directly. We have used the name, func and description attributes where we have defined the name of the tool, the function to be called by the Agent when it wants to use it, and then the description, which is read by the agent to make a judgment on when to use this tool.
We have used the initialize_agent function to initialize the agent, we are using ChatOpenAI as the LLM model and OPENAI_FUNCTIONS as the AgentType. We have written a System message prompt for our agent which are nothing but a series of instructions that the Agent is supposed to follow. We have passed the prompt to our agent in the agent_kwargs argument which is an optional argument.
Agents In Action
Till now we have defined our custom tools and Agents, now it's time to see our agent in action. Let's give our agent the task of building apps using Langchain agents.
response = agent.run("Langchain agents for building apps")
Below is the screenshot of our Agent execution logs:
The agent starts its action with the search task where it queries Google with the given query. It gets one url in response and then uses the scraper tool to scrape that page and summarise it. Once it gets the summary it invokes the file appender tool to write the summary in the file. Following is the screenshot for the same.
I trust you discovered this blog post on building a custom Langchain tool to be beneficial. I continuously produce similar content covering topics related to Langchain, LLM, and AI. If you wish to receive more articles of this nature, consider subscribing to my blog.
If you're in the Langchain space or LLM domain, let's connect on Linkedin! I'd love to stay connected and continue the conversation. Reach me at: linkedin.com/in/ritobrotoseth