本文目录导读：
我来详细介绍如何搭建一个Python脚本数据问答系统,包括几种常见的架构方案。
基础架构方案
基于规则匹配（简单场景）
# 基础问答系统
class SimpleQASystem:
def __init__(self):
self.qa_pairs = {
"你好": "你好！有什么可以帮助你的？",
"天气": "请告诉我您想查询哪个城市的天气？",
"退出": "感谢使用，再见！"
}
def get_answer(self, question):
# 关键词匹配
for key, answer in self.qa_pairs.items():
if key in question:
return answer
return "抱歉，我没有理解您的问题"
基于TF-IDF+余弦相似度
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.metrics.pairwise import cosine_similarity
import jieba
class TFIDFQASystem:
def __init__(self, data_path):
self.vectorizer = TfidfVectorizer(tokenizer=self.tokenize)
self.questions = []
self.answers = []
self.load_data(data_path)
self.build_index()
def tokenize(self, text):
return list(jieba.cut(text))
def load_data(self, path):
import json
with open(path, 'r', encoding='utf-8') as f:
data = json.load(f)
for item in data:
self.questions.append(item['question'])
self.answers.append(item['answer'])
def build_index(self):
self.question_vectors = self.vectorizer.fit_transform(self.questions)
def get_answer(self, question):
question_vec = self.vectorizer.transform([question])
similarities = cosine_similarity(question_vec, self.question_vectors)[0]
best_idx = similarities.argmax()
if similarities[best_idx] > 0.3: # 相似度阈值
return self.answers[best_idx]
return "未找到匹配答案"
进阶架构：结合向量数据库
from sentence_transformers import SentenceTransformer
import chromadb
from chromadb.config import Settings
class VectorQASystem:
def __init__(self, collection_name="qa_data"):
# 初始化向量模型
self.model = SentenceTransformer('paraphrase-multilingual-MiniLM-L12-v2')
# 初始化ChromaDB
self.client = chromadb.Client(Settings(
chroma_db_impl="duckdb+parquet",
persist_directory="./chromadb"
))
self.collection = self.client.get_or_create_collection(
name=collection_name,
embedding_function=self.get_embedding
)
def get_embedding(self, texts):
embeddings = self.model.encode(texts)
return embeddings.tolist()
def add_data(self, questions, answers, ids=None):
embeddings = self.get_embedding(questions)
if ids is None:
ids = [f"doc_{i}" for i in range(len(questions))]
metadatas = [{"answer": ans} for ans in answers]
self.collection.add(
embeddings=embeddings,
documents=questions,
metadatas=metadatas,
ids=ids
)
def query(self, question, n_results=3):
results = self.collection.query(
query_texts=[question],
n_results=n_results
)
return results
完整项目结构
qa_system/
├── data/
│ ├── raw/ # 原始数据
│ ├── processed/ # 处理后数据
│ └── embeddings/ # 向量缓存
├── src/
│ ├── __init__.py
│ ├── data_loader.py # 数据加载模块
│ ├── preprocessor.py # 文本预处理
│ ├── retriever.py # 检索模块
│ ├── generator.py # 生成模块（可选）
│ └── qa_system.py # 主系统
├── models/
│ └── vector_model/ # 模型文件
├── config/
│ └── config.yaml # 配置文件
├── api/
│ ├── app.py # Flask API
│ └── requirements.txt
└── main.py # 主入口完整的问答系统实现
# qa_system.py
import yaml
import json
import logging
from typing import List, Dict, Optional
from pathlib import Path
class QASystem:
def __init__(self, config_path: str = "config/config.yaml"):
self.config = self.load_config(config_path)
self.logger = self.setup_logger()
self.retriever = None
self.generator = None
self.initialize_components()
def load_config(self, path):
with open(path, 'r', encoding='utf-8') as f:
return yaml.safe_load(f)
def setup_logger(self):
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
return logging.getLogger(__name__)
def initialize_components(self):
from sentence_transformers import SentenceTransformer
import chromadb
# 初始化检索器
self.retriever = VectorRetriever(
model_name=self.config['retriever']['model'],
collection_name=self.config['retriever']['collection']
)
# 如果配置了生成器，初始化生成器
if self.config.get('generator', {}).get('enabled', False):
self.generator = ResponseGenerator(
model_name=self.config['generator']['model']
)
def load_data(self, data_path: str):
"""加载并索引数据"""
with open(data_path, 'r', encoding='utf-8') as f:
data = json.load(f)
questions = [item['question'] for item in data]
answers = [item['answer'] for item in data]
self.retriever.add_data(questions, answers)
self.logger.info(f"Loaded {len(questions)} QA pairs")
def answer(self, question: str, top_k: int = 3) -> Dict:
"""回答问题"""
# 1. 检索相关文档
results = self.retriever.query(question, top_k)
# 2. 如果有生成器，用生成器生成答案
if self.generator and results['documents']:
context = "\n".join(results['documents'][0])
answer = self.generator.generate(question, context)
else:
# 否则直接返回最佳匹配
if results['documents']:
answer = results['metadatas'][0][0]['answer']
else:
answer = "抱歉，我无法回答这个问题"
return {
'question': question,
'answer': answer,
'confidence': results['distances'][0][0] if results['distances'] else 0,
'sources': results['documents'][0][:3] if results['documents'] else []
}
# 向量检索器
class VectorRetriever:
def __init__(self, model_name: str, collection_name: str):
from sentence_transformers import SentenceTransformer
import chromadb
self.model = SentenceTransformer(model_name)
self.client = chromadb.PersistentClient(path="./chromadb")
self.collection = self.client.get_or_create_collection(
name=collection_name
)
def add_data(self, texts: List[str], metadatas: List[Dict]):
embeddings = self.model.encode(texts).tolist()
ids = [f"doc_{i}" for i in range(len(texts))]
self.collection.add(
embeddings=embeddings,
documents=texts,
metadatas=[{"answer": m} for m in metadatas],
ids=ids
)
def query(self, text: str, n_results: int = 5):
query_embedding = self.model.encode([text]).tolist()
results = self.collection.query(
query_embeddings=query_embedding,
n_results=n_results
)
return results
# 可选：生成器（使用本地模型）
class ResponseGenerator:
def __init__(self, model_name: str):
from transformers import pipeline
self.generator = pipeline(
'text2text-generation',
model=model_name,
max_length=200
)
def generate(self, question: str, context: str) -> str:
prompt = f"基于以下上下文回答问题：\n上下文：{context}\n问题：{question}\n答案："
result = self.generator(prompt)[0]['generated_text']
return result
API层实现
# api/app.py
from flask import Flask, request, jsonify
from flask_cors import CORS
from qa_system import QASystem
app = Flask(__name__)
CORS(app)
# 初始化问答系统
qa_system = QASystem()
@app.route('/answer', methods=['POST'])
def get_answer():
data = request.json
question = data.get('question', '')
if not question:
return jsonify({'error': '问题不能为空'}), 400
result = qa_system.answer(question)
return jsonify(result)
@app.route('/batch_answer', methods=['POST'])
def batch_answer():
data = request.json
questions = data.get('questions', [])
results = []
for q in questions:
result = qa_system.answer(q)
results.append(result)
return jsonify({'results': results})
@app.route('/add_data', methods=['POST'])
def add_data():
data = request.json
qa_system.load_data(data['data_path'])
return jsonify({'status': 'success', 'message': '数据加载成功'})
if __name__ == '__main__':
app.run(debug=True, port=5000)
配置文件示例
# config/config.yaml retriever: model: "paraphrase-multilingual-MiniLM-L12-v2" collection: "qa_collection" top_k: 5 generator: enabled: false # 可选，需要大模型支持 model: "microsoft/DialoGPT-small" data: path: "data/processed/qa_pairs.json" encoding: "utf-8" api: host: "0.0.0.0" port: 5000 debug: true
使用示例
# main.py
from src.qa_system import QASystem
def main():
# 创建问答系统
qa = QASystem()
# 加载数据
qa.load_data("data/processed/qa_pairs.json")
# 问答循环
while True:
question = input("\n请输入问题（输入'quit'退出）: ")
if question.lower() == 'quit':
break
result = qa.answer(question)
print(f"答案: {result['answer']}")
print(f"置信度: {result['confidence']:.2f}")
if __name__ == "__main__":
main()
部署建议
Docker部署
FROM python:3.9-slim WORKDIR /app COPY requirements.txt . RUN pip install -r requirements.txt COPY . . EXPOSE 5000 CMD ["python", "api/app.py"]
性能优化
- 缓存机制：使用Redis缓存常见问题
- 批量处理：支持批量数据索引
- 异步处理：使用FastAPI替代Flask
- 模型量化：压缩向量化模型
这个系统可以根据你的具体需求进行调整和扩展,需要我详细讲解某个特定模块的实现吗？
