飞污熊小站
飞污熊小站

使用LangChain+LangGraph开发AI应用（6） - 检索增强RAG

XiongNeng 收录于 AI
 约 4200 字 预计阅读 9 分钟 

https://static.xiongneng.me/langchain-langgraph-cover-20261017175843.png

学习进度

模型有两个绕不开的限制。一是上下文有限，再大的窗口也塞不下整个资料库。二是知识过期，训练数据停在某个时间点，之后的它一概不知。检索增强，RAG，就是冲着这两个限制去的，回答问题之前先把相关资料找出来给它。

本篇按官方文档的检索流水线走一遍，加载、分割、嵌入、入库、检索、生成，六步每步都有对应组件。示例在 source/hello_rag.py，九个 demo。完整案例是一个能用的命令行小工具 source/kb.py，把 markdown 笔记吃进本地向量库，随时提问，回答带出处。

前言用途

模型直接回答知识类问题的样子你大概见过，问它你家项目的接口约定，它一本正经地编。上下文窗口再大也有价，每轮对话都把全部资料塞进去，token 账单先扛不住。RAG 的思路是平时把资料切好块、算好向量存起来，问题来了只检索最相关的几块，带着这几块去问模型。资料存库几乎不花钱，每次回答只带几段相关内容进上下文。

什么时候需要 RAG，一句话判断，答案在你的资料里而不在模型的训练数据里。公司内部文档、个人笔记、产品手册、聊天记录，都是典型场景。答案在模型训练数据里的通用问题，直接问就行，硬套 RAG 反而多此一举。

LangChain 把这条流水线的每个环节都做成了可替换的组件，加载器有上百种，分割器按需选，嵌入模型换个名字就能换提供商，向量库从内存版到专业数据库一条线排开。本篇用零部署的组合把整条链跑通，你换任何一环，其他代码都不用动。

原理概念

RAG 有两种常见架构。两步式，检索永远发生在生成之前，先查资料再回答，简单可控，绝大多数问答场景够用。Agentic 式，检索被包成工具交给 agent，模型自己决定查不查、查什么、查几轮，灵活但延迟不可预期。官方文档的对比表把控制力和灵活性摆在天平两端，我的建议是先用两步式，检索质量不够了再上 agent。两步式是本篇主线，Agentic 放在核心用法最后讲。

流水线上所有环节共同认的是 Document，LangChain 用它统一承载各种来源的文本。一个 Document 就两个字段，page_content 装文本本身，metadata 装来源、页码、时间这类信息。加载器把 PDF、网页、数据库行都变成 Document 列表，后面的环节只认这个形状，来源是什么根本不重要。

https://static.xiongneng.me/rag-pipeline-20261017175843.png

嵌入是把文本变成向量的过程，语义相近的文本向量也相近。Embeddings 接口只有两个方法。embed_documents 在入库时给文档批量算向量。embed_query 在提问时给问题算向量。同一个问题两种方法算出的向量必须落在同一个语义空间里，这是检索有效的前提。向量入库后，检索就是算余弦相似度，找出跟问题向量最接近的几块。

向量库是存向量并支持相似度检索的仓库。LangChain 的 InMemoryVectorStore 跑在内存里，零部署，dump 和 load 还能把库落成 JSON 文件。数据量大再换 FAISS、Milvus 这类专业选手，接口是同一套。

核心用法

九个用法，示例在 source/hello_rag.py。嵌入有两个来源，设了 YI 环境变量就用真实嵌入（EMBED_MODEL 指定，默认 text-embedding-3-small），没设就用本地的哈希替身，同样的文本永远得到同样的向量，测试可复现。

1. Document 的本质

先看它长什么样。

1
2
3
4
5
6
7

from langchain_core.documents import Document

doc = Document(
    page_content="LangChain 用 Document 统一承载各种来源的文本。",
    metadata={"source": "notes.md", "page": 3},
)
print(doc.page_content, doc.metadata)
1
2
3

== Document ==
  内容: LangChain 用 Document 统一承载各种来源的文本。
  元数据: {'source': 'notes.md', 'page': 3}

2. 加载

加载环节做的事就一句话，把任何来源变成 Document 列表。生态里有上百种现成加载器，TxtLoader、PDF 加载器、Notion 加载器，都是一个套路。自己写一个也就几行。

1
2
3
4
5
6
7

from pathlib import Path

def load_text(path: Path) -> list[Document]:
    text = Path(path).read_text(encoding="utf-8")
    return [Document(page_content=text, metadata={"source": str(path)})]

docs = load_text(Path("notes.txt"))
1
2

== 加载 ==
  notes.txt -> 1 个 Document，374 字符

3. 分割

一整个文件直接入库，检索粒度就太粗了，召回一大段里真正相关的只有两句，token 全浪费在路上。分割用 RecursiveCharacterTextSplitter，按分隔符列表递归地切，先按段落切，超长了再按句子切。

1
2
3
4
5
6
7

from langchain_text_splitters import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=90, chunk_overlap=15,
    separators=["\n\n", "\n", "。", "，"],
)
docs = splitter.create_documents([text])
1
2
3
4
5

== 分割 chunk_size=90 overlap=15 ==
  374 字符 -> 7 块
  块 1（16 字符）: LangChain 安装与初始化 ...
  块 2（56 字符）: 安装 LangChain 用 pip install ...
  块 3（37 字符）: 装完建议先跑一个 init_chat_model 的 ...

chunk_overlap 让相邻块共享一小段文字，句子正好骑在切分点上时，两边都能保住完整语义。注意块 1 只有十六个字符，它是个孤零零的标题，标题和正文被切开了，这种碎块的嵌入信息量太稀，把块放大或者入库前把标题并进正文都能解决。

4. 嵌入

Embeddings 接口就两个方法，各管一头。

1
2
3

emb = build_embeddings()   # 真实嵌入或本地替身，约定见 demo 源码
vectors = emb.embed_documents(["北京今天晴", "上海多云"])
query = emb.embed_query("天气怎么样")
1
2

== 嵌入 ==
  OpenAIEmbeddings: 文档 2 条，每条 1536 维；问题向量 1536 维

embed_documents 是入库时批量算向量用的，embed_query 是每次提问算问题向量用的。两个方法背后是同一个模型，换嵌入模型等于换语义空间，这件事的坑后面有专条。

5. 向量库

InMemoryVectorStore 从 Document 列表一步建库，dump 和 load 还能把库落成 JSON 文件带走。

1
2
3
4
5
6

from langchain_core.vectorstores import InMemoryVectorStore

store = InMemoryVectorStore.from_documents(docs, emb)
store.dump("kb_store.json")

reloaded = InMemoryVectorStore.load("kb_store.json", embedding=emb)
1
2
3

== 向量库 ==
  库内条目: 2
  落盘重载后条目: 2

库内部的结构是个字典，每一项带 id、vector、text、metadata 四个键，dump 出来的 JSON 就是这个形状，纯文本可查，排查问题时很方便。

6. 相似检索

接下来是检索环节，RAG 里最值得花心思的一步。similarity_search 返回 Document 列表，similarity_search_with_score 附带相似度。

1
2
3
4
5

store = InMemoryVectorStore.from_texts(
    ["今天北京天气晴朗。", "上海明天有雨。", "Python 是一门编程语言。"], emb,
)
hits = store.similarity_search("今天北京天气晴朗。", k=2)
scored = store.similarity_search_with_score("今天北京天气晴朗。", k=1)
1
2
3
4

== 相似检索 k=2 ==
  命中: 今天北京天气晴朗。
  命中: 上海明天有雨。
  首位相似度: 1.0000（1.0 表示完全一致）

两个天气句子排前面，编程语言那句被排到最后，语义检索干了关键词匹配干不了的活。with_score 的第二个值是相似度，1.0 表示完全一致，数值越大越相关。网上不少文章把它当距离讲，方向反了，照着抄会把过滤条件写反。

7. 检索器

as_retriever 把向量库包成统一接口，后面换向量库，检索代码不用动。参数经 search_kwargs 透传。

1
2
3
4
5

retriever = store.as_retriever(search_kwargs={"k": 2})
hits = retriever.invoke("LangChain 提供模型抽象。")

mmr = store.as_retriever(search_type="mmr", search_kwargs={"k": 2, "fetch_k": 3})
hits2 = mmr.invoke("LangChain 提供模型抽象。")
1
2
3

== MMR 检索 ==
  命中: LangChain 提供模型抽象。
  命中: 今天的午餐是面条。

库里两条 LangChain 的句子只回来了第一条，另补了一条午餐句子。这就是 MMR 的作用，在相关性和多样性之间做平衡，避免 top-k 全是同一句话的变体。资料来源多、内容重复度高时，MMR 比纯相似度实用。

8. 两步式 RAG

把检索结果拼进提示词，让模型只依据料回答。

1
2
3
4
5
6
7
8
9

question = "装 LangChain 对 Python 版本有什么要求？"
context_docs = store.similarity_search(question, k=1)
context = "\n".join(d.page_content for d in context_docs)

prompt = (
    "只依据下面的资料回答问题，资料里没有就直说不知道。\n\n"
    f"资料：\n{context}\n\n问题：{question}"
)
answer = model.invoke([SystemMessage("你是简洁的问答助手"), HumanMessage(prompt)])
1
2
3

== 两步式 RAG ==
  检索到 1 块: 安装 LangChain 需要 Python 3.9 以上版本。
  回答: 根据笔记，Python 3.9 以上版本即可安装 LangChain。

提示词里那句「资料里没有就直说不知道」是防幻觉的关键，别省。RAG 系统胡说八道，一半的锅在这句缺失上。

9. Agentic RAG

两步式是无脑先查再答，Agentic 把检索器包成工具，查不查由模型自己决定。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

from langchain_core.tools import create_retriever_tool

retriever = store.as_retriever(search_kwargs={"k": 1})
search_tool = create_retriever_tool(
    retriever, "search_notes", "搜索 LangChain 学习笔记，返回相关段落",
)
agent = create_agent(
    model=model, tools=[search_tool],
    system_prompt="回答问题前先查笔记，只依据笔记内容作答。",
)
result = agent.invoke({"messages": [HumanMessage("装 LangChain 要什么版本的 Python？")]})
1
2
3
4

== Agentic RAG ==
  模型发起检索: search_notes({'query': 'Python 版本要求'})
  工具返回: 安装 LangChain 需要 Python 3.9 以上版本。
  回答: 根据笔记，安装 LangChain 需要 Python 3.9 以上版本。

第五篇提过的 create_retriever_tool 在这里用上了，检索器包成工具后，agent 的完整循环、错误处理、中间件全都对它生效。闲聊不用查资料的问题，模型直接回答不调工具，token 就省下来了。

完整案例

九个 demo 摊开讲完了。收个尾，把它们装进一个能日常用的命令行工具，kb.py，个人知识库的雏形。它吃 markdown 和 txt 笔记，建本地向量库，之后随时提问，回答附带每条引用的来源和相似度。

1

uv run python kb.py ingest rag_notes/
1

吃进 2 个文件，切成 2 块，库里现在有 2 块

ingest 读目录下所有 txt 和 md，按 chunk_size=200, overlap=30 分割，嵌入后存进本地 JSON 文件。list 命令看库存。

1

uv run python kb.py list
1
2
3

库内共 2 块
  1. [basics.md] LangChain 安装  安装 LangChain 用 pip insta
  2. [graph.md] LangGraph 要点  LangGraph 用节点和边组织流程，条件边让

ask 是主功能，检索 top-k 拼提示词，模型只依据料回答。

1

uv run python kb.py ask "LangChain 怎么安装"
1
2
3
4
5

问题: LangChain 怎么安装
检索到 2 块:
  (0.620) [basics.md] LangChain 安装  安装 LangChain 用 pip
  (0.307) [graph.md] LangGraph 要点  LangGraph 用节点和边组织流程，条件边让
回答: 根据笔记，安装 LangChain 用 pip install langchain，需要 Python 3.9 以上版本。

basics.md 的相似度 0.620，明显高于 graph.md 的 0.307，检索排序和人的直觉一致。接上真实模型再问一个需要判断的问题。

1

AGENT_MODEL=openai:deepseek-v4.1-flash uv run python kb.py ask "LangGraph 的条件边是干嘛的" --k 2
1
2
3
4
5

问题: LangGraph 的条件边是干嘛的
检索到 2 块:
  (0.626) [graph.md] LangGraph 要点  LangGraph 用节点和边组织流程，条件边让
  (0.257) [basics.md] LangChain 安装  安装 LangChain 用 pip
回答: 根据资料，LangGraph 的条件边用于让流程按状态分叉。

相关性排序正确，回答贴着资料走。整个案例不含任何数据库服务，落盘就是一个 JSON 文件，拷到别的机器 load 回来接着用。

测试在 tests/test_hello_rag.py 十条加 tests/test_kb.py 十三条，分割边界、相似度语义、落盘往返、ingest 与 ask 流程都钉住了。加上前五篇的，全仓九十四条，全绿。

源码

https://github.com/yidao620c/xneng-langchain-langgraph

参考

更新于 2026-10-17 
LangChainLangGraph
返回 | 主页

目录