# FastAPI + LangChain 政务政策 RAG 后端项目实战：从零构建生产级智能问答系统

FastAPI + LangChain 政务政策 RAG 后端项目实战：从零构建生产级智能问答系统

摘要：本文基于 FastAPI 框架和 LangChain，手把手教你从零构建一个完整的政务政策知识库问答后端系统。涵盖应用工厂模式、Pydantic 数据模型、依赖注入、文件上传、CORS 配置、路由拆分、LangChain 集成、Embedding 向量检索、RAG 问答 API 等完整生产级能力。所有代码可直接复制运行，附踩坑清单和上线检查表。

目录

• 1. 项目概述
• 2. 环境准备
• 3. 应用工厂与配置管理
• 4. 配置 CORS 与安全
• 5. 路由拆分与 API 模块化
• 6. Pydantic 数据模型
• 7. 依赖注入
• 8. 文件上传与处理
• 9. LangChain 集成：模型与 Embedding
• 10. RAG 知识库问答 API
• 11. 健康检查与优雅关闭
• 12. 项目完整结构
• 13. 踩坑清单
• 14. 本章重点
• 15. 封面图提示词

1. 项目概述

本项目是一个基于 FastAPI + LangChain 的政务政策知识库问答后端系统，具备以下能力：

• 基于 FastAPI 应用工厂模式构建后端服务
• 支持文件上传（PDF/TXT/MD）构建知识库索引
• 集成 DeepSeek 大语言模型进行智能问答
• 使用 Embedding 模型 + Milvus 向量数据库实现语义检索
• 提供完整的 RESTful API（上传文档、问答、健康检查）
• 配置 CORS 支持前后端联调
• 自动 Swagger 文档

技术栈：

2. 环境准备

2.1 创建项目目录

mkdir policy-rag-backend
cd policy-rag-backend

2.2 创建虚拟环境

python -m venv .venv

# Windows
.\\.venv\\Scripts\\activate

# macOS/Linux
source .venv/bin/activate

2.3 安装依赖

创建 requirements.txt：

# Web 框架
fastapi[standard]>=0.115.0
uvicorn[standard]>=0.30.0
python-multipart>=0.0.9

# 配置与环境
python-dotenv>=1.0.0
pydantic>=2.9.0
pydantic-settings>=2.5.0

# LangChain
langchain>=0.3.0
langchain-openai>=0.2.0
langchain-community>=0.3.0
langchain-text-splitters>=0.3.0

# 向量数据库
pymilvus>=2.5.0

# 文档解析
PyPDF2>=3.0.0
python-docx>=1.1.0

安装：

pip install -r requirements.txt

2.4 创建 .env 文件

# DeepSeek 配置
DEEPSEEK_API_KEY=sk-your-api-key-here
DEEPSEEK_BASE_URL=https://api.deepseek.com

# Milvus 配置
MILVUS_URI=http://localhost:19530
MILVUS_COLLECTION_NAME=policy_knowledge

# 上传目录
UPLOAD_DIR=uploads

# 应用配置
APP_TITLE=政务政策 RAG 系统
APP_DESCRIPTION=基于 FastAPI + LangChain 的智能政策问答系统
APP_VERSION=1.0.0
CORS_ORIGINS=http://localhost:3000,http://localhost:5173

安全提示：.env 文件包含 API Key，必须加入 .gitignore，不要提交到 Git 仓库。

3. 应用工厂与配置管理

3.1 为什么要用应用工厂

在大型项目中，直接写 app = FastAPI() 会遇到问题：

• 配置分散在多个文件
• 测试时难以替换依赖
• 启动时无法注入配置

应用工厂模式把创建应用的逻辑封装到函数中，支持：

• 集中管理配置
• 按需注册路由
• 灵活控制启动行为

3.2 配置管理

创建 backend/core/config.py：

from pydantic_settings import BaseSettings
from typing import List

class Settings(BaseSettings):
# DeepSeek 配置
deepseek_api_key: str = ""
deepseek_base_url: str = "https://api.deepseek.com"

# Milvus 配置
milvus_uri: str = "http://localhost:19530"
milvus_collection_name: str = "policy_knowledge"

# 应用配置
app_title: str = "政务政策 RAG 系统"
app_description: str = "FastAPI + LangChain 政策知识库问答系统"
app_version: str = "1.0.0"

# CORS 配置
cors_origins: str = "http://localhost:3000"

# 上传目录
upload_dir: str = "uploads"

@property
def cors_origin_list(self) –> List[str]:
return [origin.strip() for origin in self.cors_origins.split(",")]

class Config:
env_file = ".env"
env_file_encoding = "utf-8"

settings = Settings()

关键点：

• BaseSettings 自动从 .env 读取配置
• cors_origin_list 将逗号分隔的字符串转成列表
• 所有配置项都有默认值，开发时不填 .env 也能运行

3.3 应用工厂

创建 backend/app/factory.py：

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from pathlib import Path

from core.config import settings
from api.routes import router

def create_app() –> FastAPI:
# 1. 创建 FastAPI 应用
app = FastAPI(
title=settings.app_title,
description=settings.app_description,
version=settings.app_version,
docs_url="/docs",
redoc_url="/redoc",
)

# 2. 配置 CORS
app.add_middleware(
CORSMiddleware,
allow_origins=settings.cors_origin_list,
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)

# 3. 创建上传目录
upload_path = Path(settings.upload_dir)
upload_path.mkdir(exist_ok=True)

# 4. 注册路由
app.include_router(router, prefix="/api/v1")

# 5. 根路径
@app.get("/")
async def root():
return {
"name": settings.app_title,
"version": settings.app_version,
"docs": "/docs",
"message": "服务已启动",
}

return app

3.4 启动入口

创建 backend/main.py：

from app.factory import create_app

app = create_app()

启动：

cd backend
python -m uvicorn main:app –reload –host 0.0.0.0 –port 8000

4. 配置 CORS 与安全

4.1 什么是 CORS

**CORS（跨源资源共享）**解决浏览器同源策略限制。当前端（localhost:3000）调用后端（localhost:8000）时，浏览器会阻止请求，需要后端明确允许。

4.2 FastAPI 中配置 CORS

在 factory.py 中已经通过 CORSMiddleware 配置：

app.add_middleware(
CORSMiddleware,
allow_origins=settings.cors_origin_list,
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)

参数说明：

4.3 生产环境建议

# 开发环境
CORS_ORIGINS=http://localhost:3000,http://localhost:5173

# 生产环境（只放实际域名）
CORS_ORIGINS=https://your-domain.com

5. 路由拆分与 API 模块化

5.1 为什么要拆分路由

小型项目把所有接口写在 main.py 中没问题。但项目变大后：

• 文件过长，难以维护
• 多人协作容易冲突
• 测试时难以隔离

APIRouter 允许把路由按功能拆分到不同模块。

5.2 路由拆分结构

backend/
├── api/
│ ├── __init__.py
│ ├── routes.py # 总路由入口
│ ├── endpoints/
│ │ ├── __init__.py
│ │ ├── upload.py # 文件上传接口
│ │ └── chat.py # 问答接口

5.3 创建子路由

创建 backend/api/endpoints/upload.py：

from fastapi import APIRouter, UploadFile, File, HTTPException
from pathlib import Path
from core.config import settings

router = APIRouter(prefix="/upload", tags=["文件上传"])

@router.post("/")
async def upload_file(file: UploadFile = File(...)):
"""上传政策文档"""
# 校验文件类型
allowed_extensions = {".pdf", ".txt", ".md"}
file_ext = Path(file.filename).suffix.lower()

if file_ext not in allowed_extensions:
raise HTTPException(
status_code=400,
detail=f"不支持的文件类型：{file_ext}，仅支持 {allowed_extensions}"
)

# 保存文件
upload_path = Path(settings.upload_dir) / file.filename
content = await file.read()
upload_path.write_bytes(content)

return {
"filename": file.filename,
"size": len(content),
"path": str(upload_path),
"message": "上传成功",
}

创建 backend/api/endpoints/chat.py：

from fastapi import APIRouter, HTTPException
from pydantic import BaseModel

router = APIRouter(prefix="/chat", tags=["智能问答"])

class ChatRequest(BaseModel):
question: str
session_id: str = "default"

class ChatResponse(BaseModel):
answer: str
sources: list = []
session_id: str

@router.post("/", response_model=ChatResponse)
async def chat(request: ChatRequest):
"""政策知识库问答"""
# 这里先返回占位回答，后面集成 RAG
return ChatResponse(
answer="正在建设中，请稍后重试",
sources=[],
session_id=request.session_id,
)

5.4 注册总路由

创建 backend/api/routes.py：

from fastapi import APIRouter
from api.endpoints import upload, chat

router = APIRouter()
router.include_router(upload.router)
router.include_router(chat.router)

在 factory.py 中注册：

from api.routes import router

app.include_router(router, prefix="/api/v1")

5.5 访问路径

6. Pydantic 数据模型

6.1 为什么用 Pydantic

FastAPI 使用 Pydantic 做数据校验、序列化、文档生成。请求参数和响应数据通过 Pydantic 模型自动校验类型。

6.2 请求模型

from pydantic import BaseModel, Field

class ChatRequest(BaseModel):
question: str = Field(
..., # 必填
min_length=1,
max_length=500,
description="用户问题",
)
session_id: str = Field(
default="default",
description="会话 ID",
)

6.3 响应模型

class DocumentSource(BaseModel):
file_name: str
page: int
content: str

class ChatResponse(BaseModel):
answer: str = Field(description="模型回答")
sources: list[DocumentSource] = Field(
default=[],
description="引用来源",
)
session_id: str

6.4 自动校验效果

如果请求参数不合法，FastAPI 自动返回 422 错误：

{
"detail": [
{
"loc": ["body", "question"],
"msg": "ensure this value has at least 1 characters",
"type": "value_error.any_str.min_length"
}
]
}

7. 依赖注入

7.1 什么是依赖注入

依赖注入（DI）是把需要的资源（数据库连接、配置、服务等）通过参数传入，而不是在函数内部创建。

7.2 FastAPI 中定义依赖

创建 backend/core/deps.py：

from fastapi import Depends
from core.config import settings
from services.rag_service import RAGService

# 全局单例
_rag_service: RAGService | None = None

def get_settings():
"""获取配置"""
return settings

def get_rag_service() –> RAGService:
"""获取 RAG 服务（懒加载单例）"""
global _rag_service
if _rag_service is None:
_rag_service = RAGService()
return _rag_service

7.3 在路由中使用依赖

from fastapi import APIRouter, Depends
from core.deps import get_rag_service
from services.rag_service import RAGService

router = APIRouter()

@router.post("/")
async def chat(
request: ChatRequest,
rag_service: RAGService = Depends(get_rag_service),
):
"""问答接口"""
answer = rag_service.ask(request.question)
return ChatResponse(
answer=answer,
sources=[],
session_id=request.session_id,
)

优点：

• 测试时可以 mock 依赖
• 依赖的生命周期由 FastAPI 管理
• 代码解耦，职责清晰

8. 文件上传与处理

8.1 上传接口

前面已经创建了 upload.py，现在补充文档解析和向量入库。

8.2 文档解析服务

创建 backend/services/document_service.py：

from pathlib import Path
from typing import List
from langchain_core.documents import Document
from langchain_community.document_loaders import PyPDFLoader, TextLoader

class DocumentService:
def __init__(self, upload_dir: str):
self.upload_dir = Path(upload_dir)

def parse_file(self, file_path: Path) –> List[Document]:
"""解析文档为文本块"""
suffix = file_path.suffix.lower()

if suffix == ".pdf":
loader = PyPDFLoader(str(file_path))
elif suffix in {".txt", ".md"}:
loader = TextLoader(str(file_path), encoding="utf-8")
else:
raise ValueError(f"不支持的文件类型：{suffix}")

return loader.load()

def get_file_path(self, filename: str) –> Path:
return self.upload_dir / filename

8.3 完整上传流程

修改 upload.py：

from fastapi import APIRouter, UploadFile, File, HTTPException, Depends
from pathlib import Path
from core.config import settings
from core.deps import get_document_service
from services.document_service import DocumentService

router = APIRouter(prefix="/upload", tags=["文件上传"])

@router.post("/")
async def upload_file(
file: UploadFile = File(...),
doc_service: DocumentService = Depends(get_document_service),
):
"""上传并解析政策文档"""
# 校验文件类型
allowed_extensions = {".pdf", ".txt", ".md"}
file_ext = Path(file.filename).suffix.lower()

if file_ext not in allowed_extensions:
raise HTTPException(
status_code=400,
detail=f"不支持的文件类型：{file_ext}"
)

# 保存文件
upload_path = Path(settings.upload_dir) / file.filename
content = await file.read()
upload_path.write_bytes(content)

# 解析文档
try:
documents = doc_service.parse_file(upload_path)
return {
"filename": file.filename,
"size": len(content),
"pages": len(documents),
"message": "上传并解析成功",
}
except Exception as e:
raise HTTPException(status_code=500, detail=f"解析失败：{str(e)}")

9. LangChain 集成：模型与 Embedding

9.1 模型工厂

创建 backend/services/model_factory.py：

from langchain_openai import ChatOpenAI
from langchain_community.embeddings import DashScopeEmbeddings
from core.config import settings

def get_chat_model():
"""获取聊天模型"""
return ChatOpenAI(
model="deepseek-chat",
api_key=settings.deepseek_api_key,
base_url=settings.deepseek_base_url,
temperature=0.7,
)

def get_embedding_model():
"""获取 Embedding 模型"""
return DashScopeEmbeddings(
model="text-embedding-v4",
dashscope_api_key=settings.deepseek_api_key,
)

9.2 向量数据库连接

创建 backend/services/vector_store.py：

from langchain_community.vectorstores import Milvus
from core.config import settings
from services.model_factory import get_embedding_model
from pymilvus import connections, utility

def get_vector_store(collection_name: str = None):
"""获取 Milvus 向量库实例"""
if collection_name is None:
collection_name = settings.milvus_collection_name

# 连接 Milvus
if not connections.has_connection("default"):
connections.connect("default", uri=settings.milvus_uri)

embeddings = get_embedding_model()

return Milvus(
embedding_function=embeddings,
collection_name=collection_name,
connection_args={"uri": settings.milvus_uri},
index_params={
"index_type": "HNSW",
"metric_type": "COSINE",
"params": {"M": 16, "efConstruction": 128},
},
auto_id=True,
)

10. RAG 知识库问答 API

10.1 RAG 服务

创建 backend/services/rag_service.py：

from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_core.documents import Document
from services.vector_store import get_vector_store
from services.model_factory import get_chat_model
from services.document_service import DocumentService
from core.config import settings
from pathlib import Path

class RAGService:
def __init__(self):
self.vector_store = get_vector_store()
self.chat_model = get_chat_model()
self.document_service = DocumentService(settings.upload_dir)
self.text_splitter = RecursiveCharacterTextSplitter(
chunk_size=500,
chunk_overlap=50,
separators=["\\n\\n", "\\n", "。", "！", "？"],
)

def build_index(self, file_path: Path):
"""构建知识库索引"""
# 1. 解析文档
documents = self.document_service.parse_file(file_path)

# 2. 切分文档
chunks = self.text_splitter.split_documents(documents)

# 3. 写入向量库
self.vector_store.add_documents(chunks)

return len(chunks)

def ask(self, question: str) –> str:
"""问答"""
# 1. 检索相关文档
results = self.vector_store.similarity_search(question, k=3)

if not results:
return "未找到相关政策内容，请尝试其他问题。"

# 2. 构建上下文
context = "\\n\\n".join([doc.page_content for doc in results])

# 3. 调用模型
prompt = f"""基于以下政策资料回答问题：

资料：
{context}

问题：{question}

请根据资料回答，如果资料中没有相关信息，请说明。"""

response = self.chat_model.invoke(prompt)
return response.content

10.2 问答接口

修改 chat.py：

from fastapi import APIRouter, Depends
from pydantic import BaseModel, Field
from core.deps import get_rag_service
from services.rag_service import RAGService

router = APIRouter(prefix="/chat", tags=["智能问答"])

class ChatRequest(BaseModel):
question: str = Field(..., min_length=1, max_length=500)
session_id: str = "default"

class DocumentSource(BaseModel):
file_name: str
content: str

class ChatResponse(BaseModel):
answer: str
sources: list[DocumentSource] = []
session_id: str

@router.post("/", response_model=ChatResponse)
async def chat(
request: ChatRequest,
rag_service: RAGService = Depends(get_rag_service),
):
"""政策知识库问答"""
answer = rag_service.ask(request.question)

return ChatResponse(
answer=answer,
sources=[],
session_id=request.session_id,
)

10.3 索引构建接口

在 upload.py 中添加构建索引逻辑：

@router.post("/{filename}/index")
async def build_index(
filename: str,
rag_service: RAGService = Depends(get_rag_service),
):
"""为已上传的文档构建向量索引"""
file_path = Path(settings.upload_dir) / filename

if not file_path.exists():
raise HTTPException(status_code=404, detail="文件不存在")

try:
chunk_count = rag_service.build_index(file_path)
return {
"filename": filename,
"chunks": chunk_count,
"message": "索引构建成功",
}
except Exception as e:
raise HTTPException(status_code=500, detail=f"索引构建失败：{str(e)}")

11. 健康检查与优雅关闭

11.1 健康检查接口

在 factory.py 中添加：

@app.get("/health")
async def health_check():
return {
"status": "healthy",
"version": settings.app_version,
"timestamp": datetime.now().isoformat(),
}

11.2 优雅关闭

@app.on_event("shutdown")
async def shutdown_event():
"""应用关闭时清理资源"""
from pymilvus import connections
connections.disconnect("default")
print("已断开 Milvus 连接")

12. 项目完整结构

policy-rag-backend/
├── .env # 环境变量（密钥）
├── .gitignore # Git 忽略文件
├── requirements.txt # 依赖清单
├── backend/
│ ├── main.py # 启动入口
│ ├── app/
│ │ └── factory.py # 应用工厂
│ ├── api/
│ │ ├── __init__.py
│ │ ├── routes.py # 总路由入口
│ │ └── endpoints/
│ │ ├── __init__.py
│ │ ├── upload.py # 文件上传
│ │ └── chat.py # 智能问答
│ ├── core/
│ │ ├── __init__.py
│ │ ├── config.py # 配置管理
│ │ └── deps.py # 依赖注入
│ └── services/
│ ├── __init__.py
│ ├── model_factory.py # 模型工厂
│ ├── vector_store.py # 向量库连接
│ ├── document_service.py # 文档解析
│ └── rag_service.py # RAG 服务
└── uploads/ # 上传文件目录

13. 踩坑清单

14. 本章重点

本章需要重点掌握：

• 应用工厂：create_app() 封装创建逻辑，支持配置注入和灵活扩展
• 配置管理：Pydantic Settings 自动读取 .env，开发/生产环境切换
• 路由拆分：APIRouter 按功能模块拆分，代码结构清晰
• Pydantic 模型：请求/响应自动校验，类型安全
• 依赖注入：Depends() 解耦服务，支持测试 mock
• 文件上传：UploadFile 处理大文件，支持流式读取
• CORS 配置：前后端联调必备，生产环境严格限制域名
• LangChain 集成：模型工厂 + Embedding + 向量库，构建 RAG 服务

完整后端问答流程：

用户上传文档
-> 解析为文本块
-> Embedding 向量化
-> 存入 Milvus

用户提问
-> Embedding 向量化
-> Milvus 相似度检索
-> 获取上下文
-> DeepSeek 生成回答
-> 返回 JSON 响应

如果本文对你有帮助，欢迎点赞收藏。如有问题，欢迎在评论区留言交流。