rimsha-sudo/chatbot
0
1import streamlit as st
2from sentence_transformers import SentenceTransformer
3import faiss
4import numpy as np
5import logging
6import re
7from typing import List
8
9logger = logging.getLogger(__name__)
10
11class ChatBot:
12 def __init__(self, document_chunks):
13 self.document_chunks = document_chunks
14 self.full_text = " ".join(document_chunks)
15
16 # Initializing embedding model
17 self.embedding_model = SentenceTransformer('all-MiniLM-L6-v2')
18
19 # Created vector store if multiple chunks
20 if len(document_chunks) > 1:
21 self._create_vector_store()
22 else:
23 self.use_vector_search = False
24
25 def _create_vector_store(self):
26 #creating a vector store and using FAISS
27 try:
28 embeddings = self.embedding_model.encode(self.document_chunks)
29 dimension = embeddings.shape[1]
30 self.index = faiss.IndexFlatL2(dimension)
31 self.index.add(embeddings.astype('float32'))
32 self.use_vector_search = True
33 logger.info(f"Vector store created with {len(self.document_chunks)} chunks")
34 except Exception as e:
35 logger.error(f"Error creating vector store: {e}")
36 self.use_vector_search = False
37
38 def _find_relevant_content(self, query: str) -> str:
39 """Find relevant content using multiple strategies"""
40 query_lower = query.lower()
41
42 #to get better responses using different techniques and making sure
43 # the chatbot handles different types of questions accuractely
44 # Vector search if available
45 if hasattr(self, 'use_vector_search') and self.use_vector_search:
46 try:
47 query_embedding = self.embedding_model.encode([query])
48 k = min(3, len(self.document_chunks))
49 distances, indices = self.index.search(query_embedding.astype('float32'), k)
50 relevant_chunks = [self.document_chunks[i] for i in indices[0]]
51 return " ".join(relevant_chunks)
52 except:
53 pass
54
55 # Strategy 2: Keyword matching with context
56 return self._keyword_search_with_context(query_lower)
57
58 def _keyword_search_with_context(self, query_lower: str) -> str:
59 """Smart keyword search that finds relevant sections with context"""
60
61 # Extract important keywords from query
62 query_words = [word for word in query_lower.split() if len(word) > 2]
63
64 # Split document into sentences
65 sentences = re.split(r'[.!?]+', self.full_text)
66 sentences = [s.strip() for s in sentences if s.strip()]
67
68 # Score each sentence based on keyword matches
69 sentence_scores = []
70 for i, sentence in enumerate(sentences):
71 sentence_lower = sentence.lower()
72 score = 0
73
74 # Count keyword matches
75 for word in query_words:
76 if word in sentence_lower:
77 score += 1
78
79 # Bonus for exact phrase matches
80 if any(phrase in sentence_lower for phrase in [query_lower, " ".join(query_words[:2])]):
81 score += 2
82
83 sentence_scores.append((score, i, sentence))
84
85 # Sort by score and get top sentences
86 sentence_scores.sort(reverse=True, key=lambda x: x[0])
87
88 # Get best matching sentences with context
89 relevant_sentences = []
90 for score, idx, sentence in sentence_scores[:3]:
91 if score > 0: # Only include sentences with keyword matches
92 # Add context (previous and next sentence if available)
93 context_sentences = []
94
95 # Add previous sentence if relevant
96 if idx > 0 and len(relevant_sentences) == 0:
97 context_sentences.append(sentences[idx-1])
98
99 context_sentences.append(sentence)
100
101 # Add next sentence if relevant
102 if idx < len(sentences) - 1:
103 next_sentence = sentences[idx + 1]
104 if any(word in next_sentence.lower() for word in query_words):
105 context_sentences.append(next_sentence)
106
107 relevant_sentences.extend(context_sentences)
108
109 # If no good matches, return first part of document
110 if not relevant_sentences:
111 return self.full_text[:1000]
112
113 return ". ".join(relevant_sentences[:4]) + "."
114
115 def _format_answer(self, query: str, content: str) -> str:
116 """Format the final answer based on query type and content"""
117 query_lower = query.lower()
118
119 # Handle list-type questions
120 if any(word in query_lower for word in ['types', 'list', 'applications', 'kinds', 'examples']):
121 return self._extract_list_format(content, query_lower)
122
123 # Handle definition questions
124 if any(word in query_lower for word in ['what is', 'define', 'definition', 'meaning']):
125 return self._extract_definition(content, query_lower)
126
127 # Handle "how" questions
128 if query_lower.startswith('how'):
129 return self._extract_process(content)
130
131 # Default: return relevant content cleanly
132 return self._clean_content(content)
133
134 def _extract_list_format(self, content: str, query: str) -> str:
135 """Extract and format list items"""
136 lines = content.split('\n')
137 list_items = []
138
139 # Look for numbered or bulleted lists
140 for line in lines:
141 line = line.strip()
142 if re.match(r'^[0-9]+\.', line) or line.startswith('-') or line.startswith('•'):
143 list_items.append(line)
144 elif ':' in line and any(keyword in query for keyword in ['types', 'applications']):
145 if 'include' in line.lower():
146 parts = line.split('include')
147 if len(parts) > 1:
148 items = parts[1].split(',')
149 for item in items:
150 list_items.append(f"- {item.strip()}")
151
152 if list_items:
153 return '\n'.join(list_items)
154
155 # Fallback: look for comma-separated items
156 sentences = re.split(r'[.!?]+', content)
157 for sentence in sentences:
158 if ',' in sentence and any(keyword in sentence.lower() for keyword in ['include', 'are']):
159 return sentence.strip()
160
161 return self._clean_content(content)
162
163 def _extract_definition(self, content: str, query: str) -> str:
164 """Extract clean definitions"""
165 # Look for definition patterns
166 sentences = re.split(r'[.!?]+', content)
167
168 # Find sentence with "is" or "are" that defines the term
169 for sentence in sentences:
170 sentence = sentence.strip()
171 if ' is ' in sentence.lower() or ' are ' in sentence.lower():
172 # Check if it's actually defining something
173 if len(sentence) > 20 and not sentence.lower().startswith('there'):
174 return sentence + "."
175
176 # Fallback to first substantial sentence
177 for sentence in sentences:
178 if len(sentence.strip()) > 30:
179 return sentence.strip() + "."
180
181 return self._clean_content(content)
182
183 def _extract_process(self, content: str) -> str:
184 """Extract process or method descriptions"""
185 sentences = re.split(r'[.!?]+', content)
186 relevant = []
187
188 for sentence in sentences:
189 sentence = sentence.strip()
190 if any(word in sentence.lower() for word in ['process', 'method', 'work', 'function', 'operate']):
191 relevant.append(sentence)
192
193 if relevant:
194 return ". ".join(relevant[:2]) + "."
195
196 return self._clean_content(content)
197
198 def _clean_content(self, content: str) -> str:
199 """Clean and format content for final output"""
200 # Remove extra whitespace and clean up
201 content = re.sub(r'\s+', ' ', content).strip()
202
203 # Ensure proper sentence ending
204 if content and not content.endswith(('.', '!', '?')):
205 content += "."
206
207 # Limit length
208 if len(content) > 500:
209 sentences = re.split(r'[.!?]+', content)
210 content = ". ".join(sentences[:3]) + "."
211
212 return content
213
214 def get_response(self, question: str) -> str:
215 """Generate response using improved logic"""
216 try:
217 # Find relevant content
218 relevant_content = self._find_relevant_content(question)
219
220 # Format answer based on question type
221 answer = self._format_answer(question, relevant_content)
222
223 # Final cleanup
224 if not answer or len(answer.strip()) < 5:
225 return "I couldn't find relevant information to answer your question. Try rephrasing or asking about different aspects of the document."
226
227 return answer
228
229 except Exception as e:
230 logger.error(f"Error generating response: {e}")
231 return "I encountered an error while processing your question. Please try again with a different question."