自省式 RAG 与 LangGraph:基于状态机的高效实践
检索增强生成(RAG)常面临信息滞后或私有数据缺失问题。自省式 RAG 通过引入反馈机制,利用大模型自我校正检索质量与生成内容。探讨 CRAG 与 Self-RAG 两种主流方案,并展示如何利用 LangGraph 的状态机特性构建循环流程,实现动态检索评估与查询改写,从而提升回答准确性与系统鲁棒性。

检索增强生成(RAG)常面临信息滞后或私有数据缺失问题。自省式 RAG 通过引入反馈机制,利用大模型自我校正检索质量与生成内容。探讨 CRAG 与 Self-RAG 两种主流方案,并展示如何利用 LangGraph 的状态机特性构建循环流程,实现动态检索评估与查询改写,从而提升回答准确性与系统鲁棒性。

大多数大型语言模型(LLMs)通常只针对大量公共数据进行周期性训练,它们往往缺少最新信息或不能接触到无法用于训练的私有数据。检索增强生成(RAG)模式恰好解决了这个问题,它通过将大型语言模型连接到外部数据源上。
然而,基础 RAG 流程存在局限性。在实际操作中,实现 RAG 需要对上述步骤进行逻辑分析:比如,我们需要知道什么时候进行检索(基于问题和索引的构成)、何时改写问题以提升检索效率,或者何时抛弃无关的检索结果并重新检索。
因此提出了自省式 RAG(Self-Correcting RAG)这一概念。自省式 RAG 利用大型语言模型自我校正检索质量不佳或生成内容不够优质的问题。
如上所展示的基础 RAG 流程,实质上是一种链式过程:大型语言模型根据检索到的文档来决定生成的内容。有些 RAG 运作模式采用的是路由机制,大型语言模型会根据提出的问题选择不同的检索器。但是自省式 RAG 通常需要某种反馈机制,比如重新生成问题或重新检索文档。这时候,状态机制作为第三种认知架构,因其能够支持循环操作而非常适用:状态机可定义一系列步骤(例如检索、评估文档、改写问题)并设置它们的转换逻辑;比如,如果我们检索到的文档无关,我们可以重新改写问题再进行检索。
LangGraph 是一个简单的大型语言模型状态机实现工具。这为设计各种不同的 RAG 流程提供了极大的灵活性,并支持在 RAG 中进行所谓'流程工程',即在具体的决策点(如:评估文档)和循环(比如:重新检索)中进行特定操作。
为了展现 LangGraph 的灵活性,我们利用它来实现了两篇引人入胜、前沿的自省式 RAG 论文,CRAG 和 Self-RAG 中提出的思想。
纠正式 RAG(CRAG)在其论文中提出了以下鲜明的理念:
在描述流程时,我们对一些步骤进行了简化和调整,方便理解(实际应用时可以进行相应的定制和扩展):
from langgraph.graph import StateGraph, START, END
from typing import TypedDict, Annotated, List
import operator
# 定义状态
class RagState(TypedDict):
query: str
documents: List[str]
is_relevant: bool
final_answer: str
# 1. 检索节点
def retrieve(state: RagState) -> RagState:
# 调用向量数据库检索
docs = vector_store.similarity_search(state['query'])
return {"documents": docs}
# 2. 评估节点
def evaluate_documents(state: RagState) -> RagState:
# 使用 LLM 评估文档相关性
relevance = llm.evaluate_relevance(state['query'], state['documents'])
return {"is_relevant": relevance}
# 3. 网络搜索补充节点
def web_search(state: RagState) -> RagState:
if not state['is_relevant']:
# 调用搜索引擎补充信息
new_docs = search_engine.search(state['query'])
state['documents'].extend(new_docs)
return state
# 4. 生成节点
def generate_answer(state: RagState) -> RagState:
prompt = f"Query: {state['query']}\nContext: {state['documents']}"
answer = llm.generate(prompt)
return {"final_answer": answer}
# 构建图
workflow = StateGraph(RagState)
workflow.add_node("retrieve", retrieve)
workflow.add_node("evaluate", evaluate_documents)
workflow.add_node("web_search", web_search)
workflow.add_node("generate", generate_answer)
workflow.add_edge(START, "retrieve")
workflow.add_edge("retrieve", "evaluate")
workflow.add_conditional_edges(
"evaluate",
lambda x: "web_search" if not x["is_relevant"] else "generate",
{"web_search": "web_search", "generate": "generate"}
)
workflow.add_edge("web_search", "generate")
workflow.add_edge("generate", END)
app = workflow.compile()
自 RAG 是一个与之相望的解决方案,论文中提出了许多独到的 RAG 理念。框架中训练大型语言模型生成自我反思的提示符号,用以控制 RAG 流程的各个阶段。下面是提示符号的一览:
Retrieve 符号决定是否需要根据 x(问题) 或 x(问题)、y(回答) 检索 D 数据块。可能的输出结果有 yes, no, continue。ISREL 符号针对 x 问题,判断数据块 D 是否相关。输入为 (x(问题),d(数据块))。结果为 relevant(相关), irrelevant(不相关)。ISSUP 符号判断 D 中每个数据块生成的答复是否与之相关。输入包括 x, d, y。这个标记也是验证 d 是否支持 y(生成) 中的所有需要证实的陈述。可输出 fully supported(完全支持), partially supported(部分支持), no support(不支持)。ISUSE 符号评估 D 中每个数据块生成的答复是否对 x 有用。输入 x, y 对于 d 在 D 里。输出是 {5, 4, 3, 2, 1}。以下简图帮助我们理解信息流的运转机制:
我们可以在 LangGraph 中对其进行实现,为了说明需要进行了一些简化和调整(在实际需求中可以进行相应的定制和扩展):
展示的示例轨迹强调了主动 RAG 的自我纠正能力。查询的问题是 解释不同类型代理记忆是如何工作的?。在此示例中,所有四个文档都被认为相关,对照文档检查生成答案的环节顺利通过,但生成的答案未被认定完全有用。
之后,如所示,循环重新开始,问题稍微改写为:不同类型代理记忆的运作方式如何?。此时,四份文档中有一份因为无关而被筛选出去。之后的生成答案成功通过了所有检查:
不同类型的代理记忆包含感官记忆、短期记忆和长期记忆。感官记忆能够保留短暂的感觉信息。短期记忆则被用于实时学习和构建提示。而长期记忆则让代理人可以在很长的时间里保存和回忆信息,并常常依赖外部的向量存储来实现。
整体流程轨迹清晰可见,可以容易地进行审核。
自省机制可以显著提升 RAG 的功能,允许改正检索和生成过程中的质量问题。几篇最新的 RAG 论文都着重讨论了这一主题,但要将这些理念实际应用起来有着不小的难度。本文展示了如何利用 LangGraph 进行'流程工程化'地实施自反式 RAG。通过状态机的循环特性,系统能够在检索失败时自动触发修正策略,从而在保证准确性的同时优化资源消耗。

微信公众号「极客日志」,在微信中扫描左侧二维码关注。展示文案:极客日志 zeeklog
使用加密算法(如AES、TripleDES、Rabbit或RC4)加密和解密文本明文。 在线工具,加密/解密文本在线工具,online
生成新的随机RSA私钥和公钥pem证书。 在线工具,RSA密钥对生成器在线工具,online
基于 Mermaid.js 实时预览流程图、时序图等图表,支持源码编辑与即时渲染。 在线工具,Mermaid 预览与可视化编辑在线工具,online
解析常见 curl 参数并生成 fetch、axios、PHP curl 或 Python requests 示例代码。 在线工具,curl 转代码在线工具,online
将字符串编码和解码为其 Base64 格式表示形式即可。 在线工具,Base64 字符串编码/解码在线工具,online
将字符串、文件或图像转换为其 Base64 表示形式。 在线工具,Base64 文件转换器在线工具,online