一个最简单 RAG 功能的实现流程

最近在研究 RAG，顺手搭了一个最小可运行的案例，本文把整个实现过程拆成几步逐一梳理。用到的技术栈如下：

语言与框架：python + langchain@v1.2
向量数据库：redis
大语言模型：qwen3.6-plus
嵌入模型：text-embedding-v4

文档加载器

实现 RAG 的第一步，就是能把文档读进来。这里以文本文档为例，直接用社区包提供的加载器：

from langchain_community.document_loaders import TextLoader

docs = TextLoader("assets/rag_txt_demo.txt", encoding="utf-8").load()

print(docs)

# 打印结果
# [Document(metadata={'source': 'assets/rag_txt_demo.txt'}, page_content='...')]

记得先安装社区包：uv add langchain-community。

拿到的 docs 是一个列表，每一项都是 Document 类型，主要包含两个字段：

metadata：元数据，比如来源 {'source': 'assets/rag_txt_demo.txt'}
page_content：源文本内容

你也可以直接手动构造 Document：

source = "assets/rag_txt_demo.txt"
text = open(source).read()

docs = [
    Document(
        page_content=text,
        metadata={"source": source}
    )
]

按需选择就好，简单场景直接用 TextLoader 就够了。案例所用测试文档放在文章最后面了。

文档分割

加载完文档后，还需要把它切成小段，再对每一段进行向量化。

为什么需要文档分割？大模型通常有上下文窗口限制，整篇文档直接扔进去，超出长度会被截断。而且检索时也是以片段为单位匹配——用户问一个具体问题，更合适的做法是把相关的片段找出来交给模型，而不是把整篇文章都塞进去。切得合理，检索和生成效果才会好。

先安装分割工具：

uv add langchain-text-splitters

然后创建分割器，对文档进行切分：

from langchain_text_splitters import RecursiveCharacterTextSplitter

text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=100,
    chunk_overlap=50
)

splitter_documents = text_splitter.split_documents(documents=docs)

print(splitter_documents)

# 打印结果
# [Document(metadata={'source': 'assets/rag_txt_demo.txt'}, page_content='...'),
#  Document(metadata={'source': 'assets/rag_txt_demo.txt'}, page_content='...'), ...]

RecursiveCharacterTextSplitter 的两个核心参数：

chunk_size：每段的最大字符数
chunk_overlap：相邻片段之间重叠的字符数，用来避免关键信息刚好被切在边界上而丢失

分割后的 splitter_documents 是一个列表，每一项都是一段 Document。

向量化入库

现在要把切好的文档片段转成向量，存进向量数据库。

LangChain 支持多种嵌入模型和向量数据库，完整列表可参考官方文档：Vector store integrations。

本案例用阿里的 text-embedding-v4 做嵌入，Redis 做存储。阿里嵌入模型通过社区包的 DashScopeEmbeddings 创建，Redis 需要额外装一个支持包：

uv add langchain-redis

同时确保本地有可用的 Redis 服务，比如我这边就跑在 6379 端口。

流程如下：

from langchain_community.embeddings import DashScopeEmbeddings
from langchain_redis import RedisVectorStore, RedisConfig

embeddings = DashScopeEmbeddings(model="text-embedding-v4")

config = RedisConfig(
    index_name="my_vec_db",
    redis_url="redis://localhost:6379",
)

vector_store = RedisVectorStore(
    embeddings=embeddings,
    config=config,
)

vector_store.add_documents(splitter_documents)

把上一步的文档列表传进去，就自动完成向量化并入库了。可以用 Redis Insight 查看每个片段的信息：

redis 存储向量片段案例

每个片段会存下这些字段：

_index_name：所属向量索引名称
source：metadata 中的来源字段
embedding：文本对应的向量
text：chunk 的实际文本内容
_metadata_json：完整 metadata 的 JSON

不同向量数据库存储的字段略有差异，但大体结构都差不多。

检索器

向量数据入库后，就可以用来检索了。

RedisVectorStore 可以直接转成一个检索器：

retriever = vector_store.as_retriever()

as_retriever 支持一些配置项，比如通过 search_kwargs={"k": 4} 控制返回的文档数量。这里先用默认值。

尝试检索一个问题：

question = "Aurora Notes 使用什么技术实现语义搜索？"

ret = retriever.invoke(question)

for item in ret:
    print(item)

打印结果会包含 4 条相关的 Document 片段——这是检索器默认返回的数量。

接入大模型检索

有了检索器，接下来把检索到的内容交给大模型，让它基于这些信息来回答问题。

我用的是阿里百炼平台的 qwen3.6-plus：

from langchain.chat_models import init_chat_model
from dotenv import load_dotenv
import os

load_dotenv()

llm = init_chat_model(
    model="qwen3.6-plus",
    model_provider="openai",
    base_url=os.getenv("DASHSCOPE_BASE_URL"),
    api_key=os.getenv("DASHSCOPE_API_KEY"),
)

模型本身不知道要去检索，需要用提示词告诉它数据从哪来：

from langchain_core.prompts import PromptTemplate

prompt = PromptTemplate.from_template(
    """
请使用以下提供的文本内容来回答问题。仅使用提供的文本信息，
如果文本中没有相关信息，请回答"抱歉，提供的文本中没有这个信息"。

文本内容：
{context}

问题：
{question}
""",
)

提示词里预留了两个变量：context 是检索到的文本内容，question 是用户的问题。

把模型和检索串起来跑一遍：

from langchain_core.runnables import RunnablePassthrough


def format_docs(docs):
    return "\n\n".join(doc.page_content for doc in docs)


chain = prompt | llm

question = "Aurora Notes 使用什么技术实现语义搜索？"

docs = retriever.invoke(question)
context = format_docs(docs)

res = chain.invoke({"question": question, "context": context})

print(res.content)

# 打印结果
# 根据提供的文本，Aurora Notes 为了实现语义搜索功能，接入了**向量数据库能力**，
# 并使用 **OpenAI 的 text-embedding-3-small 模型**生成文本向量。

RAG 基本流程就走通了。效果不一定最佳——这个案例的目标只是跑通链路，没做调优。

上面的写法可以进一步简化，把检索和格式化也链进来：

chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | prompt
    | llm
)

question = "Aurora Notes 使用什么技术实现语义搜索？"

res = chain.invoke(question)
print(res.content)

这样整个流程（检索 → 拼提示词 → 调用模型）就串联成了一条链，一步到位。

总结

RAG 的核心思路很简单：让大模型在回答问题之前，先去自己的知识库里查一下相关材料，再基于这些材料来生成答案。整篇流程可以概括为四步：

加载 —— 把文档读进来，变成 Document 对象
分割 —— 将长文档切成合适大小的片段，方便检索和适配上下文窗口
向量化入库 —— 用嵌入模型把每个片段转成向量，存入向量数据库
检索 + 生成 —— 用户提问时，从向量库召回相关片段，连同问题一起交给大模型，生成最终回答

这个案例只是最简实现，实际生产环境还有不少可以优化的地方，比如：选择更合适的分割策略、调优 chunk_size 和 chunk_overlap、尝试不同的嵌入模型、引入 reranker 做二次排序，等等。不过先把链路跑通，是理解 RAG 的最佳起点。

测试数据

文中案例测试用的文本：

# 极光笔记（Aurora Notes）项目说明  
  
极光笔记（Aurora Notes）是一个面向开发者的轻量级知识管理系统，由林远在 2024 年创建。项目最初只是一个 Markdown 编辑器，但后来逐渐加入了 AI 能力，包括文本摘要、语义搜索以及自动标签生成。  
  
整个系统主要由 Go 和 Vue 构建，其中后端使用 Gin 框架，数据库采用 PostgreSQL，缓存层则使用 Redis。为了实现语义搜索功能，项目在 2025 年接入了向量数据库能力，并使用 OpenAI 的 text-embedding-3-small 模型生成文本向量。  
  
项目的知识库内容会被切分成多个 chunk，每个 chunk 大约 500 字符，然后再写入向量索引中。开发团队发现，如果 chunk 太大，会导致召回结果不准确；但 chunk 太小，又会丢失上下文信息。  
  
Aurora Notes 的 AI 助手支持 RAG 模式。当用户提问时，系统会先进行相似度检索，再将检索结果和用户问题一起发送给大语言模型生成答案。为了提升速度，团队还增加了 Redis 缓存层，用于缓存 embedding 结果与历史问答。  
  
2025 年 3 月，团队尝试加入多模态能力，希望系统未来不仅能检索文本，还能检索图片和 PDF 文档。目前该功能仍处于实验阶段。  
  
项目部署在东京的一台 Ubuntu 服务器上，平均每天处理大约 2 万次搜索请求。

文档加载器

文档分割

向量化入库

检索器

接入大模型检索

总结

测试数据

评论