Building a Knowledge Base Chatbot with LangGraph: Complete Guide

Building a Knowledge Base Chatbot with LangGraph: Complete Guide

Large language models are powerful, but they have a fundamental limitation: they only know what they were trained on. When users ask questions about private documents, recent events, or proprietary knowledge, the model either hallucinates or refuses to answer. A knowledge base chatbot solves this by combining an LLM with a retrieval system that grounds responses in your own data. LangGraph, built by the team behind LangChain, gives you a structured way to orchestrate this kind of workflow as a stateful, cyclic graph of nodes and edges.

In this guide, you'll build a production-style retrieval-augmented generation (RAG) chatbot that maintains conversation history, decides when to retrieve documents, and grades retrieved content for relevance before generating an answer. By the end, you'll understand not just the code, but the architectural decisions that make a chatbot reliable.

What Is LangGraph and Why Use It for a Knowledge Base Chatbot?

LangGraph is a library for building stateful, multi-actor applications with LLMs. It models your application as a directed graph where each node is a function or callable, and edges define the flow of control between them. State is passed between nodes as a typed object that can be updated at each step.

Compared to writing a RAG pipeline as a single linear script, LangGraph offers several advantages:

• Cycles and conditional routing: A chatbot often needs to loop — retrieve, evaluate, retrieve again, then answer. LangGraph supports cycles natively, while plain LangChain chains are mostly linear.
• State management: Conversation history, retrieved documents, and intermediate reasoning are stored in a shared state object, eliminating the need to thread variables through every function.
• Persistence: Built-in checkpointers let you save and resume conversations, which is essential for multi-turn chatbots deployed in production.
• Human-in-the-loop: You can pause execution at any node to request human approval, useful when the chatbot is about to take an action like sending an email.
• Streaming and observability: LangGraph integrates with LangSmith for tracing, and supports streaming of tokens, state updates, and events.

For a knowledge base chatbot specifically, the cyclic nature is the killer feature. You want the bot to decide whether retrieval is needed, check whether retrieved documents actually answer the question, and rewrite the query if they don't — all before responding to the user. That's a graph, not a chain.

Prerequisites and Project Setup

Before writing code, set up a clean environment. You'll need Python 3.10 or later, an OpenAI API key (or any LLM provider you prefer), and a vector store. This guide uses OpenAI for embeddings and chat models, and FAISS for the vector store because it requires no external service.

Create a project directory and install the dependencies:

mkdir kb-chatbot && cd kb-chatbot
python -m venv .venv
source .venv/bin/activate

pip install langgraph langchain langchain-openai \
            langchain-community faiss-cpu python-dotenv \
            tiktoken pydantic


Create a .env file to hold your API key:

OPENAI_API_KEY=sk-your-key-here


And a small loader at the top of your main script:

import os
from dotenv import load_dotenv
load_dotenv()


Designing the Chatbot Architecture

The chatbot you're building follows the classic "Corrective RAG" (CRAG) pattern, extended with a retrieval decision step. Here's the flow:


  
Decide to retrieve: Given the user's question and conversation history, an LLM decides whether the question requires looking up documents or can be answered directly.
  
Retrieve documents: If retrieval is needed, search the vector store for relevant chunks.
  
Grade documents: For each retrieved document, an LLM grades whether it is relevant to the question. Irrelevant documents are filtered out.
  
Decide to generate or rewrite: If at least one relevant document was found, proceed to generation. If none were relevant, rewrite the query and try retrieval again.
  
Generate answer: Produce a final answer grounded in the relevant documents, citing sources.
  
Check hallucination: Verify the generated answer is supported by the documents. If not, regenerate.


This design avoids two common failure modes: answering from parametric memory when documents exist (causing hallucination), and answering from irrelevant documents (causing confusion).

Building the Knowledge Base

First, create the vector store that will serve as your knowledge base. You'll load documents, split them into chunks, embed them, and store them in FAISS.

from langchain_community.document_loaders import DirectoryLoader, TextLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import FAISS

def build_vectorstore(docs_dir: str = "./docs") -> FAISS:
    loader = DirectoryLoader(
        docs_dir,
        glob="**/*.txt",
        loader_cls=TextLoader,
        loader_kwargs={"encoding": "utf-8"},
    )
    documents = loader.load()

    splitter = RecursiveCharacterTextSplitter(
        chunk_size=1000,
        chunk_overlap=200,
        separators=["\n\n", "\n", ". ", " ", ""],
    )
    chunks = splitter.split_documents(documents)

    embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
    vectorstore = FAISS.from_documents(chunks, embeddings)
    return vectorstore


Place a few text files in a ./docs directory. For testing, you might include product documentation, internal policies, or FAQ files. The chunk size of 1000 characters with 200 overlap is a reasonable default for prose; adjust based on your document structure.

For production, consider persisting the vectorstore to disk so you don't re-embed on every startup:

vectorstore.save_local("faiss_index")
# Later:
# vectorstore = FAISS.load_local("faiss_index", embeddings, allow_dangerous_deserialization=True)


Defining the Graph State

The state object is the backbone of a LangGraph application. Every node receives the current state and returns a partial update. Define it using Pydantic or TypedDict:

from typing import TypedDict, List, Annotated
from langchain_core.documents import Document
from langgraph.graph.message import add_messages

class ChatState(TypedDict):
    messages: Annotated[list, add_messages]   # conversation history
    question: str                              # current user question
    documents: List[Document]                  # retrieved + graded docs
    rewritten_question: str                    # query after rewriting
    retrieval_needed: bool                     # decision flag
    generation: str                            # final answer
    retries: int                               # rewrite attempt counter


The add_messages reducer appends new messages to the list rather than replacing it, which is what you want for conversation history. Other fields use the default behavior of overwriting.

Implementing the Nodes

Each node is a function that takes ChatState and returns a dict with the fields it wants to update. Let's implement them one by one.

Node 1: Decide Whether to Retrieve

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from pydantic import BaseModel, Field

class RetrievalDecision(BaseModel):
    retrieval_needed: bool = Field(description="Whether to retrieve documents")

def decide_retrieval(state: ChatState) -> dict:
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
    structured_llm = llm.with_structured_output(RetrievalDecision)

    prompt = ChatPromptTemplate.from_messages([
        ("system", (
            "You decide whether a user question requires retrieving documents "
            "from a knowledge base. Answer 'true' if the question asks about "
            "specific facts, policies, products, or content that would be in "
            "documents. Answer 'false' for greetings, small talk, or general "
            "knowledge questions."
        )),
        ("human", "{question}"),
    ])

    chain = prompt | structured_llm
    result = chain.invoke({"question": state["question"]})
    return {"retrieval_needed": result.retrieval_needed}


Using structured output ensures the LLM returns a clean boolean rather than free text you'd have to parse.

Node 2: Retrieve Documents

def retrieve(state: ChatState) -> dict:
    vectorstore = build_vectorstore()  # cache this in production
    question = state.get("rewritten_question") or state["question"]
    retriever = vectorstore.as_retriever(search_kwargs={"k": 4})
    documents = retriever.invoke(question)
    return {"documents": documents}


In production, build the vectorstore once and pass it in via a closure or a global, not on every call. Retrieving the top 4 chunks is a starting point; tune k based on your chunk size and recall needs.

Node 3: Grade Documents

class GradeResult(BaseModel):
    score: str = Field(description="'yes' if relevant, 'no' if not")

def grade_documents(state: ChatState) -> dict:
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
    structured_llm = llm.with_structured_output(GradeResult)

    prompt = ChatPromptTemplate.from_messages([
        ("system", (
            "You are a relevance grader. Given a user question and a document, "
            "decide whether the document contains information relevant to "
            "answering the question. Respond with 'yes' or 'no'."
        )),
        ("human", "Question: {question}\n\nDocument: {document}"),
    ])

    chain = prompt | structured_llm
    question = state["question"]
    filtered = []
    for doc in state["documents"]:
        result = chain.invoke({
            "question": question,
            "document": doc.page_content[:2000],
        })
        if result.score == "yes":
            filtered.append(doc)
    return {"documents": filtered}


Grading each document individually prevents one irrelevant chunk from polluting the context. The trade-off is cost — you make one LLM call per document. For large retrieval sets, consider batching or using a smaller, faster model.

Node 4: Rewrite the Query

def rewrite_query(state: ChatState) -> dict:
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)
    prompt = ChatPromptTemplate.from_messages([
        ("system", (
            "You are a query rewriter. The user's question did not retrieve "
            "relevant documents. Reformulate the question to be more specific, "
            "use synonyms, or rephrase it to improve retrieval. Output only "
            "the rewritten question."
        )),
        ("human", "{question}"),
    ])
    chain = prompt | llm
    response = chain.invoke({"question": state["question"]})
    return {
        "rewritten_question": response.content.strip(),
        "retries": state.get("retries", 0) + 1,
    }


A higher temperature here encourages creative reformulations. The retry counter prevents infinite loops — you'll cap it in the routing logic.

Node 5: Generate the Answer

def generate(state: ChatState) -> dict:
    llm = ChatOpenAI(model="gpt-4o", temperature=0.3)
    context = "\n\n".join(
        f"[Source {i+1}]: {doc.page_content}" 
        for i, doc in enumerate(state["documents"])
    )

    prompt = ChatPromptTemplate.from_messages([
        ("system", (
            "You are a knowledgeable assistant. Answer the user's question "
            "using ONLY the provided sources. If the sources do not contain "
            "enough information, say you don't know. Cite sources by their "
            "number, e.g. [Source 1]."
        )),
        ("human", "Sources:\n{context}\n\nQuestion: {question}"),
    ])

    chain = prompt | llm
    response = chain.invoke({
        "context": context,
        "question": state["question"],
    })
    return {"generation": response.content}


Node 6: Check for Hallucination

class HallucinationCheck(BaseModel):
    grounded: bool = Field(description="True if answer is supported by sources")

def check_hallucination(state: ChatState) -> dict:
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
    structured_llm = llm.with_structured_output(HallucinationCheck)

    context = "\n\n".join(doc.page_content for doc in state["documents"])
    prompt = ChatPromptTemplate.from_messages([
        ("system", (
            "You check whether an answer is fully supported by the provided "
            "sources. Respond 'true' if every claim in the answer can be "
            "verified against the sources, 'false' otherwise."
        )),
        ("human", "Sources:\n{context}\n\nAnswer: {answer}"),
    ])

    chain = prompt | structured_llm
    result = chain.invoke({
        "context": context,
        "answer": state["generation"],
    })
    return {"hallucination_free": result.grounded}


If the check fails, you route back to generate for another attempt. Add a counter to avoid infinite regeneration loops.

Wiring the Graph Together

Now connect the nodes with edges and conditional routing. This is where LangGraph's power becomes visible.

from langgraph.graph import StateGraph, START, END

def route_after_decision(state: ChatState) -> str:
    if state["retrieval_needed"]:
        return "retrieve"
    return "generate"

def route_after_grading(state: ChatState) -> str:
    if len(state["documents"]) > 0:
        return "generate"
    if state.get("retries", 0) >= 2:
        return "generate"  # give up and answer with what we have
    return "rewrite"

def route_after_check(state: ChatState) -> str:
    if state.get("hallucination_free", False):
        return END
    if state.get("gen_retries", 0) >= 2:
        return END
    return "generate"

graph_builder = StateGraph(ChatState)

graph_builder.add_node("decide_retrieval", decide_retrieval)
graph_builder.add_node("retrieve", retrieve)
graph_builder.add_node("grade_documents", grade_documents)
graph_builder.add_node("rewrite", rewrite_query)
graph_builder.add_node("generate", generate)
graph_builder.add_node("check_hallucination", check_hallucination)

graph_builder.add_edge(START, "decide_retrieval")
graph_builder.add_conditional_edges("decide_retrieval", route_after_decision)
graph_builder.add_edge("retrieve", "grade_documents")
graph_builder.add_conditional_edges("grade_documents", route_after_grading)
graph_builder.add_edge("rewrite", "retrieve")
graph_builder.add_edge("generate", "check_hallucination")
graph_builder.add_conditional_edges("check_hallucination", route_after_check)

app = graph_builder.compile()


Notice the cycle: retrieve → grade_documents → rewrite → retrieve. This loop lets the bot refine its search when the first attempt fails, but the retry counter in route_after_grading guarantees termination.

Adding Memory with a Checkpointer

A chatbot without memory is just a search interface. LangGraph's checkpointer persists state across invocations, keyed by a thread ID. Add SQLite persistence:

from langgraph.checkpoint.sqlite import SqliteSaver
import sqlite3

conn = sqlite3.connect("chatbot.db", check_same_thread=False)
checkpointer = SqliteSaver(conn)
app = graph_builder.compile(checkpointer=checkpointer)


Now each call needs a configuration object with a thread ID:

config = {"configurable": {"thread_id": "user-123-session-1"}}


The same thread ID retrieves the same conversation history on subsequent calls, enabling true multi-turn chat.

Running the Chatbot

Here's a simple interactive loop that ties everything together:

def chat():
    config = {"configurable": {"thread_id": "session-1"}}
    print("Knowledge Base Chatbot ready. Type 'quit' to exit.\n")

    while True:
        user_input = input("You: ").strip()
        if user_input.lower() in ("quit", "exit"):
            break

        result = app.invoke(
            {"question": user_input, "messages": [("user", user_input)]},
            config=config,
        )

        print(f"\nBot: {result['generation']}\n")
        if result.get("documents"):
            print(f"  (Based on {len(result['documents'])} source(s))\n")

if __name__ == "__main__":
    chat()


For streaming responses, replace invoke with astream and iterate over the events:

import asyncio

async def chat_streaming():
    config = {"configurable": {"thread_id": "session-1"}}
    user_input = input("You: ")
    
    async for event in app.astream(
        {"question": user_input, "messages": [("user", user_input)]},
        config=config,
        stream_mode="messages",
    ):
        if event[0].content:
            print(event[0].content, end="", flush=True)
    print()

asyncio.run(chat_streaming())


Best Practices

Chunk Your Documents Thoughtfully

Retrieval quality depends more on chunking than on the embedding model. Use RecursiveCharacterTextSplitter with separators that match your document structure. For Markdown, split on headers. For code, split on function boundaries. Aim for chunks that are semantically complete — a chunk that cuts off mid-sentence will retrieve poorly.

Use Metadata for Filtering

Attach metadata to your documents so you can pre-filter retrieval by source, date, category, or access level:

from langchain_core.documents import Document

doc = Document(
    page_content="Our refund policy allows returns within 30 days.",
    metadata={"source": "policy.pdf", "category": "refunds", "version": "2024-01"}
)


Then filter at retrieval time:

retriever = vectorstore.as_retriever(
    search_kwargs={"k": 4, "filter": {"category": "refunds"}}
)


Cache the Vectorstore and LLM Clients

Building the vectorstore and instantiating LLM clients are expensive operations. Create them once at application startup and share them across nodes via closures, a config object, or a simple module-level singleton. Never rebuild them inside a node function in production.

Set Retry and Token Limits

Every cycle in your graph needs a termination condition. The rewrite loop and the hallucination-check loop both have counters in this guide. Without them, a poorly behaved LLM response could send your graph into an infinite loop, burning API credits.

Evaluate Retrieval Quality

Before deploying, build a small evaluation set of question-answer pairs and measure retrieval recall and answer faithfulness. LangSmith and Ragas are excellent tools for this. A chatbot that retrieves the wrong documents will produce confident, wrong answers — the worst possible failure mode.

Handle the "No Documents" Case Gracefully

When retrieval finds nothing relevant after all retries, the bot should say so honestly rather than hallucinate. Modify the generate node to check for empty documents and return a fallback message:

def generate(state: ChatState) -> dict:
    if not state.get("documents"):
        return {"generation": (
            "I couldn't find relevant information in the knowledge base "
            "to answer your question. Could you rephrase it or provide "
            "more context?"
        )}
    # ... rest of generation logic


Log and Trace Everything

Enable LangSmith tracing by setting environment variables. This lets you inspect every LLM call, every retrieval, and every routing decision after the fact — invaluable for debugging why the bot gave a particular answer.

import os
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "ls-your-key"
os.environ["LANGCHAIN_PROJECT"] = "kb-chatbot"


Conclusion

Building a knowledge base chatbot with LangGraph gives you a level of control that linear pipelines simply can't match. By modeling the workflow as a graph with conditional edges and cycles, you can implement sophisticated behaviors like retrieval decision-making, document grading, query rewriting, and hallucination checking — all while keeping the code modular and the state explicit. The architecture you've built here is a solid foundation: swap FAISS for Pinecone or pgvector when you need scale, replace the grading model with a fine-tuned classifier when latency matters, and add human-in-the-loop checkpoints when the bot's answers drive consequential decisions. Start with this graph, evaluate it against your real questions, and iterate on the nodes that matter most for your domain. The graph structure means you can improve one piece at a time without rewriting the whole system.