RAG项目从Streamlit迁移到FastAPI：架构改造与Docker部署实战

1. 为什么我要把 RAG 项目从 Streamlit 迁到 FastAPI

1.1 起因：一个 Demo 撑不起真实业务

最开始做这个 RAG 知识库项目的时候，我选的是 Streamlit。理由很简单：上手快，几十行代码就能跑出一个带对话界面的问答 Demo，文件上传、聊天记录、流式输出这些都有现成组件，用来验证检索增强的效果再合适不过。那段时间我几乎每天都在用streamlit run app.py启动项目，改几行代码刷新页面就能看到效果，开发体验确实舒服。

但问题很快就来了。第一个真实用户把接口接进来的时候，我意识到 Streamlit 本质上是一个“面向单机演示的前端框架”，它把 UI 和业务逻辑死死绑在一起。别人想调用我的 RAG 能力，只能通过浏览器打开页面手动操作，没法用程序化方式批量提问；我想给不同用户做权限隔离、想记录每次检索的耗时和命中片段、想对并发请求做限流，这些在 Streamlit 里都极其别扭。更致命的是，Streamlit 的脚本模型是“每次交互重跑整个脚本”，一旦检索链路里涉及模型加载、向量库连接，性能开销会被反复放大。

所以这次改造的核心目标很明确：把 RAG 的检索与生成能力从 UI 里剥离出来，做成一套标准的 HTTP 服务，用 FastAPI 承载，用 Docker 打包，用 loguru 做结构化日志。Streamlit 不是被彻底抛弃，而是降级成一个纯粹的“调试前端”，真正对外提供服务的是 FastAPI。下面我把整个改造过程拆开讲，包括目录结构怎么设计、检索链路怎么解耦、Docker 怎么编排、日志怎么补全，以及我踩过的那些坑。

1.2 改造后到底解决了什么问题

先给一个直观的对比，方便你判断自己的项目是不是也到了该改造的时候。

这张表基本概括了我这次改造的全部动机。如果你的 RAG 项目还停留在“自己玩玩”的阶段，Streamlit 完全够用；但只要有人要接你的能力，或者你要做多用户、要做监控、要做灰度，那就必须把服务层独立出来。FastAPI 是 Python 生态里做这件事最顺手的框架，异步原生支持、Pydantic 校验、自动生成 OpenAPI 文档，配合 uvicorn 跑起来非常轻。

2. 整体架构设计与目录结构拆解

2.1 分层思路：把“检索”和“生成”拆成独立模块

改造前我的代码基本是一坨：app.py里既有文件解析、又有向量化、又有检索、又有 prompt 拼接、又有模型调用，最后还塞了 Streamlit 的 UI 代码。这种结构在 Demo 阶段没问题，但一旦要改检索策略，就得在 UI 代码里翻半天。

改造后我按职责做了四层拆分：

• 接口层（api）：只负责接收请求、参数校验、返回响应，不写任何业务逻辑。
• 服务层（services）：编排检索和生成的完整流程，是 RAG 链路的核心。
• 组件层（core）：封装向量库客户端、嵌入模型、大模型调用、文档解析这些可复用能力。
• 配置层（config）：集中管理环境变量、模型参数、路径配置。

这样拆的好处是，检索策略要换（比如从纯向量检索换成混合检索），我只动 services 层；换嵌入模型，我只动 core 层；接口协议要加字段，我只动 api 层。各层之间通过明确的函数签名交互，测试也好写。

2.2 我实际使用的目录结构

下面是我最终落地的目录结构，这个结构在 FastAPI 项目里算是比较通用的，你可以直接抄：

rag-service/ ├── app/ │ ├── __init__.py │ ├── main.py # FastAPI 入口，注册路由和中间件 │ ├── config.py # 配置读取，基于 pydantic-settings │ ├── api/ │ │ ├── __init__.py │ │ ├── routes_chat.py # 问答接口 │ │ ├── routes_doc.py # 文档上传与管理接口 │ │ └── deps.py # 依赖注入，如鉴权、获取服务实例 │ ├── services/ │ │ ├── __init__.py │ │ ├── rag_service.py # RAG 主流程编排 │ │ └── doc_service.py # 文档入库流程 │ ├── core/ │ │ ├── __init__.py │ │ ├── vector_store.py # 向量库封装 │ │ ├── embedder.py # 嵌入模型封装 │ │ ├── llm.py # 大模型调用封装 │ │ └── parser.py # 文档解析 │ ├── schemas/ │ │ ├── __init__.py │ │ ├── chat.py # 请求/响应模型 │ │ └── doc.py │ └── utils/ │ ├── __init__.py │ └── logger.py # loguru 配置 ├── data/ # 向量库持久化目录 ├── logs/ # 日志落盘目录 ├── tests/ ├── Dockerfile ├── docker-compose.yml ├── requirements.txt └── .env.example

这个结构里有两个点值得单独说。第一，schemas单独抽出来，是因为 FastAPI 的 Pydantic 模型既是校验工具又是文档来源，集中管理比散落在路由里清晰得多。第二，deps.py做依赖注入，把“获取 RAG 服务实例”这件事从路由里解耦，方便测试时替换成 mock。

2.3 为什么不用 Flask

热词里有人问 Flask 和 FastAPI 的比较，这里顺带说下我的选择逻辑。Flask 更成熟、生态更老，但它是同步框架，做流式输出和并发请求时要么上 gevent 要么自己折腾线程池。RAG 场景里大模型生成动辄几秒到几十秒，同步阻塞会迅速吃满 worker。FastAPI 基于 Starlette，原生 async，配合StreamingResponse做流式输出非常自然，而且 Pydantic 校验和自动文档对接口协作太友好了。所以这个项目我没有任何犹豫就选了 FastAPI。

3. 核心模块的实操实现

3.1 配置管理：别再把密钥写死在代码里

改造第一步是把所有硬编码的配置抽出来。我用的是pydantic-settings，它能从环境变量和.env文件读取配置，还带类型校验。

# app/config.py from pydantic_settings import BaseSettings class Settings(BaseSettings): app_name: str = "rag-service" vector_store_path: str = "./data/vector" embed_model: str = "bge-small-zh" llm_base_url: str = "http://localhost:11434" llm_model: str = "qwen2:7b" top_k: int = 5 chunk_size: int = 500 chunk_overlap: int = 50 log_level: str = "INFO" class Config: env_file = ".env" settings = Settings()

这里llm_base_url指向本地模型服务，比如用 Ollama 跑一个 qwen2，FastAPI 通过 HTTP 调用它。这样模型和业务解耦，换模型只改配置。top_k、chunk_size这些检索参数也放进配置，方便调优时不用改代码。

注意：.env文件一定要加进.gitignore，我见过太多人把密钥和内部地址提交到仓库里，后面清理起来非常麻烦。

3.2 文档解析与切分：RAG 效果的地基

RAG 效果好不好，七成看切分。我踩过的最大坑就是一开始按固定字符数硬切，结果把一句话从中间劈开，检索出来的片段语义残缺，模型答非所问。

我的做法是：先按段落和标题切，再对超长段落做二次切分，同时保留一定的重叠。chunk_overlap设成chunk_size的 10% 左右，是为了防止关键信息正好落在切分边界上被割裂。

# app/core/parser.py from langchain.text_splitter import RecursiveCharacterTextSplitter def split_text(text: str, chunk_size: int, chunk_overlap: int): splitter = RecursiveCharacterTextSplitter( chunk_size=chunk_size, chunk_overlap=chunk_overlap, separators=["\n\n", "\n", "。", "！", "？", "；", " ", ""], ) return splitter.split_text(text)

separators的顺序很关键，它代表切分的优先级：先按空行（段落）切，再按换行，再按中文句号、感叹号这些句子边界切，最后才退化成按空格和字符切。这样能最大程度保证每个 chunk 是一个语义完整的单元。

实操心得：中文文档一定要把中文标点加进 separators，默认的英文分隔符对中文几乎没用，切出来的片段经常是半句话。

3.3 向量库与嵌入：本地化方案的选择

向量库我选的是 Chroma，原因是它支持本地持久化、零额外部署、Python 原生，适合中小规模知识库。如果你的数据量到了百万级 chunk，再考虑 Milvus 或 Qdrant 这类专门的向量数据库。

# app/core/vector_store.py import chromadb from app.config import settings class VectorStore: def __init__(self): self.client = chromadb.PersistentClient(path=settings.vector_store_path) self.collection = self.client.get_or_create_collection( name="knowledge", metadata={"hnsw:space": "cosine"}, ) def add(self, ids, documents, embeddings, metadatas): self.collection.add( ids=ids, documents=documents, embeddings=embeddings, metadatas=metadatas, ) def query(self, query_embedding, top_k): return self.collection.query( query_embeddings=[query_embedding], n_results=top_k, )

hnsw:space设成cosine是因为文本嵌入通常用余弦相似度衡量，这个参数在建 collection 时就要定好，后面改不了，建库前想清楚。

嵌入模型我用的是本地部署的中文小模型，通过 HTTP 或本地加载调用。选小模型的原因是 RAG 检索对嵌入质量的要求没有生成那么高，小模型速度快、显存占用低，性价比更高。

3.4 RAG 主流程编排：检索与生成的串联

这是整个服务的核心，把“用户提问 → 向量化 → 检索 → 拼 prompt → 调模型 → 返回”串起来。

# app/services/rag_service.py from app.core.vector_store import VectorStore from app.core.embedder import embed from app.core.llm import chat from app.config import settings class RagService: def __init__(self): self.store = VectorStore() def retrieve(self, question: str): q_vec = embed(question) result = self.store.query(q_vec, settings.top_k) docs = result["documents"][0] metas = result["metadatas"][0] return list(zip(docs, metas)) def answer(self, question: str): hits = self.retrieve(question) context = "\n\n".join([f"[来源{i+1}] {d}" for i, (d, _) in enumerate(hits)]) prompt = ( "你是一个严谨的知识库助手，请仅根据下面的资料回答问题，" "资料中没有的信息不要编造。\n\n" f"资料：\n{context}\n\n问题：{question}" ) return chat(prompt), hits

这里有个关键设计：prompt 里明确要求“仅根据资料回答，没有就说不知道”。这是抑制大模型幻觉最有效的手段之一。我实测下来，加了这句约束后，模型编造答案的比例明显下降。

注意：top_k不是越大越好。设太大，context 里塞满无关片段，反而干扰模型；设太小，可能漏掉关键信息。中文知识库一般 3 到 5 比较合适，具体要拿真实问题测。

3.5 接口层：用 Pydantic 定义清晰的契约

接口层只做三件事：校验入参、调用服务、返回结果。

# app/schemas/chat.py from pydantic import BaseModel, Field class ChatRequest(BaseModel): question: str = Field(..., min_length=1, max_length=2000) top_k: int | None = Field(default=None, ge=1, le=20) class SourceItem(BaseModel): content: str source: str | None = None class ChatResponse(BaseModel): answer: str sources: list[SourceItem]

# app/api/routes_chat.py from fastapi import APIRouter, Depends from app.schemas.chat import ChatRequest, ChatResponse, SourceItem from app.services.rag_service import RagService from app.api.deps import get_rag_service router = APIRouter(prefix="/api/v1", tags=["chat"]) @router.post("/chat", response_model=ChatResponse) def chat_endpoint(req: ChatRequest, svc: RagService = Depends(get_rag_service)): answer, hits = svc.answer(req.question) sources = [SourceItem(content=d, source=m.get("source")) for d, m in hits] return ChatResponse(answer=answer, sources=sources)

response_model指定后，FastAPI 会自动过滤多余字段并生成文档。Depends注入服务实例，测试时可以轻松替换。

3.6 流式输出：让用户不用干等

大模型生成慢，如果等全部生成完再返回，用户要盯着转圈十几秒。FastAPI 的StreamingResponse可以边生成边推。

# app/api/routes_chat.py from fastapi.responses import StreamingResponse @router.post("/chat/stream") def chat_stream(req: ChatRequest, svc: RagService = Depends(get_rag_service)): def event_gen(): for token in svc.stream_answer(req.question): yield f"data: {token}\n\n" yield "data: [DONE]\n\n" return StreamingResponse(event_gen(), media_type="text/event-stream")

这里用的是 SSE（Server-Sent Events）格式，前端用EventSource就能接收。注意每个消息要以\n\n结尾，这是 SSE 协议的硬性要求，少一个换行前端就收不到。

4. Docker 打包与部署实战

4.1 为什么要用 Docker

改造前我部署的方式是“在服务器上装 Python、装依赖、跑脚本”，换台机器就得重来一遍，依赖版本还经常打架。Docker 把运行环境和代码一起打包成镜像，到哪都能跑，这是生产级部署的基本要求。

4.2 Dockerfile 怎么写才不臃肿

FROM python:3.11-slim WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY app ./app EXPOSE 8000 CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "2"]

几个要点：基础镜像用slim版本，体积小；先 COPY requirements 再安装依赖，这样改代码时不会让依赖层缓存失效，构建快很多；--workers 2开两个进程，充分利用多核。

实操心得：--workers不是越多越好。每个 worker 都会独立加载模型和向量库连接，开太多会吃爆内存。一般设成 CPU 核数或核数的一半。

4.3 docker-compose 编排多服务

RAG 服务通常还要配向量库、缓存、模型服务，用 compose 统一编排最省心。

version: "3.9" services: rag-api: build: . ports: - "8000:8000" volumes: - ./data:/app/data - ./logs:/app/logs environment: - LLM_BASE_URL=http://host.docker.internal:11434 depends_on: - redis redis: image: redis:7-alpine ports: - "6379:6379" volumes: - redis_data:/data volumes: redis_data:

host.docker.internal是容器访问宿主机服务的地址，模型服务跑在宿主机上时用它。volumes把数据和日志挂出来，容器删了数据还在。

4.4 Windows 上装 Docker 的常见坑

热词里很多人问 Windows 安装 Docker 的问题，我踩过几个典型的：

• 虚拟化没开：报错virtualization support not detected，要去 BIOS 里开 VT-x 或 AMD-V，Windows 功能里开 Hyper-V 和 WSL2。
• WSL2 没装：Docker Desktop 依赖 WSL2 后端，先wsl --install再装 Docker。
• 端口占用：8000 被占就换端口，docker compose里改映射即可。

注意：Windows 上跑 Docker 性能比 Linux 差一截，文件挂载尤其慢。生产环境还是建议 Linux 服务器。

5. 日志体系：用 loguru 补上可观测性

5.1 为什么放弃 print 和标准 logging

Streamlit 阶段我全靠 print 调试，日志和页面输出混在一起，出了问题根本没法追溯。标准 logging 配置繁琐，格式也不够灵活。loguru 开箱即用，一行配置就能同时输出到控制台和文件，还支持自动轮转。

5.2 loguru 配置与关键日志埋点

# app/utils/logger.py import sys from loguru import logger from app.config import settings logger.remove() logger.add( sys.stdout, level=settings.log_level, format="<green>{time:YYYY-MM-DD HH:mm:ss}</green> | <level>{level}</level> | {message}", ) logger.add( "logs/rag_{time:YYYY-MM-DD}.log", rotation="00:00", retention="7 days", level="INFO", encoding="utf-8", )

在 RAG 主流程里，我会记录每次请求的关键指标：

from loguru import logger import time def answer(self, question: str): start = time.time() hits = self.retrieve(question) retrieve_cost = time.time() - start logger.info(f"retrieve done | q_len={len(question)} | hits={len(hits)} | cost={retrieve_cost:.2f}s") ...

这样每次请求的检索耗时、命中数量都落盘，出问题能快速定位是检索慢还是生成慢。

5.3 uvicorn 日志丢失问题

热词里有人问uvicorn fastapi 日志丢失，我也遇到过。原因是 uvicorn 默认会接管 logging 配置，把 loguru 的 handler 覆盖掉。解决办法是在启动时禁用 uvicorn 的默认日志配置：

uvicorn app.main:app --log-config none

或者在代码里重新 add handler。我实测下来，用--log-config none最干净，之后所有日志都走 loguru。

6. 常见问题与排查速查

6.1 我踩过的典型坑

6.2 性能调优的几个方向

第一，嵌入和检索是瓶颈时，考虑给嵌入结果加缓存，相同问题直接命中。第二，向量库查询可以加元数据过滤，先缩小范围再算相似度。第三，生成阶段用流式输出，用户感知的等待时间大幅缩短。第四，如果并发高，把模型服务独立部署，API 层只做编排，方便各自扩容。

6.3 关于 RAG 瓶颈的判断

很多人问 RAG 的瓶颈在哪。我的经验是：检索质量是最大瓶颈，不是模型。模型再强，检索回来的资料是错的，答案必然错。所以调优优先级应该是：切分策略 > 嵌入模型 > top_k 和重排 > 生成模型。先把检索调准，再考虑换更大的模型。

7. 后续可以怎么扩展

这套改造完成后，我陆续加了几个东西。一是加了重排（rerank），检索先召回 20 条，再用重排模型精排取前 5，效果提升明显。二是加了多路检索，向量检索和关键词检索并行，结果融合，对专有名词多的场景特别有用。三是把文档管理做成了异步任务，上传大文件时先返回任务 ID，后台慢慢解析入库，不阻塞接口。

如果你也在做 RAG 项目，我的建议是：Demo 阶段用 Streamlit 快速验证没问题，但只要有人要接你的能力，就尽早把服务层用 FastAPI 独立出来。这个改造越早做，后面越省事。我拖到用户开始抱怨才动手，回头补日志、补鉴权、补并发，工作量比一开始就分层大得多。