LangGraph 构建 RAG

RAG（Retrieval-Augmented Generation，检索增强生成）是让大模型"先查资料再回答"的经典方案。本篇带你用 LangGraph 从零搭一个可控、可观测的 RAG 流程，并理解它比 LangChain 链式 RetrievalQA 强在哪里。

一、为什么用 LangGraph 构建 RAG

先回忆一下最朴素的 RAG：用户提问 → 去向量库检索相关文档 → 把文档塞进 prompt → 让 LLM 生成回答。流程简单，但一旦上线就会冒出一堆问题：

检索回来的文档真的相关吗？要不要打分过滤？
用户问题太笼统，要不要先改写再检索？
回答引用了不存在的内容（幻觉）怎么办？
某些问题根本不需要检索（比如"你好"），能不能跳过？
关键场景要不要让人工看一眼再发出去？

这些问题用一条线性的 Chain 很难优雅处理，因为它没有"分支""循环""条件判断"的概念。而 LangGraph 天生就是为带控制流的状态机设计的：

能力	LangChain RetrievalQA	LangGraph
流程结构	固定线性链	图，支持分支/循环/并行
状态管理	隐式，链路中间值难取	显式 State，每步可读可改
路由	不支持	条件边随意切
纠错重试	几乎不行	加边回到上一步即可
人工干预	不支持	interrupt 一行搞定
可观测	只能看最终结果	每个节点的输入输出都能拿

一句话：LangGraph 把 RAG 从"一条流水线"升级成"一个可编排的工作流"。对有 Java/LangGraph4j 背景的同学，可以把它类比为用状态机替代一条 Stream pipeline——你能精确控制每一步的走向。

二、基础 RAG 流程总览

最基础的 RAG 只有两个核心节点：

mermaid

flowchart LR
    U([用户提问]) --> R[retrieve 检索节点]
    R --> G[generate 生成节点]
    G --> A([最终回答])

retrieve：用问题向量去向量库检索 Top-K 文档片段。
generate：把问题和检索到的文档拼成 prompt，交给 LLM 生成回答。

我们要在 LangGraph 里把这两个节点串起来，并定义一个状态在它们之间流动。状态至少要包含：问题、检索到的文档、最终回答。

三、状态设计

RAG 的状态可以用 TypedDict 定义。新手记住一点：状态就是节点之间传递的"行李箱"，每个节点从里面拿东西、放东西。

python

from typing import TypedDict, List
from langchain_core.documents import Document

class RAGState(TypedDict):
    question: str            # 用户问题
    documents: List[Document]  # 检索到的文档片段
    answer: str              # 最终回答

小贴士：如果你用过 LangChain 的 Runnable，会发现这里的状态等价于链路里手动维护的中间 dict，只不过 LangGraph 把它"显式化"了，调试时能清楚看到每一步状态长什么样。

四、节点实现

1. 检索节点

检索节点的职责：从状态里拿 question，去向量库检索，把结果写回 documents。

python

def retrieve(state: RAGState) -> dict:
    """根据问题检索相关文档片段"""
    question = state["question"]
    # retriever 是预先构造好的向量库检索器
    docs = retriever.invoke(question)
    print(f"[retrieve] 检索到 {len(docs)} 条文档")
    return {"documents": docs}

2. 生成节点

生成节点把文档和问题拼成 prompt，调用 LLM 生成回答。

python

from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI

# 复用 prompt 和模型，避免每次节点调用都重建
prompt = ChatPromptTemplate.from_template(
    "你是一个严谨的问答助手。请只根据下面提供的参考资料回答问题，"
    "不要编造。如果资料不足以回答，请直接说'根据已知资料无法回答'。\n\n"
    "参考资料：\n{context}\n\n"
    "问题：{question}\n"
    "回答："
)
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

def generate(state: RAGState) -> dict:
    """根据检索到的文档生成回答"""
    question = state["question"]
    docs = state["documents"]
    # 把每段文档拼成一长串文本作为上下文
    context = "\n\n".join(d.page_content for d in docs)
    messages = prompt.invoke({"context": context, "question": question})
    response = llm.invoke(messages)
    print(f"[generate] 生成回答，长度 {len(response.content)}")
    return {"answer": response.content}

注意 prompt 里那句"如果资料不足以回答就直说"——这是抑制幻觉最简单有效的手段，比花里胡哨的技巧都管用。

五、组装 StateGraph

有了状态和节点，把它们连起来：

python

from langgraph.graph import StateGraph, START, END

graph_builder = StateGraph(RAGState)
graph_builder.add_node("retrieve", retrieve)
graph_builder.add_node("generate", generate)

graph_builder.add_edge(START, "retrieve")      # 起点 → 检索
graph_builder.add_edge("retrieve", "generate")  # 检索 → 生成
graph_builder.add_edge("generate", END)         # 生成 → 终点

rag_app = graph_builder.compile()

完整流程图：

mermaid

flowchart LR
    S([START]) --> retrieve
    retrieve --> generate
    generate --> E([END])

六、完整可运行示例

下面是一个能直接跑的完整例子。我们用一段内置文本做知识源，InMemoryVectorStore 当向量库，不需要装 FAISS/Chroma，也不依赖外部数据文件。

6.1 安装依赖

bash

# 国内镜像加速
pip install langgraph langchain-openai langchain-text-splitters -i https://pypi.tuna.tsinghua.edu.cn/simple

6.2 完整代码

python

import os
from typing import TypedDict, List

from langchain_core.documents import Document
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.vectorstores import InMemoryVectorStore
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langgraph.graph import StateGraph, START, END

# ============ 1. 准备知识库 ============
# 一段示例文本（实际项目里可以从文件/数据库加载）
sample_text = """
LangGraph 是一个用于构建有状态、多角色应用的库，核心是把流程建模成一张图。
图的节点是普通的 Python 函数，边决定执行的先后顺序。
状态在节点之间流动，每个节点接收状态、返回状态的更新。
条件边可以根据当前状态选择下一个要执行的节点，从而实现分支。
LangGraph 由 LangChain 团队开发，但可以脱离 LangChain 单独使用。
它特别适合构建智能体（Agent）、RAG、多智能体协作等需要复杂控制流的应用。
"""

# 切片：把长文本切成小块
splitter = RecursiveCharacterTextSplitter(chunk_size=80, chunk_overlap=10)
chunks = splitter.split_text(sample_text)
documents = [Document(page_content=c) for c in chunks]

# 构建向量库（InMemoryVectorStore 适合demo，生产用 FAISS/Chroma）
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vector_store = InMemoryVectorStore(embeddings)
vector_store.add_documents(documents)
retriever = vector_store.as_retriever(search_kwargs={"k": 2})

# ============ 2. 定义状态 ============
class RAGState(TypedDict):
    question: str
    documents: List[Document]
    answer: str

# ============ 3. 定义节点 ============
prompt = ChatPromptTemplate.from_template(
    "你是严谨的问答助手。只根据下面资料回答，不要编造。"
    "资料不足时回答'根据已知资料无法回答'。\n\n资料：\n{context}\n\n问题：{question}\n回答："
)
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

def retrieve(state: RAGState) -> dict:
    question = state["question"]
    docs = retriever.invoke(question)
    return {"documents": docs}

def generate(state: RAGState) -> dict:
    question = state["question"]
    docs = state["documents"]
    context = "\n\n".join(d.page_content for d in docs)
    response = llm.invoke(prompt.invoke({"context": context, "question": question}))
    return {"answer": response.content}

# ============ 4. 构建图 ============
gb = StateGraph(RAGState)
gb.add_node("retrieve", retrieve)
gb.add_node("generate", generate)
gb.add_edge(START, "retrieve")
gb.add_edge("retrieve", "generate")
gb.add_edge("generate", END)
rag_app = gb.compile()

# ============ 5. 运行 ============
if __name__ == "__main__":
    result = rag_app.invoke({"question": "LangGraph 适合用来做什么？"})
    print("问题：", "LangGraph 适合用来做什么？")
    print("回答：", result["answer"])
    print("\n引用文档数：", len(result["documents"]))

运行后你会看到模型基于那段示例文本回答出"适合构建智能体、RAG、多智能体协作等"。把 question 换成"今天天气如何？"，模型会老老实实回答"根据已知资料无法回答"——这正是 RAG 抑制幻觉的体现。

七、与 LangChain RetrievalQA 对比

很多新手第一反应是"这跟 RetrievalQA.from_chain_type(...) 有啥区别？"。看下对比：

python

# LangChain 经典写法：一行搞定，但你只能拿到最终 answer
from langchain.chains import RetrievalQA
qa = RetrievalQA.from_chain_type(llm=llm, retriever=retriever)
print(qa.invoke("LangGraph 适合做什么？"))

python

# LangGraph 写法：多几行，但每一步都可控
result = rag_app.invoke({"question": "LangGraph 适合做什么？"})
print(result["documents"])  # 能拿到检索到了啥
print(result["answer"])     # 也能拿到回答

差别在三点：

可观测：LangGraph 能拿到每个中间状态（检索了哪几段、拼了什么 prompt），RetrievalQA 默认拿不到。
可扩展：想加"文档评分过滤""问题改写""回答校验"？LangGraph 加个节点+一条边就行；RetrievalQA 得继承重写一堆类。
可控流：LangGraph 能做条件分支、循环纠正、人工干预，RetrievalQA 做不到。

所以本系列后续的自适应 RAG、自纠正 RAG、检索增强智能体全部基于 LangGraph 来演进，不会再用链式写法。

八、常见踩坑

1. 检索结果质量差

表现：检索回来的文档"看起来沾边但答非所问"。原因往往是：

切片太大或太小：太大塞不进 prompt 还稀释信号，太小丢失上下文。经验值 300~500 字符起步，按文档类型调。
Embedding 模型太弱：中文场景别用纯英文 embedding，建议 text-embedding-3-small 或国产 BGE/M3E。
Top-K 设置不合理：K 太小漏召回，太大塞噪声。先从 3~5 起步，配合后面的相关性评分过滤。

2. 上下文组装方式

新手常犯的错是把文档直接 str(docs) 拼进去，结果带了一堆 Document(page_content='...') 的字面量。正确做法是取 d.page_content，并加分隔符，必要时带上来源元数据 d.metadata 方便溯源。

3. 状态字段漏写导致 KeyError

节点返回的 dict 只需要包含"要更新的字段"，但消费方读的时候必须保证字段已存在。比如 generate 读 state["documents"]，如果某个分支跳过了 retrieve，就会 KeyError。解决办法：要么保证流程必经检索节点，要么用 state.get("documents", []) 防御。后续自适应 RAG 里会重点处理这个。

4. InMemoryVectorStore 不持久化

InMemoryVectorStore 进程一退出数据就没了，每次启动都要重新 embedding，费钱费时。它只适合 demo 和单测。生产环境请用 FAISS（本地文件）或 Chroma/Milvus（服务化）。

九、小结

RAG 的本质是"检索 + 生成"两步，LangGraph 用 StateGraph 把它建模成 retrieve → generate 两个节点。
相比 LangChain 的链式 RetrievalQA，LangGraph 的优势是状态显式、流程可控、可观测可扩展。
状态要包含 question / documents / answer；节点只返回要更新的字段。
抑制幻觉从 prompt 抓起：明确告诉模型"资料不足就承认不知道"。
这一篇是地基，后面的自适应 RAG、CRAG 都是在这张图上加节点和条件边。

下一篇我们让 RAG "聪明"起来：自适应 RAG ——让模型自己决定要不要检索、检索结果相不相关、回答要不要重写。

LangGraph 构建 RAG ​

一、为什么用 LangGraph 构建 RAG ​

二、基础 RAG 流程总览 ​

三、状态设计 ​

四、节点实现 ​

1. 检索节点 ​

2. 生成节点 ​

五、组装 StateGraph ​

六、完整可运行示例 ​

6.1 安装依赖 ​

6.2 完整代码 ​

七、与 LangChain RetrievalQA 对比 ​

八、常见踩坑 ​

1. 检索结果质量差 ​

2. 上下文组装方式 ​

3. 状态字段漏写导致 KeyError ​

4. InMemoryVectorStore 不持久化 ​

九、小结 ​

LangGraph 构建 RAG

一、为什么用 LangGraph 构建 RAG

二、基础 RAG 流程总览

三、状态设计

四、节点实现

1. 检索节点

2. 生成节点

五、组装 StateGraph

六、完整可运行示例

6.1 安装依赖

6.2 完整代码

七、与 LangChain RetrievalQA 对比

八、常见踩坑

1. 检索结果质量差

2. 上下文组装方式

3. 状态字段漏写导致 KeyError

4. InMemoryVectorStore 不持久化

九、小结