LLM教育游戏开发实战:从架构设计到代码实现

发布时间:2026/7/27 3:53:06
LLM教育游戏开发实战:从架构设计到代码实现 LLM驱动的教育游戏下一代学习体验的技术实现与实战指南在数字化教育快速发展的今天如何将先进的人工智能技术与教育游戏相结合创造出真正个性化、互动性强的学习体验成为教育技术领域的重要课题。基于大型语言模型的教育游戏不仅能动态生成教学内容还能实时适应学生的学习进度和认知水平这正是传统教育游戏所欠缺的核心能力。本文将完整介绍LLM在教育游戏中的应用方案从技术选型、架构设计到具体代码实现为开发者提供一套可落地的实战指南。无论你是教育游戏开发者、AI技术爱好者还是想要了解LLM应用场景的技术人员都能从中获得实用的技术见解和代码示例。1. LLM在教育游戏中的核心价值与技术基础1.1 为什么LLM适合教育游戏场景大型语言模型在教育游戏中的应用价值主要体现在三个维度内容生成的动态性、交互体验的个性化以及学习路径的自适应性。与传统预设脚本的教育游戏不同LLM能够根据学生的实时反馈生成全新的题目、对话和挑战真正实现千人千面的学习体验。从技术角度看LLM的核心优势在于其强大的自然语言理解和生成能力。在教育游戏中这种能力可以转化为动态题目生成根据学生当前水平和学习进度实时生成难度适宜的练习题智能对话系统创建虚拟教师角色能够自然回答学生问题并提供个性化指导情境化学习内容将抽象知识点嵌入到游戏故事情境中提升学习趣味性1.2 主流LLM技术选型分析目前市面上主流的LLM方案各有特点需要根据教育游戏的具体需求进行选择开源模型方案Llama系列Meta开源的Llama 2/3模型在学术和商业使用上较为灵活适合需要自定义训练的场景ChatGLM系列中英双语优化对中文教育内容支持较好推理效率较高Qwen系列阿里通义千问开源版本在中文理解和生成方面表现优秀API服务方案OpenAI GPT系列生成质量高API稳定但成本需要考虑国内大模型API如文心一言、通义千问等网络延迟较低符合国内监管要求选择时需要综合考虑成本、延迟、数据隐私以及特定教育领域的专业性需求。对于教育游戏项目建议先从API方案开始验证核心价值再根据需求考虑私有化部署。2. 教育游戏LLM系统架构设计2.1 整体技术架构一个完整的LLM教育游戏系统通常包含以下核心组件游戏前端界面 → 交互逻辑层 → LLM服务层 → 内容数据库 → 用户数据分析前端界面层负责游戏画面渲染、用户输入采集和结果展示。可以是Web应用、移动App或桌面程序。交互逻辑层核心业务逻辑处理游戏规则、学习进度跟踪、难度调整策略等。LLM服务层封装与大模型的交互包括提示词工程、响应处理和内容过滤。数据持久层存储用户学习数据、生成的内容模板、评估标准等。2.2 环境准备与依赖配置以下以Python技术栈为例展示基础环境配置# requirements.txt openai1.3.0 langchain0.0.347 fastapi0.104.1 uvicorn0.24.0 pydantic2.5.0 sqlalchemy2.0.23 redis5.0.1# config.py import os from typing import Optional class GameConfig: 教育游戏配置类 def __init__(self): self.llm_api_key os.getenv(LLM_API_KEY, ) self.llm_base_url os.getenv(LLM_BASE_URL, https://api.openai.com/v1) self.llm_model os.getenv(LLM_MODEL, gpt-3.5-turbo) # 游戏相关配置 self.max_questions_per_session 10 self.difficulty_adjustment_threshold 0.7 # 正确率阈值 self.content_safety_filter True def validate_config(self) - bool: 验证配置完整性 if not self.llm_api_key: raise ValueError(LLM API密钥未配置) return True3. 核心功能模块实现3.1 动态题目生成引擎题目生成是教育游戏的核心功能需要平衡教育性和趣味性# question_generator.py import json from typing import List, Dict, Any from openai import OpenAI class QuestionGenerator: 基于LLM的智能题目生成器 def __init__(self, config: GameConfig): self.client OpenAI(api_keyconfig.llm_api_key, base_urlconfig.llm_base_url) self.model config.llm_model self.safety_filter config.content_safety_filter def generate_math_question(self, grade_level: int, topic: str, difficulty: str) - Dict[str, Any]: 生成数学题目 prompt f 你是一个专业的数学老师需要为{grade_level}年级学生生成一道关于{topic}的数学题。 难度级别{difficulty} 要求 1. 题目描述清晰有趣最好能结合生活场景 2. 提供4个选项其中只有1个正确答案 3. 标明解题思路和知识点 4. 题目长度适中适合游戏界面显示 请以JSON格式返回 {{ question: 题目内容, options: [A. 选项1, B. 选项2, C. 选项3, D. 选项4], correct_answer: A, explanation: 解题思路说明, knowledge_points: [知识点1, 知识点2] }} try: response self.client.chat.completions.create( modelself.model, messages[{role: user, content: prompt}], temperature0.7, # 控制创造性 max_tokens800 ) result json.loads(response.choices[0].message.content) # 内容安全校验 if self.safety_filter: self._validate_content_safety(result) return result except Exception as e: print(f题目生成失败: {e}) return self._get_fallback_question(grade_level, topic) def _validate_content_safety(self, content: Dict[str, Any]): 内容安全校验 # 检查是否包含不当内容 sensitive_keywords [暴力, 歧视, 不当内容] text_to_check json.dumps(content, ensure_asciiFalse) for keyword in sensitive_keywords: if keyword in text_to_check: raise ValueError(f内容包含敏感词: {keyword}) def _get_fallback_question(self, grade_level: int, topic: str) - Dict[str, Any]: 备用题目生成 # 当LLM服务不可用时使用预设题目 fallback_questions { 小学数学: { question: 小明有5个苹果小红有3个苹果他们一共有多少个苹果, options: [A. 7个, B. 8个, C. 9个, D. 10个], correct_answer: B, explanation: 5 3 8所以一共有8个苹果, knowledge_points: [加法运算, 数量概念] } } return fallback_questions.get(f{grade_level}数学, fallback_questions[小学数学])3.2 智能学习伙伴实现虚拟学习伙伴能够提供个性化的学习指导和鼓励# learning_companion.py import random from datetime import datetime from typing import Dict, List class LearningCompanion: 智能学习伙伴类 def __init__(self, config: GameConfig): self.client OpenAI(api_keyconfig.llm_api_key, base_urlconfig.llm_base_url) self.model config.llm_model self.conversation_history [] def get_encouragement(self, performance_data: Dict) - str: 根据学习表现提供鼓励 accuracy performance_data.get(accuracy, 0) questions_answered performance_data.get(questions_answered, 0) learning_time performance_data.get(learning_time, 0) prompt f 你是一个友善的学习伙伴需要根据学生的学习表现给予适当的鼓励和指导。 学生数据 - 答题正确率{accuracy * 100}% - 已回答题目数{questions_answered} - 学习时长{learning_time}分钟 请生成一段鼓励性的话语要求 1. 根据表现给予具体表扬如果正确率高或鼓励如果正确率低 2. 提供实用的学习建议 3. 语气亲切自然适合学生接受 4. 长度在50-100字之间 直接返回鼓励话语不要包含其他内容。 response self.client.chat.completions.create( modelself.model, messages[{role: user, content: prompt}], temperature0.8, max_tokens150 ) encouragement response.choices[0].message.content.strip() self._save_conversation(system, encouragement) return encouragement def answer_question(self, student_question: str, context: Dict) - str: 回答学生问题 prompt f 你是一个专业的辅导老师需要回答学生的问题。 学习上下文 - 当前学科{context.get(subject, 数学)} - 学习进度{context.get(progress, 初级)} - 学生年龄{context.get(age, 10)}岁 学生问题{student_question} 回答要求 1. 用简单易懂的语言解释 2. 可以举例说明 3. 如果问题超出范围礼貌地引导到相关知识点 4. 保持回答的专业性和准确性 请直接回答问题。 response self.client.chat.completions.create( modelself.model, messages[{role: user, content: prompt}], temperature0.3, # 较低温度保证准确性 max_tokens300 ) answer response.choices[0].message.content.strip() self._save_conversation(student, student_question) self._save_conversation(assistant, answer) return answer def _save_conversation(self, role: str, content: str): 保存对话历史 self.conversation_history.append({ role: role, content: content, timestamp: datetime.now().isoformat() }) # 限制历史记录长度 if len(self.conversation_history) 20: self.conversation_history self.conversation_history[-20:]4. 自适应学习算法实现4.1 难度动态调整策略基于学生表现的实时难度调整是教育游戏的核心智能特性# adaptive_learning.py from typing import Dict, List import numpy as np from dataclasses import dataclass dataclass class StudentProfile: 学生能力档案 ability_level: float 0.5 # 能力值 0-1 knowledge_gaps: List[str] None learning_speed: float 1.0 # 学习速度系数 recent_performance: List[float] None # 最近表现记录 def __post_init__(self): if self.knowledge_gaps is None: self.knowledge_gaps [] if self.recent_performance is None: self.recent_performance [] class AdaptiveLearningEngine: 自适应学习引擎 def __init__(self): self.difficulty_levels [easy, medium, hard, expert] def calculate_next_difficulty(self, student: StudentProfile, recent_accuracy: float) - str: 计算下一个题目的难度级别 # 更新学生能力评估 self._update_ability_assessment(student, recent_accuracy) # 基于IRT项目反应理论的难度调整 if len(student.recent_performance) 5: # 初始阶段使用简单题目建立信心 return easy # 计算能力与难度的匹配度 ability student.ability_level if ability 0.3: target_difficulty easy elif ability 0.6: target_difficulty medium elif ability 0.8: target_difficulty hard else: target_difficulty expert # 考虑学习曲线适当挑战学生 if recent_accuracy 0.8: # 表现优秀适当增加难度 current_index self.difficulty_levels.index(target_difficulty) if current_index len(self.difficulty_levels) - 1: target_difficulty self.difficulty_levels[current_index 1] return target_difficulty def _update_ability_assessment(self, student: StudentProfile, recent_accuracy: float): 更新学生能力评估 student.recent_performance.append(recent_accuracy) # 只保留最近20次表现 if len(student.recent_performance) 20: student.recent_performance student.recent_performance[-20:] # 使用加权平均计算能力值近期表现权重更高 weights np.linspace(0.5, 1.5, len(student.recent_performance)) weighted_accuracy np.average(student.recent_performance, weightsweights) student.ability_level weighted_accuracy def identify_knowledge_gaps(self, student: StudentProfile, wrong_questions: List[Dict]) - List[str]: 识别知识薄弱点 gaps [] for question in wrong_questions: knowledge_points question.get(knowledge_points, []) for point in knowledge_points: if point not in student.knowledge_gaps: gaps.append(point) student.knowledge_gaps.extend(gaps) # 去重 student.knowledge_gaps list(set(student.knowledge_gaps)) return gaps4.2 学习路径个性化推荐# learning_path_recommender.py from typing import List, Dict from enum import Enum class LearningStyle(Enum): VISUAL visual # 视觉型 AUDITORY auditory # 听觉型 KINESTHETIC kinesthetic # 动觉型 READING reading # 阅读型 class LearningPathRecommender: 学习路径推荐引擎 def __init__(self): self.knowledge_graph self._build_knowledge_graph() def recommend_next_topic(self, student_profile: Dict, completed_topics: List[str]) - Dict: 推荐下一个学习主题 learning_style student_profile.get(learning_style, LearningStyle.VISUAL) ability_level student_profile.get(ability_level, 0.5) # 获取可选主题 available_topics self._get_available_topics(completed_topics) # 根据学习风格和能力水平排序 scored_topics [] for topic in available_topics: score self._calculate_topic_score(topic, learning_style, ability_level) scored_topics.append((topic, score)) # 按分数降序排列 scored_topics.sort(keylambda x: x[1], reverseTrue) best_topic scored_topics[0][0] if scored_topics else None return { recommended_topic: best_topic, reasoning: f基于你的学习风格({learning_style.value})和能力水平推荐, alternative_topics: [t[0] for t in scored_topics[1:4]] } def _build_knowledge_graph(self) - Dict: 构建知识点关系图 # 简化的知识点图谱 return { 小学数学: { 基础运算: [加法, 减法, 乘法, 除法], 几何初步: [图形认识, 周长面积, 立体图形], 应用题: [一步应用题, 两步应用题, 复杂应用题], 前置要求: {}, 后续主题: [初中数学] } } def _get_available_topics(self, completed_topics: List[str]) - List[str]: 获取可学习的主题列表 available [] for topic, info in self.knowledge_graph.items(): prerequisites info.get(前置要求, []) if all(req in completed_topics for req in prerequisites): available.append(topic) return available def _calculate_topic_score(self, topic: str, learning_style: LearningStyle, ability: float) - float: 计算主题匹配分数 base_score 0.5 # 根据学习风格调整 style_weights { LearningStyle.VISUAL: 0.3, LearningStyle.AUDITORY: 0.2, LearningStyle.KINESTHETIC: 0.4, LearningStyle.READING: 0.1 } # 根据能力水平调整确保难度适中 difficulty_match 1 - abs(ability - 0.6) # 理想难度在0.6左右 return base_score style_weights.get(learning_style, 0.2) difficulty_match * 0.35. 完整游戏系统集成示例5.1 游戏主循环实现# educational_game.py import time import json from datetime import datetime from typing import Dict, List class EducationalGame: 教育游戏主类 def __init__(self, config: GameConfig): self.config config self.question_generator QuestionGenerator(config) self.learning_companion LearningCompanion(config) self.adaptive_engine AdaptiveLearningEngine() self.path_recommender LearningPathRecommender() self.student_profile StudentProfile() self.game_state { current_topic: 小学数学, session_start_time: datetime.now(), questions_answered: 0, correct_answers: 0, current_streak: 0 # 连续答对题数 } def start_learning_session(self) - Dict: 开始学习会话 print( 欢迎来到智能学习游戏) print(f当前主题: {self.game_state[current_topic]}) # 获取学习伙伴的欢迎语 welcome_message self.learning_companion.get_encouragement({ accuracy: self.student_profile.ability_level, questions_answered: self.game_state[questions_answered], learning_time: 0 }) print(f学习伙伴: {welcome_message}) return { welcome_message: welcome_message, current_topic: self.game_state[current_topic], ability_level: self.student_profile.ability_level } def generate_next_question(self) - Dict: 生成下一个问题 # 计算合适难度 recent_accuracy self._calculate_recent_accuracy() difficulty self.adaptive_engine.calculate_next_difficulty( self.student_profile, recent_accuracy ) # 生成题目 question self.question_generator.generate_math_question( grade_level3, topicself.game_state[current_topic], difficultydifficulty ) question[difficulty] difficulty question[question_id] fq_{int(time.time())} return question def submit_answer(self, question_id: str, user_answer: str, time_taken: float) - Dict: 提交答案并处理结果 # 这里应该从数据库获取原题目信息 # 简化处理假设我们能获取到原题目 original_question self._get_question_by_id(question_id) is_correct (user_answer.upper() original_question[correct_answer].upper()) # 更新游戏状态 self.game_state[questions_answered] 1 if is_correct: self.game_state[correct_answers] 1 self.game_state[current_streak] 1 else: self.game_state[current_streak] 0 # 更新学生档案 self.student_profile.recent_performance.append(1.0 if is_correct else 0.0) # 获取反馈信息 feedback self.learning_companion.get_encouragement({ accuracy: self._calculate_overall_accuracy(), questions_answered: self.game_state[questions_answered], learning_time: self._get_session_duration() }) result { is_correct: is_correct, correct_answer: original_question[correct_answer], explanation: original_question[explanation], feedback: feedback, new_streak: self.game_state[current_streak], overall_accuracy: self._calculate_overall_accuracy() } return result def _calculate_recent_accuracy(self) - float: 计算近期正确率 if not self.student_profile.recent_performance: return 0.5 # 默认值 recent self.student_profile.recent_performance[-5:] # 最近5题 return sum(recent) / len(recent) if recent else 0.5 def _calculate_overall_accuracy(self) - float: 计算总体正确率 if self.game_state[questions_answered] 0: return 0.0 return self.game_state[correct_answers] / self.game_state[questions_answered] def _get_session_duration(self) - float: 获取会话持续时间分钟 duration datetime.now() - self.game_state[session_start_time] return duration.total_seconds() / 60 def _get_question_by_id(self, question_id: str) - Dict: 根据ID获取题目简化实现 # 实际项目中应该查询数据库 return { correct_answer: B, explanation: 示例解释 } # 使用示例 def demo_game_flow(): 演示游戏流程 config GameConfig() game EducationalGame(config) # 开始游戏 session_info game.start_learning_session() print(session_info) # 生成并回答几个问题 for i in range(3): question game.generate_next_question() print(f\n问题 {i1}: {question[question]}) print(f选项: {question[options]}) # 模拟用户回答实际中从界面获取 user_answer input(请输入你的答案 (A/B/C/D): ) result game.submit_answer(question[question_id], user_answer, time_taken30) print(f结果: {正确 if result[is_correct] else 错误}) print(f解释: {result[explanation]}) print(f反馈: {result[feedback]}) if __name__ __main__: demo_game_flow()5.2 前端界面集成示例HTML/JavaScript!DOCTYPE html html langzh-CN head meta charsetUTF-8 meta nameviewport contentwidthdevice-width, initial-scale1.0 title智能学习游戏/title style .game-container { max-width: 800px; margin: 0 auto; padding: 20px; font-family: Arial, sans-serif; } .question-card { background: #f5f5f5; padding: 20px; border-radius: 10px; margin: 20px 0; } .options-container { display: grid; gap: 10px; margin: 20px 0; } .option-btn { padding: 15px; border: 2px solid #ddd; border-radius: 5px; background: white; cursor: pointer; transition: all 0.3s; } .option-btn:hover { border-color: #4CAF50; } .feedback-area { margin: 20px 0; padding: 15px; border-radius: 5px; background: #e8f5e8; } .stats-bar { display: flex; justify-content: space-between; margin: 10px 0; } /style /head body div classgame-container h1 智能学习游戏/h1 div classstats-bar span正确率: span idaccuracy0%/span/span span连续正确: span idstreak0/span/span span已答题: span idanswered0/span/span /div div idquestion-area div classquestion-card h3 idquestion-text加载题目中.../h3 div classoptions-container idoptions-container !-- 选项将通过JavaScript动态生成 -- /div /div /div div idfeedback-area classfeedback-area styledisplay: none; h4 学习反馈/h4 p idfeedback-text/p p idexplanation-text/p button onclicknextQuestion()下一题/button /div /div script class EducationalGameUI { constructor() { this.currentQuestion null; this.gameState { questionsAnswered: 0, correctAnswers: 0, currentStreak: 0 }; } async loadQuestion() { try { const response await fetch(/api/game/question, { method: GET }); this.currentQuestion await response.json(); this.displayQuestion(this.currentQuestion); } catch (error) { console.error(加载题目失败:, error); } } displayQuestion(question) { document.getElementById(question-text).textContent question.question; const optionsContainer document.getElementById(options-container); optionsContainer.innerHTML ; question.options.forEach((option, index) { const button document.createElement(button); button.className option-btn; button.textContent option; button.onclick () this.submitAnswer(option.charAt(0)); optionsContainer.appendChild(button); }); // 隐藏反馈区域 document.getElementById(feedback-area).style.display none; } async submitAnswer(selectedAnswer) { try { const response await fetch(/api/game/answer, { method: POST, headers: { Content-Type: application/json }, body: JSON.stringify({ questionId: this.currentQuestion.question_id, answer: selectedAnswer, timeTaken: 30 // 简化处理 }) }); const result await response.json(); this.displayFeedback(result); this.updateGameStats(result); } catch (error) { console.error(提交答案失败:, error); } } displayFeedback(result) { document.getElementById(feedback-text).textContent result.feedback; document.getElementById(explanation-text).textContent 正确答案: ${result.correct_answer} | 解释: ${result.explanation}; document.getElementById(feedback-area).style.display block; } updateGameStats(result) { this.gameState.questionsAnswered; if (result.is_correct) { this.gameState.correctAnswers; this.gameState.currentStreak; } else { this.gameState.currentStreak 0; } const accuracy (this.gameState.correctAnswers / this.gameState.questionsAnswered * 100).toFixed(1); document.getElementById(accuracy).textContent ${accuracy}%; document.getElementById(streak).textContent this.gameState.currentStreak; document.getElementById(answered).textContent this.gameState.questionsAnswered; } } // 全局函数供按钮调用 function nextQuestion() { gameUI.loadQuestion(); } // 初始化游戏 const gameUI new EducationalGameUI(); window.onload () gameUI.loadQuestion(); /script /body /html6. 性能优化与工程实践6.1 LLM API调用优化在实际生产环境中LLM API调用是主要的性能瓶颈和成本中心需要精心优化# api_optimizer.py import asyncio import aiohttp from typing import List, Dict import time from dataclasses import dataclass from concurrent.futures import ThreadPoolExecutor dataclass class APICallConfig: API调用配置 max_retries: int 3 timeout: int 30 batch_size: int 5 rate_limit_per_minute: int 60 class LLMAPIOptimizer: LLM API调用优化器 def __init__(self, config: APICallConfig): self.config config self.semaphore asyncio.Semaphore(config.rate_limit_per_minute // 60) # 每秒限制 self.retry_delays [1, 2, 5] # 重试延迟秒 async def batch_generate_questions(self, prompts: List[str]) - List[Dict]: 批量生成题目异步优化 async with aiohttp.ClientSession() as session: tasks [] for i in range(0, len(prompts), self.config.batch_size): batch prompts[i:i self.config.batch_size] task self._process_batch(session, batch) tasks.append(task) results await asyncio.gather(*tasks, return_exceptionsTrue) # 扁平化结果 flat_results [] for batch_result in results: if isinstance(batch_result, Exception): # 处理异常使用备用方案 flat_results.extend(self._get_fallback_questions(len(batch_result))) else: flat_results.extend(batch_result) return flat_results async def _process_batch(self, session: aiohttp.ClientSession, prompts: List[str]) - List[Dict]: 处理一个批次的提示词 async with self.semaphore: for attempt in range(self.config.max_retries): try: # 实际的API调用逻辑 results await self._make_api_call(session, prompts) return results except Exception as e: if attempt self.config.max_retries - 1: raise e await asyncio.sleep(self.retry_delays[attempt]) return [] async def _make_api_call(self, session: aiohttp.ClientSession, prompts: List[str]) - List[Dict]: 执行API调用简化示例 # 这里应该是实际的API调用代码 await asyncio.sleep(0.1) # 模拟网络延迟 return [{question: f基于提示生成的问题 {i}} for i in range(len(prompts))] def _get_fallback_questions(self, count: int) - List[Dict]: 获取备用题目 return [{question: f备用题目 {i}, is_fallback: True} for i in range(count)] # 使用示例 async def demo_optimized_generation(): 演示优化后的生成流程 config APICallConfig() optimizer LLMAPIOptimizer(config) prompts [f生成数学题提示 {i} for i in range(10)] results await optimizer.batch_generate_questions(prompts) print(f成功生成 {len(results)} 个题目) return results6.2 缓存策略实现# cache_manager.py import redis import json import hashlib from typing import Any, Optional from datetime import timedelta class QuestionCacheManager: 题目缓存管理器 def __init__(self, redis_url: str redis://localhost:6379): self.redis_client redis.from_url(redis_url) self.default_ttl timedelta(hours24) # 默认缓存24小时 def get_cache_key(self, prompt: str, parameters: Dict) - str: 生成缓存键 content f{prompt}{json.dumps(parameters, sort_keysTrue)} return hashlib.md5(content.encode()).hexdigest() def get_cached_question(self, prompt: str, parameters: Dict) - Optional[Dict]: 获取缓存的题目 cache_key self.get_cache_key(prompt, parameters) cached_data self.redis_client.get(cache_key) if cached_data: return json.loads(cached_data) return None def cache_question(self, prompt: str, parameters: Dict, question_data: Dict): 缓存题目 cache_key self.get_cache_key(prompt, parameters) self.redis_client.setex( cache_key, self.default_ttl, json.dumps(question_data) ) def get_or_generate_question(self, prompt: str, parameters: Dict, generator_func) - Dict: 获取或生成题目缓存优先 # 先尝试从缓存获取 cached self.get_cached_question(prompt, parameters) if cached: cached[from_cache] True return cached # 缓存未命中生成新题目 new_question generator_func(prompt, parameters) new_question[from_cache] False # 缓存新题目 self.cache_question(prompt, parameters, new_question) return new_question7. 安全与内容审核7.1 多层内容安全过滤在教育游戏中内容安全是重中之重需要建立多层防护# content_safety.py import re from typing import List, Dict from abc import ABC, abstractmethod class ContentSafetyFilter(ABC): 内容安全过滤器抽象基类 abstractmethod def filter_content(self, content: str) - Dict: 过滤内容返回过滤结果 pass class KeywordFilter(ContentSafetyFilter): 关键词过滤 def __init__(self): self.sensitive_keywords self._load_sensitive_keywords() def filter_content(self, content: str) - Dict: 关键词过滤 found_keywords [] for