Python脚本数据问答系统如何搭建

本文目录导读：

1. 基础架构方案
2. 进阶架构：结合向量数据库
3. 完整项目结构
4. 完整的问答系统实现
5. API层实现
6. 配置文件示例
7. 使用示例
8. 部署建议

我来详细介绍如何搭建一个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"]

性能优化

1. 缓存机制：使用Redis缓存常见问题
2. 批量处理：支持批量数据索引
3. 异步处理：使用FastAPI替代Flask
4. 模型量化：压缩向量化模型

这个系统可以根据你的具体需求进行调整和扩展,需要我详细讲解某个特定模块的实现吗？