Corrective-RAG 原理与实现：提升大模型检索生成鲁棒性

Corrective Retrieval Augmented Generation (CRAG) 旨在提升生成的鲁棒性，其核心是通过轻量级检索评估器来区分和触发三种不同的知识检索操作。借助 Web 搜索的扩展和优化知识利用，CRAG 显著增强了自动自我纠正的能力，并有效地利用检索到的文档信息。实验结果广泛证明了 CRAG 对 RAG 方法的适应性，以及在短格式和长格式生成任务中的泛化能力。

LangChain 实现细节

虽然 CRAG 主要从纠错的角度对 RAG 框架进行改进，并能与各种 RAG 方法无缝结合，但仍需要对外部检索评估器进行微调。未来的研究将集中于如何淘汰这一外部评估器，为 LLMs 配备更强大的检索评估能力，以进一步提升系统的智能性和性能。

LangChain 框架实现了 CRAG 的应用，其中 LangChain 框架被用来处理检索增强生成（RAG）的复杂流程，LangGraph 则是用于从头构建图工作流的工具。在 CRAG 的操作流程中，如果至少有一篇文档的相关性超过预设阈值，那么系统就会继续生成响应。在生成之前，还会执行知识细化步骤，将文档分割为'知识片段'，对每个片段进行评分，并过滤掉不相关的内容。

如果所有文档的相关性都低于阈值，或者评估器无法确定相关性，系统将寻求额外的数据源进行补充。这时，CRAG 会使用网络搜索来增强原有的检索结果，从而提高信息的全面性和准确性。

在实现过程中，一些步骤可以被简化或调整。例如，初次尝试时可以跳过知识细化阶段，如果需要，可以在后续版本中作为独立节点添加回去。当某些文档被判定为不相关时，可以选择通过网络搜索来补充检索，优化查询以获得更相关的结果。

关键代码实现

以下是基于 LangChain 和 LangGraph 的关键代码实现，展示了状态管理、检索、评估及决策逻辑。

from langchain.schema import Document
from typing import Dict, Any

# 定义检索节点
def retrieve(state: Dict[str, Any]) -> Dict[str, Any]:
    """
    Retrieve documents from the vector store.
    
    Args:
        state (dict): The current graph state containing 'question'.
    Returns:
        dict: New key added to state, 'documents', that contains retrieved documents.
    """
    print("---RETRIEVE---")
    question = state["question"]
    # Retrieval logic here
    documents = retriever.get_relevant_documents(question)
    return {"documents": documents, "question": question}

# 定义生成节点
def generate(state: Dict[str, Any]) -> Dict[str, Any]:
    """
    Generate answer using LLM and context.
    
    Args:
        state (dict): The current graph state.
    Returns:
        dict: New key added to state, 'generation', that contains LLM generation.
    """
    print("---GENERATE---")
    question = state["question"]
    documents = state["documents"]
    # RAG generation logic
    generation = rag_chain.invoke({"context": documents, "question": question})
    return {"documents": documents, "question": question, "generation": generation}

# 定义文档相关性评估节点
def grade_documents(state: Dict[str, Any]) -> Dict[str, Any]:
    """
    Determines whether the retrieved documents are relevant to the question.
    
    Args:
        state (dict): The current graph state.
    Returns:
        dict: Updates documents key with only filtered relevant documents.
    """
    print("---CHECK DOCUMENT RELEVANCE TO QUESTION---")
    question = state["question"]
    documents = state["documents"]
    filtered_docs = []
    web_search_needed = False
    
    for d in documents:
        score = retrieval_grader.invoke({
            "question": question, 
            "document": d.page_content
        })
        grade = score.binary_score
        if grade == "yes":
            print("---GRADE: DOCUMENT RELEVANT---")
            filtered_docs.append(d)
        else:
            print("---GRADE: DOCUMENT NOT RELEVANT---")
            web_search_needed = True
            continue
    
    return {
        "documents": filtered_docs, 
        "question": question, 
        "web_search": "Yes" if web_search_needed else "No"
    }

# 定义查询转换节点
def transform_query(state: Dict[str, Any]) -> Dict[str, Any]:
    """
    Transform the query to produce a better question for web search.
    
    Args:
        state (dict): The current graph state.
    Returns:
        dict: Updates question key with a re-phrased question.
    """
    print("---TRANSFORM QUERY---")
    question = state["question"]
    documents = state["documents"]
    # Re-write question for broader search
    better_question = question_rewriter.invoke({"question": question})
    return {"documents": documents, "question": better_question}

# 定义网络搜索节点
def web_search(state: Dict[str, Any]) -> Dict[str, Any]:
    """
    Web search based on the re-phrased question.
    
    Args:
        state (dict): The current graph state.
    Returns:
        dict: Updates documents key with appended web results.
    """
    print("---WEB SEARCH---")
    question = state["question"]
    documents = state["documents"]
    # Web search execution
    docs = web_search_tool.invoke({"query": question})
    web_results = "\n".join([d["content"] for d in docs])
    web_doc = Document(page_content=web_results)
    documents.append(web_doc)
    return {"documents": documents, "question": question}

# 定义决策边缘逻辑
def decide_to_generate(state: Dict[str, Any]) -> str:
    """
    Determines whether to generate an answer, or re-generate a question.
    
    Args:
        state (dict): The current graph state.
    Returns:
        str: Binary decision for next node to call ('generate' or 'transform_query').
    """
    print("---ASSESS GRADED DOCUMENTS---")
    web_search = state["web_search"]
    
    if web_search == "Yes":
        # All documents have been filtered check_relevance
        # We will re-generate a new query via web search
        print("---DECISION: ALL DOCUMENTS ARE NOT RELEVANT TO QUESTION, TRANSFORM QUERY---")
        return "transform_query"
    else:
        # We have relevant documents, so generate answer
        print("---DECISION: GENERATE---")
        return "generate"

CRAG 通过这些步骤，显著提升了系统在复杂信息环境下的鲁棒性和灵活性，增强了对检索文档的有效利用。项目开源地址为 https://github.com/HuskyInSalt/CRAG，该项目运行需要 Python 3.11 环境。

总结与展望

CRAG 代表了检索增强生成技术的一个重要演进方向，即从被动检索转向主动纠错。通过引入自我反思与自我评估机制，它有效缓解了传统 RAG 在面对噪声数据时的脆弱性。在实际生产环境中部署 CRAG 时，建议关注评估器的准确率调优以及 Web 搜索的成本控制。未来随着多模态检索技术的发展，CRAG 有望进一步融合图像、视频等多源信息，构建更加全面的知识问答系统。