PDF Chatbot That Don’t Forget: Using Pickle for Persistent Chat

AI Generated

Ever wondered how to build your own local AI assistant that can read your PDFs, remember your conversations, and answer your questions contextually using Gemini and Sentence Transformers? Here’s a complete walkthrough

How It Works:

1. PDF Processing Pipeline: The code creates a system that extracts text from multiple PDF documents, splits it into manageable chunks, and tracks which file each chunk came from.
2. Vector Embeddings: It uses SentenceTransformer to convert text chunks into numerical vector representations that capture semantic meaning.
3. Semantic Search: When given a question, the system calculates cosine similarity between the question’s embedding and all document chunks to find the most relevant information.
4. Conversation Memory: The ConversationMemory class maintains a history of previous interactions, allowing the system to provide contextually relevant responses.
5. Persistent Storage: Processed data (chunks, embeddings, conversation history) is saved to disk using pickle, preventing redundant processing when the application restarts.
6. LLM Integration: Google’s Gemini model is used to generate natural language responses based on relevant document chunks and conversation history.
7. Interactive Interface: The system provides a simple command-line interface where users can ask questions about the content of their PDFs and receive AI-generated answers.
8. Source Attribution: Responses explicitly mention which PDF file(s) contained the relevant information, providing transparency and traceability.

Code for PDF Chatbot That Don’t Forget: Using Pickle for Persistent Chat

from sentence_transformers import SentenceTransformer
import fitz  # PyMuPDF
from langchain.text_splitter import RecursiveCharacterTextSplitter
import numpy as np
import google.generativeai as genai
import os
from typing import List, Dict
import pickle
from datetime import datetime

def save_processed_data(chunks_with_sources, all_chunks, chunk_vectors, memory, folder_path):
    """Save processed data and conversation history to local storage"""
    data = {
        'chunks_with_sources': chunks_with_sources,
        'all_chunks': all_chunks,
        'chunk_vectors': chunk_vectors,
        'conversation_history': memory.history,  # Add conversation history
        'timestamp': datetime.now(),
        'folder_path': folder_path
    }
    with open('processed_data.pkl', 'wb') as f:
        pickle.dump(data, f)
    print("Data and conversation history saved locally")

def load_processed_data(folder_path):
    """Load processed data from local storage"""
    try:
        with open('processed_data.pkl', 'rb') as f:
            data = pickle.load(f)

        # Check if the data is from the same folder
        if data['folder_path'] != folder_path:
            return None

        return data
    except FileNotFoundError:
        return None

class ChunkWithSource:
    def __init__(self, text: str, source: str):
        self.text = text
        self.source = source

def process_pdfs(folder_path: str) -> List[ChunkWithSource]:
    chunks_with_sources = []
    pdf_files = get_pdf_files(folder_path)

    for pdf_path in pdf_files:
        filename = os.path.basename(pdf_path)
        text = extract_text_from_pdf(pdf_path)
        chunks = split_text_into_chunks(text)

        for chunk in chunks:
            chunks_with_sources.append(ChunkWithSource(chunk, filename))

    return chunks_with_sources

def get_pdf_files(folder_path: str) -> List[str]:
    """Get all PDF files from the specified folder."""
    pdf_files = []
    for file in os.listdir(folder_path):
        if file.endswith('.pdf'):
            pdf_files.append(os.path.join(folder_path, file))
    return pdf_files

class ConversationMemory:
    def __init__(self, max_history: int = 5):
        self.history: List[Dict] = []
        self.max_history = max_history

    def add_interaction(self, query: str, response: str, context: str):
        self.history.append({
            "query": query,
            "response": response,
            "context": context
        })
        if len(self.history) > self.max_history:
            self.history.pop(0)

    def get_formatted_history(self) -> str:
        formatted = ""
        for interaction in self.history:
            formatted += f"Question: {interaction['query']}\n"
            formatted += f"Answer: {interaction['response']}\n"
        return formatted

def extract_text_from_pdf(pdf_path):
    text = ""
    with fitz.open(pdf_path) as doc:
        for page_num, page in enumerate(doc, start=1):
            text += page.get_text()
    return text

def split_text_into_chunks(text, chunk_size=1000, chunk_overlap=200):
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=chunk_size,
        chunk_overlap=chunk_overlap,
        separators=["\n\n", "\n", ".", "!", "?", " ", ""]
    )
    return splitter.split_text(text)

def sentence_encode(sentences):
    model = SentenceTransformer('sentence-transformers/all-MiniLM-L6-v2')
    embeddings = model.encode(sentences)
    return embeddings

def cosine_similarity(a, b):
    a = np.array(a)
    b = np.array(b)
    return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))

def has_folder_changed(folder_path, last_processed_time):
    """Check if any PDF in the folder has been modified since last processing"""
    for file in os.listdir(folder_path):
        if file.endswith('.pdf'):
            file_path = os.path.join(folder_path, file)
            if os.path.getmtime(file_path) > last_processed_time.timestamp():
                return True
    return False
if __name__ == "__main__":
    # Specify your PDFs folder path
    folder_path = "/Users/rudrabhaskar/Desktop/Python_Env/ML/Folder_with_pdfs"

    # Initialize conversation memory first
    memory = ConversationMemory()

     # Try to load existing processed data
    loaded_data = load_processed_data(folder_path)

    if loaded_data is not None:
        print("Loading previously processed data...")
        chunks_with_sources = loaded_data['chunks_with_sources']
        all_chunks = loaded_data['all_chunks']
        chunk_vectors = loaded_data['chunk_vectors']
        if 'conversation_history' in loaded_data:  # Load conversation history
            memory.history = loaded_data['conversation_history']
        print(f"Loaded {len(all_chunks)} chunks from local storage")
    else:
        print("Processing PDFs...")
        chunks_with_sources = process_pdfs(folder_path)
        if not chunks_with_sources:
            print("No PDF files found in the specified folder!")
            exit()

    # Process all PDFs with source tracking
    chunks_with_sources = process_pdfs(folder_path)
    if not chunks_with_sources:
        print("No PDF files found in the specified folder!")
        exit()

    # Extract just the text for embeddings
    all_chunks = [chunk.text for chunk in chunks_with_sources]
    print(f"Total chunks created: {len(all_chunks)}")

    # Create embeddings for all chunks
    chunk_vectors = sentence_encode(all_chunks)

    while True:
        # Get user input
        query = input("\nEnter your question (or 'quit' to exit): ")

        if query.lower() == 'quit':
            break

        query_vector = sentence_encode([query])
        top_k = 3

        similarities = []
        for idx, chunk_vec in enumerate(chunk_vectors):
            sim = cosine_similarity(chunk_vec, query_vector[0])
            similarities.append((sim, idx))

        print("Similarities:", similarities)

        print("==" * 20)

        # Sort by similarity descending and get top_k indices
        top_chunks = sorted(similarities, reverse=True)[:top_k]
        top_indices = [idx for _, idx in top_chunks]

        print("Top chunk indices:", top_indices)

        new_context = ""
        for i in top_indices:
            new_context += all_chunks[i] + "\n"

        GOOGLE_API_KEY = "Put Your gemini 2.0(Flash) API key here"

          # Create history-aware prompt
        conversation_history = memory.get_formatted_history()
        prompt_template = f"""You are a helpful assistant with access to previous conversation context and the current question.

Previous Conversation:
{conversation_history}

Current Context (from {chunks_with_sources[top_indices[0]].source}):
{new_context}

Current Question: {query}

Please provide a coherent answer that takes into account both the conversation history and the current context. Also mention which PDF file(s) contained the relevant information."""
        try:
                # Configure the API
                genai.configure(api_key=GOOGLE_API_KEY)

                # Initialize the model correctly
                model = genai.GenerativeModel('gemini-2.0-flash')

                # Generate response with the actual prompt
                response = model.generate_content(prompt_template)
                print("\nResponse:")
                print(response.text)
                # Store interaction in memory
                memory.add_interaction(query, response.text, new_context)
                # Save updated data including conversation
                save_processed_data(chunks_with_sources, all_chunks, chunk_vectors, memory, folder_path)
        except Exception as e:
                print(f"Error generating response: {str(e)}")

Line by Line breakdown of the code:

The Imports Section

from sentence_transformers import SentenceTransformer
import fitz  # PyMuPDF
from langchain.text_splitter import RecursiveCharacterTextSplitter
import numpy as np
import google.generativeai as genai
import os
from typing import List, Dict
import pickle
from datetime import datetime

• SentenceTransformer: Provides pre-trained models to convert text into numerical embeddings (vector representations)
• fitz (PyMuPDF): Allows reading and processing PDF files
• RecursiveCharacterTextSplitter: From LangChain library, helps divide text into manageable chunks
• numpy: For mathematical operations on embeddings
• google.generativeai: Google’s Generative AI API (Gemini model)
• os: For file and directory operations
• typing: Provides type hints for better code documentation
• pickle: For serializing and deserializing Python objects (saving/loading processed data)
• datetime: To track when data was processed

Data Management Functions

Save Processed Data

def save_processed_data(chunks_with_sources, all_chunks, chunk_vectors, memory, folder_path):
    """Save processed data and conversation history to local storage"""
    data = {
        'chunks_with_sources': chunks_with_sources,
        'all_chunks': all_chunks,
        'chunk_vectors': chunk_vectors,
        'conversation_history': memory.history,  # Add conversation history
        'timestamp': datetime.now(),
        'folder_path': folder_path
    }
    with open('processed_data.pkl', 'wb') as f:
        pickle.dump(data, f)
    print("Data and conversation history saved locally")

This function:

• Creates a dictionary containing all processed data (chunks, vectors, conversation history)
• Records the current time and folder path
• Uses pickle to save this data to a file called ‘processed_data.pkl’
• Confirms successful storage with a message

Load Processed Data

def load_processed_data(folder_path):
    """Load processed data from local storage"""
    try:
        with open('processed_data.pkl', 'rb') as f:
            data = pickle.load(f)

        # Check if the data is from the same folder
        if data['folder_path'] != folder_path:
            return None

        return data
    except FileNotFoundError:
        return None

This function:

• Attempts to read the previously saved data file
• Validates that the saved data matches the current folder path
• Returns either the loaded data or None if the file doesn’t exist or folder doesn’t match

Document Processing Classes and Functions

ChunkWithSource Class

class ChunkWithSource:
    def __init__(self, text: str, source: str):
        self.text = text
        self.source = source

A simple class that:

• Stores a chunk of text
• Keeps track of which file it came from

PDF Processing Function

def process_pdfs(folder_path: str) -> List[ChunkWithSource]:
    chunks_with_sources = []
    pdf_files = get_pdf_files(folder_path)

    for pdf_path in pdf_files:
        filename = os.path.basename(pdf_path)
        text = extract_text_from_pdf(pdf_path)
        chunks = split_text_into_chunks(text)

        for chunk in chunks:
            chunks_with_sources.append(ChunkWithSource(chunk, filename))

    return chunks_with_sources

This function:

• Gets a list of PDF files from the specified folder
• For each PDF, extracts all text content
• Splits the text into manageable chunks
• Creates ChunkWithSource objects for each chunk, noting which file it came from
• Returns all chunks with their source information

PDF File Discover

def get_pdf_files(folder_path: str) -> List[str]:
    """Get all PDF files from the specified folder."""
    pdf_files = []
    for file in os.listdir(folder_path):
        if file.endswith('.pdf'):
            pdf_files.append(os.path.join(folder_path, file))
    return pdf_files

A utility function that:

• Lists all files in the specified folder
• Filters to include only PDF files
• Returns full paths to those files

Conversation Management

class ConversationMemory:
    def __init__(self, max_history: int = 5):
        self.history: List[Dict] = []
        self.max_history = max_history
    def add_interaction(self, query: str, response: str, context: str):
        self.history.append({
            "query": query,
            "response": response,
            "context": context
        })
        if len(self.history) > self.max_history:
            self.history.pop(0)
    def get_formatted_history(self) -> str:
        formatted = ""
        for interaction in self.history:
            formatted += f"Question: {interaction['query']}\n"
            formatted += f"Answer: {interaction['response']}\n"
        return formatted

This class:

• Maintains a conversation history as a list of dictionaries
• Limits history to a maximum number of entries (default 5)
• Each entry contains the user’s query, the system’s response, and the context used
• Provides a formatted string representation of the conversation history

Text Processing Functions

PDF Text Extraction

def extract_text_from_pdf(pdf_path):
    text = ""
    with fitz.open(pdf_path) as doc:
        for page_num, page in enumerate(doc, start=1):
            text += page.get_text()
    return text

This function:

• Opens a PDF document using PyMuPDF
• Extracts text from each page
• Combines all text into a single string

Text Chunking

def split_text_into_chunks(text, chunk_size=1000, chunk_overlap=200):
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=chunk_size,
        chunk_overlap=chunk_overlap,
        separators=["\n\n", "\n", ".", "!", "?", " ", ""]
    )
    return splitter.split_text(text)

This function:

• Uses LangChain’s RecursiveCharacterTextSplitter
• Splits text into chunks of approximately 1000 characters
• Uses 200 characters of overlap between chunks to avoid losing context
• Tries to split at natural boundaries like paragraphs, sentences, or spaces

Semantic Search Functions

Embedding Generation

def sentence_encode(sentences):
    model = SentenceTransformer('sentence-transformers/all-MiniLM-L6-v2')
    embeddings = model.encode(sentences)
    return embeddings

This function:

• Loads a pre-trained sentence embedding model (all-MiniLM-L6-v2)
• Converts text into dense vector embeddings
• Returns these embeddings for later similarity comparison

Similarity Calculation

def cosine_similarity(a, b):
    a = np.array(a)
    b = np.array(b)
    return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))

This function:

• Converts inputs to numpy arrays
• Calculates cosine similarity between two vectors
• Higher values (closer to 1) indicate greater similarity

Folder Change Detection

def has_folder_changed(folder_path, last_processed_time):
    """Check if any PDF in the folder has been modified since last processing"""
    for file in os.listdir(folder_path):
        if file.endswith('.pdf'):
            file_path = os.path.join(folder_path, file)
            if os.path.getmtime(file_path) > last_processed_time.timestamp():
                return True
    return False

This function:

• Checks if any PDF file has been modified since the last processing time
• Returns True if changes are detected
• Used to determine if reprocessing is needed

Main Execution Block

if __name__ == "__main__":
    # Specify your PDFs folder path
    folder_path = "/Users/rudrabhaskar/Desktop/Python_Env/ML/Folder_with_pdfs"
# Initialize conversation memory first
    memory = ConversationMemory()
    # Try to load existing processed data
    loaded_data = load_processed_data(folder_path)

    if loaded_data is not None:
        print("Loading previously processed data...")
        chunks_with_sources = loaded_data['chunks_with_sources']
        all_chunks = loaded_data['all_chunks']
        chunk_vectors = loaded_data['chunk_vectors']
        if 'conversation_history' in loaded_data:  # Load conversation history
            memory.history = loaded_data['conversation_history']
        print(f"Loaded {len(all_chunks)} chunks from local storage")
    else:
        print("Processing PDFs...")
        chunks_with_sources = process_pdfs(folder_path)
        if not chunks_with_sources:
            print("No PDF files found in the specified folder!")
            exit()

This section:

• Sets the folder path containing PDFs
• Initializes the conversation memory
• Attempts to load previously processed data
• If successful, restores chunks, vectors, and conversation history
• If unsuccessful, processes PDFs from scratch
• Exits if no PDFs are found

# Process all PDFs with source tracking
    chunks_with_sources = process_pdfs(folder_path)
    if not chunks_with_sources:
        print("No PDF files found in the specified folder!")
        exit()

    # Extract just the text for embeddings
    all_chunks = [chunk.text for chunk in chunks_with_sources]
    print(f"Total chunks created: {len(all_chunks)}")

    # Create embeddings for all chunks
    chunk_vectors = sentence_encode(all_chunks)

This section:

• Processes PDFs and extracts chunks with their sources
• (Note: This code appears redundant as it’s already done in the else block above)
• Creates a list of just the text from each chunk
• Creates vector embeddings for all chunks

while True:
        # Get user input
        query = input("\nEnter your question (or 'quit' to exit): ")

        if query.lower() == 'quit':
            break

        query_vector = sentence_encode([query])
        top_k = 3

This section:

• Starts an interactive loop
• Gets user input (their question)
• Exits if the user types “quit”
• Creates a vector embedding for the user’s query
• Sets top_k to 3 (will retrieve 3 most relevant chunks)

similarities = []
        for idx, chunk_vec in enumerate(chunk_vectors):
            sim = cosine_similarity(chunk_vec, query_vector[0])
            similarities.append((sim, idx))

        print("Similarities:", similarities)
print("==" * 20)
        # Sort by similarity descending and get top_k indices
        top_chunks = sorted(similarities, reverse=True)[:top_k]
        top_indices = [idx for _, idx in top_chunks]
        print("Top chunk indices:", top_indices)

This section:

• Calculates similarity between the query and each chunk
• Stores each similarity score with its chunk index
• Sorts by similarity (highest first)
• Gets the indices of the top k most similar chunks
• Prints debugging information

new_context = ""
        for i in top_indices:
            new_context += all_chunks[i] + "\n"
GOOGLE_API_KEY = "Put Your gemini 2.0(Flash) API key here"

This section:

• Builds a context string from the top k most relevant chunks
• Sets the Google API key for accessing the Gemini model

# Create history-aware prompt
        conversation_history = memory.get_formatted_history()
        prompt_template = f"""You are a helpful assistant with access to previous conversation context and the current question.Previous Conversation:
{conversation_history}
Current Context (from {chunks_with_sources[top_indices[0]].source}):
{new_context}
Current Question: {query}
Please provide a coherent answer that takes into account both the conversation history and the current context. Also mention which PDF file(s) contained the relevant information."""

This section:

• Gets the formatted conversation history
• Creates a prompt template that includes:
• Previous conversation history
• Context from the relevant chunks
• The current question
• Instructions for the AI to reference the source files

try:
                # Configure the API
                genai.configure(api_key=GOOGLE_API_KEY)

                # Initialize the model correctly
                model = genai.GenerativeModel('gemini-2.0-flash')

                # Generate response with the actual prompt
                response = model.generate_content(prompt_template)
                print("\nResponse:")
                print(response.text)
                # Store interaction in memory
                memory.add_interaction(query, response.text, new_context)
                # Save updated data including conversation
                save_processed_data(chunks_with_sources, all_chunks, chunk_vectors, memory, folder_path)
        except Exception as e:
                print(f"Error generating response: {str(e)}")

This final section:

• Configures the Google Generative AI API with the API key
• Initializes the Gemini model
• Generates a response based on the prompt
• Prints the response
• Adds this interaction to the conversation memory
• Saves all updated data, including the new conversation history
• Handles any errors that might occur during response generation

Conclusion

This code creates a sophisticated PDF-based question-answering system that:

1. Processes PDF documents into searchable chunks
2. Stores conversation history for contextual awareness
3. Finds relevant document sections using semantic search
4. Generates contextually appropriate responses using Google’s Gemini model
5. Persists data to avoid redundant processing

The implementation demonstrates effective use of embedding models for semantic search and large language models for generating human-like responses based on document content.

Thank you for reading, I hope this article helped you understand the concepts.

Connect with me:

Linkedin | Email | Medium | X