[Python] How to design a RAG system to retrieve data from Washington State Tax Determinations...

I am having trouble extracting metadata from the WTD. They are available to the public. For example, if I ask specific questions about a specific WTD let's say 34 WTD 571, my chatbot can't pick up the name or certain information about the document. Here are some example questions I would like to ask a document and answers I would expect:

• What is the citation to this document? o Determination No. 14-0259. • What year was the case, ETA, interim guidance, or determination published? o 4/29/16. • What is the service provided by the Taxpayer? o In person and online training. • Provide a summary of the service provided by Taxpayer? o Training. • Who is the customer? o Employees of businesses that needed training on how to procure and contract for software. • What industry was the customer in? o Various industries which needed to buy software (most companies) • Provide a summary of the customer o Miscellaneous industries • What is the customer’s related business activity? o The related business activity was utilizing the licensing knowledge learned during training workshops. • Where does the customer’s related business activity occur? o Where the customer’s employees worked utilizing the licensing knowledge learned during training workshops. • Provide a summary of the customer’s related activity o Utilizing training knowledge. • Where are the benefits of the service received? o Where customer’s employees work. • What data should the taxpayer use to create a method of apportionment? o Specific data: if available specific oQice locations where the employees work. • Was section WAC 458-20-19402(301)(a) used to attribute the receipts? o No • If section WAC 458-20-19402(301)(a) was not used, what section was used and why? o Section WAC 458-20-19402(301)(b) was used because Taxpayer did not have access to information addressing where the customer’s employees worked. Therefore, Taxpayer used WAC 458

from flask import Flask, request, jsonify
from flask_cors import CORS
import pdfplumber
import re
from langchain.schema import Document
import os
import socket
import platform
from collections import deque
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.vectorstores import FAISS
from langchain_community.document_loaders import PyPDFLoader, DirectoryLoader
from langchain.prompts import PromptTemplate
from langchain_huggingface import HuggingFaceEmbeddings
from langchain_ollama import OllamaLLM # ✅ Using Ollama

app = Flask(__name__)
CORS(app)

# Initialize global variables
vector_store = None
pdf_directory = "pdfs"
conversation_history = deque(maxlen=5)

# ✅ Use Ollama's LLaMA 3 model
llm = OllamaLLM(model="llama3")

def generate_prompt(user_query, context):
"""Creates a structured and **highly detailed** prompt for the chatbot."""
return f"""
You are an expert Washington State tax consultant specializing in Washington tax laws.
Your role is to analyze various tax documents, court opinions, administrative determinations, and other legal guidance provided by the Washington State Department of Revenue.
Provide stuctured, legally defensible answers strictly based on published tax law with citations.

### **Guiding Principles for Responses:**
1. **Legal Foundation:** Always cite relevant RCW (statutes), WAC codes (administrative codes),
Determinations (Department of Revenue administrative cases), ETAs (excise tax advisories),
and tax topics (website-based documents published by the Department of Revenue).
2. **Yes/No Answers:** If a yes/no question is asked, clearly state:
- **"Yes, because [reason]"** or **"No, because [reason]"**
-When you answer, provide all relevant legal citations associated with your answer. For example:
-"XYZ's services do not relate to real property or tangible personal property but they do relate to its business activity,
and they must attribute their receipts to where their related business activity occurs" (WAC 458-20-19402(303)(c)).
-If the document does not address it directly say:
-"No direct reference found, but similar cases suggest [analysis]."
3. **Legal Changes Over Time:** Compare past and present versions of a law if asked about updates (e.g.,Rule 19402).
4. **Tax Apportionment Cases:**
-**Start with WAC 458-20-19402(301)(a)**: Check if the **customer's business location is available**.
-If unavailable, apply **WAC 458-20-19402(301)(b)** (location from which service was ordered).
-Ask follow-up questions if **data for (301)(a) is missing**.
5. **Query Refinement:** If insufficient data is found, suggest alternative phrasing or details to clarify the query.


**❓ User's Question:**
{user_query}

** Retrieved Tax Determinations:**
{context}

---

** Expert Response:**
"""

def ollama_generate(user_query, context):
"""Generates a structured response using the Llama3 model."""
formatted_prompt = generate_prompt(user_query, context)
return llm.invoke(formatted_prompt) # ✅ Uses the refined structured prompt

def normalize_determination_number(det_number):
"""Normalizes determination numbers for comparison."""
return det_number.replace(" ", "").replace("-", "")


def extract_text_from_pdf(pdf_path):
"""Extracts text content from a PDF file."""
text =[]
with pdfplumber.open(pdf_path) as pdf:
for page in pdf.pages:
text.append(page.extract_text())
return "\f".join(text)

def extract_metadata_from_text(full_text, first_page_text):
"""Extracts key metadata fields from the document text."""
metadata = {}

### ✅ Improved regex pattern for determination numbers
det_number_match = re.search(
r"(?:Cite\s+as\s+Det\.\s+No\.\s+|Det\.\s+No\.\s+)?(\d{1,4}-\d{4,5})?,?\s*(\d{1,2}\s*WTD\s*\d{1,4})",
first_page_text,
re.IGNORECASE
)

if det_number_match:
primary_det_number = det_number_match.group(1) or "N/A"
wtd_number = det_number_match.group(2) or "N/A"
metadata["determination_number"] = f"{primary_det_number}, {wtd_number}".strip(", ")
else:
metadata["determination_number"] = "N/A"

### ✅ Extract Publication Date ###
date_pattern = re.search(r'(\w+\s+\d{1,2},\s+\d{4})', first_page_text)
metadata["publication_date"] = date_pattern.group(1) if date_pattern else "Not listed"

### ✅ Extract Petition Status ###
case_status_match = re.search(
r"(petition\s+(?:is|was)\s+(granted|denied|approved|dismissed|partially granted|not granted|not approved)(\s+in\s+part)?)",
full_text,
re.IGNORECASE
)
metadata["case_status"] = case_status_match.group(1).capitalize if case_status_match else "Not specified"

### ✅ Extract Hearing Officer ###
hearing_officer_match = re.search(
r"(?:Hearing\s+Officer[:\s]+|By[:\s]+|Signed\s+by[:\s]+|Adjudicated\s+by[:\s]+)([\w\s]+,?\s*A\.L\.J\.)",
full_text,
re.IGNORECASE
)
metadata["hearing_officer"] = hearing_officer_match.group(1) if hearing_officer_match else "Not listed"

### ✅ Extract Business Activity ###
business_match = re.search(r"Taxpayer\s+operates\s+a\s+(.*?)\sin\s+\[Washington\]", full_text, re.IGNORECASE)
metadata["business_activity"] = business_match.group(1) if business_match else "Not specified"

### ✅ Extract Legal References (RCW, WAC Codes) ###
legal_references_match = re.findall(r"(RCW|WAC|ETA)\s\d{2,}-\d{2,}-\d+", full_text)
metadata["legal_references"] = ", ".join(legal_references_match) if legal_references_match else "Not specified"

### ✅ Extract Service Provided by Taxpayer ###
taxpayer_service_match = re.search(r"(?:Service\s+provided\s+by\s+Taxpayer[:\s]+)([\w\s]+)", full_text, re.IGNORECASE)
metadata["taxpayer_service"] = taxpayer_service_match.group(1) if taxpayer_service_match else "Not specified"

# Extract Case Decision (if explicitly mentioned)
case_status_match = re.search(
r"(petition\s+(?:is|was)\s+(granted|denied|approved|dismissed|not granted|not approved)(\s+in\s+part)?)",
full_text,
re.IGNORECASE
)
metadata["case_status"] = case_status_match.group(1).capitalize() if case_status_match and case_status_match.group(1) else "Not specified"

# Extract Taxpayer name
taxpayer_match = re.search(r"Taxpayer:\s+([^\n]+)", full_text, re.IGNORECASE)
if not taxpayer_match:
taxpayer_match = re.search(r"In\s+the\s+Matter\s+of\s+the\s+Petition\s+for\s+Correction\s+of\s+Assessment\s+of\s+([^\n]+)", full_text, re.IGNORECASE)
metadata["taxpayer"] = taxpayer_match.group(1) if taxpayer_match else "Not specified"

customer_industry_match = re.search(r"Customer's\s+Industry[:\s]+([\w\s]+)", full_text)
metadata["customer_industry"] = customer_industry_match.group(1) if customer_industry_match else "Not specified"

return metadata

def debug_check_index():
"""Prints all indexed determination numbers to verify correct indexing."""
if not vector_store:
print("⚠️ FAISS vector store is not initialized.")
return
print(" Determination Numbers in FAISS Index:")
for i, doc_id in enumerate(vector_store.docstore._dict.keys()):
doc = vector_store.docstore.search(doc_id)
if doc:
determination_number = doc.metadata.get("determination_number", "N/A")
source = doc.metadata.get("source", "Unknown")
print(f"{i+1}. {determination_number} - {source}")

def initialize_chatbot():
"""Loads and processes PDFs into FAISS vector database."""
global vector_store

if not os.path.exists(pdf_directory):
os.makedirs(pdf_directory)
print(f" Created directory: {pdf_directory}")
return False

pdf_files = [f for f in os.listdir(pdf_directory) if f.lower().endswith(".pdf")]
if not pdf_files:
return False

print(f" Found {len(pdf_files)} PDF files. Processing...")

try:
loader = DirectoryLoader(pdf_directory, glob="**/*.pdf", loader_cls=PyPDFLoader)
documents = loader.load()

# Extract metadata from each document
processed_docs = []
for doc in documents:
pages = doc.page_content.split("\f") # Splitting by form feed to get individual pages
first_page_text = pages[0] if pages else doc.page_content # Get only the first page
full_text = doc.page_content # Use full document for other metadata

metadata = extract_metadata_from_text(full_text, first_page_text)

# Add metadata to document object
processed_docs.append(Document(
page_content=doc.page_content,
metadata={
"source": doc.metadata.get("source", "Unknown"),
"determination_number": metadata["determination_number"],
"publication_date": metadata["publication_date"],
"legal_references": metadata["legal_references"],
"hearing_officer": metadata["hearing_officer"],
"taxpayer_service": metadata["taxpayer_service"],
"customer_industry": metadata["customer_industry"],
"business_activity": metadata["business_activity"],
"case_status": metadata["case_status"],
"taxpayer": metadata.get("taxpayer", "Not specified")
}
))

text_splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=100)
chunks = text_splitter.split_documents(processed_docs)

embeddings = HuggingFaceEmbeddings(model_name="sentence-transformers/all-MiniLM-L6-v2")
vector_store = FAISS.from_documents(chunks, embeddings)

print("✅ Chatbot initialized successfully with document retrieval.")
# Run debug check to verify indexing
debug_check_index()
return True

except Exception as e:
print(f"❌ Error initializing chatbot: {str(e)}")
return False

conversation_history = deque(maxlen=5)

@app.route("/ask", methods=["POST"])
def ask_question():
"""API endpoint to handle user queries."""
global conversation_history
data = request.json
user_query = data.get("query", "")

if not user_query:
return jsonify({"error": "No query provided."}), 400

if not vector_store:
return jsonify({"error": "Chatbot is not initialized."}), 500

# ✅ Improved pattern for extracting WTD references from queries
det_number_match = re.search(
r"(\d{1,2}\s*WTD\s*\d{1,4})",
user_query,
re.IGNORECASE
)
retrieved_docs = []
if det_number_match:
requested_determination = det_number_match.group(1).replace(" ", "")
print(f" Looking for specific determination: {requested_determination}")

# First try exact match

for doc in vector_store.similarity_search(user_query, k=10):
indexed_det_number = doc.metadata.get("determination_number", "")
if requested_determination in indexed_det_number:
retrieved_docs.append(doc)


# If no exact matches, fall back to similarity search
if not retrieved_docs:
print(f"⚠️ No exact match found for {requested_determination}. Using similarity search...")
retrieved_docs = vector_store.similarity_search(user_query, k=5)
else:
# Retrieve relevant tax determinations
retrieved_docs = vector_store.similarity_search(user_query, k=5)
if not retrieved_docs:
return jsonify({"response": "No relevant tax determinations found."})

conversation_context = "\n".join(conversation_history)
context = "\n\n".join([doc.page_content[:300] for doc in retrieved_docs])

full_prompt = f"Previous Conversations:\n{conversation_context}\n\nUser's Question:\n{user_query}\n\nRetrieved Information:\n{context}"
response = ollama_generate(user_query, full_prompt)

conversation_history.append(f"User: {user_query}\nChatbot: {response}")

# Extract sources (file names) & content
sources = set()
document_snippets = []

for doc in retrieved_docs:
source = doc.metadata.get('source', 'Unknown')
determination_number = doc.metadata.get("determination_number", "N/A")
title = os.path.basename(source)
publication_date = doc.metadata.get("publication_date", "Not listed")
case_status = doc.metadata.get("case_status", "Not specified")
legal_references = doc.metadata.get("legal_references", "Not specified")
taxpayer_service = doc.metadata.get("taxpayer_service", "Not specified")
customer_industry = doc.metadata.get("customer_industry", "Not specified")
business_activity = doc.metadata.get("business_activity", "Not listed")
hearing_officer = doc.metadata.get("hearing_officer", "Not listed")
taxpayer = doc.metadata.get("taxpayer", "Not specified")
source_type = "Court Case" if "Court of Appeal" in title else "DOR Determination"

sources.add(title)
document_snippets.append(
f" **{source_type}: {determination_number}** - {title}\n"
f"‍ **Taxpayer:** {taxpayer}\n"
f" **Publication Date:** {publication_date}\n"
f" **Excerpt:** {doc.page_content[:300]}...\n"
f" **Key Legal References:** {legal_references}\n"
f"‍ **Taxpayer Service:** {taxpayer_service}\n"
f" **Customer Industry:** {customer_industry}\n"
f" **Business Activity:** {business_activity}\n"
f"‍⚖️ **Hearing Officer:** {hearing_officer}\n"
f" **Case Status:** {case_status}\n"
)

# Generate response with enhanced formatting
context = "\n\n".join(document_snippets)
response = ollama_generate(user_query, context)

# Format the response with citations and indicate if specific document was found
if det_number_match:
requested_determination = det_number_match.group(1)
if any(normalize_determination_number(requested_determination) in
normalize_determination_number(doc.metadata.get("determination_number", ""))
for doc in retrieved_docs):
response_with_sources = f" **Analysis of {requested_determination}**\n\n{response}\n\n **Referenced Tax Determinations:**\n" + "\n\n".join(document_snippets)
else:
response_with_sources = f"⚠️ Could not find an exact match for {requested_determination}. Here are the most relevant results:\n\n{response}\n\n **Referenced Tax Determinations:**\n" + "\n\n".join(document_snippets)
else:
response_with_sources = f"{response}\n\n **Referenced Tax Determinations:**\n" + "\n\n".join(document_snippets)

return jsonify({"response": response_with_sources})

if __name__ == "__main__":
print(" Initializing chatbot...")
success = initialize_chatbot()

if success:
print("✅ Chatbot initialized successfully.")

# Extract metadata from 34 WTD 541.pdf
pdf_path = os.path.join(pdf_directory, "34 WTD 541.pdf")
if os.path.exists(pdf_path):
print(f" Extracting text from PDF: {pdf_path}")

with pdfplumber.open(pdf_path) as pdf:
full_text = "\n".join([page.extract_text() for page in pdf.pages if page.extract_text()])
first_page_text = pdf.pages[0].extract_text() if pdf.pages else ""

metadata = extract_metadata_from_text(full_text, first_page_text)
print(" Extracted Metadata:", metadata)
else:
print(f"⚠️ Warning: {pdf_path} not found. Ensure it is in the correct directory.")

# Interactive terminal mode
print("\n Chatbot is ready. Type your questions below (or type 'exit' to quit).")

while True:
user_query = input("\n You: ")
if user_query.lower() == "exit":
print(" Exiting chatbot.")
break

# Perform document search
retrieved_docs = []
det_number_match = re.search(r"(\d{1,2}\s*WTD\s*\d{1,4})", user_query, re.IGNORECASE)

if det_number_match:
requested_determination = det_number_match.group(1).replace(" ", "")
print(f" Looking for specific determination: {requested_determination}")

for doc in vector_store.similarity_search(user_query, k=10):
indexed_det_number = doc.metadata.get("determination_number", "")
if requested_determination in indexed_det_number:
retrieved_docs.append(doc)

if not retrieved_docs:
print(f"⚠️ No exact match found for {requested_determination}. Using similarity search...")
retrieved_docs = vector_store.similarity_search(user_query, k=5)
else:
retrieved_docs = vector_store.similarity_search(user_query, k=5)

if not retrieved_docs:
print(" Chatbot: No relevant tax determinations found.")
continue

# Generate response
context = "\n\n".join([doc.page_content[:300] for doc in retrieved_docs])
response = ollama_generate(user_query, context)

print(f"\n Chatbot:\n{response}")

else:
print("⚠️ Chatbot initialization failed. Check logs for errors.")

Continue reading...