Building AI Chatbots with RAG — Knowledge-Grounded Conversational Agents

from langchain_community.document_loaders import DirectoryLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma

# Load documents
loader = DirectoryLoader(
    "./docs/",
    glob="**/*.md",
    show_progress=True
)
documents = loader.load()
print(f"Loaded {len(documents)} documents")

# Split into chunks
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,
    chunk_overlap=100,
    separators=["\n## ", "\n### ", "\n", ". ", " "]
)
chunks = text_splitter.split_documents(documents)
print(f"Created {len(chunks)} chunks")

# Create vector store
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vectorstore = Chroma.from_documents(
    documents=chunks,
    embedding=embeddings,
    persist_directory="./chroma_db"
)
print(f"Vector store created with {vectorstore._collection.count()} vectors")

from langchain.chains import create_retrieval_chain
from langchain.chains.history_aware_retriever import (
    create_history_aware_retriever
)
from langchain.chains.combine_documents import create_stuff_documents_chain
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o", temperature=0.3)

# Prompt to rephrase query with history context
contextualize_prompt = ChatPromptTemplate.from_messages([
    ("system", "Given chat history and a user question, rephrase the question to be standalone."),
    ("placeholder", "{chat_history}"),
    ("human", "{input}")
])

history_aware_retriever = create_history_aware_retriever(
    llm, vectorstore.as_retriever(
        search_kwargs={"k": 4}
    ), contextualize_prompt
)

# Prompt to answer with sources
answer_prompt = ChatPromptTemplate.from_messages([
    ("system", "Answer based on the context. Cite sources. Say 'I cannot find this' if unsure.\n\n{context}"),
    ("placeholder", "{chat_history}"),
    ("human", "{input}")
])

document_chain = create_stuff_documents_chain(llm, answer_prompt)
rag_chain = create_retrieval_chain(
    history_aware_retriever, document_chain
)
print("RAG chain initialized successfully")

from langchain.memory import ChatMessageHistory

sessions = {}

def chat(session_id: str, message: str) -> str:
    if session_id not in sessions:
        sessions[session_id] = ChatMessageHistory()

    history = sessions[session_id]
    result = rag_chain.invoke({
        "input": message,
        "chat_history": history.messages
    })

    history.add_user_message(message)
    history.add_ai_message(result["answer"])

    sources = list(set(
        doc.metadata.get("source", "unknown")
        for doc in result["context"]
    ))

    return {
        "answer": result["answer"],
        "sources": sources,
        "session_length": len(history.messages) // 2
    }

# Test conversation
response1 = chat("user-1", "What are the API rate limits?")
print(f"Q: What are the API rate limits?")
print(f"A: {response1['answer'][:150]}...")
print(f"Sources: {response1['sources']}")

response2 = chat("user-1", "What happens if I exceed them?")
print(f"\nQ: What happens if I exceed them?")
print(f"A: {response2['answer'][:150]}...")

from pydantic import BaseModel, Field

class GuardrailResult(BaseModel):
    is_allowed: bool = Field(description="Whether the query is allowed")
    reason: str = Field(description="Explanation of the decision")

guardrail_prompt = ChatPromptTemplate.from_messages([
    ("system", """Determine if the query is relevant to the chatbot's
documented knowledge base. Block queries about:
- Personal advice, medical, legal, or financial decisions
- Instructions for harmful or illegal activities
- Queries completely unrelated to the documentation"""),
    ("human", "{query}")
])

guardrail_llm = llm.with_structured_output(GuardrailResult)

def guarded_chat(session_id: str, message: str) -> dict:
    check = guardrail_llm.invoke(
        guardrail_prompt.format(query=message)
    )

    if not check.is_allowed:
        return {
            "answer": f"I cannot answer that question. {check.reason}",
            "sources": [],
            "blocked": True
        }

    return chat(session_id, message)

response = guarded_chat("user-1", "How do I hack a website?")
print(response["answer"])
print(f"Blocked: {response.get('blocked', False)}")