在大模型时代，RAG（Retrieval-Augmented Generation） 已成为构建智能问答系统的标准范式。它巧妙融合了向量检索的精准性与大语言模型的生成能力，既能避免模型“幻觉”，又能动态利用私有知识库。

本文将手把手带你从零搭建一个端到端可运行的 RAG 服务，技术栈选用 FastAPI（高性能异步框架） + Milvus（开源向量数据库） + Llama3（Meta 最新开源大模型），并提供完整可部署的 Docker 方案。

无论你是想快速验证 RAG 架构，还是为生产环境打基础，本文都适合你——尤其适合具备 Go/Python 双栈能力、熟悉微服务与高并发的全栈后端工程师。

---

一、整体架构设计

我们采用模块化设计，将系统拆分为四个核心目录：

• /api：对外暴露的 FastAPI 接口
• /ingest：文档预处理与入库逻辑
• /retriever：基于 Milvus 的向量检索
• /generator：调用 Llama3 生成答案

整个流程如下：

用户提问 → FastAPI 接收 → 文本嵌入 → Milvus 检索 Top-K 相关片段 → 拼接 Prompt → Llama3 生成答案 → 返回结果

为了清晰展示数据流向，我们绘制了以下系统流程图：

---

二、环境准备与依赖安装

1. 基础依赖

• Python ≥ 3.10
• Docker + Docker Compose（用于部署 Milvus）
• Hugging Face Token（用于下载 Llama3）

2. Python 依赖（requirements.txt）

fastapi==0.110.0uvicorn[standard]==0.29.0pymilvus==2.4.3sentence-transformers==2.2.2transformers==4.38.0torch==2.2.0accelerate==0.27.2bitsandbytes==0.43.0  # 用于 4-bit 量化节省显存python-dotenv

提示：Llama3 虽为开源模型，但需在 Hugging Face 官网申请访问权限。建议使用 meta-llama/Meta-Llama-3-8B-Instruct 版本。

---

三、文档预处理与向量入库（/ingest）

1. 文档加载与分块

我们以 PDF/Markdown/TXT 为例，使用 unstructured 或 PyPDF2 加载文本，再通过 RecursiveCharacterTextSplitter 分块：

from langchain.text_splitter import RecursiveCharacterTextSplitterdef split_text(text: str, chunk_size=512, chunk_overlap=50):    splitter = RecursiveCharacterTextSplitter(        chunk_size=chunk_size,        chunk_overlap=chunk_overlap,        separators=["\n\n", "\n", "。", " ", ""]    )    return splitter.split_text(text)

2. 异步嵌入生成

使用 sentence-transformers 的 all-MiniLM-L6-v2 模型（轻量高效），通过异步方式批量处理嵌入，避免阻塞主线程：

from sentence_transformers import SentenceTransformerimport asynciomodel = SentenceTransformer('all-MiniLM-L6-v2')async def embed_texts(texts: list[str]) -> list[list[float]]:    loop = asyncio.get_event_loop()    embeddings = await loop.run_in_executor(None, model.encode, texts)    return embeddings.tolist()

关键优化：模型加载一次后全局复用，避免重复初始化。

3. 存入 Milvus

首先通过 docker-compose 启动 Milvus（见第五部分），然后创建集合：

from pymilvus import connections, Collection, FieldSchema, CollectionSchema, DataTypeconnections.connect("default", host="localhost", port="19530")fields = [    FieldSchema(name="id", dtype=DataType.INT64, is_primary=True, auto_id=True),    FieldSchema(name="text", dtype=DataType.VARCHAR, max_length=65535),    FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=384)]schema = CollectionSchema(fields, "RAG knowledge base")collection = Collection("rag_docs", schema)# 创建索引（IVF_FLAT 适合中小规模）index_params = {"index_type": "IVF_FLAT", "metric_type": "L2", "params": {"nlist": 128}}collection.create_index("embedding", index_params)collection.load()

入库时调用 embed_texts 获取向量，再批量插入：

embeddings = await embed_texts(chunks)entities = [    {"text": chunks},    {"embedding": embeddings}]collection.insert(entities)

---

四、FastAPI 接口与错误处理

1. 核心接口 /query

from fastapi import FastAPI, HTTPExceptionfrom pydantic import BaseModelapp = FastAPI()class QueryRequest(BaseModel):    question: str@app.post("/query")async def query(request: QueryRequest):    try:        # 1. 嵌入问题        query_emb = await embed_texts([request.question])        # 2. Milvus 检索        results = collection.search(            data=query_emb,            anns_field="embedding",            param={"metric_type": "L2", "params": {"nprobe": 10}},            limit=3,            output_fields=["text"]        )        if not results[0]:            raise HTTPException(status_code=404, detail="未检索到相关文档")        context = "\n".join([hit.entity.get("text") for hit in results[0]])        # 3. 调用 Llama3 生成        answer = await generate_answer(request.question, context)        return {"answer": answer}    except asyncio.TimeoutError:        raise HTTPException(status_code=408, detail="请求超时")    except Exception as e:        raise HTTPException(status_code=500, detail=f"内部错误: {str(e)}")

2. Llama3 生成答案（4-bit 量化节省显存）

from transformers import AutoTokenizer, AutoModelForCausalLM, pipelineimport torchtokenizer = AutoTokenizer.from_pretrained("meta-llama/Meta-Llama-3-8B-Instruct")model = AutoModelForCausalLM.from_pretrained(    "meta-llama/Meta-Llama-3-8B-Instruct",    device_map="auto",    load_in_4bit=True  # 关键：4-bit 量化)pipe = pipeline("text-generation", model=model, tokenizer=tokenizer, max_new_tokens=512)async def generate_answer(question: str, context: str) -> str:    prompt = f"""你是一个专业助手，请根据以下上下文回答问题。上下文：{context}问题：{question}答案："""    loop = asyncio.get_event_loop()    response = await loop.run_in_executor(None, lambda: pipe(prompt))    return response[0]['generated_text'].split("答案：")[-1].strip()

错误处理覆盖三大场景：

• 超时（asyncio.TimeoutError）
• 空检索（返回 404）
• LLM 异常（如 OOM、网络中断，返回 500）

---

五、Docker 化部署

1. docker-compose.yml（含 Milvus）

version: '3.8'services:  etcd:    image: quay.io/coreos/etcd:latest    environment:      - ETCD_AUTO_COMPACTION_MODE=revision      - ETCD_AUTO_COMPACTION_RETENTION=1000      - ETCD_QUOTA_BACKEND_BYTES=4294967296    volumes:      - ./volumes/etcd:/etcd    command: etcd -advertise-client-urls=http://0.0.0.0:2379 -listen-client-urls http://0.0.0.0:2379 --data-dir /etcd  minio:    image: minio/minio:latest    environment:      MINIO_ROOT_USER: minioadmin      MINIO_ROOT_PASSWORD: minioadmin    volumes:      - ./volumes/minio:/data    command: minio server /data --console-address ":9001"  milvus-standalone:    image: milvusdb/milvus:v2.4.3    environment:      ETCD_ENDPOINTS: etcd:2379      MINIO_ADDRESS: minio:9000    ports:      - "19530:19530"    depends_on:      - etcd      - minio

2. 应用 Dockerfile

FROM python:3.10-slimWORKDIR /appCOPY requirements.txt .RUN pip install --no-cache-dir -r requirements.txtCOPY . .CMD ["uvicorn", "api.main:app", "--host", "0.0.0.0", "--port", "8000"]

---

六、完整项目与总结

关键技术亮点回顾：

• ✅异步嵌入：避免 I/O 阻塞，提升吞吐
• ✅连接池复用：Milvus 连接全局初始化，避免频繁创建
• ✅4-bit 量化：在消费级 GPU 上运行 Llama3
• ✅完备错误处理：覆盖超时、空检索、LLM 失败三大场景
• ✅一键部署：Docker Compose 一键启动 Milvus + 应用

RAG 不是魔法，而是工程。只有将检索、生成、部署、可观测性打通，才能真正落地。希望本文能成为你构建企业级 RAG 系统的坚实起点。

延伸建议：生产环境中可加入 Redis 缓存高频问题、Prometheus 监控 QPS/延迟、以及基于用户反馈的自动负样本挖掘机制。

想入门 AI 大模型却找不到清晰方向？备考大厂 AI 岗还在四处搜集零散资料？别再浪费时间啦！2025 年 AI 大模型全套学习资料已整理完毕，从学习路线到面试真题，从工具教程到行业报告，一站式覆盖你的所有需求，现在全部免费分享！

👇👇扫码免费领取全部内容👇👇

​

一、学习必备：100+本大模型电子书+26 份行业报告 + 600+ 套技术PPT，帮你看透 AI 趋势

想了解大模型的行业动态、商业落地案例？大模型电子书？这份资料帮你站在 “行业高度” 学 AI：

1. 100+本大模型方向电子书

2. 26 份行业研究报告：覆盖多领域实践与趋势

报告包含阿里、DeepSeek 等权威机构发布的核心内容，涵盖：

• 职业趋势：《AI + 职业趋势报告》《中国 AI 人才粮仓模型解析》；
• 商业落地：《生成式 AI 商业落地白皮书》《AI Agent 应用落地技术白皮书》；
• 领域细分：《AGI 在金融领域的应用报告》《AI GC 实践案例集》；
• 行业监测：《2024 年中国大模型季度监测报告》《2025 年中国技术市场发展趋势》。

3. 600+套技术大会 PPT：听行业大咖讲实战

PPT 整理自 2024-2025 年热门技术大会，包含百度、腾讯、字节等企业的一线实践：

• 安全方向：《端侧大模型的安全建设》《大模型驱动安全升级（腾讯代码安全实践）》；
• 产品与创新：《大模型产品如何创新与创收》《AI 时代的新范式：构建 AI 产品》；
• 多模态与 Agent：《Step-Video 开源模型（视频生成进展）》《Agentic RAG 的现在与未来》；
• 工程落地：《从原型到生产：AgentOps 加速字节 AI 应用落地》《智能代码助手 CodeFuse 的架构设计》。

二、求职必看：大厂 AI 岗面试 “弹药库”，300 + 真题 + 107 道面经直接抱走

想冲字节、腾讯、阿里、蔚来等大厂 AI 岗？这份面试资料帮你提前 “押题”，拒绝临场慌！

1. 107 道大厂面经：覆盖 Prompt、RAG、大模型应用工程师等热门岗位

面经整理自 2021-2025 年真实面试场景，包含 TPlink、字节、腾讯、蔚来、虾皮、中兴、科大讯飞、京东等企业的高频考题，每道题都附带思路解析：

2. 102 道 AI 大模型真题：直击大模型核心考点

针对大模型专属考题，从概念到实践全面覆盖，帮你理清底层逻辑：

3. 97 道 LLMs 真题：聚焦大型语言模型高频问题

专门拆解 LLMs 的核心痛点与解决方案，比如让很多人头疼的 “复读机问题”：

三、路线必明： AI 大模型学习路线图，1 张图理清核心内容

刚接触 AI 大模型，不知道该从哪学起？这份「AI大模型 学习路线图」直接帮你划重点，不用再盲目摸索！

路线图涵盖 5 大核心板块，从基础到进阶层层递进：一步步带你从入门到进阶，从理论到实战。

L1阶段:启航篇丨极速破界AI新时代

L1阶段：了解大模型的基础知识，以及大模型在各个行业的应用和分析，学习理解大模型的核心原理、关键技术以及大模型应用场景。

L2阶段：攻坚篇丨RAG开发实战工坊

L2阶段：AI大模型RAG应用开发工程，主要学习RAG检索增强生成：包括Naive RAG、Advanced-RAG以及RAG性能评估，还有GraphRAG在内的多个RAG热门项目的分析。

L3阶段：跃迁篇丨Agent智能体架构设计

L3阶段：大模型Agent应用架构进阶实现，主要学习LangChain、 LIamaIndex框架，也会学习到AutoGPT、 MetaGPT等多Agent系统，打造Agent智能体。

L4阶段：精进篇丨模型微调与私有化部署

L4阶段：大模型的微调和私有化部署，更加深入的探讨Transformer架构，学习大模型的微调技术，利用DeepSpeed、Lamam Factory等工具快速进行模型微调，并通过Ollama、vLLM等推理部署框架，实现模型的快速部署。

L5阶段：专题集丨特训篇 【录播课】

四、资料领取：全套内容免费抱走，学 AI 不用再找第二份

不管你是 0 基础想入门 AI 大模型，还是有基础想冲刺大厂、了解行业趋势，这份资料都能满足你！
现在只需按照提示操作，就能免费领取：

👇👇扫码免费领取全部内容👇👇

​

2025 年想抓住 AI 大模型的风口？别犹豫，这份免费资料就是你的 “起跑线”！