在技术团队和知识密集型项目中如何高效地整理个人笔记、管理技术文档和电子书并让这些知识资产能够被智能检索和复用一直是一个核心痛点。传统的笔记软件、网盘和文档库往往相互割裂而 AI 知识库AI KnowledgeBase正是为了解决信息孤岛和知识检索效率问题而出现的工程实践方向。一个理想的 AI 知识库应该能够将笔记记录、书籍管理、内容关联和智能问答融为一体让知识真正流动起来。本文将围绕构建一个集笔记、书库和 AI 能力于一体的知识库系统从技术选型、架构设计、核心功能实现到生产环境部署提供一个完整的工程实践指南。无论你是独立开发者希望搭建个人知识中枢还是团队技术负责人规划内部知识平台都能从中获得可落地的方案。1. 理解 AI 知识库的核心组件与技术栈一个完整的 AI 知识库系统通常由三个层次构成数据存储层、知识处理层和智能交互层。每一层都有明确的技术职责和选型考量。1.1 数据存储层如何组织笔记和书籍的元数据与内容笔记和书籍虽然都是知识载体但它们的结构、使用频率和存储需求有所不同。笔记更偏向碎片化、高频更新而书籍则是整体化、相对稳定的内容。在数据库设计上建议将元数据标题、作者、标签、创建时间等与内容数据分开存储。元数据使用关系型数据库如 PostgreSQL 或 MySQL便于复杂查询和统计而书籍全文、笔记大文本内容则可存入文档数据库如 MongoDB或专用全文检索引擎如 Elasticsearch以获得更好的检索性能。以下是一个简化的笔记元数据表结构示例CREATE TABLE notes ( id BIGINT AUTO_INCREMENT PRIMARY KEY, title VARCHAR(500) NOT NULL, content_text TEXT, -- 纯文本内容用于快速预览和检索 content_html TEXT, -- 富文本内容用于展示 book_id BIGINT, -- 关联的书籍ID如果该笔记属于某本书 user_id BIGINT NOT NULL, tags JSON, -- 标签数组如 [AI, 编程, 机器学习] created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, FOREIGN KEY (book_id) REFERENCES books(id), INDEX idx_user_created (user_id, created_at), FULLTEXT INDEX idx_content_text (content_text) -- 全文索引 );书籍表结构则需要包含更多出版信息和文件信息CREATE TABLE books ( id BIGINT AUTO_INCREMENT PRIMARY KEY, isbn VARCHAR(20), title VARCHAR(500) NOT NULL, author VARCHAR(200), publisher VARCHAR(200), publish_year INT, file_path VARCHAR(1000), -- 电子书文件存储路径 file_type ENUM(pdf, epub, mobi), -- 文件格式 cover_image VARCHAR(1000), -- 封面图片路径 summary TEXT, -- 书籍摘要 user_id BIGINT NOT NULL, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, INDEX idx_user_title (user_id, title) );1.2 知识处理层从原始内容到可检索的知识单元原始文档和笔记内容需要经过处理才能被 AI 模型有效理解。处理流程通常包括文本提取、分块、向量化和索引构建。文本提取对于 PDF、EPUB 等格式的电子书需要使用相应的解析库如 PyPDF2、epub-li提取纯文本内容。内容分块大文档需要切分成适当大小的文本块通常 200-1000 字符以便后续的向量化和检索。分块时要注意保持语义完整性比如按章节、段落自然边界切割。向量化使用文本嵌入模型如 OpenAI text-embedding-ada-002、BGE、M3E将文本块转换为高维向量。这些向量能够捕捉文本的语义信息相似的内容在向量空间中距离更近。向量数据库将文本向量存入专门的向量数据库如 Pinecone、Chroma、Weaviate 或 PGVector以便后续进行相似性检索。1.3 智能交互层实现自然语言问答和知识发现这是用户直接感知的 AI 能力层核心是检索增强生成RAG技术。当用户提出问题时系统不是直接让大语言模型LLM凭空生成答案而是先从未向量库中检索最相关的知识片段然后将这些片段作为上下文提供给 LLM 生成精准的答案。RAG 流程的关键步骤问题向量化将用户查询转换为向量。相似性检索在向量数据库中查找与查询向量最相似的文本块。上下文组装将检索到的相关文本块组合成提示词上下文。智能生成LLM 基于上下文生成自然语言回答。2. 环境准备与核心技术依赖构建 AI 知识库需要协调多个组件从基础的后端框架到专门的 AI 服务。以下是推荐的技术栈和版本要求。2.1 后端技术栈选择与版本对齐对于大多数知识库项目建议选择成熟稳定的技术组合Python 3.9AI 生态最完善的语言拥有丰富的文本处理和机器学习库。FastAPI 或 DjangoFastAPI 适合需要高性能 API 的场景Django 适合需要强大后台管理功能的项目。PostgreSQL 15可靠的关系数据库配合 PGVector 扩展可以同时处理结构化数据和向量数据。Redis用于缓存会话、频繁查询结果和任务队列。关键 Python 依赖库及其版本# requirements.txt fastapi0.104.1 uvicorn0.24.0 sqlalchemy2.0.23 psycopg2-binary2.9.9 langchain0.0.350 langchain-community0.0.10 openai1.3.0 chromadb0.4.15 pypdf23.0.1 python-multipart0.0.6 jwt1.3.1 python-dotenv1.0.02.2 AI 模型服务配置根据项目需求和预算可以选择不同的 AI 模型服务模型服务适用场景成本考量部署方式OpenAI GPT-4高精度问答、复杂推理API 调用按 token 计费云端 APIOpenAI GPT-3.5-Turbo一般问答、内容生成成本较低性能足够云端 API开源模型Llama 2、ChatGLM数据隐私要求高、需要定制需要自建 GPU 服务器本地部署Azure OpenAI企业级合规要求与微软生态集成云端 API对于本地部署的开源模型可以使用 Ollama、Text Generation InferenceTGI等工具简化部署流程。2.3 向量数据库选型与配置向量数据库的选择直接影响检索性能和系统复杂度向量数据库特点部署复杂度适合规模Chroma轻量级、易于集成低可嵌入应用个人或小团队Weaviate功能丰富、支持混合检索中需要单独部署中小型企业PGVectorPostgreSQL 扩展数据一致性好低与主数据库一体已有 PostgreSQL 的项目Pinecone全托管、无需运维低但依赖云服务快速原型和中小项目对于大多数项目从 Chroma 开始是合理的选择它简单易用且功能足够。以下是在 Python 中初始化 Chroma 客户端的示例import chromadb from chromadb.config import Settings # 初始化 Chroma 客户端 chroma_client chromadb.Client(Settings( chroma_db_implduckdbparquet, persist_directory./chroma_db # 向量数据持久化目录 )) # 创建或获取集合类似数据库中的表 collection chroma_client.create_collection( nameknowledge_base, metadata{description: 存储笔记和书籍知识的向量集合} )3. 构建核心知识管理功能有了技术基础后我们需要实现用户直接交互的核心功能笔记管理、书籍上传和内容检索。3.1 笔记系统的数据结构与 API 设计笔记系统应该支持富文本编辑、标签分类和版本管理。使用 JSON API 设计 RESTful 接口from fastapi import FastAPI, HTTPException, Depends from pydantic import BaseModel from typing import List, Optional from datetime import datetime app FastAPI() class NoteCreate(BaseModel): title: str content_html: str content_text: str tags: List[str] [] book_id: Optional[int] None class NoteResponse(BaseModel): id: int title: str content_html: str tags: List[str] created_at: datetime updated_at: datetime app.post(/api/notes, response_modelNoteResponse) async def create_note(note: NoteCreate, user_id: int Depends(get_current_user)): 创建新笔记 # 数据库插入逻辑 db_note create_note_in_db(user_id, note.dict()) # 异步处理向量化 await process_note_for_vectorization(db_note) return NoteResponse(**db_note.__dict__) app.get(/api/notes, response_modelList[NoteResponse]) async def list_notes( skip: int 0, limit: int 100, tag: Optional[str] None, user_id: int Depends(get_current_user) ): 分页获取笔记列表支持按标签过滤 notes get_notes_from_db(user_id, skip, limit, tag) return [NoteResponse(**note.__dict__) for note in notes]3.2 电子书上传与解析流程电子书处理是比较复杂的部分需要处理多种文件格式和解析文本内容import os from pathlib import Path from fastapi import UploadFile, File import PyPDF2 from ebooklib import epub app.post(/api/books/upload) async def upload_book( file: UploadFile File(...), user_id: int Depends(get_current_user) ): 上传电子书并解析内容 # 验证文件类型 allowed_types [application/pdf, application/epubzip] if file.content_type not in allowed_types: raise HTTPException(400, 不支持的文件格式) # 保存文件 upload_dir Path(f./uploads/{user_id}) upload_dir.mkdir(parentsTrue, exist_okTrue) file_path upload_dir / file.filename with open(file_path, wb) as buffer: content await file.read() buffer.write(content) # 解析书籍内容 book_info await parse_book_content(file_path, file.content_type) # 保存书籍元数据到数据库 book_record create_book_in_db(user_id, book_info, str(file_path)) # 异步处理书籍内容的向量化 await process_book_for_vectorization(book_record) return {id: book_record.id, title: book_info.title} async def parse_book_content(file_path: Path, content_type: str): 解析电子书内容 if content_type application/pdf: return parse_pdf(file_path) elif content_type application/epubzip: return parse_epub(file_path) def parse_pdf(file_path: Path): 解析PDF文件 text_content with open(file_path, rb) as file: pdf_reader PyPDF2.PdfReader(file) for page in pdf_reader.pages: text_content page.extract_text() \n # 提取元数据 with open(file_path, rb) as file: pdf_reader PyPDF2.PdfReader(file) metadata pdf_reader.metadata return { title: metadata.title if metadata and metadata.title else file_path.stem, author: metadata.author if metadata and metadata.author else 未知, content: text_content }3.3 基于向量的语义检索实现语义检索是 AI 知识库的核心能力以下实现基于 Chroma 的相似性搜索from langchain.embeddings import OpenAIEmbeddings from langchain.vectorstores import Chroma from langchain.text_splitter import RecursiveCharacterTextSplitter class VectorSearchService: def __init__(self, persist_directory: str ./chroma_db): self.embeddings OpenAIEmbeddings( modeltext-embedding-ada-002, openai_api_keyos.getenv(OPENAI_API_KEY) ) self.vector_store Chroma( persist_directorypersist_directory, embedding_functionself.embeddings ) self.text_splitter RecursiveCharacterTextSplitter( chunk_size500, chunk_overlap50 ) async def add_document(self, content: str, metadata: dict): 添加文档到向量数据库 # 文本分块 chunks self.text_splitter.split_text(content) # 为每个块生成向量并存储 for i, chunk in enumerate(chunks): chunk_metadata metadata.copy() chunk_metadata[chunk_index] i self.vector_store.add_texts( texts[chunk], metadatas[chunk_metadata] ) async def search_similar(self, query: str, k: int 5): 语义搜索相似内容 results self.vector_store.similarity_search(query, kk) return [ { content: doc.page_content, source: doc.metadata.get(source, 未知), score: doc.metadata.get(score, 0) } for doc in results ] # 在 FastAPI 中提供搜索接口 app.get(/api/search) async def search_knowledge( q: str, search_type: str semantic, # semantic 或 keyword user_id: int Depends(get_current_user) ): 知识库搜索接口 vector_service VectorSearchService() if search_type semantic: results await vector_service.search_similar(q) else: # 关键字搜索 fallback results await keyword_search(q, user_id) return {query: q, results: results}4. 集成 AI 问答与知识推理能力单纯的检索还不够我们需要让系统能够理解问题并生成自然语言的答案。4.1 RAG 流程的完整实现检索增强生成RAG将检索和生成两个步骤有机结合from langchain.chat_models import ChatOpenAI from langchain.schema import HumanMessage, SystemMessage from langchain.prompts import ChatPromptTemplate class RAGService: def __init__(self): self.llm ChatOpenAI( model_namegpt-3.5-turbo, temperature0.1, # 低温度保证答案稳定性 openai_api_keyos.getenv(OPENAI_API_KEY) ) self.vector_search VectorSearchService() # 定义系统提示词模板 self.prompt_template ChatPromptTemplate.from_messages([ (system, 你是一个专业的知识库助手基于用户提供的上下文信息回答问题。 请遵循以下规则 1. 只基于提供的上下文信息回答不要使用外部知识 2. 如果上下文信息不足以回答问题如实告知用户 3. 回答要准确、简洁、专业 4. 引用上下文中的关键信息支持你的回答 上下文信息 {context}), (human, {question}) ]) async def ask_question(self, question: str, user_id: int): 基于知识库回答问题 # 1. 检索相关文档 search_results await self.vector_search.search_similar(question, k3) if not search_results: return 抱歉知识库中没有找到相关信息来回答这个问题。 # 2. 构建上下文 context \n\n.join([ f来源{result[source]}\n内容{result[content]} for result in search_results ]) # 3. 构建提示词 prompt self.prompt_template.format_messages( contextcontext, questionquestion ) # 4. 调用 LLM 生成答案 response await self.llm.agenerate([prompt]) answer response.generations[0][0].text return { answer: answer, sources: [result[source] for result in search_results], context: context # 调试用生产环境可移除 } # 问答接口 app.post(/api/ask) async def ask_question(request: dict, user_id: int Depends(get_current_user)): 智能问答接口 question request.get(question, ).strip() if not question: raise HTTPException(400, 问题不能为空) rag_service RAGService() response await rag_service.ask_question(question, user_id) # 记录问答历史用于改进系统 await save_question_history(user_id, question, response) return response4.2 处理 AI 幻觉与知识边界问题AI 模型有时会产生幻觉Hallucination即生成看似合理但实际错误的信息。在知识库系统中这是需要重点防范的风险。防范策略包括严格的上下文限制明确要求模型只基于提供的上下文回答。置信度评估对模型的回答进行置信度评分低置信度的回答需要特殊标记。多源验证从多个相关文档中检索信息交叉验证一致性。用户反馈机制允许用户标记答案质量用于改进系统。class HallucinationChecker: 简单的幻觉检测器 async def check_answer_consistency(self, question: str, answer: str, context: str): 检查答案与上下文的一致性 prompt f 请判断以下答案是否严格基于提供的上下文信息。如果答案中的关键信息在上下文中没有明确依据请标记为不一致。 问题{question} 上下文{context} 答案{answer} 请用JSON格式回复{{consistent: true/false, reason: 不一致的原因如果存在}} response await self.llm.agenerate([[HumanMessage(contentprompt)]]) result_text response.generations[0][0].text try: import json return json.loads(result_text) except: return {consistent: True, reason: 解析失败默认通过}4.3 对话历史与上下文管理对于复杂的多轮对话需要维护对话历史以确保上下文连贯class ConversationManager: 对话历史管理器 def __init__(self, max_history: int 10): self.max_history max_history self.conversations {} # user_id - conversation_history def get_conversation_history(self, user_id: int): 获取用户的对话历史 return self.conversations.get(user_id, []) def add_to_history(self, user_id: int, question: str, answer: str): 添加对话到历史 if user_id not in self.conversations: self.conversations[user_id] [] self.conversations[user_id].append({ question: question, answer: answer, timestamp: datetime.now() }) # 保持历史长度不超过限制 if len(self.conversations[user_id]) self.max_history: self.conversations[user_id] self.conversations[user_id][-self.max_history:] def build_context_with_history(self, current_question: str, history: list): 构建包含历史上下文的提示词 if not history: return current_question history_context 之前的对话历史\n for i, exchange in enumerate(history[-3:]): # 只使用最近3轮历史 history_context fQ{i1}: {exchange[question]}\n history_context fA{i1}: {exchange[answer]}\n\n return f{history_context}当前问题{current_question}5. 前端界面与用户体验优化技术实现完成后需要提供友好的用户界面来展示 AI 知识库的能力。5.1 主要功能模块的界面设计知识库前端应该包含以下核心页面仪表盘显示最近笔记、阅读进度、搜索统计等概览信息。笔记管理支持创建、编辑、分类、搜索笔记的界面。书库管理书籍上传、列表展示、阅读器集成。智能问答聊天式界面显示问答历史和参考来源。知识图谱可视化展示知识点之间的关联关系。使用现代前端框架如 React、Vue.js构建单页面应用确保响应式设计支持多设备访问。5.2 搜索与发现功能的交互设计优秀的搜索体验是知识库系统的关键// React 组件示例智能搜索框 import { useState, useCallback } from react; import { debounce } from lodash; const SmartSearch ({ onSearch, onResultClick }) { const [query, setQuery] useState(); const [results, setResults] useState([]); const [searchType, setSearchType] useState(semantic); // semantic 或 keyword // 防抖搜索避免频繁请求 const debouncedSearch useCallback( debounce(async (searchQuery, type) { if (!searchQuery.trim()) { setResults([]); return; } try { const response await fetch(/api/search?q${encodeURIComponent(searchQuery)}search_type${type}); const data await response.json(); setResults(data.results || []); } catch (error) { console.error(搜索失败:, error); } }, 300), [] ); const handleInputChange (e) { const newQuery e.target.value; setQuery(newQuery); debouncedSearch(newQuery, searchType); }; return ( div classNamesearch-container div classNamesearch-header input typetext value{query} onChange{handleInputChange} placeholder搜索笔记、书籍或直接提问... classNamesearch-input / select value{searchType} onChange{(e) setSearchType(e.target.value)} classNamesearch-type-select option valuesemantic语义搜索/option option valuekeyword关键词搜索/option /select /div {results.length 0 ( div classNamesearch-results {results.map((result, index) ( div key{index} classNamesearch-result-item onClick{() onResultClick(result)} div classNameresult-source{result.source}/div div classNameresult-content{result.content}/div /div ))} /div )} /div ); };5.3 响应式设计与移动端适配确保知识库在手机、平板等设备上也能正常使用/* 响应式设计示例 */ .search-container { max-width: 800px; margin: 0 auto; padding: 20px; } .search-input { width: 100%; padding: 12px; font-size: 16px; border: 1px solid #ddd; border-radius: 8px; } /* 移动端适配 */ media (max-width: 768px) { .search-container { padding: 10px; } .search-header { flex-direction: column; } .search-type-select { margin-top: 10px; width: 100%; } }6. 部署与生产环境考量开发完成后需要将系统部署到生产环境并确保稳定运行。6.1 容器化部署与配置管理使用 Docker 容器化部署可以保证环境一致性# Dockerfile FROM python:3.9-slim WORKDIR /app # 安装系统依赖 RUN apt-get update apt-get install -y \ gcc \ rm -rf /var/lib/apt/lists/* # 复制依赖文件并安装 COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt # 复制应用代码 COPY . . # 创建非root用户 RUN useradd --create-home --shell /bin/bash app USER app EXPOSE 8000 CMD [uvicorn, main:app, --host, 0.0.0.0, --port, 8000]使用 Docker Compose 编排多个服务# docker-compose.yml version: 3.8 services: web: build: . ports: - 8000:8000 environment: - DATABASE_URLpostgresql://user:passworddb:5432/knowledgebase - REDIS_URLredis://redis:6379 depends_on: - db - redis db: image: postgres:15 environment: - POSTGRES_DBknowledgebase - POSTGRES_USERuser - POSTGRES_PASSWORDpassword volumes: - postgres_data:/var/lib/postgresql/data redis: image: redis:7-alpine volumes: postgres_data:6.2 性能优化与缓存策略生产环境需要关注性能优化数据库查询优化为常用查询字段建立索引避免 N1 查询问题。向量检索优化使用 HNSW 等高效索引算法设置合适的检索参数。多级缓存使用 Redis 缓存频繁访问的查询结果和会话数据。异步处理将耗时的操作如文档解析、向量化放入任务队列异步处理。# 使用 Redis 缓存示例 import redis import json from functools import wraps redis_client redis.Redis(hostlocalhost, port6379, db0) def cache_result(expire_seconds: int 300): 缓存装饰器 def decorator(func): wraps(func) async def wrapper(*args, **kwargs): # 生成缓存键 cache_key f{func.__name__}:{str(args)}:{str(kwargs)} # 尝试从缓存获取 cached redis_client.get(cache_key) if cached: return json.loads(cached) # 执行函数并缓存结果 result await func(*args, **kwargs) redis_client.setex(cache_key, expire_seconds, json.dumps(result)) return result return wrapper return decorator # 使用缓存 cache_result(expire_seconds600) # 缓存10分钟 async def search_knowledge_cached(q: str, user_id: int): 带缓存的搜索函数 return await search_knowledge(q, user_id)6.3 监控、日志与错误处理完善的监控体系是生产系统的保障import logging from logging.handlers import RotatingFileHandler import sentry_sdk from sentry_sdk.integrations.fastapi import FastApiIntegration # 配置结构化日志 logging.basicConfig( levellogging.INFO, format%(asctime)s - %(name)s - %(levelname)s - %(message)s, handlers[ RotatingFileHandler(app.log, maxBytes10485760, backupCount5), # 10MB per file logging.StreamHandler() ] ) # 错误监控Sentry sentry_sdk.init( dsnos.getenv(SENTRY_DSN), integrations[FastApiIntegration()], traces_sample_rate1.0 ) # 自定义中间件记录请求日志 app.middleware(http) async def log_requests(request: Request, call_next): start_time time.time() response await call_next(request) process_time time.time() - start_time logging.info( f{request.method} {request.url} - Status: {response.status_code} - Time: {process_time:.2f}s ) return response7. 常见问题排查与优化建议在实际使用中可能会遇到各种问题以下是典型问题的排查路径。7.1 向量检索效果不佳的调优方法如果语义搜索返回的结果不相关可以从以下几个方面排查问题现象可能原因检查方式解决方案检索结果完全不相关文本分块大小不合适检查分块后文本的语义完整性调整分块大小200-1000字符尝试按段落或章节分块部分相关但排名不对嵌入模型不适合领域测试不同嵌入模型在同一数据集上的表现尝试领域特定的嵌入模型如针对中文的BGE模型检索遗漏关键信息检索数量k值太小检查top k个结果是否包含正确答案增加k值或使用重排序re-ranking技术相似问题结果不一致向量索引参数需要优化检查向量索引的构建参数调整HNSW参数ef_construction、M或尝试其他索引算法7.2 AI 回答质量问题的诊断流程当 AI 生成的答案质量不高时按以下步骤诊断检查检索质量首先确认检索到的上下文是否包含正确答案。验证提示词设计检查系统提示词是否清晰传达了约束条件。分析模型温度设置过高的温度可能导致答案不稳定建议设置为0.1-0.3。检查上下文长度过长的上下文可能让模型忽略关键信息尝试精简上下文。测试不同模型某些问题可能更适合特定的模型版本。async def diagnose_answer_quality(question: str, answer: str, context: str): 答案质量诊断工具 diagnostics {} # 检查上下文相关性 relevance_check await check_context_relevance(question, context) diagnostics[context_relevance] relevance_check # 检查答案与上下文一致性 consistency_check await check_answer_consistency(question, answer, context) diagnostics[answer_consistency] consistency_check # 检查答案完整性 completeness_check await check_answer_completeness(question, answer) diagnostics[answer_completeness] completeness_check return diagnostics7.3 系统性能瓶颈的识别与解决随着数据量增长系统可能遇到性能瓶颈数据库性能优化为常用查询字段添加索引定期清理无用数据使用连接池避免频繁建立连接向量检索优化使用更高效的索引算法HNSW考虑向量量化PQ减少内存占用对向量进行归一化处理缓存策略优化对频繁访问的查询结果设置缓存使用多级缓存内存缓存 Redis合理设置缓存过期时间8. 安全性与权限管理实践知识库可能包含敏感信息需要完善的安全措施。8.1 用户认证与授权设计实现基于 JWT 的认证系统from jose import JWTError, jwt from passlib.context import CryptContext from datetime import datetime, timedelta pwd_context CryptContext(schemes[bcrypt], deprecatedauto) # 密码哈希和验证 def verify_password(plain_password, hashed_password): return pwd_context.verify(plain_password, hashed_password) def get_password_hash(password): return pwd_context.hash(password) # JWT token 生成和验证 def create_access_token(data: dict, expires_delta: timedelta None): to_encode data.copy() if expires_delta: expire datetime.utcnow() expires_delta else: expire datetime.utcnow() timedelta(minutes15) to_encode.update({exp: expire}) encoded_jwt jwt.encode(to_encode, SECRET_KEY, algorithmALGORITHM) return encoded_jwt async def get_current_user(token: str Depends(oauth2_scheme)): credentials_exception HTTPException( status_code401, detailCould not validate credentials, ) try: payload jwt.decode(token, SECRET_KEY, algorithms[ALGORITHM]) username: str payload.get(sub) if username is None: raise credentials_exception except JWTError: raise credentials_exception user get_user_from_db(username) if user is None: raise credentials_exception return user8.2 数据加密与隐私保护对敏感数据进行加密存储from cryptography.fernet import Fernet class DataEncryptor: def __init__(self, key: str None): self.key key or os.getenv(ENCRYPTION_KEY) self.fernet Fernet(self.key) def encrypt_data(self, data: str) - str: 加密数据 if not data: return data encrypted self.fernet.encrypt(data.encode()) return encrypted.decode() def decrypt_data(self, encrypted_data: str) - str: 解密数据 if not encrypted_data: return encrypted_data decrypted self.fernet.decrypt(encrypted_data.encode()) return decrypted.decode() # 在存储敏感信息时使用加密 encryptor DataEncryptor() class UserCreate(BaseModel): username: str email: str password: str async def create_user(user: UserCreate): 创建用户密码加密存储 hashed_password get_password_hash(user.password) encrypted_email encryptor.encrypt_data(user.email) # 存储加密后的数据 user_record User( usernameuser.username, emailencrypted_email, hashed_passwordhashed_password ) # ... 保存到数据库8.3 API 安全最佳实践确保 API 接口的安全性速率限制防止 API 被滥用输入验证对所有输入数据进行严格验证SQL 注入防护使用参数化查询或 ORMCORS 配置合理配置跨域请求策略敏感信息过滤避免在日志和响应中泄露敏感信息from slowapi import Limiter, _rate_limit_exceeded_handler from