Build intelligent QA systems using extractive and generative approaches with BERT-based models and retrieval-augmented generation.
Question answering (QA) systems automatically answer questions posed in natural language. From search engines and virtual assistants to customer support chatbots, QA is one of the most practical NLP applications.
Modern QA systems fall into two categories: extractive QA identifies answer spans from given context, while generative QA produces answers in natural language. The rise of transformer models like BERT has dramatically improved QA accuracy, and retrieval-augmented generation (RAG) combines the best of both retrieval and generation.
In this lesson, you'll build QA systems that can comprehend text and answer questions accurately.
Extractive QA finds answer spans within provided context.
Architecture:
1class ExtractiveQA:
2 """BERT-based extractive QA"""
3
4 def __init__(self, model_name='bert-base-uncased'):
5 self.tokenizer = AutoTokenizer.from_pretrained(model_name)
6 self.model = AutoModelForQuestionAnswering.from_pretrained(model_name)
7
8 def predict(self, question, context):
9 # Tokenize question and context
10 inputs = self.tokenizer(
11 question,
12 context,
13 return_tensors='pt',
14 max_length=512,
15 truncation=True
16 )
17
18 # Get start and end logits
19 outputs = self.model(**inputs)
20 start_logits = outputs.start_logits
21 end_logits = outputs.end_logits
22
23 # Find best answer span
24 start_idx = torch.argmax(start_logits)
25 end_idx = torch.argmax(end_logits)
26
27 # Decode answer
28 answer_tokens = inputs['input_ids'][0][start_idx:end_idx+1]
29 answer = self.tokenizer.decode(answer_tokens)
30
31 return answerThe model learns to predict start and end positions of the answer span.
RAG combines document retrieval with generative models.
1class RAGSystem:
2 """Retrieval-Augmented Generation for QA"""
3
4 def __init__(self):
5 self.retriever = DenseRetriever() # FAISS + embeddings
6 self.generator = T5ForConditionalGeneration.from_pretrained('t5-base')
7
8 def answer(self, question, documents):
9 # Step 1: Retrieve relevant passages
10 relevant_docs = self.retriever.retrieve(question, documents, top_k=5)
11
12 # Step 2: Concatenate context
13 context = " ".join([doc['text'] for doc in relevant_docs])
14
15 # Step 3: Generate answer
16 prompt = f"question: {question} context: {context}"
17 inputs = self.tokenizer(prompt, return_tensors='pt')
18
19 outputs = self.generator.generate(
20 inputs['input_ids'],
21 max_length=50,
22 num_beams=4
23 )
24
25 answer = self.tokenizer.decode(outputs[0], skip_special_tokens=True)
26 return answer, relevant_docsRAG is powerful for open-domain QA where answers aren't in a fixed context.
Dense retrieval uses embeddings for semantic similarity.
1class DenseRetriever:
2 """Dense passage retrieval with FAISS"""
3
4 def __init__(self, model_name='sentence-transformers/all-MiniLM-L6-v2'):
5 self.encoder = SentenceTransformer(model_name)
6 self.index = None
7 self.documents = []
8
9 def index_documents(self, documents):
10 """Create FAISS index for documents"""
11 self.documents = documents
12
13 # Encode all documents
14 embeddings = self.encoder.encode(
15 [doc['text'] for doc in documents],
16 show_progress_bar=True
17 )
18
19 # Build FAISS index
20 dimension = embeddings.shape[1]
21 self.index = faiss.IndexFlatIP(dimension) # Inner product
22 faiss.normalize_L2(embeddings)
23 self.index.add(embeddings)
24
25 def retrieve(self, query, top_k=5):
26 """Retrieve top-k relevant documents"""
27 query_embedding = self.encoder.encode([query])
28 faiss.normalize_L2(query_embedding)
29
30 scores, indices = self.index.search(query_embedding, top_k)
31
32 results = []
33 for score, idx in zip(scores[0], indices[0]):
34 results.append({
35 'document': self.documents[idx],
36 'score': float(score)
37 })
38
39 return resultsStandard metrics for QA evaluation:
1def exact_match(prediction, ground_truth):
2 """Exact match score"""
3 def normalize(text):
4 text = text.lower().strip()
5 # Remove articles and punctuation
6 text = re.sub(r'\b(a|an|the)\b', ' ', text)
7 text = re.sub(r'[^\w\s]', '', text)
8 return ' '.join(text.split())
9
10 return int(normalize(prediction) == normalize(ground_truth))
11
12def f1_score(prediction, ground_truth):
13 """Token-level F1 score"""
14 pred_tokens = prediction.lower().split()
15 truth_tokens = ground_truth.lower().split()
16
17 common = set(pred_tokens) & set(truth_tokens)
18
19 if len(common) == 0:
20 return 0
21
22 precision = len(common) / len(pred_tokens)
23 recall = len(common) / len(truth_tokens)
24
25 return 2 * precision * recall / (precision + recall)
26
27# Example evaluation
28predictions = ["Albert Einstein", "1879", "Germany"]
29ground_truths = ["Albert Einstein", "March 14, 1879", "Ulm, Germany"]
30
31for pred, truth in zip(predictions, ground_truths):
32 em = exact_match(pred, truth)
33 f1 = f1_score(pred, truth)
34 print(f"Pred: '{pred}' | Truth: '{truth}'")
35 print(f" EM: {em}, F1: {f1:.3f}")Create a question answering system with extractive and retrieval components.
1import numpy as np
2from collections import Counter
3import re
4
5class SimpleQA:
6 """Simple question answering system"""
7
8 def __init__(self):
9 self.documents = []
10 self.doc_vectors = []
11 self.vocabulary = set()
12
13 def preprocess(self, text):
14 """Clean and tokenize text"""
15 text = text.lower()
16 text = re.sub(r'[^\w\s]', '', text)
17 return text.split()
18
19 def build_tfidf(self, documents):
20 """Build TF-IDF vectors for documents"""
21 self.documents = documents
22
23 # Build vocabulary
24 all_tokens = []
25 doc_tokens = []
26 for doc in documents:
27 tokens = self.preprocess(doc['text'])
28 doc_tokens.append(tokens)
29 all_tokens.extend(tokens)
30
31 self.vocabulary = list(set(all_tokens))
32 vocab_idx = {word: i for i, word in enumerate(self.vocabulary)}
33
34 # Compute IDF
35 n_docs = len(documents)
36 doc_freq = Counter()
37 for tokens in doc_tokens:
38 for word in set(tokens):
39 doc_freq[word] += 1
40
41 self.idf = {word: np.log(n_docs / (df + 1)) for word, df in doc_freq.items()}
42
43 # Compute TF-IDF vectors
44 self.doc_vectors = []
45 for tokens in doc_tokens:
46 tf = Counter(tokens)
47 vector = np.zeros(len(self.vocabulary))
48 for word, count in tf.items():
49 if word in vocab_idx:
50 vector[vocab_idx[word]] = count * self.idf.get(word, 0)
51 # Normalize
52 norm = np.linalg.norm(vector)
53 if norm > 0:
54 vector = vector / norm
55 self.doc_vectors.append(vector)
56
57 self.vocab_idx = vocab_idx
58
59 def retrieve(self, query, top_k=3):
60 """Retrieve most relevant documents"""
61 # Query vector
62 tokens = self.preprocess(query)
63 tf = Counter(tokens)
64 query_vec = np.zeros(len(self.vocabulary))
65 for word, count in tf.items():
66 if word in self.vocab_idx:
67 query_vec[self.vocab_idx[word]] = count * self.idf.get(word, 0)
68
69 # Normalize
70 norm = np.linalg.norm(query_vec)
71 if norm > 0:
72 query_vec = query_vec / norm
73
74 # Compute similarities
75 scores = []
76 for i, doc_vec in enumerate(self.doc_vectors):
77 sim = np.dot(query_vec, doc_vec)
78 scores.append((sim, i))
79
80 # Sort and return top-k
81 scores.sort(reverse=True)
82 return [(self.documents[i], score) for score, i in scores[:top_k]]
83
84 def extract_answer(self, question, context):
85 """Extract answer span from context"""
86 question_words = set(self.preprocess(question))
87
88 # Split context into sentences
89 sentences = re.split(r'[.!?]', context)
90
91 # Score each sentence
92 best_sentence = ""
93 best_score = 0
94
95 for sentence in sentences:
96 if not sentence.strip():
97 continue
98
99 sent_words = set(self.preprocess(sentence))
100 # Score by word overlap with question
101 overlap = len(question_words & sent_words)
102
103 if overlap > best_score:
104 best_score = overlap
105 best_sentence = sentence.strip()
106
107 return best_sentence
108
109 def answer(self, question):
110 """Full QA pipeline"""
111 # Retrieve relevant documents
112 retrieved = self.retrieve(question, top_k=2)
113
114 # Combine contexts
115 combined_context = " ".join([doc['text'] for doc, _ in retrieved])
116
117 # Extract answer
118 answer = self.extract_answer(question, combined_context)
119
120 return {
121 'answer': answer,
122 'retrieved_docs': retrieved,
123 'confidence': retrieved[0][1] if retrieved else 0
124 }
125
126# Knowledge base
127documents = [
128 {
129 'id': 1,
130 'title': 'Albert Einstein',
131 'text': 'Albert Einstein was born on March 14, 1879 in Ulm, Germany. He developed the theory of relativity. Einstein received the Nobel Prize in Physics in 1921.'
132 },
133 {
134 'id': 2,
135 'title': 'Theory of Relativity',
136 'text': 'The theory of relativity was published by Einstein in 1905 (special) and 1915 (general). E=mc² is the famous equation from special relativity.'
137 },
138 {
139 'id': 3,
140 'title': 'Nobel Prize',
141 'text': 'The Nobel Prize is awarded annually. Einstein won for the photoelectric effect, not relativity. Marie Curie won two Nobel Prizes.'
142 },
143 {
144 'id': 4,
145 'title': 'Marie Curie',
146 'text': 'Marie Curie was born in Poland in 1867. She discovered radium and polonium. Curie won Nobel Prizes in Physics (1903) and Chemistry (1911).'
147 }
148]
149
150# Initialize QA system
151qa = SimpleQA()
152qa.build_tfidf(documents)
153
154print("QUESTION ANSWERING SYSTEM")
155print("=" * 60)
156
157# Test questions
158questions = [
159 "When was Einstein born?",
160 "What is the famous equation from relativity?",
161 "Who won two Nobel Prizes?",
162 "What did Marie Curie discover?"
163]
164
165print("\nAnswering Questions:")
166print("-" * 60)
167
168for question in questions:
169 result = qa.answer(question)
170 print(f"\nQ: {question}")
171 print(f"A: {result['answer']}")
172 print(f"Confidence: {result['confidence']:.3f}")
173 print(f"Source: {result['retrieved_docs'][0][0]['title']}")This demonstrates a complete QA pipeline with document retrieval using TF-IDF and simple extractive answering based on sentence scoring.
In our next lesson, "Text Summarization," you'll learn to build systems that condense documents while preserving key information.
Your progress is saved in this browser, on this device only.