把 RAG 系统封装成RESTful API可被前端 / 小程序 / App 调用给 API 加文档、异常处理、日志、配置化把整套服务Docker 容器化一键部署、到处运行接口支持上传 PDF、构建知识库、提问、清空知识库先讲核心新知识必须掌握1. FastAPI 是什么高性能 Python Web 框架用来把你的 AI 服务变成 API 接口自动生成 API 文档/docs支持异步、高并发类型提示、自动校验参数适合做 AI 后端服务2. Docker 是什么容器化工具把你的代码 环境 依赖打包成一个镜像解决「在我电脑能跑你那不能跑」问题一键部署到服务器 / 云平台隔离环境稳定不冲突3. 我们要做的架构用户/前端 → FastAPI接口 → RAG系统 → Chroma向量库 → 返回回答 第一步项目结构标准工程化plaintextrag_api_project/ ├── .env # 配置文件 ├── main.py # FastAPI 主程序 ├── requirements.txt # 依赖 ├── Dockerfile # 构建镜像 ├── .dockerignore # Docker 忽略文件 └── uploads/ # 上传PDF目录 └── chroma_db/ # 向量库自动生成 第二步完整代码功能极强、稳定、可直接部署1.requirements.txtfastapi uvicorn[standard] python-multipart python-dotenv langchain langchain-openai langchain-community pypdf chromadb2..envDOUBAO_API_KEY你的豆包key CHROMA_DB_DIR./chroma_db UPLOAD_DIR./uploads HOST0.0.0.0 PORT80003.main.py核心FastAPI RAGimport os import logging from fastapi import FastAPI, UploadFile, File, HTTPException from fastapi.responses import JSONResponse from dotenv import load_dotenv import shutil # RAG 相关 from langchain_community.document_loaders import PyPDFLoader from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain_openai import OpenAIEmbeddings, ChatOpenAI from langchain_community.vectorstores import Chroma from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser from langchain_core.runnables import RunnablePassthrough # 日志配置 logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) # 加载配置 load_dotenv() app FastAPI(titleRAG 文档问答 API, version2.0) # 创建目录 os.makedirs(os.getenv(UPLOAD_DIR), exist_okTrue) # 模型初始化 llm ChatOpenAI( api_keyos.getenv(DOUBAO_API_KEY), base_urlhttps://ark.cn-beijing.volces.com/api/v3, modeldoubao-pro-32k, temperature0.0 ) embedding OpenAIEmbeddings( api_keyos.getenv(DOUBAO_API_KEY), base_urlhttps://ark.cn-beijing.volces.com/api/v3 ) # 全局变量 vector_db None rag_chain None # RAG 工具函数 def load_pdf_and_build_rag(pdf_path): global vector_db, rag_chain # 加载 loader PyPDFLoader(pdf_path) docs loader.load() # 分割 splitter RecursiveCharacterTextSplitter( chunk_size1000, chunk_overlap200, separators[\n\n, \n, 。, , , ] ) chunks splitter.split_documents(docs) # 构建向量库 vector_db Chroma.from_documents( documentschunks, embeddingembedding, persist_directoryos.getenv(CHROMA_DB_DIR) ) vector_db.persist() # 构建 RAG retriever vector_db.as_retriever(k3) prompt ChatPromptTemplate.from_template( 你是严格的文档问答助手只根据下面的文档内容回答。 不知道就说文档中没有相关内容。 文档 {context} 问题 {question} ) rag_chain ( {context: retriever, question: RunnablePassthrough()} | prompt | llm | StrOutputParser() ) return len(chunks) def clear_knowledge_base(): global vector_db, rag_chain db_path os.getenv(CHROMA_DB_DIR) if os.path.exists(db_path): shutil.rmtree(db_path) vector_db None rag_chain None # API 接口 app.get(/, summary健康检查) def health_check(): return {status: running, message: RAG API 服务正常} app.post(/upload-pdf, summary上传PDF并构建知识库) def upload_pdf(file: UploadFile File(...)): try: if not file.filename.endswith(.pdf): raise HTTPException(status_code400, detail只支持PDF文件) save_path os.path.join(os.getenv(UPLOAD_DIR), file.filename) with open(save_path, wb) as f: f.write(file.file.read()) chunk_num load_pdf_and_build_rag(save_path) return JSONResponse({ status: success, filename: file.filename, chunk_count: chunk_num, message: PDF上传并构建知识库成功 }) except Exception as e: logger.error(f上传失败{str(e)}) raise HTTPException(status_code500, detailf处理失败{str(e)}) app.post(/ask, summary向RAG提问) def ask_question(question: str): try: if not vector_db or not rag_chain: raise HTTPException(status_code400, detail请先上传PDF构建知识库) answer rag_chain.invoke(question) return JSONResponse({ status: success, question: question, answer: answer }) except Exception as e: logger.error(f提问失败{str(e)}) raise HTTPException(status_code500, detailf处理失败{str(e)}) app.post(/clear, summary清空知识库) def clear_db(): clear_knowledge_base() return {status: success, message: 向量库已清空} # 启动 if __name__ __main__: import uvicorn uvicorn.run( main:app, hostos.getenv(HOST), portint(os.getenv(PORT)), reloadTrue )4.Dockerfile容器化打包dockerfileFROM python:3.11-slim WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY . . EXPOSE 8000 CMD [python, main.py]5..dockerignore__pycache__ .env uploads/ chroma_db/ *.pyc *.pyo *.pyd 第三步详细功能说明超级重要1. 提供4 个生产级 APIGET /→ 健康检查POST /upload-pdf→ 上传 PDF自动构建向量库POST /ask?questionxxx→ 提问返回 RAG 回答POST /clear→ 清空知识库2. 自动生成 API 文档可视化启动后访问http://localhost:8000/docs你会看到Swagger UI 自动接口文档可以直接在线测试3. 工程化特性全局异常捕获日志系统类型校验文件上传安全校验环境变量配置向量库持久化状态管理是否已上传 PDF 第四步运行方式2 种方式 1直接运行本地测试python main.py方式 2Docker 容器运行生产部署# 构建镜像 docker build -t rag-api . # 运行容器 docker run -p 8000:8000 rag-api访问http://localhost:8000/docs 第五步API 调用示例任何语言都能调用用 curl 调用# 上传PDF curl -X POST http://localhost:8000/upload-pdf -F filetest.pdf # 提问 curl -X POST http://localhost:8000/ask?question文档主要讲什么Python 调用python运行import requests requests.post(http://localhost:8000/ask, params{question:什么是RAG})前端 / 小程序 / APP 都能调用 第六步你今天学到的硬核新知识✅ FastAPI 集成 LangChain✅ 文件上传 PDF 处理✅ RAG 系统接口化✅ 全局异常处理✅ 日志系统✅ 环境变量配置✅ Docker 容器化✅ 生产级项目结构✅ Swagger API 文档