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进行网页搜索。
我们将使用查询重写来优化网页搜索的查询。
环境
! pip install langchain_community tiktoken langchain-openai langchainhub chromadb langchain langgraph tavily-python
LLMs
import osos.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’— -CHECK DOCUMENT RELEVANCE TO QUESTION — -— -GRADE: DOCUMENT NOT RELEVANT — -— -GRADE: DOCUMENT NOT RELEVANT — -— -GRADE: DOCUMENT RELEVANT — -— -GRADE: DOCUMENT RELEVANT — -“Node ‘grade_documents’:”‘\n — -\n’— -ASSESS GRADED DOCUMENTS — -— -DECISION: ALL DOCUMENTS ARE NOT RELEVANT TO QUESTION, TRANSFORM QUERY — -— -TRANSFORM QUERY — -“Node ‘transform_query’:”‘\n — -\n’— -WEB SEARCH — -“Node ‘web_search_node’:”‘\n — -\n’— -GENERATE — -“Node ‘generate’:”‘\n — -\n’“Node ‘__end__’:”‘\n — -\n’(‘Agents possess short-term memory, which is utilized for in-context learning, ‘‘and long-term memory, allowing them to retain and recall vast amounts of ‘‘information over extended periods. Some experts also classify working memory ‘‘as a distinct type, although it can be considered a part of short-term ‘‘memory in many cases.’)
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’— -CHECK DOCUMENT RELEVANCE TO QUESTION — -— -GRADE: DOCUMENT NOT RELEVANT — -— -GRADE: DOCUMENT NOT RELEVANT — -— -GRADE: DOCUMENT NOT RELEVANT — -— -GRADE: DOCUMENT RELEVANT — -“Node ‘grade_documents’:”‘\n — -\n’— -ASSESS GRADED DOCUMENTS — -— -DECISION: ALL DOCUMENTS ARE NOT RELEVANT TO QUESTION, TRANSFORM QUERY — -— -TRANSFORM QUERY — -“Node ‘transform_query’:”‘\n — -\n’— -WEB SEARCH — -“Node ‘web_search_node’:”‘\n — -\n’— -GENERATE — -“Node ‘generate’:”‘\n — -\n’“Node ‘__end__’:”‘\n — -\n’(‘The AlphaCodium paper functions by proposing a code-oriented iterative flow ‘‘that involves repeatedly running and fixing generated code against ‘‘input-output tests. Its key mechanisms include generating additional data ‘‘like problem reflection and test reasoning to aid the iterative process, as ‘‘well as enriching the code generation process. AlphaCodium aims to improve ‘‘the performance of Large Language Models on code problems by following a ‘‘test-based, multi-stage approach.’)
