wsk/zhuangang

0 Fork 0

Files

T

Shikui Wang 0bae9916a9 内容结构化改造（阶段 1）：新增 topics/scenarios 目录、内容规范、frontmatter schema、校验与目录生成脚本、31 个技术栈首页骨架，周文件加 frontmatter

2026-10-23 15:05:36 +08:00

48 KiB

Raw Blame History

id, title, type, domain, stack, week, est_hours, order

title

type

domain

stack

week

est_hours

order

plan.week06

第 6 周：RAG 知识库问答（下）——评测、混合检索、重排、记忆、缓存、LangChain、部署

week

plan

postgresql

redis

docker

rag

vector-db

langchain

第 6 周：RAG 知识库问答（下）——评测、混合检索、重排、记忆、缓存、LangChain、部署

项目二 rag-qa 的第 2 周。上周 week05-rag-1.md 完成了手写主流程；下周 week07-frontend.md 做前端界面。

本周目标：从“能跑”到“能用数字证明好不好”。先建评测集，再逐个加混合检索、重排、多轮改写、缓存，每加一项都跑一次评测并记录结果；最后用 LangChain 重写一遍做对照，了解 Milvus / pgvector，用 Docker Compose 部署。

本周新增文件

rag-qa/
├── src/rag_qa/
│   ├── config.py               # 覆盖：新增检索模式、Redis 配置
│   ├── container.py            # 覆盖：按配置组装混合检索、缓存、会话
│   ├── eval_metrics.py         # Hit@K、MRR、拒答准确率
│   ├── retrieval/bm25.py       # 关键词检索（jieba 分词）
│   ├── retrieval/hybrid.py     # RRF 融合 + 混合检索问答服务
│   ├── retrieval/rerank.py     # 重排（接口实现 + 假实现）
│   ├── qa/rewrite.py           # 多轮追问改写
│   ├── memory/__init__.py
│   ├── memory/session.py       # 会话短期记忆（内存 / Redis）
│   ├── cache.py                # 向量缓存、答案缓存
│   ├── api/chat.py             # 多轮对话接口
│   └── lc_pipeline.py          # LangChain LCEL 版对照
├── eval/
│   ├── questions.jsonl
│   ├── run_eval.py
│   └── results.md
├── Dockerfile
├── docker-compose.yml
└── tests/
    ├── test_eval.py
    ├── test_hybrid.py
    ├── test_rerank.py
    ├── test_memory.py
    ├── test_cache.py
    └── test_lc_pipeline.py

uv add rank-bm25 jieba "redis>=5" langchain-core

第 1 天：评测集和指标

学习内容

说明：
- 没有评测，所有优化都是感觉。面试官最爱问“你怎么知道改完变好了？”——答案就是评测集 + 指标 + 对比记录。
- 检索指标（不需要调用大模型，便宜、稳定，先做这个）：
  - Hit@K：前 K 个结果里有没有至少一个相关片段。
  - MRR（平均倒数排名）：第一个相关片段排第 1 得 1 分，排第 2 得 0.5 分，没找到得 0 分，取平均。看的是“排得够不够靠前”。
  - 拒答准确率：本来就没有答案的问题，系统是否正确地一个片段都不返回。
- 怎么判断“相关”：每道题写一个必须出现在片段里的关键句（不用片段 ID，因为改切分参数后 ID 会变）。
- 生成质量（忠实度、是否答对）可以人工抽查，也可以用模型打分（LLM-as-judge），本周人工抽查即可，第 13 周 Agent 评测再系统做。
案例：

# 文件：rag-qa/src/rag_qa/eval_metrics.py
import json
from pathlib import Path

from pydantic import BaseModel

from rag_qa.vectorstore.base import ScoredChunk


class EvalItem(BaseModel):
    q: str
    must_contain: str = ""   # 相关片段必须包含的关键句；为空表示这题应该拒答


class EvalReport(BaseModel):
    hit_at_k: float
    mrr: float
    refuse_acc: float
    n_answerable: int
    n_refuse: int
    misses: list[str]


def load_items(path: Path) -> list[EvalItem]:
    return [EvalItem.model_validate_json(line) for line in path.read_text(encoding="utf-8").splitlines() if line.strip()]


def first_relevant_rank(item: EvalItem, hits: list[ScoredChunk]) -> int | None:
    for rank, h in enumerate(hits, start=1):
        if item.must_contain in h.chunk.text:
            return rank
    return None


def evaluate(items: list[EvalItem], results: list[list[ScoredChunk]]) -> EvalReport:
    hit = rr = refuse_ok = 0.0
    answerable = [(i, r) for i, r in zip(items, results) if i.must_contain]
    refuse = [(i, r) for i, r in zip(items, results) if not i.must_contain]
    misses = []
    for item, hits in answerable:
        rank = first_relevant_rank(item, hits)
        if rank is None:
            misses.append(item.q)
        else:
            hit += 1
            rr += 1 / rank
    for item, hits in refuse:
        if hits:
            misses.append(f"（应拒答）{item.q}")
        else:
            refuse_ok += 1
    na, nr = len(answerable), len(refuse)
    return EvalReport(hit_at_k=round(hit / na, 3) if na else 0, mrr=round(rr / na, 3) if na else 0,
                      refuse_acc=round(refuse_ok / nr, 3) if nr else 0, n_answerable=na, n_refuse=nr, misses=misses)


def to_markdown_row(name: str, r: EvalReport) -> str:
    return f"| {name} | {r.hit_at_k:.3f} | {r.mrr:.3f} | {r.refuse_acc:.3f} | {len(r.misses)} |"


if __name__ == "__main__":
    print(json.dumps({"q": "Redis 旁路缓存怎么更新", "must_contain": "先写数据库"}, ensure_ascii=False))

# 文件：rag-qa/tests/test_eval.py
from rag_qa.eval_metrics import EvalItem, evaluate
from rag_qa.ingest.chunker import Chunk
from rag_qa.vectorstore.base import ScoredChunk


def hit(text: str) -> ScoredChunk:
    return ScoredChunk(chunk=Chunk(chunk_id=text, doc_id="d", filename="f.md", text=text, index=0), score=0.5)


def test_metrics():
    items = [EvalItem(q="q1", must_contain="甲"), EvalItem(q="q2", must_contain="乙"),
             EvalItem(q="q3", must_contain="丙"), EvalItem(q="q4")]
    results = [[hit("甲是答案")], [hit("无关"), hit("乙在第二")], [hit("无关")], []]
    r = evaluate(items, results)
    assert r.hit_at_k == round(2 / 3, 3)
    assert r.mrr == round((1 + 0.5 + 0) / 3, 3)
    assert r.refuse_acc == 1.0 and r.misses == ["q3"]

评测集文件（每行一条 JSON，至少 30 条，其中 5~8 条是文档里没有答案的）：

{"q": "Redis 旁路缓存更新数据时的顺序是什么", "must_contain": "先写数据库"}

{"q": "goroutine 之间推荐用什么通信", "must_contain": "channel"}

{"q": "公司今年的年终奖怎么算"}

# 文件：rag-qa/eval/run_eval.py
"""用法：uv run python eval/run_eval.py   （先 rag-qa ingest samples/）"""
import asyncio
from pathlib import Path

from rag_qa.container import build_container
from rag_qa.eval_metrics import evaluate, load_items, to_markdown_row


async def main() -> None:  # 不自动运行
    items = load_items(Path(__file__).with_name("questions.jsonl"))
    print("| 方案 | Hit@K | MRR | 拒答准确率 | 失败数 |\n| --- | --- | --- | --- | --- |")
    for mode in ["vector", "hybrid", "hybrid_rerank"]:
        c = build_container(mode=mode)
        results = [await c.qa.retrieve(i.q) for i in items]
        report = evaluate(items, results)
        print(to_markdown_row(mode, report))
        for m in report.misses:
            print(f"<!-- {mode} 失败：{m} -->")


if __name__ == "__main__":
    asyncio.run(main())

易错点 / 面试常问（补充）

易错点：评测集要包含“本来就没有答案”的问题，否则看不出系统会不会硬答。
易错点：评测集写好后不要按评测结果去改题目；改了系统再跑同一套题才有可比性。评测结果要记下当时的参数（Top-K、阈值、检索方式）。
面试常问：Hit@K 和 MRR 分别衡量什么（前 K 个里有没有命中 / 命中的排名靠不靠前）；检索指标和回答质量指标的区别（后者可以用人工或模型评分）。

要建的文件

eval_metrics.py、tests/test_eval.py、eval/questions.jsonl（从上周 8 条人工试验扩充到 30 条以上）、eval/run_eval.py、eval/results.md

运行方式

uv run pytest tests/test_eval.py -v
uv run python eval/run_eval.py      # 第 2、3 天的代码写完后才能跑全部三种方案

完成标准

评测集 ≥ 30 条；eval/results.md 记下基线（纯向量检索）的三个指标。每次改动后追加一行，写清改了什么、指标变化、原因分析。
实验一：把 chunk_size 分别设为 300 / 500 / 800，重新入库并评测，记录哪个最好（数值取决于你的文档和模型，以你自己的结果为准）。

第 2 天：BM25 关键词检索 + RRF 混合检索

学习内容

说明：
- 向量检索擅长“意思相近”，但对专有名词、错误码、函数名、版本号这类精确匹配不敏感；BM25 是经典关键词检索算法（Elasticsearch 默认算法），正好互补。两者一起用叫混合检索。
- 中文要先分词（jieba），英文按单词。
- 两路结果的分数尺度不同（余弦 0~~1，BM25 可能是 0~~20），不能直接相加。**RRF（倒数排名融合）**只看排名：每个片段得分 = Σ 1/(k + 排名)，k 一般取 60。简单、稳定、不用调权重。
- 坑：rank-bm25 的 BM25Okapi 在文档很少时，出现在一半以上文档里的词 IDF 会接近 0 甚至为负，测试时语料要多准备几条。
案例：

# 文件：rag-qa/src/rag_qa/retrieval/bm25.py
import logging
import re

import jieba
from rank_bm25 import BM25Okapi

from rag_qa.ingest.chunker import Chunk
from rag_qa.vectorstore.base import ScoredChunk

jieba.setLogLevel(logging.WARNING)
STOP = set("的 了 是 在 和 与 及 或 也 就 都 而 吗 呢 啊 怎么 什么 如何 哪些 为什么 一个".split())


def tokenize(text: str) -> list[str]:
    words = jieba.lcut(text.lower())
    return [w for w in words if w.strip() and w not in STOP and not re.fullmatch(r"\W+", w)]


class BM25Index:
    def __init__(self, chunks: list[Chunk]):
        self.chunks = chunks
        self.bm25 = BM25Okapi([tokenize(c.heading_path + " " + c.text) for c in chunks]) if chunks else None

    def search(self, query: str, top_k: int) -> list[ScoredChunk]:
        if self.bm25 is None:
            return []
        scores = self.bm25.get_scores(tokenize(query))
        ranked = sorted(range(len(scores)), key=lambda i: scores[i], reverse=True)[:top_k]
        return [ScoredChunk(chunk=self.chunks[i], score=float(scores[i])) for i in ranked if scores[i] > 0]

# 文件：rag-qa/src/rag_qa/retrieval/hybrid.py
from rag_qa.llm.base import ChatModel, Embedder
from rag_qa.qa.answer import QAService
from rag_qa.retrieval.bm25 import BM25Index
from rag_qa.retrieval.rerank import Reranker
from rag_qa.vectorstore.base import ScoredChunk, VectorStore


def rrf(*ranked_lists: list[ScoredChunk], k: int = 60) -> list[ScoredChunk]:
    scores: dict[str, float] = {}
    by_id: dict[str, ScoredChunk] = {}
    for lst in ranked_lists:
        for rank, h in enumerate(lst, start=1):
            cid = h.chunk.chunk_id
            scores[cid] = scores.get(cid, 0) + 1 / (k + rank)
            by_id.setdefault(cid, h)
    order = sorted(scores, key=scores.get, reverse=True)
    return [ScoredChunk(chunk=by_id[cid].chunk, score=round(scores[cid], 6)) for cid in order]


class HybridQAService(QAService):
    """只覆盖 retrieve()，回答、流式、引用逻辑全部复用父类。"""

    def __init__(self, store: VectorStore, embedder: Embedder, chat: ChatModel, *,
                 top_k: int = 4, min_score: float = 0.2, max_context_chars: int = 6000,
                 candidates: int = 20, reranker: Reranker | None = None, rerank_min_score: float = 0.0):
        super().__init__(store, embedder, chat, top_k, min_score, max_context_chars)
        self.candidates, self.reranker, self.rerank_min_score = candidates, reranker, rerank_min_score
        self._bm25: BM25Index | None = None
        self._version: tuple = ()

    def bm25(self) -> BM25Index:
        version = tuple((d.doc_id, d.content_hash) for d in self.store.list_docs())
        if self._bm25 is None or version != self._version:  # 文档有变化才重建索引
            self._bm25, self._version = BM25Index(self.store.all_chunks()), version
        return self._bm25

    async def retrieve(self, question: str) -> list[ScoredChunk]:
        [qv] = await self.embedder.embed([question])
        vec = [h for h in self.store.query(qv, self.candidates) if h.score >= self.min_score]
        kw = self.bm25().search(question, self.candidates)
        fused = rrf(vec, kw)
        if self.reranker and fused:
            fused = [h for h in await self.reranker.rerank(question, fused) if h.score >= self.rerank_min_score]
        return fused[: self.top_k]

# 文件：rag-qa/tests/test_hybrid.py
from rag_qa.ingest.chunker import Chunk
from rag_qa.ingest.loaders import load_bytes
from rag_qa.ingest.pipeline import ingest_document
from rag_qa.llm.fake import FakeChat, FakeEmbedder
from rag_qa.retrieval.bm25 import BM25Index
from rag_qa.retrieval.hybrid import HybridQAService, rrf
from rag_qa.vectorstore.base import ScoredChunk
from rag_qa.vectorstore.chroma_store import ChromaStore

DOC = """# 运维手册
## 错误码
错误码 E1024 表示磁盘空间不足，需要清理日志目录。
## 缓存
缓存命中率下降时，先检查过期时间设置。
## 数据库
慢查询要先看执行计划，再考虑加索引。
## 部署
部署前先跑完测试，再构建镜像。
## 监控
监控面板要包含请求量、错误率、延迟。
"""


def sc(cid: str) -> ScoredChunk:
    return ScoredChunk(chunk=Chunk(chunk_id=cid, doc_id="d", filename="f", text=cid, index=0), score=1)


def test_rrf_prefers_items_in_both_lists():
    fused = rrf([sc("a"), sc("b"), sc("c")], [sc("c"), sc("d")])
    assert fused[0].chunk.chunk_id == "c"      # 两路都出现，排第一
    assert {h.chunk.chunk_id for h in fused} == {"a", "b", "c", "d"}


def test_bm25_exact_term():
    chunks = [Chunk(chunk_id=str(i), doc_id="d", filename="f", text=t, index=i)
              for i, t in enumerate(["错误码 E1024 表示磁盘不足", "缓存过期", "执行计划", "镜像构建", "监控面板"])]
    hits = BM25Index(chunks).search("E1024 是什么错误", top_k=3)
    assert hits[0].chunk.text.startswith("错误码 E1024")


async def test_hybrid_service(tmp_path):
    store, emb = ChromaStore(str(tmp_path)), FakeEmbedder()
    await ingest_document(load_bytes("ops.md", DOC.encode()), store, emb)
    svc = HybridQAService(store, emb, FakeChat("见 [1]"), top_k=2)
    hits = await svc.retrieve("E1024")
    assert "磁盘空间不足" in hits[0].chunk.text
    ans = await svc.answer("E1024")
    assert ans.citations[0].cited

要建的文件

retrieval/bm25.py、retrieval/hybrid.py、tests/test_hybrid.py（retrieval/rerank.py 明天写，今天可以先写一个只有 Reranker 协议的版本）

运行方式

uv run pytest tests/test_hybrid.py -v
uv run python eval/run_eval.py

完成标准

测试通过；评测结果追加 hybrid 一行。如果评测集里有专有名词、错误码类问题，通常能看到混合检索的提升——以你自己的数据为准，没有提升也要记录并分析原因。
面试常问：为什么要混合检索？RRF 的公式和优点？

第 3 天：重排（Rerank）

学习内容

说明：
- 向量检索用的是“双塔”模型：问题和片段分别变成向量再比较，快但粗。**重排模型（cross-encoder）**把“问题 + 片段”放在一起打分，准但慢。所以流程是：先粗召回 20~50 个候选，再用重排模型精排，取前 K 个。
- 常见重排模型：bge-reranker 系列等；很多模型服务提供 /rerank 接口，请求一般是 {model, query, documents, top_n}，返回每个文档的 index 和 relevance_score。不同服务商的字段可能不同，以服务商文档为准。
- 重排分数可以再设一个阈值，进一步减少“不相关片段混进上下文”。
案例：

# 文件：rag-qa/src/rag_qa/retrieval/rerank.py
from typing import Protocol

import httpx

from rag_qa.llm.base import LLMError
from rag_qa.llm.fake import tokens
from rag_qa.vectorstore.base import ScoredChunk


class Reranker(Protocol):
    async def rerank(self, query: str, hits: list[ScoredChunk]) -> list[ScoredChunk]: ...


class ApiReranker:
    """调用 /rerank 接口。请求和响应字段以你所用服务商的文档为准。"""

    def __init__(self, client: httpx.AsyncClient, model: str, top_n: int = 8):
        self.client, self.model, self.top_n = client, model, top_n

    async def rerank(self, query: str, hits: list[ScoredChunk]) -> list[ScoredChunk]:
        resp = await self.client.post("/rerank", json={
            "model": self.model, "query": query,
            "documents": [h.chunk.text for h in hits], "top_n": min(self.top_n, len(hits)),
        })
        if resp.status_code >= 400:
            raise LLMError("RERANK", f"重排服务返回 {resp.status_code}")
        results = sorted(resp.json()["results"], key=lambda r: r["relevance_score"], reverse=True)
        return [ScoredChunk(chunk=hits[r["index"]].chunk, score=r["relevance_score"]) for r in results]


class OverlapReranker:
    """假实现：按“问题词在片段里出现的比例”打分。只用于测试和没有重排服务时的演示。"""

    async def rerank(self, query: str, hits: list[ScoredChunk]) -> list[ScoredChunk]:
        q = set(tokens(query))
        scored = [ScoredChunk(chunk=h.chunk, score=len(q & set(tokens(h.chunk.text))) / (len(q) or 1)) for h in hits]
        return sorted(scored, key=lambda h: h.score, reverse=True)

# 文件：rag-qa/tests/test_rerank.py
import json

import httpx

from rag_qa.ingest.chunker import Chunk
from rag_qa.retrieval.rerank import ApiReranker, OverlapReranker
from rag_qa.vectorstore.base import ScoredChunk


def hits(*texts: str) -> list[ScoredChunk]:
    return [ScoredChunk(chunk=Chunk(chunk_id=t, doc_id="d", filename="f", text=t, index=i), score=0.5)
            for i, t in enumerate(texts)]


async def test_api_reranker_reorders():
    def handler(request: httpx.Request) -> httpx.Response:
        body = json.loads(request.content)
        assert body["documents"] == ["甲", "乙"] and body["top_n"] == 2
        return httpx.Response(200, json={"results": [{"index": 0, "relevance_score": 0.1},
                                                     {"index": 1, "relevance_score": 0.9}]})

    async with httpx.AsyncClient(transport=httpx.MockTransport(handler), base_url="https://r.test") as c:
        out = await ApiReranker(c, "reranker").rerank("问题", hits("甲", "乙"))
    assert [h.chunk.text for h in out] == ["乙", "甲"] and out[0].score == 0.9


async def test_overlap_reranker():
    out = await OverlapReranker().rerank("索引 执行计划", hits("部署镜像", "先看执行计划再加索引"))
    assert out[0].chunk.text.startswith("先看执行计划")

要建的文件

retrieval/rerank.py、tests/test_rerank.py

运行方式

uv run pytest tests/test_rerank.py -v
uv run python eval/run_eval.py

完成标准

评测结果追加 hybrid_rerank 一行；记下重排带来的延迟增加（在 rag-qa ask 里看 latency_ms），说明“效果和速度的取舍”。
面试常问：双塔和交叉编码器的区别？为什么不直接用重排模型检索全部文档？（太慢，每个片段都要和问题一起过一遍模型）

第 4 天：多轮追问——问题改写 + 会话短期记忆（Redis）

学习内容

说明：
- 用户第二轮常问“那它的缺点呢？”——直接拿去检索肯定找不到。做法：用模型把追问改写成独立问题（“Redis 旁路缓存的缺点是什么”），再检索。
- 会话历史存 Redis 列表：RPUSH 追加、LTRIM 只保留最近 N 轮、EXPIRE 设置过期。这是“短期记忆”；跨会话的“长期记忆”放到第 12 周 Agent 项目。
- 历史只用于改写和展示，回答时仍然只给模型当前检索到的资料，避免历史里的错误被反复放大。
案例：

# 文件：rag-qa/src/rag_qa/memory/__init__.py

# 文件：rag-qa/src/rag_qa/memory/session.py
import json
from typing import Protocol

from rag_qa.llm.base import Message


class SessionStore(Protocol):
    async def history(self, session_id: str) -> list[Message]: ...
    async def append(self, session_id: str, question: str, answer: str) -> None: ...


class MemorySessionStore:
    def __init__(self, max_turns: int = 6):
        self.max_turns = max_turns
        self.data: dict[str, list[Message]] = {}

    async def history(self, session_id: str) -> list[Message]:
        return list(self.data.get(session_id, []))

    async def append(self, session_id: str, question: str, answer: str) -> None:
        items = self.data.setdefault(session_id, [])
        items += [{"role": "user", "content": question}, {"role": "assistant", "content": answer}]
        del items[: -self.max_turns * 2]


class RedisSessionStore:
    def __init__(self, redis, max_turns: int = 6, ttl_s: int = 3600):
        self.r, self.max_turns, self.ttl_s = redis, max_turns, ttl_s

    def _key(self, session_id: str) -> str:
        return f"rag:session:{session_id}"

    async def history(self, session_id: str) -> list[Message]:
        return [json.loads(x) for x in await self.r.lrange(self._key(session_id), 0, -1)]

    async def append(self, session_id: str, question: str, answer: str) -> None:
        key = self._key(session_id)
        async with self.r.pipeline(transaction=True) as p:  # 一次往返完成四个命令
            p.rpush(key, json.dumps({"role": "user", "content": question}, ensure_ascii=False),
                    json.dumps({"role": "assistant", "content": answer}, ensure_ascii=False))
            p.ltrim(key, -self.max_turns * 2, -1)
            p.expire(key, self.ttl_s)
            await p.execute()

# 文件：rag-qa/src/rag_qa/qa/rewrite.py
from rag_qa.llm.base import ChatModel, Message
from rag_qa.memory.session import SessionStore
from rag_qa.qa.answer import Answer, QAService

REWRITE = """根据对话历史，把用户最后的追问改写成一个不依赖上下文、可以单独检索的完整问题。
只输出改写后的问题，不要回答，不要解释。"""


async def rewrite_question(chat: ChatModel, history: list[Message], question: str) -> str:
    if not history:
        return question  # 第一轮不需要改写，省一次模型调用
    convo = "\n".join(f"{m['role']}: {m['content'][:300]}" for m in history[-6:])
    out = (await chat.complete([
        {"role": "system", "content": REWRITE},
        {"role": "user", "content": f"对话历史：\n{convo}\n\n追问：{question}"},
    ])).strip()
    return out if 0 < len(out) <= 200 else question  # 改写结果异常就退回原问题


class ConversationalQA:
    def __init__(self, qa: QAService, sessions: SessionStore, rewriter: ChatModel):
        self.qa, self.sessions, self.rewriter = qa, sessions, rewriter

    async def ask(self, session_id: str, question: str) -> tuple[str, Answer]:
        history = await self.sessions.history(session_id)
        standalone = await rewrite_question(self.rewriter, history, question)
        ans = await self.qa.answer(standalone)
        await self.sessions.append(session_id, question, ans.text)
        return standalone, ans

# 文件：rag-qa/tests/test_memory.py
import os

import pytest

from rag_qa.ingest.loaders import load_bytes
from rag_qa.ingest.pipeline import ingest_document
from rag_qa.llm.fake import FakeChat, FakeEmbedder
from rag_qa.memory.session import MemorySessionStore, RedisSessionStore
from rag_qa.qa.answer import QAService
from rag_qa.qa.rewrite import ConversationalQA
from rag_qa.vectorstore.chroma_store import ChromaStore


class ScriptedChat(FakeChat):
    """改写请求返回固定问题，其余请求返回固定回答。"""

    async def complete(self, messages):
        self.received.append(messages)
        if "改写" in messages[0]["content"]:
            return "旁路缓存的缺点是什么"
        return "可能出现短暂不一致 [1]。"


async def test_follow_up_is_rewritten(tmp_path):
    store, emb, chat = ChromaStore(str(tmp_path)), FakeEmbedder(), ScriptedChat()
    doc = "# 缓存\n## 旁路缓存\n旁路缓存的缺点是可能出现短暂的数据不一致。\n## 其他\n无关内容。"
    await ingest_document(load_bytes("cache.md", doc.encode()), store, emb)
    conv = ConversationalQA(QAService(store, emb, chat), MemorySessionStore(), chat)

    q1, _ = await conv.ask("s1", "什么是旁路缓存")
    q2, ans = await conv.ask("s1", "那它的缺点呢")
    assert q1 == "什么是旁路缓存"            # 第一轮不改写
    assert q2 == "旁路缓存的缺点是什么" and ans.citations


async def test_memory_store_keeps_last_turns():
    s = MemorySessionStore(max_turns=2)
    for i in range(3):
        await s.append("s", f"q{i}", f"a{i}")
    assert [m["content"] for m in await s.history("s")] == ["q1", "a1", "q2", "a2"]


@pytest.mark.skipif(not os.getenv("REDIS_URL"), reason="需要 REDIS_URL，例如 redis://localhost:6379/0")
async def test_redis_session_store():
    import redis.asyncio as redis

    r = redis.from_url(os.environ["REDIS_URL"], decode_responses=True)
    s = RedisSessionStore(r, max_turns=2, ttl_s=60)
    await r.delete("rag:session:test")
    for i in range(3):
        await s.append("test", f"q{i}", f"a{i}")
    assert [m["content"] for m in await s.history("test")] == ["q1", "a1", "q2", "a2"]
    assert 0 < await r.ttl("rag:session:test") <= 60
    await r.aclose()

易错点 / 面试常问（补充）

易错点：多轮追问时直接拿“它呢？”去检索会什么都找不到，要先结合历史把问题改写成完整的问题，再检索。
易错点：会话历史要限制条数或长度，并设置过期时间（本项目用 Redis 过期），否则越聊越慢、越贵。
面试常问：短期记忆和长期记忆的区别（当前会话上下文 / 跨会话的用户偏好和事实）；问题改写失败会有什么后果，怎么兜底（改写结果为空时用原问题）。

要建的文件

memory/__init__.py、memory/session.py、qa/rewrite.py、tests/test_memory.py

运行方式

docker run -d --name rag-redis -p 6379:6379 redis:7     # 第 3 周装过可以复用
REDIS_URL=redis://localhost:6379/0 uv run pytest tests/test_memory.py -v

完成标准

三个测试都通过（Redis 测试没有跳过）；用真实模型连续追问 3 轮，看改写结果是否合理。

第 5 天：缓存、配置组装、多轮对话接口

学习内容

说明：
- 向量缓存：同一段文字的向量永远一样，按“模型名 + 文字哈希”缓存，重复入库、重复问题都省钱。键里必须带模型名，换模型就自动失效。
- 答案缓存：同一问题短时间内重复问，直接返回。难点是失效：文档一更新，旧答案可能就错了。做法是把“索引版本”（全部文档内容哈希的摘要）放进缓存键，文档一变键就变，旧缓存自然不再命中，过期后自动清理。
- 这里用“装饰器模式”：CachedEmbedder 实现同样的 Embedder 协议，包在真实实现外面，业务代码完全不用改。
- 第 3 周学过的缓存问题（穿透、击穿、雪崩）在这里同样存在，面试时可以结合讲。
案例：

# 文件：rag-qa/src/rag_qa/cache.py
import hashlib
import json
import time
from typing import Protocol

from rag_qa.llm.base import Embedder
from rag_qa.qa.answer import Answer, QAService


class KV(Protocol):
    async def get(self, key: str) -> str | None: ...
    async def set(self, key: str, value: str, ttl_s: int) -> None: ...


class MemoryKV:
    def __init__(self):
        self.data: dict[str, tuple[float, str]] = {}

    async def get(self, key: str) -> str | None:
        item = self.data.get(key)
        if item is None or item[0] < time.time():
            return None
        return item[1]

    async def set(self, key: str, value: str, ttl_s: int) -> None:
        self.data[key] = (time.time() + ttl_s, value)


class RedisKV:
    def __init__(self, redis):
        self.r = redis

    async def get(self, key: str) -> str | None:
        return await self.r.get(key)

    async def set(self, key: str, value: str, ttl_s: int) -> None:
        await self.r.set(key, value, ex=ttl_s)


def sha(text: str) -> str:
    return hashlib.sha1(text.encode()).hexdigest()


class CachedEmbedder:
    def __init__(self, inner: Embedder, kv: KV, model: str, ttl_s: int = 7 * 86400):
        self.inner, self.kv, self.model, self.ttl_s = inner, kv, model, ttl_s

    async def embed(self, texts: list[str]) -> list[list[float]]:
        keys = [f"rag:emb:{self.model}:{sha(t)}" for t in texts]
        cached = [await self.kv.get(k) for k in keys]          # 生产环境用 MGET / pipeline 一次取回
        missing = [i for i, v in enumerate(cached) if v is None]
        fresh = await self.inner.embed([texts[i] for i in missing]) if missing else []
        for i, vec in zip(missing, fresh):
            await self.kv.set(keys[i], json.dumps(vec), self.ttl_s)
            cached[i] = json.dumps(vec)
        return [json.loads(v) for v in cached]


class CachedQA:
    def __init__(self, qa: QAService, kv: KV, ttl_s: int = 600):
        self.qa, self.kv, self.ttl_s = qa, kv, ttl_s

    def _key(self, question: str) -> str:
        version = sha("|".join(f"{d.doc_id}:{d.content_hash}" for d in self.qa.store.list_docs()))
        return f"rag:ans:{version[:12]}:{sha(' '.join(question.lower().split()))}"

    async def answer(self, question: str) -> Answer:
        key = self._key(question)
        if (hit := await self.kv.get(key)) is not None:
            return Answer.model_validate_json(hit)
        ans = await self.qa.answer(question)
        await self.kv.set(key, ans.model_dump_json(), self.ttl_s)
        return ans

# 文件：rag-qa/tests/test_cache.py
from rag_qa.cache import CachedEmbedder, CachedQA, MemoryKV
from rag_qa.ingest.loaders import load_bytes
from rag_qa.ingest.pipeline import ingest_document
from rag_qa.llm.fake import FakeChat, FakeEmbedder
from rag_qa.qa.answer import QAService
from rag_qa.vectorstore.chroma_store import ChromaStore


async def test_cached_embedder_only_embeds_missing():
    inner = FakeEmbedder()
    emb = CachedEmbedder(inner, MemoryKV(), model="fake")
    a = await emb.embed(["甲", "乙"])
    b = await emb.embed(["乙", "丙"])
    assert inner.calls == 2 and a[1] == b[0]


async def test_answer_cache_invalidated_by_doc_change(tmp_path):
    store, emb, chat = ChromaStore(str(tmp_path)), FakeEmbedder(), FakeChat("答 [1]")
    await ingest_document(load_bytes("a.md", "# 甲\n甲的内容是一".encode()), store, emb)
    qa = CachedQA(QAService(store, emb, chat), MemoryKV())
    await qa.answer("甲的内容")
    await qa.answer("  甲的内容 ")          # 空白不同也命中缓存
    assert len(chat.received) == 1
    await ingest_document(load_bytes("a.md", "# 甲\n甲的内容是二".encode()), store, emb)
    await qa.answer("甲的内容")
    assert len(chat.received) == 2           # 文档更新后不再使用旧答案

把各部分按配置组装起来（覆盖第 5 周的 config.py 和 container.py）：

# 文件：rag-qa/src/rag_qa/config.py
from typing import Literal

from pydantic_settings import BaseSettings, SettingsConfigDict


class Settings(BaseSettings):
    model_config = SettingsConfigDict(env_file=".env", extra="ignore")

    llm_base_url: str = ""
    llm_api_key: str = ""
    chat_model: str = ""
    embed_model: str = ""
    rerank_model: str = ""          # 为空时用假重排
    embed_batch: int = 32
    chroma_path: str = "data/chroma"
    collection: str = "rag_chunks"
    chunk_size: int = 500
    chunk_overlap: int = 80
    top_k: int = 4
    min_score: float = 0.2
    max_context_chars: int = 6000
    max_upload_mb: int = 10
    retrieval_mode: Literal["vector", "hybrid", "hybrid_rerank"] = "hybrid"
    redis_url: str = ""             # 为空时缓存和会话放内存（只适合单进程开发）

    @property
    def use_fake(self) -> bool:
        return not self.llm_api_key

# 文件：rag-qa/src/rag_qa/container.py
from dataclasses import dataclass

import httpx

from rag_qa.cache import KV, CachedEmbedder, CachedQA, MemoryKV, RedisKV
from rag_qa.config import Settings
from rag_qa.llm.base import ChatModel, Embedder
from rag_qa.llm.fake import FakeChat, FakeEmbedder
from rag_qa.llm.openai_compat import OpenAICompatChat, OpenAICompatEmbedder
from rag_qa.memory.session import MemorySessionStore, RedisSessionStore, SessionStore
from rag_qa.qa.answer import QAService
from rag_qa.qa.rewrite import ConversationalQA
from rag_qa.retrieval.hybrid import HybridQAService
from rag_qa.retrieval.rerank import ApiReranker, OverlapReranker, Reranker
from rag_qa.vectorstore.base import VectorStore
from rag_qa.vectorstore.chroma_store import ChromaStore


@dataclass
class Container:
    settings: Settings
    store: VectorStore
    embedder: Embedder
    chat: ChatModel
    qa: QAService
    kv: KV
    sessions: SessionStore
    conv: ConversationalQA
    cached_qa: CachedQA


def build_container(settings: Settings | None = None, store: VectorStore | None = None,
                    embedder: Embedder | None = None, chat: ChatModel | None = None,
                    mode: str | None = None) -> Container:
    s = settings or Settings()
    mode = mode or s.retrieval_mode
    client = None
    if not s.use_fake:
        client = httpx.AsyncClient(base_url=s.llm_base_url, timeout=60,
                                   headers={"Authorization": f"Bearer {s.llm_api_key}"})
    embedder = embedder or (OpenAICompatEmbedder(client, s.embed_model, s.embed_batch) if client else FakeEmbedder())
    chat = chat or (OpenAICompatChat(client, s.chat_model) if client else FakeChat())

    if s.redis_url:
        import redis.asyncio as redis

        r = redis.from_url(s.redis_url, decode_responses=True)
        kv: KV = RedisKV(r)
        sessions: SessionStore = RedisSessionStore(r)
    else:
        kv, sessions = MemoryKV(), MemorySessionStore()

    embedder = CachedEmbedder(embedder, kv, s.embed_model or "fake")
    store = store or ChromaStore(s.chroma_path, s.collection)

    if mode == "vector":
        qa = QAService(store, embedder, chat, s.top_k, s.min_score, s.max_context_chars)
    else:
        reranker: Reranker | None = None
        if mode == "hybrid_rerank":
            reranker = ApiReranker(client, s.rerank_model) if client and s.rerank_model else OverlapReranker()
        qa = HybridQAService(store, embedder, chat, top_k=s.top_k, min_score=s.min_score,
                             max_context_chars=s.max_context_chars, reranker=reranker)
    return Container(s, store, embedder, chat, qa, kv, sessions,
                     ConversationalQA(qa, sessions, chat), CachedQA(qa, kv))

多轮对话接口单独放一个模块，在 create_app 的 return app 之前加一行注册：

# 片段：rag-qa/src/rag_qa/api/app.py 的 create_app() 里
    from rag_qa.api.chat import register_chat_routes

    register_chat_routes(app, c)
    return app

# 文件：rag-qa/src/rag_qa/api/chat.py
from fastapi import FastAPI
from pydantic import BaseModel, Field

from rag_qa.container import Container
from rag_qa.qa.answer import Answer


class ChatIn(BaseModel):
    session_id: str = Field(min_length=1, max_length=64)
    question: str = Field(min_length=1, max_length=500)


class ChatOut(BaseModel):
    standalone_question: str
    answer: Answer


def register_chat_routes(app: FastAPI, c: Container) -> None:
    @app.post("/chat", response_model=ChatOut)
    async def chat(body: ChatIn):
        q, ans = await c.conv.ask(body.session_id, body.question)
        return ChatOut(standalone_question=q, answer=ans)

    @app.get("/chat/{session_id}/history")
    async def history(session_id: str):
        return await c.sessions.history(session_id)

# 文件：rag-qa/tests/test_chat_api.py
from fastapi.testclient import TestClient

from rag_qa.api.app import create_app
from rag_qa.api.chat import register_chat_routes
from rag_qa.config import Settings
from rag_qa.container import build_container
from rag_qa.vectorstore.chroma_store import ChromaStore


def test_chat_routes(tmp_path):
    c = build_container(Settings(llm_api_key="", redis_url=""), store=ChromaStore(str(tmp_path)))
    app = create_app(c)
    register_chat_routes(app, c)
    with TestClient(app) as client:
        flow(client)


def flow(client: TestClient):
    client.post("/documents", files={"file": ("go.md", "# Go\ngoroutine 用 channel 通信".encode(), "text/markdown")})
    r = client.post("/chat", json={"session_id": "s1", "question": "goroutine 怎么通信"})
    assert r.status_code == 200 and r.json()["answer"]["citations"]
    assert len(client.get("/chat/s1/history").json()) == 2

易错点 / 面试常问（补充）

易错点：答案缓存的键要包含“知识库版本”，否则文档更新后还会返回旧答案（本项目用文档哈希算版本）。
易错点：流式接口和多轮对话的回答一般不缓存，或者只缓存嵌入向量（同一段文字的向量是固定的）。
面试常问：缓存什么最划算（嵌入向量、高频问题的答案）；缓存穿透、击穿、雪崩分别是什么（第 14 周有专题）。

要建的文件

cache.py、config.py（覆盖）、container.py（覆盖）、api/chat.py、在 api/app.py 里注册路由、tests/test_cache.py、tests/test_chat_api.py

运行方式

uv run pytest -v
REDIS_URL=redis://localhost:6379/0 uv run rag-qa ask "……"    # 问两次，第二次看 latency_ms
redis-cli --scan --pattern 'rag:*' | head

完成标准

全部测试通过；能讲清“答案缓存怎么失效”和“为什么向量缓存键要带模型名”。

第 6 天：LangChain 对照实现 + Milvus / pgvector

学习内容

1. LangChain LCEL

说明：LangChain 的 LCEL 用 | 把“检索 → 提示词 → 模型 → 解析”串成一条链，每个环节都是 Runnable，自带 invoke / ainvoke / stream / batch。它帮你省掉的是胶水代码；检索策略、阈值、拒答、引用这些核心逻辑仍然要你自己设计——这就是先手写的意义，面试时这样讲很加分。
接入真实模型可以用 init_chat_model 或 langchain-openai 的 ChatOpenAI(base_url=...)；向量库有 langchain-chroma、langchain-milvus 等集成包。LangChain 版本更新快，接口以官方文档为准：https://python.langchain.com/ （LCEL 概念：https://python.langchain.com/docs/concepts/lcel/ ）
案例：复用我们自己的检索和提示词，只把编排换成 LCEL。

# 文件：rag-qa/src/rag_qa/lc_pipeline.py
from langchain_core.language_models import BaseChatModel
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import Runnable, RunnableLambda, RunnablePassthrough

from rag_qa.qa.answer import QAService
from rag_qa.qa.prompts import NOT_FOUND, SYSTEM, build_context


def build_chain(qa: QAService, llm: BaseChatModel) -> Runnable:
    async def retrieve(question: str) -> str:
        hits = await qa.retrieve(question)
        context, _ = build_context(hits, qa.max_context_chars)
        return context

    prompt = ChatPromptTemplate.from_messages([
        ("system", SYSTEM),
        ("human", "<资料>\n{context}\n</资料>\n\n问题：{question}"),  # {} 是模板变量
    ])
    answer_chain = prompt | llm | StrOutputParser()

    async def route(inputs: dict) -> str:  # 没检索到资料就不调模型
        if not inputs["context"]:
            return NOT_FOUND
        return await answer_chain.ainvoke(inputs)

    return {"context": RunnableLambda(retrieve), "question": RunnablePassthrough()} | RunnableLambda(route)

# 文件：rag-qa/tests/test_lc_pipeline.py
from langchain_core.language_models.fake_chat_models import FakeListChatModel

from rag_qa.ingest.loaders import load_bytes
from rag_qa.ingest.pipeline import ingest_document
from rag_qa.lc_pipeline import build_chain
from rag_qa.llm.fake import FakeChat, FakeEmbedder
from rag_qa.qa.answer import QAService
from rag_qa.qa.prompts import NOT_FOUND
from rag_qa.vectorstore.chroma_store import ChromaStore


async def test_lcel_chain(tmp_path):
    store, emb = ChromaStore(str(tmp_path)), FakeEmbedder()
    await ingest_document(load_bytes("k8s.md", "# K8s\nPod 是最小调度单位。".encode()), store, emb)
    chain = build_chain(QAService(store, emb, FakeChat()), FakeListChatModel(responses=["Pod 是最小调度单位 [1]"]))
    assert await chain.ainvoke("Pod 是什么") == "Pod 是最小调度单位 [1]"
    assert await chain.ainvoke("今天天气") == NOT_FOUND

对照表（写进 README）

环节	手写版	LangChain 版
模型调用	`OpenAICompatChat`（httpx）	`ChatOpenAI` / `init_chat_model`
编排	普通 async 函数	LCEL `\|` 组合 Runnable
流式	自己解析 `data:` 行	`astream()`
检索、阈值、拒答、引用	自己写	仍然自己写（或用现成 Retriever 再定制）
优点	每一步可控、好调试	生态集成多、换组件快
缺点	胶水代码多	抽象层多，出问题要读源码；版本变化快

2. Milvus 和 pgvector

说明：
- Milvus：专门的分布式向量数据库，适合千万级以上向量、需要水平扩展的场景，部署比 Chroma 重（单机版也要 etcd、MinIO 等组件）；有 Milvus Lite 可以本地体验。
- pgvector：PostgreSQL 的向量扩展。好处是向量和业务数据在同一个库里，可以一条 SQL 同时做向量检索和权限过滤、事务一致；规模中等时很实用。
- Chroma：上手最快，适合原型和中小规模。
- 换库只需要再写一个实现 VectorStore 协议的类，其他代码不动——面试时讲“我是怎么为替换做准备的”。
案例：pgvector 版的核心 SQL（需要装好 pgvector 扩展，例如用 pgvector/pgvector:pg17 镜像）：

CREATE EXTENSION IF NOT EXISTS vector;
CREATE TABLE rag_chunks (
  chunk_id     text PRIMARY KEY,
  doc_id       text NOT NULL,
  filename     text NOT NULL,
  heading_path text NOT NULL DEFAULT '',
  text         text NOT NULL,
  embedding    vector(1024) NOT NULL          -- 维度必须和向量模型一致
);
CREATE INDEX ON rag_chunks USING hnsw (embedding vector_cosine_ops);
CREATE INDEX ON rag_chunks (doc_id);

-- <=> 是余弦距离，相似度 = 1 - 距离；$1 是问题向量
SELECT chunk_id, text, 1 - (embedding <=> $1) AS score
FROM rag_chunks
ORDER BY embedding <=> $1
LIMIT 4;

Milvus 的 Python 用法示意（pymilvus 的 MilvusClient，接口以官方文档为准：https://milvus.io/docs ）：

# 不自动运行：需要 uv add pymilvus 和一个 Milvus 服务（或 Milvus Lite 本地文件）
from pymilvus import MilvusClient

client = MilvusClient("milvus_demo.db")  # 本地 Milvus Lite；连服务器时传 uri="http://host:19530"
client.create_collection(collection_name="rag_chunks", dimension=4, metric_type="COSINE")
client.insert("rag_chunks", [{"id": 1, "vector": [0.1, 0.2, 0.3, 0.4], "text": "示例", "doc_id": "d1"}])
res = client.search("rag_chunks", data=[[0.1, 0.2, 0.3, 0.4]], limit=3, output_fields=["text", "doc_id"])
print(res)

	Chroma	Milvus	pgvector
定位	轻量、嵌入式或单服务	专业分布式向量库	PostgreSQL 扩展
规模	中小	大	中
运维	最简单	最重	复用已有 PostgreSQL
和业务数据联查	弱	弱	强（同一条 SQL、同一个事务）
本项目	默认	了解 + 选做	选做

易错点 / 面试常问（补充）

易错点：LangChain 的包拆得很细（langchain-core、langchain-openai、各种集成包），网上旧教程的导入路径经常已经失效，以官方文档为准：https://docs.langchain.com/oss/python/langchain/overview 。
易错点：换向量库（Milvus、pgvector）时，距离类型（余弦 / 内积 / 欧氏）要和嵌入模型的要求一致，否则分数没有意义。
面试常问：Chroma、Milvus、pgvector 怎么选（原型和小规模用 Chroma；数据量大、要分布式用 Milvus；已经有 PostgreSQL 且规模不大用 pgvector，事务和过滤方便）；什么是 HNSW 索引（近似最近邻，用一点精度换速度）。

要建的文件

lc_pipeline.py、tests/test_lc_pipeline.py；README 加对照表和向量库对比
选做：vectorstore/pgvector_store.py（复用第 2 周的 PostgreSQL 知识，uv add "psycopg[binary]" pgvector）

运行方式

uv run pytest tests/test_lc_pipeline.py -v

完成标准

能回答：“LangChain 帮你做了什么，没做什么？”“Chroma、Milvus、pgvector 怎么选？”

第 7 天：Docker Compose 部署 + 复盘

学习内容

说明：
- Python 镜像用 python:3.12-slim + uv：先只复制依赖文件安装依赖（利用缓存层），再复制代码。
- 数据卷：Chroma 数据（/app/data）必须挂卷，否则容器重建数据全丢。
- Chroma 也可以作为独立服务运行（官方镜像 chromadb/chroma），这时代码改用 chromadb.HttpClient(host=..., port=...)；本项目单实例先用嵌入式。
- 多个 API 实例时：会话和缓存必须放 Redis（内存版各实例互相看不到），嵌入式 Chroma 不能被多个进程同时写——要么单实例，要么换成独立的向量库服务。
案例：

FROM python:3.12-slim
COPY --from=ghcr.io/astral-sh/uv:latest /uv /usr/local/bin/uv
WORKDIR /app
ENV UV_COMPILE_BYTECODE=1 UV_LINK_MODE=copy
COPY pyproject.toml uv.lock README.md ./
RUN uv sync --frozen --no-dev --no-install-project
COPY src ./src
RUN uv sync --frozen --no-dev
ENV PATH="/app/.venv/bin:$PATH"
EXPOSE 8000
CMD ["uvicorn", "rag_qa.api.app:create_app", "--factory", "--host", "0.0.0.0", "--port", "8000"]

services:
  api:
    build: .
    env_file: .env
    environment:
      REDIS_URL: redis://redis:6379/0
      CHROMA_PATH: /app/data/chroma
    volumes:
      - rag-data:/app/data
    ports:
      - "8000:8000"
    depends_on:
      redis:
        condition: service_healthy
    restart: unless-stopped
  redis:
    image: redis:7
    healthcheck:
      test: ["CMD", "redis-cli", "ping"]
      interval: 5s
      retries: 10
volumes:
  rag-data:

易错点 / 面试常问（补充）

易错点：Chroma 的数据目录要挂到数据卷上，否则重建容器后索引就没了。
面试常问：RAG 上线后怎么持续优化（收集答不好的问题 → 加进评测集 → 调切分 / 检索 / 提示词 → 对比指标）。

要建的文件

Dockerfile、docker-compose.yml、.dockerignore（.venv、data、.env、__pycache__）
eval/results.md 整理成表格；README 补“评测结果”“设计取舍”“对照表”

运行方式

docker compose up -d --build
curl localhost:8000/health
curl -F "file=@samples/你的文件.md" localhost:8000/documents
docker compose restart api && curl localhost:8000/documents    # 重启后数据还在

完成标准

Compose 一键启动，重启后数据不丢；部署到服务器的事放到第 14 周统一做。
能脱稿回答：
1. 你的评测集怎么建的？指标是什么？各项优化分别带来了什么变化？
2. 混合检索和重排分别解决什么问题？代价是什么？
3. 多轮追问怎么处理？会话存哪里？
4. 缓存怎么失效？
5. LangChain 帮你做了什么？
6. 向量库怎么选？

本周自查清单

评测集 ≥ 30 条（含拒答题），会算 Hit@K、MRR、拒答准确率
eval/results.md 有基线、切分实验、混合检索、重排的对比记录
混合检索（BM25 + 向量 + RRF）和重排都有测试
多轮追问改写 + Redis 会话记忆（真实 Redis 测试通过）
向量缓存、答案缓存（带失效策略）
LangChain LCEL 对照实现，能讲清手写和框架的取舍
了解 Milvus、pgvector，能说出选型理由
Docker Compose 部署，数据持久化

48 KiB Raw Blame History Unescape Escape

第 6 周：RAG 知识库问答（下）——评测、混合检索、重排、记忆、缓存、LangChain、部署

第 1 天：评测集和指标

学习内容

要建的文件

运行方式

完成标准

第 2 天：BM25 关键词检索 + RRF 混合检索

学习内容

要建的文件

运行方式

完成标准

第 3 天：重排（Rerank）

学习内容

要建的文件

运行方式

完成标准

第 4 天：多轮追问——问题改写 + 会话短期记忆（Redis）

学习内容

要建的文件

运行方式

完成标准

第 5 天：缓存、配置组装、多轮对话接口

学习内容

要建的文件

运行方式

完成标准

第 6 天：LangChain 对照实现 + Milvus / pgvector

学习内容

要建的文件

运行方式

完成标准

第 7 天：Docker Compose 部署 + 复盘

学习内容

要建的文件

运行方式

完成标准

本周自查清单

48 KiB

Raw Blame History