How to build an agent knowledge base with LangChain and Elasticsearch
Learn how to build an agent knowledge base and test its ability to query sources of information based on context, use WebSearch for out-of-scope queries, and refine recommendations based on user intention.
Applications of large language models throughout the industry.
In industry use cases, there are two primary modes of interacting with large language models (LLMs). Direct querying, ie., conversing with an LLM on an ad-hoc basis, is useful for getting assistance on tasks like summarization, proofreading, information extraction, and non-domain-specific querying.
For specific business applications, such as in customer relationship management, maintenance of IT systems, and investigative work, to name only a few examples, direct LLM usage is insufficient. Private, enterprise-specific information, or information about niche interests and topics, or even from specific documents and written sources, tends to be lacking from LLM training datasets. In addition, real-world data is constantly changing, and enterprise contexts are constantly evolving. LLMs also tend to require reinforcement of factual accuracy. All these factors limit the utility value of using LLMs directly for enterprise use-cases, especially those requiring up-to-date factual information about specific technical or business topics.
Retrieval Augmented Generation (RAG), the use of searchable data stores to retrieve information sources relevant to the context and intention of a user query, was popularized as a way to address this deficiency. A large amount of work has been done to implement, assess, and improve the quality of RAG applications, and RAG has enjoyed widespread adoption in enterprise use cases for productivity enhancement and workflow automation. However, RAG does not leverage the decision-making capacity of large language models.
Different application modes of large language models.
The agentic model revolves around the LLM being able to take specific actions in response to a user input. These actions may involve the use of tools to augment the LLM's existing capabilities. In this sense, RAG functions as a long-term memory store that the LLM agent may choose to use to augment and reinforce answers to user queries. Where the traditional RAG model involves the LLM querying one or more knowledge bases, an agentic implementation allows an LLM to choose from a set of knowledge bases. This allows for more flexible question-answering behavior, and can improve accuracy, as information from irrelevant knowledge bases is omitted, reducing potential sources of noise. We might call such a system an "agent knowledge base." Let's take a look at how to implement such a system using Elasticsearch.
Designing an agent knowledge base
All code may be found in the GitHub repo.
The agent knowledge base implemented in this article.
I recently became interested in scuba diving after trying it and realizing it could cure my persistent thalassophobia, so I decided to set up an agentic knowledge base for diving specifically.
The US Navy Dive Manual - Containing a wealth of technical detail about diving operations and equipment.
Diving Safety Manual - Containing general guidelines and procedures aimed at recreational divers.
The Google Custom Search API - Capable of searching the web for any information not contained within the two manuals.
The intention was that this Diving Assistant would be a one-stop shop for diving-related knowledge, which would be capable of responding to any query, even those out of scope of the knowledge bases ingested. The LLM would recognize the motivation behind a user query, and select the source of information most likely to be relevant. I decided to use LangChain as the agentic wrapper, and built a streamlit UI around it.
Setting up the endpoints
I start by creating a .env file and populating it with the following variables:
ELASTIC_ENDPOINT=<ELASTIC CLOUD ENDPOINT>
ELASTIC_API_KEY=<ELASTIC CLOUD API KEY>
# Enable custom search API
# https://developers.google.com/custom-search/v1/introduction/?apix=true
GCP_API_KEY=<GCP API KEY>
GCP_PSE_ID=<GCP PSE ID>
AZURE_OPENAI_SYSTEM_PROMPT="You are a helpful assistant. Be as concise and efficient as possible. Convey maximum meaning in fewest words possible."
AZURE_OPENAI_ENDPOINT=<AZURE ENDPOINT>
AZURE_OPENAI_API_VERSION=<AZURE API VERDSION>
AZURE_OPENAI_API_KEY=<AZURE API KEY>
AZURE_OPENAI_MODEL="gpt-4o-mini"This project makes use of a GPT-4o-Mini deployed on Azure OpenAI, as well as the Google Custom Search API, and an Elastic Cloud deployment to hold my data. I also add a custom system prompt encouraging the LLM to avoid wordiness as much as possible.
Ingestion and processing
The US Navy Dive Manual and Diving Safety Manual are in PDF format, so the next step was to ingest them into an Elastic Cloud deployment. I set-up this python script using Elastic's bulk API to upload documents to Elastic Cloud:
import os
from collections import defaultdict
from concurrent.futures import ThreadPoolExecutor, as_completed
from elasticsearch import Elasticsearch, helpers # elasticsearch==8.14.0
from tqdm import tqdm # tqdm==4.66.4
from llama_index.core import SimpleDirectoryReader
def bulk_upload_to_elasticsearch(data, index_name, es, batch_size=500, max_workers=10):
'''
data: [ {document} ]
document: {
"_id": str
...
}
index_name: str
es: Elasticsearch
batch_size: int
max_workers: int
'''
total_documents = len(data)
success_bar = tqdm(total=total_documents, desc="Successful uploads", colour="green")
failed_bar = tqdm(total=total_documents, desc="Failed uploads", colour="red")
def create_action(doc):
'''
Define upload action from source documents
'''
return {
"_index": index_name,
"_id": doc["id_"],
"body": doc["text"]
}
def read_and_create_batches(data):
'''
Yield document batches
'''
batch = []
for doc in data:
batch.append(create_action(doc))
if len(batch) == batch_size:
yield batch
batch = []
if batch:
yield batch
def upload_batch(batch):
'''
Make bulk call for batch
'''
try:
success, failed = helpers.bulk(es, batch, raise_on_error=False, request_timeout=45)
if isinstance(failed, list):
failed = len(failed)
return success, failed
except Exception as e:
print(f"Error during bulk upload: {str(e)}")
return 0, len(batch)
'''
Parallel execution of batch upload
'''
with ThreadPoolExecutor(max_workers=max_workers) as executor:
future_to_batch = {executor.submit(upload_batch, batch): batch for batch in read_and_create_batches(data)}
for future in as_completed(future_to_batch):
success, failed = future.result()
success_bar.update(success)
failed_bar.update(failed)
'''
Update progress bars
'''
total_uploaded = success_bar.n
total_failed = failed_bar.n
success_bar.close()
failed_bar.close()
return total_uploaded, total_failed
# This is connecting to ES Cloud via credentials stored in .env
# May have to change this to suit your env.
try:
es_endpoint = os.environ.get("ELASTIC_ENDPOINT")
es_client = Elasticsearch(
es_endpoint,
api_key=os.environ.get("ELASTIC_API_KEY")
)
except Exception as e:
es_client = None
print(es_client.ping())After downloading the US Navy Dive Manual PDF and storing it in its own folder, I use LlamaIndex's SimpleDirectoryReader to load the PDF data, then trigger a bulk upload:
reader = SimpleDirectoryReader(input_dir="./data")
documents = reader.load_data()
bulk_upload_to_elasticsearch([i.to_dict() for i in list(documents)],
"us_navy_dive_manual_raw",
es_client, batch_size=16, max_workers=10)This sends all the text content to Elastic Cloud, with each page of the PDF as a separate document, to an index called us_navy_dive_manual_raw. No further processing is done, so the process of uploading all 991 pages takes less than a second. The next step is to do semantic embedding within Elastic Cloud.
Semantic data embedding and chunking
In my Elastic Cloud DevTools console, I first deploy the ELSER v2 model using the Elastic inference API
PUT _inference/sparse_embedding/elser_v2
{
"service": "elasticsearch",
"service_settings": {
"num_allocations": 1,
"num_threads": 8,
"model_id": ".elser_model_2_linux-x86_64"
},
"chunking_settings": {
"strategy": "sentence",
"max_chunk_size": 250,
"sentence_overlap": 1
}
}I then define a simple pipeline. Each document stores the text of a page from the dive manual in the body field, so I copy the contents of body to a field called semantic_content.
PUT _ingest/pipeline/diving_pipeline
{
"processors": [
{
"set": {
"field": "semantic_content",
"copy_from": "body",
"if": "ctx.body != null"
}
}
]
}I then create a new index called us_navy_dive_manual, and set semantic_content as a semantic_text field:
PUT us_navy_dive_manual
{
"mappings": {
"properties": {
"semantic_content": {
"type": "semantic_text",
"inference_id": "elser_v2"
}
}
}
}I then trigger a reindex job. Now the data will flow from us_navy_dive_manual_raw, to be chunked and embedded using ELSER, and be reindexed into us_navy_dive_manual ready for use.
POST _reindex?slices=auto&wait_for_completion=false
{
"source": {
"index": "us_navy_dive_manual_raw",
"size": 4
},
"dest": {
"index": "us_navy_dive_manual",
"pipeline": "diving_pipeline"
},
"conflicts": "proceed"
}I repeat this process for the Diving Safety Manual, and with this simple process, data ingestion is completed.
Tooling for agentic search
This agent is relatively simple, so I make use of LangChain's AgentExecutor which creates an agent and bundles it with a set of tools. Complex decisionmaking flows can be achieved using the LangGaph implementation , which we will use in a future blog. We will focus on the parts related to the agents, so for details on the actual streamlit UI, please check out the github repo.
I create two tools for my agent to use. The first is an ElasticSearcher class, which performs a semantic search over an Elastic index, then returns the top 10 articles as text.
class ElasticSearcher:
def __init__(self):
self.client = Elasticsearch(
os.environ.get("ELASTIC_ENDPOINT"),
api_key=os.environ.get("ELASTIC_API_KEY")
)
def search(self, query, index="us_navy_dive_manual", size=10):
response = self.client.search(
index=index,
body={
"query": {
"semantic": {
"field": "semantic_content",
"query": query
}
}
},
size=size
)
return "\n".join([hit["_source"].get("body", "No Body")
for hit in response["hits"]["hits"]])The second tools is the Googler class, which calls the Google Custom Search API to perform a general web search.
class Googler:
def __init__(self):
self.service = build('customsearch', 'v1', developerKey=os.getenv("GCP_API_KEY"))
def scrape(self, url):
try:
response = requests.get(url, timeout=10)
if response.status_code == 200:
soup = BeautifulSoup(response.text, 'html.parser')
for script in soup(["script", "style"]):
script.decompose()
text = soup.get_text()
lines = (line.strip() for line in text.splitlines())
chunks = (phrase.strip() for line in lines for phrase in line.split(" "))
return '\n'.join(chunk for chunk in chunks if chunk)[:5000]
return None
except:
return None
def search(self, query, n=5):
results = self.service.cse().list(q=query, cx=os.getenv("GCP_PSE_ID"), num=n).execute()
scraped_data = []
for item in results.get('items', []):
url = item['link']
title = item['title']
content = self.scrape(url) or item['snippet']
scraped_data.append(f"Page: {title}\nURL: {url}\n\n{content}\n")
return "\n".join(scraped_data)I then create a set of tools for the agent to use. The description of each tool is an important part of the prompt engineering, as the agent will refer to it primarily when choosing which tool to use for its response to a user query.
tools = [
Tool(
name="WebSearch",
func=lambda q: googler.search(q, n=3),
description="Search the web for information. Use for current events or general knowledge or to complement with additional information."
),
Tool(
name="NavyDiveManual",
func=lambda q: elastic.search(q, index="us_navy_dive_manual"),
description="Search the Operations Dive Manual. Use for diving procedures, advanced or technical operational planning, resourcing, and technical information."
),
Tool(
name="DivingSafetyManual",
func=lambda q: elastic.search(q, index="diving_safety_manual"),
description="Search the Diving Safety Manual. Use for generic diving safety protocols and best practices."
)
]Next, I define an LLM using the AzureChatOpenAI abstraction:
llm = AzureChatOpenAI(
azure_endpoint=os.getenv("AZURE_OPENAI_ENDPOINT"),
api_key=os.getenv("AZURE_OPENAI_API_KEY"),
api_version=os.getenv("AZURE_OPENAI_API_VERSION"),
deployment_name=os.getenv("AZURE_OPENAI_MODEL"),
streaming=False
)And also create a custom prompt for the LLM, telling it how to make use of the tools and their outputs.
prompt = PromptTemplate.from_template("""Answer the following questions as best you can. You have access to the following tools:
{tools}
You should use multiple tools in conjunction to promote completeness of information.
Be comprehensive in your answer.
Use the following format:
Question: the input question you must answer
Thought: you should always think about what to do
Action: the action to take, should be one of [{tool_names}]
Action Input: the input to the action
Observation: the result of the action
... (this Thought/Action/Action Input/Observation can repeat N times)
Thought: I now know the final answer
Final Answer: the final answer to the original input question
Question: {input}
{agent_scratchpad}""")Finally, I define the agent, passing it the LLM, prompt, and toolset, and integrate it into the rest of the UI.
agent = create_react_agent(llm, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=False)And with that, we are ready to test out our agent knowledge base.
The UI implemented for the agentic knowledge base.
Test 1: Retrieve knowledge from a specific knowledge base
First, we should test that the knowledge base component is working as intended. Let's start by asking a detailed technical question. The agent should choose the right knowledge base, retrieve some information, and use it to craft a response.
List of recommended equipment for ice/cold water diving operationsThe UI displays the Agent's thought process, and the first thought it has is to consult the diving safety manual:
Thought: To provide a comprehensive list of recommended equipment for ice
or cold water diving operations, I should consult specialized diving
manuals that cover operational procedures and safety protocols.
This will ensure that I include all necessary items for safety and
efficiency in such challenging environments.
Action: NavyDiveManual
Action Input: "recommended equipment for ice cold water
diving operations"The agent uses the NavyDiveManual tool to perform a semantic search of the us_navy_dive_manual index. The result is information like this:
CHAPTER 11 — Ice and Cold Water Diving Operations 11-9
A detailed reconnaissance of the dive site will provide the
planner with information that is helpful in deciding what ancillary equipment is
required. Diving under ice will require special accessory equipment such as a line
with lights/strobes for under water navigation, ice-cutting tools, platforms, engine
protection kits, and stainless steel ice screws, quick draw, and carabineers.
The method of cutting the hole through the ice depends on ice thickness and availability
of equipment. Normally, two or more of the following tools are used: hand
ice chipper, ice handsaw, ice auger, chain saw, thermal ice cutter or blasting equipment.
In addition, equipment to lift the ice block, remove the slush, and mark ...Which the LLM then uses to craft a response:
The recommended equipment for ice and cold water diving operations includes:
Thermal Protection:
* Wet suit or Variable Volume Dry Suit (VVDS)
* Heavy insulating socks (if using a wet suit)
* Diving Equipment:
* Approved cold water open-circuit SCUBA or closed-circuit UBAThe response is exactly what we wanted. Great!
Test 2: Iteratively refine knowledge collection
The next test is to see whether the agent can select multiple sources of knowledge to refine its responses. Let's ask a more open ended question:
What are the risk factors when snorkeling?The first thought the agent has involves searching Google for information specific to risk factors and dangers:
Thought: To provide a comprehensive answer to the risk factors associated with snorkeling,
I should gather information on general safety protocols and specific risks from both diving
and snorkeling perspectives. I will start by searching for general snorkeling risks and
then look into specific safety guidelines related to diving and snorkeling.
Action: WebSearch
Action Input: "risk factors when snorkeling"This returns web results such as:
Factors Contributing to Snorkel Drowning in Hawai'i - PMC
URL: https://pmc.ncbi.nlm.nih.gov/articles/PMC8899085/
Causes of the extraordinarily high and increasing incidence of snorkeler drownings in Hawai'i
have remained unexplained for years. Defining the mechanisms and factors predisposing to
drowning while snorkeling is needed to provide recommendations to substantially mitigate
the incidence of this form of preventable death. The mechanisms of drowning are described
and insight into the predisposing factors are explored in this study...The agent then chooses to consult the diving_safety_manual, and takes the appropriate action to retrieve safety protocols and guidelines.
Action: DivingSafetyManual
Action Input: "snorkeling safety protocols"Which returns relevant information from the DivingSafetyManual tool:
Pre-Snorkel briefing
Immediately before snorkelling commences, and using site diagrams or
photos as appropriate, the snorkeller supervisor should emphasise
the following points:
• On-site environmental conditions
• Boundaries of the snorkelling area
• Times when snorkelli ng is being conducted
• Specific site hazards
• Location of lookout
• Emergency and recall signals
• Buddy and guided group proceduresFinally, the LLM provides a comprehensive answer, containing both risk factors and safety guidelines:
The risk factors when snorkeling include:
Environmental Conditions: Strong currents, waves, and marine life can pose
immediate threats. Snorkelers should be aware of local weather and water
conditions before entering.
Physical Fitness: Snorkeling is a strenuous activity that can be
challenging for individuals with pre-existing medical conditions
such as heart disease, respiratory issues (e.g., asthma), or those
who are not strong swimmers.
...
To mitigate these risks, snorkelers should engage in pre-snorkel briefings,
wear personal flotation devices, ensure proper equipment checks, and
maintain awareness of their environment and physical limits.Test 3: Searching out of scope
The final test is to see what happens when the user asks a query that is certainly out of scope of the prepared knowledge bases. Let's ask about a specific diving product:
Tell me about the Apeks XTX50 regulatorThe agent's first thought is to check the external web for relevant information:
Thought: I need to gather information about the Apeks XTX50 regulator,
including its features, specifications, and performance in diving conditions.
I will begin with a web search for comprehensive details.
Action: WebSearch
Action Input: "Apeks XTX50 regulator review features specifications"It discovers a wealth of information from commercial and diving hobbyist sites, and returns both technical specifications and user feedback:
The Apeks XTX50 is a high-performance scuba diving regulator designed for
divers who seek reliability and advanced features at a moderate price point.
Here are the key features and specifications:
Features:
First Stage:
Utilizes the DST (Diver Adjustable) first stage with an environmentally
sealed design to prevent water ingress. Over-balanced diaphragm technology
provides superior performance as divers descend, with medium pressure gas
increasing faster than ambient pressure.
...
Divers appreciate the XTX50's reliability and comfort, making it a popular
choice for both recreational and technical diving.
The regulator is lauded for its performance in colder conditions, with
many users noting its effectiveness during deep dives.
Some divers have raised concerns about the availability of parts for maintenance,
as Apeks does not sell parts directly to consumers.
...The agent did not choose to search the US Navy Dive Manual, which is concerned with operational planning, or with the DivingSafetyManual.
Conclusion
In a traditional RAG implementation, we might have chosen to force the LLM to search and use information from all three data sources simultaneously, but this would have negatively impacted accuracy by introducing noise from irrelevant information. With the agentic knowledge base, we see the LLM making targeted searches to specific sources of knowledge based on user intention and context. The agent is able to refine the knowledge collected by building upon initial searches, and combining them with information collected from other sources.
The agent is also able to handle questions out of the scope of its prepared data, and is also able to exclude knowledge bases not pertinent to the query - Significant enhancements on the traditional RAG model.
This agent knowledge base concept provides a graceful way to combine many different sources into a coherent and comprehensive system, and the next steps would be to expand the range of actions and the diversity of information that can be referred to. Introducing workflows for fact-checking and cross referencing would be a boon to overall reliability, and tools for specialized capabilities like calculation would be a very interesting direction to explore.
Related Content
