FDE Training · Module 19 · Capstone
19
毕业项目实战 · 从零搭一个生产级 RAG 问答系统
模块 19 · 毕业项目实战
本章目标:把前面 18 个模块的知识串起来,从零做一个可以给客户演示的生产级 RAG 问答系统。不是 Hello World,是包含文档解析、向量检索、LLM 生成、API 服务、简单前端、测试和部署的完整项目。
毕业项目全链路:文档 → 解析 → 入库 → 检索 → 生成
19.1 项目目标:做一个"企业知识库问答"系统
你要交付的东西:
- 用户上传 PDF/Word 文档。
- 系统自动解析、分块、入库。
- 用户用自然语言提问,系统从文档里找答案。
- 回答带来源引用(哪份文档第几页)。
- 有一个简单的网页界面能演示。
技术栈:Python + FastAPI + ChromaDB + OpenAI API + 一个简单前端。
19.2 项目结构
kb-assistant/
├── app/
│ ├── main.py # FastAPI 入口
│ ├── parser.py # 文档解析
│ ├── chunker.py # 分块
│ ├── vectorstore.py # 向量库
│ ├── rag.py # 检索+生成
│ └── config.py # 配置
├── tests/
│ ├── test_chunker.py
│ └── test_rag.py
├── static/
│ └── index.html # 简单前端
├── requirements.txt
└── README.md
19.3 核心代码骨架
19.3.1 文档解析与分块
# parser.py
import pdfplumber
def parse_pdf(path: str) -> str:
text = []
with pdfplumber.open(path) as pdf:
for page in pdf.pages:
t = page.extract_text()
if t: text.append(t)
return "\n\n".join(text)
# chunker.py
def chunk_text(text: str, max_len: int = 500, overlap: int = 50) -> list[str]:
paragraphs = [p.strip() for p in text.split("\n\n") if p.strip()]
chunks, buf = [], ""
for p in paragraphs:
if len(buf) + len(p) <= max_len:
buf += p + "\n\n"
else:
if buf: chunks.append(buf.strip())
buf = p
if buf: chunks.append(buf.strip())
return chunks
19.3.2 向量库
# vectorstore.py
import chromadb
client = chromadb.PersistentClient(path="./chroma_db")
col = client.get_or_create_collection("kb", metadata={"hnsw:space": "cosine"})
def add_docs(chunks: list[str], doc_id: str, source: str):
ids = [f"{doc_id}-{i}" for i in range(len(chunks))]
col.upsert(ids=ids, documents=chunks,
metadatas=[{"doc_id": doc_id, "source": source} for _ in chunks])
def search(query: str, top_k: int = 3):
res = col.query(query_texts=[query], n_results=top_k)
return res["documents"][0], res["metadatas"][0]
19.3.3 RAG 问答
# rag.py
from openai import OpenAI
client = OpenAI()
def ask(question: str) -> dict:
# 1. 检索
chunks, metas = search(question, top_k=3)
# 2. 组装 Prompt
ctx = "\n\n".join(f"[片段{i+1}] {c}" for i, c in enumerate(chunks))
resp = client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{"role": "system", "content": "严格依据资料回答,没有就说不知道。"},
{"role": "user", "content": f"资料:\n{ctx}\n\n问题:{question}"},
],
temperature=0.1,
)
return {
"answer": resp.choices[0].message.content,
"sources": [m["source"] for m in metas],
}
19.3.4 API 入口
# main.py
from fastapi import FastAPI, UploadFile
from parser import parse_pdf
from chunker import chunk_text
from vectorstore import add_docs
from rag import ask
app = FastAPI()
@app.post("/upload")
def upload(file: UploadFile):
text = parse_pdf(file.file)
chunks = chunk_text(text)
add_docs(chunks, doc_id=file.filename, source=file.filename)
return {"chunks": len(chunks)}
@app.get("/ask")
def query(q: str):
return ask(q)
19.4 验收清单
毕业项目要做到以下几点才算合格:
- 能上传一个 PDF,解析出文本。
- 分块入库后,问相关问题能找到对应片段。
- 回答带来源引用,不是凭空编。
- 问资料里没有的问题,会说"不知道",不瞎编。
- 至少有 5 个单元测试跑通。
- README 写清楚怎么跑起来。
19.5 动手练习(可折叠答案)
练习 1:怎么验证你的 RAG 系统"不瞎编"?
答案思路:
- 从文档里出 5 个问题,看回答对不对、有没有带来源。
- 出 3 个文档里完全没有的问题,看模型会不会说"资料中未提及"。
- 如果瞎编了,优化 Prompt(加约束)、降低 temperature、检查检索结果是不是真的没相关内容。
练习 2:客户要给这个系统加权限控制(A 部门不能看 B 部门文档),你怎么改?
答案思路:
- 入库时给每个 chunk 打 metadata:department = "sales"。
- 检索时加 where 过滤:where={"department": user_dept}。
- API 层鉴权,确认用户身份和部门。
练习 3:这个项目怎么部署给客户用?
答案思路:
- 写 Dockerfile,把 FastAPI + Chroma 打成镜像。
- docker compose 一键起。
- OpenAI API 用环境变量配 key。
- 写部署 README:怎么装 Docker、怎么跑、怎么看日志。
19.6 毕业答辩准备
做完项目,你要能对着白板讲清楚:
- 架构图:从文档上传到回答生成的完整链路。
- 为什么这么设计:为什么用 Chroma 不用 Milvus?为什么 chunk 大小是 500?
- 踩过的坑:解析脏数据、幻觉、检索不准,你怎么解决的?
- 怎么扩展:用户量上来了怎么优化?要加微服务怎么拆?
19.7 课程总结
恭喜你走完 FDE 培养 19 个模块。回顾一下你学会了什么:
| 阶段 | 模块 | 能力 |
|---|---|---|
| 工程基础 | 01-02 | 全栈开发、云原生部署 |
| AI 核心 | 03-06 | LLM 调用、RAG、Agent、Evals |
| 生产保障 | 07-10 | 安全合规、可观测、排障 |
| 业务交付 | 11-13 | 软技能、需求翻译、面试 |
| 进阶深化 | 14-19 | 微服务、LLM 工程、性能成本、测试、项目管理、毕业项目 |
接下来就是去客户现场真刀真枪干了。FDE 的成长是在实战里磨出来的,不是在教室里。
本页由 FDE 培养课程文档生成,完整课程见 FDE 培养 · 课程总览。