要构建一个具备本地题库匹配失败时自动调用DeepSeek API的Python题库系统，需要设计以下几个核心模块： ## 系统架构设计 | 模块 | 功能 | 实现方式 | |------|------|----------| | 题库存储 | 存储题目、选项和答案 | SQLite数据库或JSON文件 | | 请求处理器 | 接收题目和选项请求 | Flask/FastAPI服务端 | | 匹配引擎 | 精准匹配题库内容 | 字符串精确匹配 | | DeepSeek接口 | 外部API调用 | requests库调用 | | 答案写入器 | 将新答案写入本地 | 数据库操作或文件写入 | ## 核心代码实现 ### 1. 数据库设计（SQLite） ```python import sqlite3 import json from datetime import datetime class QuestionDatabase: def __init__(self, db_path="question_bank.db"): self.db_path = db_path self.init_database() def init_database(self): """初始化数据库表结构""" conn = sqlite3.connect(self.db_path) cursor = conn.cursor() # 创建题目表 cursor.execute(''' CREATE TABLE IF NOT EXISTS questions ( id INTEGER PRIMARY KEY AUTOINCREMENT, question_text TEXT NOT NULL, options_json TEXT NOT NULL, correct_answer TEXT NOT NULL, source TEXT DEFAULT 'manual', created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, UNIQUE(question_text, options_json) ) ''') # 创建API调用记录表 cursor.execute(''' CREATE TABLE IF NOT EXISTS api_calls ( id INTEGER PRIMARY KEY AUTOINCREMENT, question_id INTEGER, api_response TEXT, called_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (question_id) REFERENCES questions(id) ) ''') conn.commit() conn.close() def add_question(self, question_text, options, correct_answer, source='manual'): """添加题目到题库""" conn = sqlite3.connect(self.db_path) cursor = conn.cursor() try: cursor.execute(''' INSERT OR IGNORE INTO questions (question_text, options_json, correct_answer, source) VALUES (?, ?, ?, ?) ''', (question_text, json.dumps(options), correct_answer, source)) conn.commit() return cursor.lastrowid except Exception as e: print(f"添加题目失败: {e}") return None finally: conn.close() def find_question(self, question_text, options): """精准匹配题目和选项""" conn = sqlite3.connect(self.db_path) cursor = conn.cursor() cursor.execute(''' SELECT id, correct_answer FROM questions WHERE question_text = ? AND options_json = ? ''', (question_text, json.dumps(options))) result = cursor.fetchone() conn.close() return result if result else None ``` ### 2. DeepSeek API 调用模块 ```python import requests import json import time class DeepSeekAPI: def __init__(self, api_key=None, base_url="http://localhost:11434/api/generate"): """ 初始化DeepSeek API客户端 参数说明： - api_key: API密钥（如果使用在线API） - base_url: 本地部署的Ollama API地址[ref_1] """ self.api_key = api_key self.base_url = base_url self.headers = { "Content-Type": "application/json", "Authorization": f"Bearer {api_key}" if api_key else None } def get_answer(self, question_text, options): """调用DeepSeek获取答案""" # 构建提示词 prompt = self._build_prompt(question_text, options) payload = { "model": "deepseek-r1:7b", # 根据实际部署的模型调整[ref_2] "prompt": prompt, "stream": False, "options": { "temperature": 0.1, # 低温度确保答案稳定 "max_tokens": 100 } } try: response = requests.post( self.base_url, headers=self.headers, json=payload, timeout=30 ) if response.status_code == 200: result = response.json() answer = result.get("response", "").strip() return self._extract_answer(answer) else: print(f"API调用失败: {response.status_code}") return None except requests.exceptions.RequestException as e: print(f"网络请求错误: {e}") return None def _build_prompt(self, question_text, options): """构建提示词模板""" options_text = "\n".join([f"{chr(65+i)}. {option}" for i, option in enumerate(options)]) prompt = f"""请仔细分析以下选择题，并给出正确答案的字母选项。 题目：{question_text} 选项： {options_text} 请只返回正确答案的字母（如：A、B、C、D），不要包含任何解释或额外文本。""" return prompt def _extract_answer(self, response_text): """从响应中提取答案字母""" import re # 匹配A-D的字母 match = re.search(r'[A-D]', response_text.upper()) return match.group(0) if match else None ``` ### 3. 主服务模块 ```python from flask import Flask, request, jsonify import threading import logging app = Flask(__name__) logging.basicConfig(level=logging.INFO) class QuestionAnswerSystem: def __init__(self, db_path="question_bank.db"): self.db = QuestionDatabase(db_path) self.deepseek = DeepSeekAPI() self.cache = {} # 简单缓存机制 def process_question(self, question_text, options): """处理题目请求的核心逻辑""" # 1. 检查缓存 cache_key = f"{question_text}_{hash(str(options))}" if cache_key in self.cache: return self.cache[cache_key] # 2. 查询本地题库 db_result = self.db.find_question(question_text, options) if db_result: question_id, answer = db_result result = { "answer": answer, "source": "local_database", "question_id": question_id } self.cache[cache_key] = result return result # 3. 本地题库未找到，调用DeepSeek API logging.info(f"本地题库未找到，调用DeepSeek API: {question_text[:50]}...") deepseek_answer = self.deepseek.get_answer(question_text, options) if deepseek_answer: # 4. 将新答案写入本地题库 question_id = self.db.add_question( question_text=question_text, options=options, correct_answer=deepseek_answer, source="deepseek_api" ) result = { "answer": deepseek_answer, "source": "deepseek_api", "question_id": question_id } self.cache[cache_key] = result # 5. 异步记录API调用（可选） threading.Thread( target=self._log_api_call, args=(question_id, deepseek_answer) ).start() return result else: return { "answer": None, "source": "none", "error": "无法从DeepSeek获取答案" } def _log_api_call(self, question_id, response): """记录API调用日志""" conn = sqlite3.connect(self.db.db_path) cursor = conn.cursor() cursor.execute( "INSERT INTO api_calls (question_id, api_response) VALUES (?, ?)", (question_id, response) ) conn.commit() conn.close() # 全局系统实例 qa_system = QuestionAnswerSystem() @app.route('/api/get_answer', methods=['POST']) def get_answer(): """REST API接口""" data = request.json if not data or 'question' not in data or 'options' not in data: return jsonify({"error": "缺少必要参数"}), 400 question_text = data['question'] options = data['options'] if not isinstance(options, list) or len(options) < 2: return jsonify({"error": "选项必须是至少包含2个元素的列表"}), 400 result = qa_system.process_question(question_text, options) return jsonify(result) @app.route('/api/add_question', methods=['POST']) def add_question(): """手动添加题目到题库""" data = request.json required_fields = ['question', 'options', 'answer'] for field in required_fields: if field not in data: return jsonify({"error": f"缺少字段: {field}"}), 400 question_id = qa_system.db.add_question( question_text=data['question'], options=data['options'], correct_answer=data['answer'], source='manual' ) if question_id: return jsonify({ "success": True, "question_id": question_id, "message": "题目添加成功" }) else: return jsonify({"error": "添加题目失败"}), 500 ``` ### 4. 客户端调用示例 ```python import requests import json class QuestionClient: def __init__(self, server_url="http://localhost:5000"): self.server_url = server_url def ask_question(self, question, options): """发送题目请求""" payload = { "question": question, "options": options } try: response = requests.post( f"{self.server_url}/api/get_answer", json=payload, timeout=10 ) return response.json() except Exception as e: return {"error": str(e)} def batch_process(self, questions_list): """批量处理题目""" results = [] for question, options in questions_list: result = self.ask_question(question, options) results.append({ "question": question, "options": options, "result": result }) # 避免频繁请求 import time time.sleep(0.5) return results # 使用示例 if __name__ == "__main__": # 启动服务端 # app.run(host='0.0.0.0', port=5000, debug=True) # 客户端调用 client = QuestionClient() # 示例1：本地题库已有的题目 result1 = client.ask_question( question="Python中用于创建虚拟环境的标准库是什么？", options=["venv", "virtualenv", "pipenv", "conda"] ) print(f"结果1: {result1}") # 示例2：本地题库没有的题目（会调用DeepSeek） result2 = client.ask_question( question="深度学习中，Transformer模型的核心机制是什么？", options=["卷积神经网络", "循环神经网络", "自注意力机制", "反向传播算法"] ) print(f"结果2: {result2}") ``` ## 系统优化建议 1. **缓存策略优化**：使用Redis或内存缓存提高频繁查询题目的响应速度 2. **匹配算法增强**：除了精确匹配，可添加模糊匹配支持（使用文本相似度算法） 3. **API限流处理**：对DeepSeek API调用添加速率限制和失败重试机制 4. **批量处理支持**：实现异步批量题目处理，提高效率 5. **监控和日志**：添加详细的运行日志和性能监控 ## 部署配置示例 ```yaml # config.yaml database: path: "data/question_bank.db" backup_interval: 3600 # 每小时备份一次 deepseek_api: local_mode: true # 使用本地部署[ref_4] model: "deepseek-r1:7b" temperature: 0.1 max_tokens: 100 server: host: "0.0.0.0" port: 5000 workers: 4 cache: enabled: true ttl: 3600 # 缓存1小时 ``` 这个系统实现了完整的题库管理、智能匹配和外部API调用集成，可以根据实际需求进行扩展和优化[ref_3]。本地部署DeepSeek模型时，需要确保Ollama服务正常运行[ref_1][ref_2]。