15 · 完整后端项目：把 14 篇的零件拼成生产系统

本片目标：本系列最终交付——把前 13 篇学的全部整合成一个知识问答 RAG 机器人：配置化、索引管线、会话持久化、引用溯源、错误处理、健康检查、流式。 新增规定性：15（工程化约定：配置层 / 组件分层 / 索引管线 / 错误边界） 数据字典：项目目录结构、配置模型、各组件职责。 进程线程模型：异步路由 + 单例组件 + 线程安全注意。 网络模型：REST + SSE；外网模型调用 + 本地检索。

1. 全集回顾

每一篇解决一个问题，但现在代码是”散装”的：

已学	生产还缺
第 06 篇会话记忆（内存版）	重启丢失 → 要持久化
第 08 篇向量库（手工建）	文档更新要重入库 → 要索引管线
第 09 篇 RAG 链	模型名/k 值写死在代码 → 要配置化
第 10 篇重排	可选环节要能开关 → 要配置开关
第 13 篇全地图	知道了模块但没用到工程上 → 本篇全部落地
第 14 篇 FastAPI	返回没有来源 → 要引用溯源
全程	出错就崩 → 要错误处理

本篇把它们全部收进一个分层项目。

2. 项目结构（数据字典：目录即架构）

enterprise-rag-course/project/
├── config.py          # 配置层：所有可调参数集中在这（读 .env）
├── components.py      # 组件层：模型/embedding/向量库/重排 单例工厂
├── pipeline.py        # 管线层：RAG 链 + 对话式改写 + 重排 + 溯源
├── storage.py         # 存储层：会话记忆持久化（文件版，接口兼容内存版）
├── indexer.py         # 索引管线：文档 → 切分 → 入库（可重复执行）
├── api.py             # 接口层：FastAPI 路由（问答/流式/健康）
└── requirements.txt   # 依赖清单

分层原则（规定性 14）：每层只依赖下层，不跨层。改配置不动代码、换存储不动管线、换模型不动接口。

3. 各层数据字典

3.1 配置层 `config.py`

# 所有可调参数集中一处 —— 换模型/调 k 值只改这里
class Settings(BaseModel):
    # 模型
    model_name: str = "glm-4.7"
    base_url: str = "https://open.bigmodel.cn/api/anthropic"
    api_key: str = Field(default="", repr=False)
    max_tokens: int = 512
    temperature: float = 0.3
    # 检索
    retriever_k: int = 8          # 粗排召回数（第 10 篇）
    final_k: int = 3              # 精排后取前几（第 10 篇）
    use_rerank: bool = True       # 重排开关
    use_history_rewrite: bool = True  # 对话改写开关（第 10 篇）
    # 路径
    data_dir: str = "../data"
    chroma_db: str = "../data/chroma_db"
    rerank_model: str = "../models/bge-reranker-base"
    # 服务
    host: str = "0.0.0.0"
    port: int = 8000

3.2 组件层 `components.py`（单例工厂）

_model: ChatAnthropic | None = None
def get_model() -> ChatAnthropic:
    global _model
    if _model is None:
        _model = ChatAnthropic(model=settings.model_name, ...)
    return _model
# get_embeddings() / get_retriever() / get_reranker() 同理

为什么单例：ChatAnthropic 内部持有 httpx 客户端（连接池）、重排模型占 1.1GB 显存/内存。每个请求都新建 = 灾难。模块级缓存，进程内一份。

3.3 管线层 `pipeline.py`（核心逻辑）

def build_rag_answer(question: str, history: list[BaseMessage]) -> AnswerResult:
    # ① 可选：对话改写（第 10 篇）
    if settings.use_history_rewrite and history:
        question = rewrite_query(question, history)
    # ② 粗排（第 08 篇）
    docs = retriever.invoke(question)
    # ③ 可选：精排（第 10 篇）
    if settings.use_rerank:
        docs = rerank(question, docs)[:settings.final_k]
    # ④ 生成（第 09 篇）
    context = format_docs(docs)
    answer = (prompt | model | StrOutputParser()).invoke(
        {"context": context, "question": question})
    # ⑤ 溯源：从 metadata 提取来源
    sources = [d.metadata.get("source", "未知") for d in docs]
    return AnswerResult(answer=answer, sources=sources)

溯源的关键：第 07 篇学的 Document.metadata 在这里兑现——metadata["source"] 告诉你答案来自哪份文档。

3.4 存储层 `storage.py`（会话持久化）

# 内存版接口：BaseChatMessageHistory（第 06 篇学的接口）
# 文件版实现同样接口 —— 换实现不用改管线！

class FileChatMessageHistory(BaseChatMessageHistory):
    def __init__(self, file_path: str):
        self._file = Path(file_path)
        self._messages = self._load()     # JSON 读盘
    def add_message(self, message): ...   # append + 写盘
    @property
    def messages(self): ...               # 读内存列表

sessions_dir.mkdir(exist_ok=True)
def get_history(session_id: str) -> BaseChatMessageHistory:
    return FileChatMessageHistory(sessions_dir / f"{session_id}.json")

接口即契约：InMemoryChatMessageHistory → FileChatMessageHistory → 生产换 RedisChatMessageHistory，管线层一行不改——因为它们都实现 BaseChatMessageHistory。这是第 06 篇”规定性 6”的工程回报。

3.5 索引管线 `indexer.py`（文档更新）

def index_documents(files: list[Path]) -> None:
    # ① 读取 + 切分（第 07 篇）
    # ② 删旧 collection + 重建（第 08 篇的 delete_collection）
    # ③ add_texts 入库（第 08 篇）
    # 幂等：可重复执行，每次重建

为什么单独成层：建索引和问答是两个独立生命周期——文档更新时跑 indexer，平时只跑 API。生产用 cron/CI 定时跑，或文档变更钩子触发。

3.6 接口层 `api.py`（第 14 篇的增强版）

@app.post("/chat")          # 问答 + 溯源
@app.post("/chat/stream")   # 流式 SSE
@app.get("/health")         # 健康检查（容器探活用）
@app.post("/index")         # 触发索引重建（管理员用）

4. 进程线程模型（最终全景）

uvicorn 进程（异步）
  │
  ├─ 单例组件（模块级缓存，进程内一份）：
  │    ChatAnthropic（httpx 连接池）
  │    Chroma（本地 sqlite + HNSW）
  │    CrossEncoder（1.1GB 模型常驻内存）
  │
  ├─ POST /chat → async 路由 → await run_in_executor(管线) → 线程池执行同步链
  │    ├─ 检索（本地）
  │    ├─ 重排（本地 GPU/CPU）
  │    └─ 模型调用（httpx AsyncClient 非阻塞）
  │
  ├─ POST /chat/stream → async 生成器 → astream 逐块 → SSE
  │
  └─ 多 worker 注意：--workers N 时每进程一份单例 + 一份内存
       会话文件存储跨进程安全（各写各的 session 文件）

5. 网络模型（最终全景）

前端
 ├─ POST /chat → JSON
 │    ├─ 服务内部：Chroma 本地（无网络）
 │    ├─ CrossEncoder 本地（无网络）
 │    └─ 智谱 API（HTTPS POST，外网）
 │    └─ 返回 JSON（answer + sources）
 ├─ POST /chat/stream → SSE
 └─ GET /health → JSON

两条网络路径：外部（浏览器↔你的服务，HTTP）+ 内部（你的服务↔智谱 API，HTTPS）。你的服务是中间层，把本地知识检索和云端模型组装成对外服务。

6. 错误处理（生产级的”边界”）

错误	处理
API key 缺失	启动时校验配置，直接报错退出
向量库不存在	`/health` 返回 503 + 提示先跑 indexer
模型调用超时	ChatAnthropic 自带重试（max_retries）；管线层 try/except 返回友好错误
API 限流 429	SDK 自动重试；重试仍失败 → 返回 503
会话文件损坏	存储层 try/except，损坏则重建空历史

7. 验证：跑起来

① 建索引：

cd enterprise-rag-course\project
python indexer.py

② 启动 API：

uvicorn api:app --port 8000

③ 测试：

# 问答 + 溯源
curl -X POST http://localhost:8000/chat -H "Content-Type: application/json" `
  -d '{"question":"转正需要什么条件？","session_id":"u1"}'

# 流式
curl -N -X POST http://localhost:8000/chat/stream -H "Content-Type: application/json" `
  -d '{"question":"日总结什么时候发？","session_id":"u1"}'

# 健康
curl http://localhost:8000/health

预期输出：

{
  "answer": "根据手册，转正需试用期满、累计至少 60 篇日总结、答辩平均分 75 分以上。",
  "sources": ["新人培训手册.md"],
  "rewritten_question": null
}

8. 边界与下一步

本系列到此结束，但生产级不止于此。真实的更大规模系统还会加：

方向	怎么做
认证鉴权	FastAPI 中间件 + JWT/OAuth
日志与监控	LangSmith 追踪（RunnableConfig 已支持）+ 结构化日志
超大知识库	Chroma 换 Milvus/ES（接口约定相近）
会话存 Redis	换 `RedisChatMessageHistory`（接口不变）
多模型容灾	配置层加模型池，失败自动切换
检索再提升	混合检索（BM25+向量）、多路召回、HyDE

但地基已经打牢：你理解了每个环节的数据形态（数据字典）、调度方式（进程线程）、通信方式（网络模型），以及为什么每一步必然这样展开（规定性链条）。加任何新东西，都是在熟悉的地基上扩展。

9. 全系列回顾：规定性展开总图

规定性 1  模型调用（invoke）            —— 第 01 篇
规定性 2  消息类（9 字段信封）          —— 第 02 篇
规定性 3  模板（结构/数据分离）        —— 第 03 篇
规定性 4  输出解析 + 流式（SSE）       —— 第 04 篇
规定性 5  结构化输出（Pydantic 契约）  —— 第 05 篇
规定性 6  记忆（历史存储接口）        —— 第 06 篇
规定性 7  文档（Document + 切分）     —— 第 07 篇
规定性 8  向量化（Embedding + Chroma） —— 第 08 篇
规定性 9  组合（LCEL 管道）           —— 第 09 篇
规定性 10 检索质量（重排 + 改写）     —— 第 10 篇
规定性 11 工具调用（bind_tools 协议）  —— 第 11 篇
规定性 12 执行模型（线程/事件循环）   —— 第 12 篇
规定性 13 地基全貌（langchain_core 地图）—— 第 13 篇（盘点，不新增规定性）
规定性 14 服务化（FastAPI + SSE）     —— 第 14 篇
规定性 15 工程化（配置/分层/管线）    —— 第 15 篇（本篇）

每一层都是上一层逼出来的：模型有状态差异→统一消息；消息手拼脆弱→模板；文本输出难用→解析器；业务要 JSON→结构化；模型金鱼脑→记忆；记忆救不了公司知识→文档；文档太大→切分；切分后要召回→向量；召回要串链路→LCEL；链路不准→重排；并发不够→异步；地基摸透→服务化；要对外→API；要生产→工程化。 这就是”逻辑必然性”——不是 LangChain 发明了这些概念，是你的需求一步步逼出来的。 恭喜你走完全程。

RAG 教程

小林学 LangChain

15 · 完整后端项目：把 14 篇的零件拼成生产系统

15 · 完整后端项目：把 14 篇的零件拼成生产系统

1. 全集回顾

2. 项目结构（数据字典：目录即架构）

3. 各层数据字典

3.1 配置层 `config.py`

3.2 组件层 `components.py`（单例工厂）

3.3 管线层 `pipeline.py`（核心逻辑）

3.4 存储层 `storage.py`（会话持久化）

3.5 索引管线 `indexer.py`（文档更新）

3.6 接口层 `api.py`（第 14 篇的增强版）

4. 进程线程模型（最终全景）

5. 网络模型（最终全景）

6. 错误处理（生产级的”边界”）

7. 验证：跑起来

8. 边界与下一步

推荐资料（延伸阅读）

9. 全系列回顾：规定性展开总图

​15 · 完整后端项目：把 14 篇的零件拼成生产系统

​1. 全集回顾

​2. 项目结构（数据字典：目录即架构）

​3. 各层数据字典

​3.1 配置层 config.py

​3.2 组件层 components.py（单例工厂）

​3.3 管线层 pipeline.py（核心逻辑）

​3.4 存储层 storage.py（会话持久化）

​3.5 索引管线 indexer.py（文档更新）

​3.6 接口层 api.py（第 14 篇的增强版）

​4. 进程线程模型（最终全景）

​5. 网络模型（最终全景）

​6. 错误处理（生产级的”边界”）

​7. 验证：跑起来

​8. 边界与下一步

​推荐资料（延伸阅读）

​9. 全系列回顾：规定性展开总图

15 · 完整后端项目：把 14 篇的零件拼成生产系统

1. 全集回顾

2. 项目结构（数据字典：目录即架构）

3. 各层数据字典

3.1 配置层 `config.py`

3.2 组件层 `components.py`（单例工厂）

3.3 管线层 `pipeline.py`（核心逻辑）

3.4 存储层 `storage.py`（会话持久化）

3.5 索引管线 `indexer.py`（文档更新）

3.6 接口层 `api.py`（第 14 篇的增强版）

4. 进程线程模型（最终全景）

5. 网络模型（最终全景）

6. 错误处理（生产级的”边界”）

7. 验证：跑起来

8. 边界与下一步

推荐资料（延伸阅读）

9. 全系列回顾：规定性展开总图