📖 第9章:问答系统¶
学习时间:8小时 难度星级:⭐⭐⭐⭐ 前置知识:文本分类、预训练模型、信息检索基础 学习目标:掌握检索式和生成式问答系统,理解阅读理解和对话系统
📋 目录¶
- 1. 问答系统概述
- 2. 检索式问答
- 3. 阅读理解
- 4. 知识库问答(KBQA)
- 5. 生成式问答
- 6. 开放域对话系统
- 7. RAG问答系统
- 8. 实战:构建智能问答系统
- 9. 面试要点
- 10. 练习题
1. 问答系统概述¶
1.1 问答系统的分类¶
Python
qa_taxonomy = {
"按数据源分": {
"文本QA": "从文档/段落中找答案(阅读理解)",
"知识库QA": "从结构化知识库中查询(KBQA)",
"表格QA": "从表格数据中回答(Table QA)",
"多模态QA": "结合图片、视频回答(VQA)",
},
"按回答方式分": {
"抽取式": "从文本中抽取答案片段",
"生成式": "生成自然语言答案",
"选择式": "从候选答案中选择",
},
"按领域分": {
"开放域": "任何问题(如搜索引擎)",
"封闭域": "特定领域(如医疗问答、法律咨询)",
},
}
qa_pipeline = """
问答系统架构演进:
1. 传统Pipeline:
问题 → 问题分析 → 信息检索 → 答案抽取 → 答案
2. 阅读理解:
问题 + 段落 → BERT → 答案span
3. RAG (当前主流):
问题 → 检索相关文档 → LLM生成答案
"""
print(qa_pipeline)
2. 检索式问答¶
2.1 基于TF-IDF的检索¶
Python
import numpy as np
from collections import Counter
import math
import jieba
class TFIDFRetriever:
"""基于TF-IDF的文档检索器"""
def __init__(self):
self.documents = []
self.doc_vectors = []
self.idf = {}
self.vocab = {}
def fit(self, documents):
"""建立索引"""
self.documents = documents
tokenized_docs = [jieba.lcut(doc) for doc in documents]
# 建立词汇表
all_words = set()
for doc in tokenized_docs:
all_words.update(doc)
self.vocab = {w: i for i, w in enumerate(sorted(all_words))} # enumerate同时获取索引和元素
# 计算IDF
n_docs = len(documents)
df = Counter() # Counter统计元素出现次数
for doc in tokenized_docs:
df.update(set(doc))
self.idf = {w: math.log(n_docs / (df[w] + 1)) for w in self.vocab}
# 计算文档TF-IDF向量
self.doc_vectors = []
for doc in tokenized_docs:
vec = self._tfidf_vector(doc)
self.doc_vectors.append(vec)
def _tfidf_vector(self, tokens):
"""计算TF-IDF向量"""
vec = np.zeros(len(self.vocab))
tf = Counter(tokens)
for word, count in tf.items():
if word in self.vocab:
idx = self.vocab[word]
vec[idx] = (count / len(tokens)) * self.idf.get(word, 0)
# L2归一化
norm = np.linalg.norm(vec) # np.linalg线性代数运算
if norm > 0:
vec = vec / norm
return vec
def search(self, query, top_k=3):
"""检索最相关的文档"""
query_tokens = jieba.lcut(query)
query_vec = self._tfidf_vector(query_tokens)
scores = []
for i, doc_vec in enumerate(self.doc_vectors):
score = np.dot(query_vec, doc_vec) # np.dot矩阵/向量点乘
scores.append((i, score))
scores.sort(key=lambda x: -x[1]) # lambda匿名函数
return [(self.documents[i], score) for i, score in scores[:top_k]]
# 构建知识库
knowledge_base = [
"Python是一种高级编程语言,由Guido van Rossum于1991年创建。Python以简洁易读著称。",
"机器学习是人工智能的一个分支,通过数据训练模型来做预测。常见算法包括决策树、SVM、神经网络等。",
"深度学习是机器学习的子领域,使用多层神经网络处理复杂任务。代表性架构有CNN、RNN、Transformer。",
"自然语言处理(NLP)是AI处理人类语言的技术。核心任务包括文本分类、命名实体识别、机器翻译等。",
"BERT是Google在2018年提出的预训练语言模型,使用Transformer编码器和双向训练目标。",
"GPT是OpenAI提出的生成式预训练模型,使用Transformer解码器和自回归语言模型目标。",
"Transformer是2017年提出的注意力机制架构,完全基于Self-Attention,取代了RNN和CNN。",
"RAG(检索增强生成)结合了检索和生成,先从知识库检索相关文档,再用大模型生成答案。",
]
retriever = TFIDFRetriever()
retriever.fit(knowledge_base)
# 测试检索
queries = ["什么是BERT?", "Python是什么语言?", "什么是RAG?"]
for query in queries:
results = retriever.search(query, top_k=2)
print(f"\n❓ {query}")
for doc, score in results:
print(f" [{score:.3f}] {doc[:50]}...") # 切片操作,取前n个元素
2.2 基于向量的语义检索¶
Python
class SemanticRetriever:
"""基于语义向量的检索器"""
def __init__(self, embedding_dim=64):
self.embedding_dim = embedding_dim
self.document_embeddings = []
self.documents = []
def encode(self, text):
"""简易文本编码(实际应使用sentence-transformers)"""
tokens = jieba.lcut(text)
np.random.seed(hash(text) % 2**31)
return np.random.randn(self.embedding_dim)
def index(self, documents):
"""建立向量索引"""
self.documents = documents
self.document_embeddings = [self.encode(doc) for doc in documents]
# 归一化
self.document_embeddings = [
e / np.linalg.norm(e) for e in self.document_embeddings
]
def search(self, query, top_k=3):
"""语义搜索"""
query_emb = self.encode(query)
query_emb = query_emb / np.linalg.norm(query_emb)
scores = [np.dot(query_emb, doc_emb) for doc_emb in self.document_embeddings]
ranked = sorted(enumerate(scores), key=lambda x: -x[1])
return [(self.documents[i], score) for i, score in ranked[:top_k]]
print("\n实际推荐使用的语义检索工具:")
print(" 1. sentence-transformers: 文本编码为向量")
print(" 2. FAISS: 高效向量相似搜索")
print(" 3. Milvus: 分布式向量数据库")
print(" 4. Chromadb: 轻量级向量数据库")
3. 阅读理解¶
3.1 抽取式阅读理解¶
Python
"""
抽取式阅读理解(Extractive MRC):
给定问题Q和段落P,从P中抽取一个连续的子串作为答案。
P: "BERT是Google在2018年提出的预训练语言模型,使用了Transformer架构。"
Q: "BERT是谁提出的?"
A: "Google"(从段落中抽取)
模型输出两个位置:答案开始位置和结束位置
"""
import torch
import torch.nn as nn
class MRCModel(nn.Module): # 继承nn.Module定义网络层
"""简化版阅读理解模型"""
def __init__(self, vocab_size, embed_dim=128, hidden_dim=256):
super().__init__() # super()调用父类方法
self.embedding = nn.Embedding(vocab_size, embed_dim, padding_idx=0)
# 问题编码
self.question_encoder = nn.LSTM(
embed_dim, hidden_dim, batch_first=True, bidirectional=True
)
# 段落编码
self.passage_encoder = nn.LSTM(
embed_dim, hidden_dim, batch_first=True, bidirectional=True
)
# 注意力
self.attention = nn.Linear(hidden_dim * 4, 1)
# 预测开始和结束位置
self.start_fc = nn.Linear(hidden_dim * 2, 1)
self.end_fc = nn.Linear(hidden_dim * 2, 1)
def forward(self, passage, question):
# 编码
p_emb = self.embedding(passage)
q_emb = self.embedding(question)
p_out, _ = self.passage_encoder(p_emb)
q_out, _ = self.question_encoder(q_emb)
# 计算问题的表示(取平均)
q_repr = q_out.mean(dim=1, keepdim=True)
# 注意力加权的段落表示
q_expanded = q_repr.expand_as(p_out)
combined = torch.cat([p_out, q_expanded], dim=-1) # torch.cat沿已有维度拼接张量
attn_scores = self.attention(combined)
# 预测开始和结束位置
start_logits = self.start_fc(p_out).squeeze(-1) # squeeze压缩维度
end_logits = self.end_fc(p_out).squeeze(-1)
return start_logits, end_logits
print("阅读理解模型创建成功")
3.2 使用BERT做阅读理解¶
Python
# 使用Hugging Face的BERT阅读理解
from transformers import AutoTokenizer, AutoModelForQuestionAnswering
import torch
def bert_qa(question, context, model_name="uer/roberta-base-chinese-extractive-qa"):
"""使用BERT进行阅读理解"""
try: # try/except捕获异常
tokenizer = AutoTokenizer.from_pretrained(model_name)
model = AutoModelForQuestionAnswering.from_pretrained(model_name)
inputs = tokenizer(question, context, return_tensors="pt",
max_length=512, truncation=True)
with torch.no_grad(): # 禁用梯度计算,节省内存
outputs = model(**inputs)
start_idx = outputs.start_logits.argmax()
end_idx = outputs.end_logits.argmax()
answer_tokens = inputs["input_ids"][0][start_idx:end_idx+1]
answer = tokenizer.decode(answer_tokens)
return answer
except Exception as e:
return f"需要下载模型: {e}"
# 测试
context = "自然语言处理是人工智能的重要分支。BERT是2018年由Google提出的预训练模型。"
questions = [
"BERT是什么时候提出的?",
"BERT是谁提出的?",
"自然语言处理是什么的分支?",
]
print("BERT阅读理解测试:")
for q in questions:
print(f" Q: {q}")
a = bert_qa(q, context)
print(f" A: {a}")
3.3 SQuAD数据集格式¶
Python
# SQuAD数据集格式
squad_example = {
"title": "自然语言处理",
"paragraphs": [
{
"context": "自然语言处理(NLP)是人工智能和计算语言学的交叉领域。"
"NLP的目标是让计算机理解、生成和处理人类语言。"
"常见的NLP任务包括文本分类、命名实体识别、机器翻译等。",
"qas": [
{
"question": "NLP是哪两个领域的交叉?",
"answers": [{"text": "人工智能和计算语言学", "answer_start": 10}],
"id": "q001",
},
{
"question": "NLP的目标是什么?",
"answers": [{"text": "让计算机理解、生成和处理人类语言", "answer_start": 35}],
"id": "q002",
},
],
}
],
}
print("SQuAD数据格式示例:")
for para in squad_example["paragraphs"]:
print(f"\n 段落: {para['context'][:50]}...")
for qa in para["qas"]:
print(f" Q: {qa['question']}")
print(f" A: {qa['answers'][0]['text']} (起始位置: {qa['answers'][0]['answer_start']})")
4. 知识库问答(KBQA)¶
Python
class SimpleKBQA:
"""简易知识库问答系统"""
def __init__(self):
# 知识库三元组
self.triples = [
("北京", "是...的首都", "中国"),
("中国", "人口", "14亿"),
("Python", "创建者", "Guido van Rossum"),
("Python", "创建年份", "1991年"),
("BERT", "提出者", "Google"),
("BERT", "发表年份", "2018年"),
("Transformer", "提出论文", "Attention Is All You Need"),
("GPT", "提出者", "OpenAI"),
("苹果公司", "CEO", "蒂姆·库克"),
("苹果公司", "创始人", "史蒂夫·乔布斯"),
]
self.entity_index = {}
for h, r, t in self.triples:
self.entity_index.setdefault(h, []).append((r, t))
self.entity_index.setdefault(t, []).append((r, h))
def parse_question(self, question):
"""问题解析:识别实体和关系意图"""
# 简单的实体识别
mentioned_entities = []
for entity in self.entity_index:
if entity in question:
mentioned_entities.append(entity)
# 关系意图识别
intent_keywords = {
"谁": ["创建者", "提出者", "CEO", "创始人"],
"什么时候": ["创建年份", "发表年份"],
"首都": ["是...的首都"],
"人口": ["人口"],
}
possible_relations = []
for keyword, relations in intent_keywords.items():
if keyword in question:
possible_relations.extend(relations)
return mentioned_entities, possible_relations
def answer(self, question):
"""回答问题"""
entities, relations = self.parse_question(question)
if not entities:
return "抱歉,我无法识别问题中的实体。"
answers = []
for entity in entities:
if entity in self.entity_index:
for rel, value in self.entity_index[entity]:
if not relations or rel in relations:
answers.append(f"{entity}的{rel}是{value}")
if answers:
return " | ".join(answers[:3])
else:
return f"抱歉,我没有找到关于{entities[0]}的相关信息。"
# 测试
kbqa = SimpleKBQA()
questions = [
"Python是谁创建的?",
"BERT是什么时候发表的?",
"中国的首都是哪里?",
"苹果公司的CEO是谁?",
"BERT的提出者是谁?",
]
print("KBQA问答测试:")
for q in questions:
a = kbqa.answer(q)
print(f"\n ❓ {q}")
print(f" 💡 {a}")
5. 生成式问答¶
Python
class GenerativeQA:
"""生成式问答(基于模板和检索)"""
def __init__(self, retriever):
self.retriever = retriever
def answer(self, question, top_k=2):
"""检索+生成答案"""
# 1. 检索相关文档
results = self.retriever.search(question, top_k=top_k)
# 2. 构建上下文
context = "\n".join([doc for doc, _ in results])
# 3. 生成答案(实际应使用LLM)
prompt = f"""基于以下参考资料回答问题。
参考资料:
{context}
问题: {question}
答案:"""
return {
"prompt": prompt,
"retrieved_docs": results,
"answer": f"[需要LLM生成] 基于检索到的文档回答问题",
}
# 测试
gqa = GenerativeQA(retriever)
result = gqa.answer("什么是Transformer?")
print("生成式问答:")
print(f" 问题: 什么是Transformer?")
print(f" 检索文档: {len(result['retrieved_docs'])}条")
print(f" Prompt:\n{result['prompt'][:200]}...")
6. 开放域对话系统¶
Python
class DialogueSystem:
"""简易对话系统"""
def __init__(self):
self.history = []
self.max_history = 10
# 意图识别规则
self.intents = {
"问候": ["你好", "嗨", "hello", "hi", "早上好", "晚上好"],
"告别": ["再见", "拜拜", "bye", "goodbye"],
"感谢": ["谢谢", "感谢", "thanks"],
"天气": ["天气", "温度", "下雨"],
"NLP知识": ["NLP", "自然语言", "BERT", "GPT", "Transformer", "机器翻译"],
}
# 回复模板
self.responses = {
"问候": ["你好!有什么可以帮助你的?", "嗨!很高兴见到你!"],
"告别": ["再见!祝你一切顺利!", "拜拜,欢迎下次再来!"],
"感谢": ["不客气!", "很高兴能帮到你!"],
"天气": ["抱歉,我目前无法查询天气信息。"],
"NLP知识": [
"这是一个很好的NLP问题!NLP(自然语言处理)是AI处理人类语言的技术。",
"关于NLP,你可以参考我们的NLP教程系列,从基础到进阶都有详细讲解。",
],
"未知": ["抱歉,我不太理解你的问题。能请你换个方式问吗?"],
}
def classify_intent(self, text):
"""意图识别"""
text_lower = text.lower()
for intent, keywords in self.intents.items():
for kw in keywords:
if kw.lower() in text_lower:
return intent
return "未知"
def respond(self, user_input):
"""生成回复"""
import random
self.history.append({"role": "user", "content": user_input})
intent = self.classify_intent(user_input)
response = random.choice(self.responses[intent])
self.history.append({"role": "assistant", "content": response})
# 保持历史长度
if len(self.history) > self.max_history * 2:
self.history = self.history[-self.max_history * 2:]
return response, intent
# 测试对话
dialogue = DialogueSystem()
test_inputs = [
"你好",
"我想了解一下NLP",
"BERT是什么?",
"谢谢",
"再见",
]
print("对话系统测试:")
print("="*50)
for user_input in test_inputs:
response, intent = dialogue.respond(user_input)
print(f" 用户: {user_input}")
print(f" 系统: {response} [意图: {intent}]")
print()
7. RAG问答系统¶
Python
class SimpleRAG:
"""简易RAG(检索增强生成)系统"""
def __init__(self):
self.retriever = TFIDFRetriever()
self.knowledge_base = []
def load_knowledge(self, documents):
"""加载知识库"""
self.knowledge_base = documents
self.retriever.fit(documents)
print(f"已加载 {len(documents)} 条知识")
def build_prompt(self, question, retrieved_docs, max_context_len=1000):
"""构建RAG prompt"""
context = "\n---\n".join([doc for doc, _ in retrieved_docs])
if len(context) > max_context_len:
context = context[:max_context_len]
prompt = f"""你是一个知识问答助手。请根据以下参考资料回答用户问题。
如果参考资料中没有相关信息,请如实告知。
### 参考资料:
{context}
### 问题:
{question}
### 回答:
"""
return prompt
def answer(self, question, top_k=3):
"""RAG问答"""
# Step 1: 检索
retrieved = self.retriever.search(question, top_k=top_k)
# Step 2: 构建Prompt
prompt = self.build_prompt(question, retrieved)
# Step 3: 生成(这里模拟,实际应调用LLM API)
# response = call_llm(prompt)
return {
"question": question,
"retrieved_docs": retrieved,
"prompt": prompt,
"answer": "[需调用LLM API生成答案]",
}
def evaluate(self, qa_pairs):
"""评估RAG系统"""
results = []
for question, expected_answer in qa_pairs:
result = self.answer(question)
# 简单评估:检索到的文档是否包含答案
has_answer = any( # any()任一为True则返回True
expected_answer in doc for doc, _ in result['retrieved_docs']
)
results.append({
"question": question,
"expected": expected_answer,
"retrieval_hit": has_answer,
})
hit_rate = sum(r['retrieval_hit'] for r in results) / len(results)
return {"hit_rate": hit_rate, "details": results}
# 构建RAG系统
rag = SimpleRAG()
rag.load_knowledge(knowledge_base)
# 测试
qa_pairs = [
("什么是BERT?", "预训练语言模型"),
("Transformer是什么时候提出的?", "2017"),
("RAG是什么?", "检索增强生成"),
]
print("\nRAG问答系统测试:")
for q, expected in qa_pairs:
result = rag.answer(q, top_k=2)
print(f"\n❓ {q}")
print(f"📚 检索到 {len(result['retrieved_docs'])} 条文档")
for doc, score in result['retrieved_docs']:
print(f" [{score:.3f}] {doc[:60]}...")
# 评估
eval_result = rag.evaluate(qa_pairs)
print(f"\n📊 检索命中率: {eval_result['hit_rate']:.1%}")
8. 实战:构建智能问答系统¶
Python
"""
实战项目:基于检索的NLP知识问答系统
"""
class NLPQASystem:
"""NLP领域智能问答系统"""
def __init__(self):
self.rag = SimpleRAG()
self.dialogue = DialogueSystem()
self.faq = self._build_faq()
def _build_faq(self):
"""构建FAQ库"""
return {
"什么是NLP": "NLP(自然语言处理)是人工智能的重要分支,旨在让计算机理解和处理人类自然语言。",
"什么是BERT": "BERT是Google在2018年提出的预训练语言模型,采用双向Transformer编码器架构。",
"什么是GPT": "GPT是OpenAI提出的生成式预训练模型,采用Transformer解码器架构,通过自回归方式生成文本。",
"什么是Transformer": "Transformer是2017年提出的注意力机制架构,核心是Self-Attention,已成为NLP的标准架构。",
"什么是RAG": "RAG(检索增强生成)先从知识库检索相关文档,再用大语言模型生成答案,是当前主流的问答范式。",
}
def answer(self, question):
"""回答问题"""
# 1. 先查FAQ
for faq_q, faq_a in self.faq.items():
if faq_q in question or question in faq_q:
return {"source": "FAQ", "answer": faq_a}
# 2. 检索知识库
if hasattr(self.rag, 'knowledge_base') and self.rag.knowledge_base: # hasattr检查对象是否有某属性
result = self.rag.answer(question, top_k=2)
if result['retrieved_docs'] and result['retrieved_docs'][0][1] > 0.1:
return {
"source": "RAG",
"answer": result['retrieved_docs'][0][0],
"confidence": result['retrieved_docs'][0][1],
}
# 3. 对话兜底
response, intent = self.dialogue.respond(question)
return {"source": "Dialogue", "answer": response, "intent": intent}
# 使用
qa_system = NLPQASystem()
qa_system.rag.load_knowledge(knowledge_base)
questions = [
"什么是BERT",
"你好",
"Transformer怎么工作的?",
"Python是什么?",
"什么是RAG",
]
print("="*50)
print("NLP智能问答系统")
print("="*50)
for q in questions:
result = qa_system.answer(q)
print(f"\n❓ {q}")
print(f"💡 [{result['source']}] {result['answer'][:80]}...")
9. 面试要点¶
🔑 面试高频考点
考点1:RAG的核心流程和优势?¶
Text Only
✅ 标准答案要点:
流程:
1. 索引阶段:将文档切块 → 编码为向量 → 存入向量数据库
2. 检索阶段:将问题编码 → 在向量库中检索Top-K相关文档
3. 生成阶段:将问题+检索到的文档作为上下文 → LLM生成答案
优势:
- 知识可更新(不需要重新训练模型)
- 减少幻觉(有据可查)
- 可溯源(知道答案来自哪个文档)
- 领域适配成本低
挑战:
- 检索质量直接影响最终效果
- 长上下文处理能力
- 检索和生成的协调
考点2:抽取式阅读理解的模型怎么训练?¶
Text Only
✅ 标准答案要点:
- 输入格式:[CLS] 问题 [SEP] 段落 [SEP]
- 输出:预测答案的开始位置和结束位置
- 损失函数:CrossEntropy(start) + CrossEntropy(end)
- 预测:选择start_logits[i] + end_logits[j]最大且i≤j的span
- 数据集:SQuAD、CMRC2018(中文)
考点3:如何提升RAG系统的效果?¶
Text Only
✅ 标准答案要点:
检索优化:
- 混合检索(BM25 + 向量检索)
- 查询重写/扩展
- 多步检索
- 重排序(Reranker)
切块优化:
- 合适的chunk_size
- 保留上下文的重叠切块
- 基于语义的切块
生成优化:
- 优化Prompt template
- 长上下文压缩
- 引用标注
评估:
- 检索准确率、召回率
- 答案正确率、忠实度
- 端到端评估(RAGAS框架)
10. 练习题¶
📝 基础题¶
- 对比抽取式和生成式问答的优缺点。
答案:抽取式问答:从文档中抽取连续文本片段作为答案(如BERT预测start/end位置)。优点——答案有据可查、不产生幻觉、可解释性强;缺点——答案必须是原文片段,无法综合多段信息或生成自然回答。生成式问答:生成自由文本作为答案。优点——可综合多源信息、回答自然灵活、能处理推理问题;缺点——可能产生幻觉、答案不可溯源、评估困难。趋势:RAG将两者结合,检索提供证据,生成模型输出答案。
- 解释RAG的完整工作流程。
答案:RAG(Retrieval-Augmented Generation)流程:①离线索引:文档分块(chunking) → Embedding模型编码为向量 → 存入向量数据库(FAISS/Milvus等);②查询处理:用户query → 向量编码 → 在向量库中检索Top-K相似文档块;③上下文增强:将检索到的文档块与query拼接构造增强Prompt;④生成回答:将增强Prompt输入LLM生成答案。关键优化点:查询改写、混合检索(稀疏+稠密)、重排序Reranker、分块策略、引用溯源等。
💻 编程题¶
- 实现一个基于TF-IDF的检索问答系统。
- 使用Hugging Face的BERT完成中文阅读理解任务。
- 构建一个简易的RAG系统(使用任意向量数据库)。
🔬 思考题¶
- 大模型时代,RAG和长上下文(Long Context)哪种方案更好?各自的适用场景是什么?
答案:取决于场景。RAG更适合:知识库频繁更新;知识库规模极大(百万文档);需要引用溯源;成本敏感(只输入相关片段)。长上下文更适合:文档少但需全局理解(分析整本书);信息密集难以检索定位;实现简单无需维护检索管道。实际趋势:两者常结合——先RAG检索候选,再用长上下文模型深度理解。长上下文能力在提升,但RAG在可扩展性和实时性上的优势仍不可替代。
✅ 自我检查清单¶
Text Only
□ 我知道问答系统的主要分类
□ 我理解检索式问答的工作流程
□ 我能实现抽取式阅读理解
□ 我理解KBQA的原理
□ 我知道RAG的完整流程和优势
□ 我完成了智能问答系统实战
□ 我完成了至少3道练习题
📚 延伸阅读¶
- SQuAD: 100,000+ Questions for Machine Comprehension of Text
- Retrieval-Augmented Generation for Knowledge-Intensive NLP Tasks
- CMRC 2018: 中文阅读理解数据集
- LangChain RAG教程
下一篇:10-预训练语言模型 — 从ELMo到BERT的预训练革命