LangChain RAG + FastAPI 接口化 + Docker 容器化
- 把 RAG 系统封装成RESTful API（可被前端 / 小程序 / App 调用）
- 给 API 加文档、异常处理、日志、配置化
- 把整套服务Docker 容器化，一键部署、到处运行
- 接口支持：上传 PDF、构建知识库、提问、清空知识库
先讲核心新知识（必须掌握）
1. FastAPI 是什么？
高性能 Python Web 框架，用来把你的 AI 服务变成 API 接口：
- 自动生成 API 文档（
/docs） - 支持异步、高并发
- 类型提示、自动校验参数
- 适合做 AI 后端服务
2. Docker 是什么？
容器化工具，把你的代码 + 环境 + 依赖打包成一个镜像：
- 解决「在我电脑能跑，你那不能跑」问题
- 一键部署到服务器 / 云平台
- 隔离环境，稳定不冲突
3. 我们要做的架构
用户/前端 → FastAPI接口 → RAG系统 → Chroma向量库 → 返回回答🚀 第一步：项目结构（标准工程化）
plaintext
rag_api_project/ ├── .env # 配置文件 ├── main.py # FastAPI 主程序 ├── requirements.txt # 依赖 ├── Dockerfile # 构建镜像 ├── .dockerignore # Docker 忽略文件 └── uploads/ # 上传PDF目录 └── chroma_db/ # 向量库（自动生成）📦 第二步：完整代码（功能极强、稳定、可直接部署）
1.requirements.txt
fastapi uvicorn[standard] python-multipart python-dotenv langchain langchain-openai langchain-community pypdf chromadb2..env
DOUBAO_API_KEY=你的豆包key CHROMA_DB_DIR=./chroma_db UPLOAD_DIR=./uploads HOST=0.0.0.0 PORT=80003.main.py（核心：FastAPI + RAG）
import 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(level=logging.INFO) logger = logging.getLogger(__name__) # ===================== 加载配置 ===================== load_dotenv() app = FastAPI(title="RAG 文档问答 API", version="2.0") # 创建目录 os.makedirs(os.getenv("UPLOAD_DIR"), exist_ok=True) # ===================== 模型初始化 ===================== llm = ChatOpenAI( api_key=os.getenv("DOUBAO_API_KEY"), base_url="https://ark.cn-beijing.volces.com/api/v3", model="doubao-pro-32k", temperature=0.0 ) embedding = OpenAIEmbeddings( api_key=os.getenv("DOUBAO_API_KEY"), base_url="https://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_size=1000, chunk_overlap=200, separators=["\n\n", "\n", "。", "！", "？", " "] ) chunks = splitter.split_documents(docs) # 构建向量库 vector_db = Chroma.from_documents( documents=chunks, embedding=embedding, persist_directory=os.getenv("CHROMA_DB_DIR") ) vector_db.persist() # 构建 RAG retriever = vector_db.as_retriever(k=3) 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_code=400, 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_code=500, detail=f"处理失败：{str(e)}") @app.post("/ask", summary="向RAG提问") def ask_question(question: str): try: if not vector_db or not rag_chain: raise HTTPException(status_code=400, 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_code=500, detail=f"处理失败：{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", host=os.getenv("HOST"), port=int(os.getenv("PORT")), reload=True )4.Dockerfile（容器化打包）
dockerfile
FROM 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 个生产级 API
GET /→ 健康检查POST /upload-pdf→ 上传 PDF，自动构建向量库POST /ask?question=xxx→ 提问，返回 RAG 回答POST /clear→ 清空知识库
2. 自动生成 API 文档（可视化！）
启动后访问：
http://localhost:8000/docs你会看到Swagger UI 自动接口文档，可以直接在线测试！
3. 工程化特性
- 全局异常捕获
- 日志系统
- 类型校验
- 文件上传安全校验
- 环境变量配置
- 向量库持久化
- 状态管理（是否已上传 PDF）
🚀 第四步：运行方式（2 种）
方式 1：直接运行（本地测试）
python main.py方式 2：Docker 容器运行（生产部署）
# 构建镜像 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 "file=@test.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 文档