LangChain 入门学习第五篇-使用本地 Embeddings 实现向量检索 RAG

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

上一篇用关键词检索实现了一个最小 RAG。

这一篇继续往前走，把检索方式换成向量检索。

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

这里有一个现实情况：当前本地 OpenAI 兼容服务只有聊天模型，没有额外的 embedding 模型。所以这一篇没有直接使用 OpenAIEmbeddings，而是写了一个教学版的 HashEmbeddings。

它不是语义向量模型，但可以完整演示 LangChain 里 Embeddings、VectorStore、Retriever 的流程。等后面有真正的 embedding 模型时，再替换掉这一层即可。

本篇目标

这一篇主要完成三件事：

把 Markdown 文章切分成 chunk。
用本地 HashEmbeddings 生成向量。
使用 InMemoryVectorStore 做向量检索，再交给模型回答。

完整流程是：

Markdown 文档
  -> split_documents
  -> HashEmbeddings
  -> InMemoryVectorStore
  -> Retriever
  -> Prompt + LLM

为什么需要 Embeddings

关键词检索是根据字面命中来判断相关性。

向量检索则是先把文本转换成向量，然后比较向量之间的相似度。

真实项目里，embedding 模型会尽量把语义相近的文本映射到距离更近的位置。比如“硬解码”和“MediaCodec 解码加速”虽然字面不完全一样，但语义上可能有关。

不过这一篇的重点不是训练或选择 embedding 模型，而是先看清 LangChain 的向量检索接口。

教学版 HashEmbeddings

代码放在 chapter05/local_embeddings.py。

核心类如下：

class HashEmbeddings(Embeddings):
    """一个本地可运行的教学版 Embeddings。

    它把文本 token hash 到固定维度向量中，适合用来演示 VectorStore /
    Retriever 的流程。真实项目里可以替换成 OpenAIEmbeddings 或其他模型。
    """

    def __init__(self, dimensions: int = 256) -> None:
        self.dimensions = dimensions

    def embed_documents(self, texts: list[str]) -> list[list[float]]:
        return [self._embed(text) for text in texts]

    def embed_query(self, text: str) -> list[float]:
        return self._embed(text)

LangChain 的 VectorStore 需要的是 Embeddings 接口。

只要实现：

embed_documents
embed_query

就可以接到 InMemoryVectorStore 上。

这里的 _embed() 做的事情是：

把文本切成 token。
把 token hash 到固定维度向量。
对向量做归一化。

它不是语义模型，但结果稳定、完全本地可运行，适合教学阶段使用。

构建向量库

第五篇使用 LangChain 自带的 InMemoryVectorStore：

def build_vector_store(chunks: list[Document]) -> InMemoryVectorStore:
    """用本地 HashEmbeddings 构建一个内存向量库。"""

    vector_store = InMemoryVectorStore(HashEmbeddings())
    vector_store.add_documents(chunks)
    return vector_store

这里把切分后的 chunk 加入内存向量库。

需要注意的是，InMemoryVectorStore 只存在于当前进程中，程序结束后索引就没了。

所以第五篇适合演示流程，第六篇会继续做持久化。

使用 Retriever

构建完向量库后，可以拿到 retriever：

retriever = vector_store.as_retriever(search_kwargs={"k": 4})
relevant_chunks = retriever.invoke(question)

这一步和第四篇自己写检索函数不同。

这里使用的是 LangChain 的标准 Retriever 接口。后续如果换成别的向量库，只要它也提供 retriever，后面的 RAG 代码就不用大改。

这也是 LangChain 这一层抽象的意义。

组装回答链路

后面的 Prompt 和第四篇差不多：

chain = prompt | create_chat_model(temperature=0.2) | StrOutputParser()
answer = chain.invoke(
    {
        "question": question,
        "context": format_context(relevant_chunks),
    }
)

可以看到，RAG 的后半段基本没有变化。

变化的是检索部分：

第四篇：关键词检索
第五篇：向量检索

这说明只要把“检索结果”整理成统一的上下文格式，后面的生成链路可以保持稳定。

运行 demo

在 code/langchain-demo 目录下运行：

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

正常输出会包含问题、来源和回答：

Question:
Python 函数式编程讲了什么？

Sources:
- blog/content/post/code/Python 函数式编程/index.md

Answer:
...

如果需要绕开外部 PYTHONPATH，可以使用：

env -u PYTHONPATH .venv/bin/python -m chapter05.vector_rag "Python 函数式编程讲了什么？"

本地 HashEmbeddings 的限制

这里要强调一下：HashEmbeddings 只是教学版实现。

它的优点是：

不需要额外模型。
完全本地可运行。
接口符合 LangChain 的 Embeddings 协议。

它的限制也很明显：

不是语义向量。
不能真正理解同义表达。
检索质量不如真实 embedding 模型。

真实项目中，可以换成类似下面的实现：

from langchain_openai import OpenAIEmbeddings

embeddings = OpenAIEmbeddings(
    model="text-embedding-3-small",
    api_key=settings.api_key,
    base_url=settings.base_url,
)

当然，前提是你的 OpenAI 兼容服务提供 embedding 模型。

小结

这一篇把第四篇的关键词检索换成了向量检索。

虽然 embedding 是本地 hash 版本，但已经把 LangChain 的核心接口串起来了：

Embeddings -> VectorStore -> Retriever -> RAG Chain

下一篇继续解决一个实际问题：每次运行都重新构建索引太慢，所以需要把索引保存下来。

参考

code/langchain-demo/chapter05
LangChain Embeddings
LangChain InMemoryVectorStore
LangChain Retriever

原创文章，转载请注明来源: LangChain 入门学习第五篇-使用本地 Embeddings 实现向量检索 RAG