自省式 RAG 与 LangGraph：基于状态机的高效实践
研究背景
大多数大型语言模型（LLMs）通常只针对大量公共数据进行周期性训练，它们往往缺少最新信息或不能接触到无法用于训练的私有数据。检索增强生成（RAG）模式恰好解决了这个问题，它通过将大型语言模型连接到外部数据源上。
然而，基础 RAG 流程存在局限性。在实际操作中，实现 RAG 需要对上述步骤进行逻辑分析：比如，我们需要知道什么时候进行检索（基于问题和索引的构成）、何时改写问题以提升检索效率，或者何时抛弃无关的检索结果并重新检索。
自省式 RAG
因此提出了自省式 RAG（Self-Correcting RAG）这一概念。自省式 RAG 利用大型语言模型自我校正检索质量不佳或生成内容不够优质的问题。
如上所展示的基础 RAG 流程，实质上是一种链式过程：大型语言模型根据检索到的文档来决定生成的内容。有些 RAG 运作模式采用的是路由机制，大型语言模型会根据提出的问题选择不同的检索器。但是自省式 RAG 通常需要某种反馈机制，比如重新生成问题或重新检索文档。这时候，状态机制作为第三种认知架构，因其能够支持循环操作而非常适用：状态机可定义一系列步骤（例如检索、评估文档、改写问题）并设置它们的转换逻辑；比如，如果我们检索到的文档无关，我们可以重新改写问题再进行检索。
利用 LangGraph 实现自省式 RAG
LangGraph 是一个简单的大型语言模型状态机实现工具。这为设计各种不同的 RAG 流程提供了极大的灵活性，并支持在 RAG 中进行所谓'流程工程'，即在具体的决策点（如：评估文档）和循环（比如：重新检索）中进行特定操作。
为了展现 LangGraph 的灵活性，我们利用它来实现了两篇引人入胜、前沿的自省式 RAG 论文，CRAG 和 Self-RAG 中提出的思想。
纠正式 RAG (CRAG)
纠正式 RAG（CRAG）在其论文中提出了以下鲜明的理念：
- 引入一种轻量级检索评估工具，用以对查询返回的文档进行整体质量评估，并为每项文档打分。
- 当检索的结果不明确或与用户的查询不够相关时，启用基于网络的文档检索来补充上下文。
- 执行知识细化：把检索的文档分成'知识条'，对每条进行评分，过滤出无关的内容。
在描述流程时，我们对一些步骤进行了简化和调整，方便理解（实际应用时可以进行相应的定制和扩展）：
- 我们省略了知识细化这一环节，尽管它代表了一个颇具价值的数据后处理方法，但在本文的流程示例中并不是必要的。
- 如果发现任意一个检索的文档不相关，我们将通过网络搜索来补充检索内容。在这里使用通用的网络搜索 API 来进行网络搜索，这既快速又方便。
- 我们还将改写查询语句，以便于网络搜索能提供更优的结果。
- 对于二选一的决策节点，我们用 Pydantic 来确定输出模型，并作为每一次执行大型语言模型的调用过程中运行的 OpenAI 工具函数。这让我们能够根据二选一逻辑确定何种逻辑路径。
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:
final_answer:
() -> RagState:
docs = vector_store.similarity_search(state[])
{: docs}
() -> RagState:
relevance = llm.evaluate_relevance(state[], state[])
{: relevance}
() -> RagState:
state[]:
new_docs = search_engine.search(state[])
state[].extend(new_docs)
state
() -> RagState:
prompt =
answer = llm.generate(prompt)
{: answer}
workflow = StateGraph(RagState)
workflow.add_node(, retrieve)
workflow.add_node(, evaluate_documents)
workflow.add_node(, web_search)
workflow.add_node(, generate_answer)
workflow.add_edge(START, )
workflow.add_edge(, )
workflow.add_conditional_edges(
,
x: x[] ,
{: , : }
)
workflow.add_edge(, )
workflow.add_edge(, END)
app = workflow.()
