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

自主RAG (Self-RAG)

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

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

1. 是否从检索器检索 - Retrieve

   • 标记：Retrieve
   • 输入：x (问题) 或 x (问题), y (生成的回答)
   • 决定何时使用R检索D个片段
   • 输出：yes, no, continue

2. 检索到的片段D是否与问题x相关 - ISREL

   • 标记：ISREL
   • 输入：(x (问题), d (片段)) 对于每个d在D中
   • d提供解决x的有用信息
   • 输出：relevant, irrelevant

3. 每个片段D生成的LLM回答是否与片段相关（如幻觉等） - ISSUP

   • 标记：ISSUP
   • 输入：x (问题), d (片段), y (生成的回答) 对于每个d在D中
   • 在y (生成的回答)中所有需要验证的陈述都由d支持
   • 输出：fully supported, partially supported, no support

4. 每个片段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, TypedDict

from langchain_core.messages import BaseMessage


class 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 structure
              that supports addition with `operator.add`. This could be used, for instance, to accumulate messages
              or other pieces of data throughout the graph.
    """

    keys: Dict[str, any]

节点和边

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

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

我们可以将自助RAG表示为一个图：

import json
import operator
from typing import Annotated, Sequence, TypedDict

from 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_score
        if grade == "yes":
            print("---GRADE: DOCUMENT RELEVANT---")
            filtered_docs.append(d)
        else:
            print("---GRADE: DOCUMENT NOT RELEVANT---")
            continue

    return {"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_tool

    score = chain.invoke({"generation": generation, "documents": documents})
    grade = score[0].binary_score

    if 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_tool

    score = chain.invoke({"generation": generation, "question": question})
    grade = score[0].binary_score

    if grade == "yes":
        print("---DECISION: USEFUL---")
        return "useful"
    else:
        print("---DECISION: NOT USEFUL---")
        return "not useful"

Graph

import pprint

from 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")

扩展知识

1. RAG (Retrieval-Augmented Generation)

RAG是一种结合了检索和生成的技术，首先从一个大型文档集合中检索相关内容，然后使用生成模型基于这些内容生成回答。这种方法能够有效结合外部知识库和生成模型的能力，提高回答的准确性和信息丰富性。

2. LangChain

LangChain是一个用于构建语言模型应用的框架，支持多种文档加载、文本分割、嵌入计算和向量存储方式。它简化了从数据准备到模型应用的整个流程。

3. OpenAI

OpenAI提供了强大的语言模型如GPT-3和GPT-4，这些模型能够理解和生成自然语言文本，广泛应用于各种NLP任务，如问答、翻译、文本生成等。

4. Chroma

Chroma是一个高效的向量数据库，支持快速的向量检索，常用于结合嵌入技术进行相似性搜索。它在处理大型文档集合时表现出色。

总结

在本文中，我们介绍了自主RAG的概念及其实现方法，使用LangChain和OpenAI工具从头开始构建一个结合检索和生成的系统。通过对多个博客文章进行索引和处理，我们展示了如何利用这些工具进行高效的信息检索和回答生成。本文还补充了相关的技术知识，帮助读者更好地理解和应用这些技术。