目录

一个专注音视频领域的小圈子

第五篇已经用本地 HashEmbeddings 和 InMemoryVectorStore 跑通了向量检索 RAG。

但它还有一个问题：每次运行都要重新读取文章、切分文章、生成向量。

文章少的时候没关系，文章一多就不适合了。

这一篇就来做一个简单的持久化索引，把生成好的 chunk 和向量保存成 JSON 文件。下次问答时直接加载索引，不需要从头构建。

本文对应代码位于 code/langchain-demo/chapter06。

本篇目标

这一篇把流程拆成两个命令：

  1. build_index：读取文章，生成索引，保存到本地。
  2. persistent_rag：加载索引，检索片段，调用模型回答。

目录结构如下：

chapter06/
  __init__.py
  config.py
  llm.py
  load_docs.py
  local_embeddings.py
  index_store.py
  build_index.py
  persistent_rag.py

和第五篇相比，新增的重点是：

  • index_store.py
  • build_index.py
  • persistent_rag.py

索引保存什么

一个 RAG 索引至少要保存三类信息：

  1. chunk 文本内容。
  2. chunk 的 metadata，比如来源文章、标题、分类。
  3. chunk 对应的 embedding 向量。

所以代码里定义了一个 IndexedChunk：

@dataclass
class IndexedChunk:
    """持久化索引中的一个文本片段。"""

    page_content: str
    metadata: dict
    embedding: list[float]

这个结构很朴素，但已经足够说明索引持久化的本质。

真实项目里可能会把这些内容存到 Chroma、FAISS、PGVector、Milvus 之类的向量库里。这里先保存成 JSON，方便直接打开查看。

构建索引

构建索引的代码在 index_store.py：

def build_index(chunks: list[Document]) -> list[IndexedChunk]:
    """给 chunks 生成向量，形成可保存的索引结构。"""

    embeddings = HashEmbeddings()
    return [
        IndexedChunk(
            page_content=chunk.page_content,
            metadata=dict(chunk.metadata),
            embedding=embeddings.embed(chunk.page_content),
        )
        for chunk in chunks
    ]

这里每个 chunk 都会生成一个向量，然后和正文、metadata 放在一起。

保存索引也很简单：

def save_index(index: list[IndexedChunk], path: Path) -> None:
    """把索引保存成 JSON 文件。"""

    path.parent.mkdir(parents=True, exist_ok=True)
    payload = [asdict(item) for item in index]
    path.write_text(json.dumps(payload, ensure_ascii=False), encoding="utf-8")

索引路径在 config.py 里：

INDEX_PATH = PROJECT_ROOT / "data" / "chapter06" / "blog_index.json"

data/ 目录已经加入 .gitignore，它属于本地生成数据，不需要提交。

构建命令

build_index.py 是一个单独命令：

def main() -> None:
    """构建本地博客文章索引，并保存到 JSON 文件。"""

    documents = load_markdown_documents(BLOG_POST_ROOT)
    chunks = split_documents(documents)
    index = build_index(chunks)
    save_index(index, INDEX_PATH)

    print(f"Documents: {len(documents)}")
    print(f"Chunks: {len(chunks)}")
    print(f"Saved index: {INDEX_PATH}")

运行：

uv run python -m chapter06.build_index

输出类似：

Documents: 194
Chunks: 1433
Saved index: .../code/langchain-demo/data/chapter06/blog_index.json

这时候本地索引就已经生成好了。

从索引检索

加载索引时，把 JSON 还原成 IndexedChunk：

def load_index(path: Path) -> list[IndexedChunk]:
    """从 JSON 文件读取索引。"""

    payload = json.loads(path.read_text(encoding="utf-8"))
    return [IndexedChunk(**item) for item in payload]

检索时，用问题生成 query embedding，再计算余弦相似度：

def retrieve(question: str, index: list[IndexedChunk], top_k: int = 4) -> list[Document]:
    """从持久化索引中检索最相关的片段。"""

    embeddings = HashEmbeddings()
    query_embedding = embeddings.embed(question)

    scored = [
        (cosine_similarity(query_embedding, item.embedding), item)
        for item in index
    ]
    scored.sort(key=lambda item: item[0], reverse=True)

这里没有用 InMemoryVectorStore，而是手写了一版余弦相似度。

这么做是为了让索引持久化过程更透明：索引里保存的是向量，检索时就是拿 query 向量和已有向量计算相似度。

运行问答

先构建索引：

uv run python -m chapter06.build_index

再运行问答：

uv run python -m chapter06.persistent_rag "Python 函数式编程讲了什么？"

如果索引不存在，程序会提示先构建：

Index not found. Build it first:
uv run python -m chapter06.build_index

这个提示很有必要。否则新用户第一次运行时，很容易不知道为什么没有结果。

需要注意的问题

这一版索引持久化只是教学实现。

它适合说明流程，但还不是生产级方案：

  1. JSON 文件会随着文章数量增大而变大。
  2. 每次查询仍然要遍历所有向量。
  3. 没有增量更新机制。
  4. 没有记录索引版本和生成参数。

真实项目里，如果数据量变大，需要换成专门的向量数据库，或者至少使用 FAISS 这类本地向量索引库。

不过从学习角度看，JSON 版本反而更容易理解。

小结

这一篇把 RAG 拆成了两个阶段：

离线阶段：读取文档 -> 切分 -> 生成向量 -> 保存索引
在线阶段：加载索引 -> 检索 -> 生成回答

这是本地知识库问答里很常见的结构。

后面第七篇会继续处理一个更细的问题：同样是 RAG，为什么有时候检索结果不准，以及怎么调参数。

参考

  • code/langchain-demo/chapter06
  • LangChain Document
  • 向量余弦相似度
  • RAG 索引持久化流程

原创文章，转载请注明来源:    LangChain 入门学习第六篇-持久化索引和本地知识库问答