CRAG通过引入检索评估器来评估检索到的文档与查询之间的关系，从而增强了传统的RAG。那么，让我们了解它的工作原理。
我会给你一个小型场景，假设我们正在参加一个开卷考试。
我们遵循的策略，比如说对于每个主题，我们查阅书籍并确定相关部分。在形成观点之前，将收集到的信息分为三组：正确，错误和模糊。分别处理每种类型的信息。然后，根据这些处理过的信息，进行编译和总结。在考试试卷上写下我们的答复。这就是CRAG所做的。
RAG的局限性/CRAG的动机：
上图说明了大多数传统RAG方法不考虑文档与问题的关联性，只是简单地合并检索到的文档。这可能会引入不相关信息，阻碍模型获取准确知识，并可能导致幻觉问题。
此外，大多数传统RAG方法将整个检索到的文档作为输入。然而，这些检索到的文档中的大部分文本通常对生成并不必要，也不应该在RAG中同等地参与。
CRAG的关键思想
有3种可能的判断结果。
如果是**
正确**，这意味着检索到的文档包含了查询所需的必要内容，然后使用知识提炼算法重写检索到的文档。如果检索到的文档是**
错误**的，这意味着查询和检索到的文档是不相关的。因此，我们不能将文档发送给LLM。在CRAG中，使用网页搜索引擎检索外部知识。对于**
模糊**的情况，这意味着检索到的文档可能接近但不足以提供答案。在这种情况下，需要通过网页搜索获取额外的信息。因此，既使用知识提炼算法也使用搜索引擎。
最后，处理过的信息被转发到LLM以生成响应。
CRAG的三个关键组成部分：检索评估器，知识提炼算法和知识搜索
检索评估器
检索评估器显著影响后续程序的结果，对确定整个系统的性能至关重要。
CRAG使用一个轻量级的T5-large模型作为检索评估器并对其进行微调。值得注意的是，在大型语言模型的时代，T5-large也被认为是轻量级的。
对于每个查询，通常检索十个文档。然后将查询与每个文档单独连接作为输入，以预测它们的相关性。在微调过程中，为正面样本分配标签1，为负面样本分配-1。在推理过程中，评估器为每个文档分配一个相关性得分，范围从-1到1。
这些得分将根据阈值被分为三个级别。显然，需要两个阈值来进行这种分类。在CRAG中，阈值设置可能会根据实验数据而有所不同：
知识提炼算法
对于检索到的相关文档，CRAG设计了一种分解然后重新组合的知识提取方法，以进一步提取最关键的知识陈述。
首先，应用启发式规则将每个文档分解为细粒度的知识条，目的是获得细粒度的结果。如果检索到的文档只包含一个或两个句子，则被视为独立单元。否则，根据总长度将文档划分为更小的单元，通常由几个句子组成。每个单元都应包含一个独立的信息。
接下来，使用检索评估器计算每个知识条的相关性得分。过滤掉相关性得分低的条。然后将剩余的相关知识条重新组合，以形成内部知识。
知识搜索
当一个块被分类为模糊或错误时，将应用此功能。当发现一个块是不相关的时，我们将其丢弃，并使用网页搜索API从互联网上找到相关结果。所以，我们不是使用错误的块，而是使用互联网上的来源进行最终答案生成。
然而，在模糊的情况下，我们同时应用知识提炼和搜索。过滤掉不相关的条，添加来自互联网的新信息。最终连接的块被发送到LLM以生成答案。
CRAG是开源的，Langchain和LlamaIndex都提供了自己的实现。
我们将使用LangGraph从头开始实现这些想法：
什么是LangGraph？
LangGraph是LangChain生态系统的扩展。LangGraph允许我们以图的形式构建AI应用程序，包括代理和RAG。它将工作流程视为循环图结构，其中每个节点代表一个函数或一个Langchain可运行对象，边是节点之间的连接。它还提供了一个有状态的解决方案，其中全局状态对象可以在节点之间共享。
LangGraph的主要特点包括：
节点：任何函数或Langchain可运行对象，如工具。
边：定义节点之间的方向。
有状态的图：主要类型的图。它旨在在通过节点处理数据时管理和更新状态对象。
我们将使用LangGraph构建我们的纠正RAG管道。
让我们开始：
作为第一次尝试，让我们跳过知识提炼阶段。如果需要，可以将其作为节点添加回来。
如果任何文档不相关，让我们选择用网页搜索来补充检索。
我们将使用Tavily Search进行网页搜索。
我们将使用查询重写来优化网页搜索的查询。
环境
!pipinstalllangchain_communitytiktokenlangchain-openailangchainhubchromadblangchainlanggraphtavily-python
LLMs
importosos.environ['OPENAI_API_KEY']=""
搜索
我们将使用Tavily Search进行网页搜索。
os.environ['TAVILY_API_KEY']=""
跟踪
使用LangSmith进行跟踪
os.environ['LANGCHAIN_TRACING_V2']='true'os.environ['LANGCHAIN_ENDPOINT']='https://api.smith.langchain.com'os.environ['LANGCHAIN_API_KEY']=""
索引
让我们索引3篇博客文章。
from langchain.text_splitter import RecursiveCharacterTextSplitterfrom langchain_community.document_loaders import WebBaseLoaderfrom langchain_community.vectorstores import Chromafrom langchain_openai import OpenAIEmbeddingsurls = ["https://lilianweng.github.io/posts/2023-06-23-agent/","https://lilianweng.github.io/posts/2023-03-15-prompt-engineering/","https://lilianweng.github.io/posts/2023-10-25-adv-attack-llm/",]docs = [WebBaseLoader(url).load() for url in urls]docs_list = [item for sublist in docs for item in sublist]text_splitter = RecursiveCharacterTextSplitter.from_tiktoken_encoder(chunk_size=250, chunk_overlap=0)doc_splits = text_splitter.split_documents(docs_list)# 添加到向量数据库vectorstore = Chroma.from_documents(documents=doc_splits,collection_name="rag-chroma",embedding=OpenAIEmbeddings(),)retriever = vectorstore.as_retriever()
LLMs
### 检索评分器from langchain_openai import ChatOpenAIfrom langchain_core.prompts import ChatPromptTemplatefrom langchain_core.pydantic_v1 import BaseModel, Field# 数据模型class GradeDocuments(BaseModel):"""对检索到的文档与用户问题的相关性进行二元评分检查的模型。"""binary_score: str = Field(description="文档与问题相关，'是'或'否'")# 带功能调用的LLMllm = ChatOpenAI(model="gpt-3.5-turbo-0125", temperature=0)structured_llm_grader = llm.with_structured_output(GradeDocuments)# 提示system = """你是评估检索到的文档与用户问题相关性的评分员。\n如果文档包含与问题相关的关键词或语义含义，将其评为相关。\n给出二元分数'是'或'否'，以指示文档是否与问题相关。"""grade_prompt = ChatPromptTemplate.from_messages([("system", system),("human", "检索到的文档：\n\n {document} \n\n 用户问题：{question}"),])retrieval_grader = grade_prompt | structured_llm_graderquestion = "agent memory"docs = retriever.get_relevant_documents(question)doc_txt = docs[1].page_contentprint(retrieval_grader.invoke({"question": question, "document": doc_txt}))
#binary_score='是'
### 生成from langchain import hubfrom langchain_core.output_parsers import StrOutputParser# 提示prompt = hub.pull("rlm/rag-prompt")# LLMllm = ChatOpenAI(model_name="gpt-3.5-turbo", temperature=0)# 后处理def format_docs(docs):return "\n\n".join(doc.page_content for doc in docs)# 链rag_chain = prompt | llm | StrOutputParser()# 运行generation = rag_chain.invoke({"context": docs, "question": question})print(generation)
#生成代理的设计结合了LLM与记忆、规划和反思机制，使代理能够根据过去的经验行为。记忆流是一个长期记忆模块，以自然语言记录代理经验的全面列表。短期记忆用于上下文学习，而长期记忆允许代理在较长时间内保留和回忆信息。
### 问题重写器# LLMllm = ChatOpenAI(model="gpt-3.5-turbo-0125", temperature=0)# 提示system = """你是一个将输入问题转换为优化的更好版本的问题的重写器，\n针对网页搜索进行优化。查看输入并尝试理解背后的语义意图/含义。"""re_write_prompt = ChatPromptTemplate.from_messages([("system", system),("human", "这是初始问题：\n\n {question} \n 制定一个改进后的问题。"),])question_rewriter = re_write_prompt | llm | StrOutputParser()question_rewriter.invoke({"question": question})
#人工智能代理中记忆的作用是什么？
网页搜索工具
### 搜索from langchain_community.tools.tavily_search import TavilySearchResultsweb_search_tool = TavilySearchResults(k=3)
图
将流程捕获为图。
图状态
from typing_extensions import TypedDictfrom typing import Listclass GraphState(TypedDict):"""表示我们的图的状态。属性：question: 问题generation: LLM生成web_search: 是否添加搜索documents: 文档列表"""question : strgeneration : strweb_search : strdocuments : List[str]
from langchain.schema import Documentdef retrieve(state):"""检索文档参数：state (dict): 当前图状态返回：state (dict): 向状态添加新键，文档，包含检索到的文档"""print("---检索---")question = state["question"]# 检索documents = retriever.get_relevant_documents(question)return {"documents": documents, "question": question}def generate(state):"""生成答案参数：state (dict): 当前图状态返回：state (dict): 向状态添加新键，生成，包含LLM生成"""print("---生成---")question = state["question"]documents = state["documents"]# RAG生成generation = rag_chain.invoke({"context": documents, "question": question})return {"documents": documents, "question": question, "generation": generation}def grade_documents(state):"""确定检索到的文档是否与问题相关。参数：state (dict): 当前图状态返回：state (dict): 更新文档键，只包含过滤后的相关文档"""print("---检查文档与问题的相关性---")question = state["question"]documents = state["documents"]# 为每个文档打分filtered_docs = []web_search = "否"for d in documents:score = retrieval_grader.invoke({"question": question, "document": d.page_content})grade = score.binary_scoreif grade == "是":print("---评分：文档相关---")filtered_docs.append(d)else:print("---评分：文档不相关---")web_search = "是"continuereturn {"documents": filtered_docs, "question": question, "web_search": web_search}def transform_query(state):"""转换查询以产生更好的问题。参数：state (dict): 当前图状态返回：state (dict): 更新问题键，用重述的问题更新"""print("---转换查询---")question = state["question"]documents = state["documents"]# 重写问题better_question = question_rewriter.invoke({"question": question})return {"documents": documents, "question": better_question}def web_search(state):"""基于重述的问题进行网页搜索。参数：state (dict): 当前图状态返回：state (dict): 更新文档键，附加网页结果"""print("---网页搜索---")question = state["question"]documents = state["documents"]# 网页搜索docs = web_search_tool.invoke({"query": question})web_results = "\n".join([d["content"] for d in docs])web_results = Document(page_content=web_results)documents.append(web_results)return {"documents": documents, "question": question}### 边def decide_to_generate(state):"""确定是否生成答案，或者重新生成问题。参数：state (dict): 当前图状态返回：str: 下一个节点调用的二元决策"""print("---评估评分文档---")question = state["question"]web_search = state["web_search"]filtered_documents = state["documents"]if web_search == "是":# 所有文档都已通过check_relevance过滤# 我们将重新生成新查询print("---决策：所有文档与问题不相关，转换查询---")return "transform_query"else:# 我们有相关文档，所以生成答案print("---决策：生成---")return "generate"
构建图
这只是按照我们上面图中概述的流程进行。
from langgraph.graph import END, StateGraphworkflow = StateGraph(GraphState)# 定义节点workflow.add_node("retrieve", retrieve)# 检索workflow.add_node("grade_documents", grade_documents)# 评分文档workflow.add_node("generate", generate)# 生成workflow.add_node("transform_query", transform_query)# 转换查询workflow.add_node("web_search_node", web_search)# 网页搜索# 构建图workflow.set_entry_point("retrieve")workflow.add_edge("retrieve", "grade_documents")workflow.add_conditional_edges("grade_documents",decide_to_generate,{"transform_query": "transform_query","generate": "generate",},)workflow.add_edge("transform_query", "web_search_node")workflow.add_edge("web_search_node", "generate")workflow.add_edge("generate", END)# 编译app = workflow.compile()
from pprint import pprint# 运行inputs = {"question": "What are the types of agent memory?"}for output in app.stream(inputs):for key, value in output.items():# 节点pprint(f"节点 '{key}':")# 可选：在每个节点打印完整状态# pprint.pprint(value["keys"], indent=2, width=80, depth=None)pprint("\n---\n")# 最终生成pprint(value["generation"])
#—-RETRIEVE—-“Node‘retrieve’:”‘\n—-\n’—-CHECKDOCUMENTRELEVANCETOQUESTION—-—-GRADEOCUMENTNOTRELEVANT—-—-GRADE
from pprint import pprint# 运行inputs = {"question": "How does the AlphaCodium paper work?"}for output in app.stream(inputs):for key, value in output.items():# 节点pprint(f"节点 '{key}':")# 可选：在每个节点打印完整状态# pprint.pprint(value["keys"], indent=2, width=80, depth=None)pprint("\n---\n")# 最终生成pprint(value["generation"])
#—-RETRIEVE—-“Node‘retrieve’:”‘\n—-\n’—-CHECKDOCUMENTRELEVANCETOQUESTION—-—-GRADE