构建LangChain应用程序的示例代码：38、自主RAG的概念及其实现方法，使用LangChain和OpenAI工具从头开始构建一个结合检索和生成的系统

# 安装必要的库
! pip install langchain_community tiktoken langchain-openai langchainhub chromadb langchain langgraph

自主RAG (Self-RAG)

自主RAG是最近的一篇论文，介绍了一种用于主动RAG的有趣方法。

该框架训练单个任意的语言模型（如LLaMA2-7b, 13b）来生成控制RAG过程的标记：

是否从检索器检索 - Retrieve
- 标记：Retrieve
- 输入：x (问题) 或 x (问题), y (生成的回答)
- 决定何时使用R检索D个片段
- 输出：yes, no, continue
检索到的片段D是否与问题x相关 - ISREL
- 标记：ISREL
- 输入：(x (问题), d (片段)) 对于每个d在D中
- d提供解决x的有用信息
- 输出：relevant, irrelevant
每个片段D生成的LLM回答是否与片段相关（如幻觉等） - ISSUP
- 标记：ISSUP
- 输入：x (问题), d (片段), y (生成的回答) 对于每个d在D中
- 在y (生成的回答)中所有需要验证的陈述都由d支持
- 输出：fully supported, partially supported, no support
每个片段D生成的LLM回答是否对x (问题)有用 - ISUSE
- 标记：ISUSE
- 输入：x (问题), y (生成的回答) 对于每个d在D中
- y (生成的回答)是否对x (问题)有用
- 输出：{5, 4, 3, 2, 1}

我们可以将其表示为一个图：

在这里插入图片描述

论文链接：https://arxiv.org/abs/2310.11511

让我们使用LangGraph从头开始实现这个过程。

检索器 (Retriever)

让我们索引三个博客文章。

from langchain.text_splitter import RecursiveCharacterTextSplitter  # 从LangChain导入递归字符文本分割器
from langchain_community.document_loaders import WebBaseLoader      # 从LangChain Community导入网页基础加载器
from langchain_community.vectorstores import Chroma                # 从LangChain Community导入Chroma向量存储
from langchain_openai import OpenAIEmbeddings                      # 从LangChain OpenAI导入OpenAI嵌入# 定义需要索引的博客文章URL列表
urls = ["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/",
]# 加载每个URL的文档内容
docs = [WebBaseLoader(url).load() for url in urls]
# 将嵌套的文档列表展开为单个列表
docs_list = [item for sublist in docs for item in sublist]# 使用递归字符文本分割器将文档分割成大小为250字符的块，且块之间没有重叠
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(),    # 使用OpenAI嵌入
)# 将向量存储转化为检索器
retriever = vectorstore.as_retriever()

状态

我们将定义一个图。

我们的状态将是一个字典。

我们可以从任何图节点访问它，使用 state[‘keys’]。

from typing import Dict, TypedDictfrom langchain_core.messages import BaseMessageclass GraphState(TypedDict):"""Represents the state of an agent in the conversation.Attributes:keys: A dictionary where each key is a string and the value is expected to be a list or another structurethat supports addition with `operator.add`. This could be used, for instance, to accumulate messagesor other pieces of data throughout the graph."""keys: Dict[str, any]

节点和边

每个节点将简单地修改状态。

每条边将选择下一个要调用的节点。

我们可以将自助RAG表示为一个图：
在这里插入图片描述

import json
import operator
from typing import Annotated, Sequence, TypedDictfrom langchain import hub
from langchain.output_parsers import PydanticOutputParser
from langchain.output_parsers.openai_tools import PydanticToolsParser
from langchain.prompts import PromptTemplate
from langchain_community.vectorstores import Chroma
from langchain_core.messages import BaseMessage, FunctionMessage
from langchain_core.output_parsers import StrOutputParser
from langchain_core.pydantic_v1 import BaseModel, Field
from langchain_core.runnables import RunnablePassthrough
from langchain_core.utils.function_calling import convert_to_openai_tool
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langgraph.prebuilt import ToolInvocation### 节点 ###def retrieve(state):"""检索文档参数:state (dict): 代理当前的状态，包括所有键。返回:dict: 向状态添加新键，documents，包含检索到的文档。"""print("---RETRIEVE---")state_dict = state["keys"]question = state_dict["question"]documents = retriever.invoke(question)return {"keys": {"documents": documents, "question": question}}def generate(state):"""生成答案参数:state (dict): 代理当前的状态，包括所有键。返回:dict: 向状态添加新键，generation，包含生成的答案。"""print("---GENERATE---")state_dict = state["keys"]question = state_dict["question"]documents = state_dict["documents"]# 提示模板prompt = hub.pull("rlm/rag-prompt")# LLM模型llm = ChatOpenAI(model="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": documents, "question": question})return {"keys": {"documents": documents, "question": question, "generation": generation}}def grade_documents(state):"""确定检索到的文档是否与问题相关。参数:state (dict): 代理当前的状态，包括所有键。返回:dict: 向状态添加新键，filtered_documents，包含相关文档。"""print("---CHECK RELEVANCE---")state_dict = state["keys"]question = state_dict["question"]documents = state_dict["documents"]# 数据模型class grade(BaseModel):"""相关性检查的二进制评分。"""binary_score: str = Field(description="相关性评分 'yes' 或 'no'")# LLM模型model = ChatOpenAI(temperature=0, model="gpt-4-0125-preview", streaming=True)# 工具grade_tool_oai = convert_to_openai_tool(grade)# LLM与工具绑定并强制调用llm_with_tool = model.bind(tools=[convert_to_openai_tool(grade_tool_oai)],tool_choice={"type": "function", "function": {"name": "grade"}},)# 解析器parser_tool = PydanticToolsParser(tools=[grade])# 提示模板prompt = PromptTemplate(template="""你是一个评估员，正在评估检索到的文档与用户问题的相关性。\n 这是检索到的文档：\n\n {context} \n\n这是用户问题：{question} \n如果文档包含与用户问题相关的关键词或语义，请将其评为相关。\n给出一个二进制评分 'yes' 或 'no'，表示文档是否与问题相关。""",input_variables=["context", "question"],)# 链chain = prompt | llm_with_tool | parser_tool# 评分filtered_docs = []for d in documents:score = chain.invoke({"question": question, "context": d.page_content})grade = score[0].binary_scoreif grade == "yes":print("---GRADE: DOCUMENT RELEVANT---")filtered_docs.append(d)else:print("---GRADE: DOCUMENT NOT RELEVANT---")continuereturn {"keys": {"documents": filtered_docs, "question": question}}def transform_query(state):"""转换查询以生成更好的问题。参数:state (dict): 代理当前的状态，包括所有键。返回:dict: 保存新问题到state。"""print("---TRANSFORM QUERY---")state_dict = state["keys"]question = state_dict["question"]documents = state_dict["documents"]# 创建一个包含格式指令和查询的提示模板prompt = PromptTemplate(template="""你正在生成针对检索进行优化的问题。\n 查看输入并尝试推理其潜在的语义意图。\n 这是初始问题：\n ------- \n{question} \n ------- \n形成一个改进的问题：""",input_variables=["question"],)# 评分模型model = ChatOpenAI(temperature=0, model="gpt-4-0125-preview", streaming=True)# 提示模板chain = prompt | model | StrOutputParser()better_question = chain.invoke({"question": question})return {"keys": {"documents": documents, "question": better_question}}def prepare_for_final_grade(state):"""准备进行最终评分，状态透传。参数:state (dict): 代理当前的状态，包括所有键。返回:state (dict): 代理当前的状态，包括所有键。"""print("---FINAL GRADE---")state_dict = state["keys"]question = state_dict["question"]documents = state_dict["documents"]generation = state_dict["generation"]return {"keys": {"documents": documents, "question": question, "generation": generation}}### 边 ###def decide_to_generate(state):"""决定是生成答案还是重新生成问题。参数:state (dict): 代理当前的状态，包括所有键。返回:dict: 向状态添加新键，filtered_documents，包含相关文档。"""print("---DECIDE TO GENERATE---")state_dict = state["keys"]question = state_dict["question"]filtered_documents = state_dict["documents"]if not filtered_documents:# 所有文档在检查相关性时都被过滤掉了# 我们将重新生成一个新查询print("---DECISION: TRANSFORM QUERY---")return "transform_query"else:# 我们有相关文档，所以生成答案print("---DECISION: GENERATE---")return "generate"def grade_generation_v_documents(state):"""确定生成的回答是否基于文档。参数:state (dict): 代理当前的状态，包括所有键。返回:str: 二进制决策评分。"""print("---GRADE GENERATION vs DOCUMENTS---")state_dict = state["keys"]question = state_dict["question"]documents = state_dict["documents"]generation = state_dict["generation"]# 数据模型class grade(BaseModel):"""相关性检查的二进制评分。"""binary_score: str = Field(description="支持评分 'yes' 或 'no'")# LLM模型model = ChatOpenAI(temperature=0, model="gpt-4-0125-preview", streaming=True)# 工具grade_tool_oai = convert_to_openai_tool(grade)# LLM与工具绑定并强制调用llm_with_tool = model.bind(tools=[convert_to_openai_tool(grade_tool_oai)],tool_choice={"type": "function", "function": {"name": "grade"}},)# 解析器parser_tool = PydanticToolsParser(tools=[grade])# 提示模板prompt = PromptTemplate(template="""你是一个评估员，正在评估答案是否基于/支持一组事实。\n 这里是事实：\n ------- \n{documents} \n ------- \n这是答案：{generation}给出一个二进制评分 'yes' 或 'no'，表示答案是否基于/支持一组事实。""",input_variables=["generation", "documents"],)# 链chain = prompt | llm_with_tool | parser_toolscore = chain.invoke({"generation": generation, "documents": documents})grade = score[0].binary_scoreif grade == "yes":print("---DECISION: SUPPORTED, MOVE TO FINAL GRADE---")return "supported"else:print("---DECISION: NOT SUPPORTED, GENERATE AGAIN---")return "not supported"def grade_generation_v_question(state):"""确定生成的回答是否解决了问题。参数:state (dict): 代理当前的状态，包括所有键。返回:str: 二进制决策评分。"""print("---GRADE GENERATION vs QUESTION---")state_dict = state["keys"]question = state_dict["question"]documents = state_dict["documents"]generation = state_dict["generation"]# 数据模型class grade(BaseModel):"""相关性检查的二进制评分。"""binary_score: str = Field(description="有用评分 'yes' 或 'no'")# LLM模型model = ChatOpenAI(temperature=0, model="gpt-4-0125-preview", streaming=True)# 工具grade_tool_oai = convert_to_openai_tool(grade)# LLM与工具绑定并强制调用llm_with_tool = model.bind(tools=[convert_to_openai_tool(grade_tool_oai)],tool_choice={"type": "function", "function": {"name": "grade"}},)# 解析器parser_tool = PydanticToolsParser(tools=[grade])# 提示模板prompt = PromptTemplate(template="""你是一个评估员，正在评估答案是否有助于解决问题。\n 这是答案：\n ------- \n{generation} \n ------- \n这是问题：{question}给出一个二进制评分 'yes' 或 'no'，表示答案是否有助于解决问题。""",input_variables=["generation", "question"],)# 提示模板chain = prompt | llm_with_tool | parser_toolscore = chain.invoke({"generation": generation, "question": question})grade = score[0].binary_scoreif grade == "yes":print("---DECISION: USEFUL---")return "useful"else:print("---DECISION: NOT USEFUL---")return "not useful"

Graph

import pprintfrom langgraph.graph import END, StateGraph# 定义工作流状态图
workflow = 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("prepare_for_final_grade", prepare_for_final_grade)  # 准备最终评分，透传状态# 构建图
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", "retrieve")  # 转换查询后返回检索节点
workflow.add_conditional_edges("generate",grade_generation_v_documents,{"supported": "prepare_for_final_grade",  # 生成答案后，根据决定转到准备最终评分或重新生成答案"not supported": "generate",},
)
workflow.add_conditional_edges("prepare_for_final_grade",grade_generation_v_question,{"useful": END,  # 准备最终评分后，根据决定结束或返回转换查询"not useful": "transform_query",},
)# 编译工作流
app = workflow.compile()

# 导入 pprint 模块，用于美化打印输出
import pprint# 定义输入，包含一个问题
inputs = {"keys": {"question": "Explain how the different types of agent memory work?"}}# 运行编译好的工作流应用
for output in app.stream(inputs):# 遍历每个节点的输出for key, value in output.items():# 打印节点名称pprint.pprint(f"Output from node '{key}':")pprint.pprint("---")# 打印输出的详细信息pprint.pprint(value["keys"], indent=2, width=80, depth=None)# 打印分隔符，区分不同节点的输出pprint.pprint("\n---\n")

import pprint# 定义输入，包含一个问题
inputs = {"keys": {"question": "Explain how chain of thought prompting works?"}}# 运行编译好的工作流应用
for output in app.stream(inputs):# 遍历每个节点的输出for key, value in output.items():# 打印节点名称pprint.pprint(f"Output from node '{key}':")pprint.pprint("---")# 打印输出的详细信息pprint.pprint(value["keys"], indent=2, width=80, depth=None)# 打印分隔符，区分不同节点的输出pprint.pprint("\n---\n")