自我反思检索

Self-RAG 教程（基于 LangGraph）

本教程将指导你使用 LangGraph 构建一个 Self-RAG（检索增强生成） 系统。该系统结合了自我反思和自我评分机制，对检索到的文档和生成的回答进行评估。以下是实现该系统的详细步骤。

1. 环境准备与依赖安装

首先，安装所需的库：

bash
pip install -U langchain_community tiktoken langchain-openai langchainhub chromadb

2. 检索器设置

检索器使用 WebBaseLoader 加载博客文章，并使用 Chroma 向量存储对文档进行索引。文档会被分割成块以便高效检索。

python
from langchain_community.document_loaders import WebBaseLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings

# 加载博客文章
loader = WebBaseLoader("https://lilianweng.github.io/posts/2023-06-23-agent/")
data = loader.load()

# 将文档分割成块
text_splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=0)
splits = text_splitter.split_documents(data)

# 创建向量存储
vectorstore = Chroma.from_documents(documents=splits, embedding=OpenAIEmbeddings())
retriever = vectorstore.as_retriever()

3. 自我评分与反思机制

Self-RAG 的核心是 自我评分 和 自我反思。我们使用 LLM（如 GPT-4）对检索到的文档和生成的回答进行评分。

3.1 文档评分

python
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.messages import HumanMessage
from langchain_openai import ChatOpenAI

# 定义文档评分提示
grade_prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个评分助手，负责评估检索到的文档是否与问题相关。"),
    ("human", "问题：{question}\n文档：{document}\n请评分（1-5）：")
])

# 初始化 LLM
llm = ChatOpenAI(model="gpt-4")

def grade_documents(state):
    question = state["question"]
    documents = state["documents"]
    scores = []
    for doc in documents:
        score = llm.invoke(grade_prompt.format(question=question, document=doc))
        scores.append(score)
    return {"scores": scores}

3.2 回答反思

python
def reflect_response(state):
    generation = state["generation"]
    reflection = llm.invoke([HumanMessage(content=f"请反思以下回答的质量：{generation}")])
    return {"reflection": reflection}

4. 构建 LangGraph 工作流

使用 LangGraph 将检索、评分、生成和反思模块连接成一个完整的工作流。

python
from langgraph.graph import Graph

# 定义节点函数
def retrieve(state):
    question = state["question"]
    documents = retriever.get_relevant_documents(question)
    return {"documents": documents}

def generate(state):
    question = state["question"]
    documents = state["documents"]
    generation = llm.invoke([HumanMessage(content=f"问题：{question}\n文档：{documents}")])
    return {"generation": generation}

def transform_query(state):
    question = state["question"]
    rewritten_question = llm.invoke([HumanMessage(content=f"请重写以下问题：{question}")])
    return {"question": rewritten_question}

# 构建工作流
workflow = Graph()
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_edge("retrieve", "grade_documents")
workflow.add_edge("grade_documents", "generate")
workflow.add_edge("generate", "transform_query")
workflow.add_edge("transform_query", "retrieve")

app = workflow.compile()

# 执行工作流
question = "智能体系统的关键组件是什么？"
for output in app.stream({"question": question}):
    for key, value in output.items():
        print(f"节点：{key}, 输出：{value}")

5. 完整代码

以下是完整的 Self-RAG 实现代码：

python
from langchain_community.document_loaders import WebBaseLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.messages import HumanMessage
from langgraph.graph import Graph

# 加载博客文章
loader = WebBaseLoader("https://lilianweng.github.io/posts/2023-06-23-agent/")
data = loader.load()

# 将文档分割成块
text_splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=0)
splits = text_splitter.split_documents(data)

# 创建向量存储
vectorstore = Chroma.from_documents(documents=splits, embedding=OpenAIEmbeddings())
retriever = vectorstore.as_retriever()

# 初始化 LLM
llm = ChatOpenAI(model="gpt-4")

# 定义文档评分提示
grade_prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个评分助手，负责评估检索到的文档是否与问题相关。"),
    ("human", "问题：{question}\n文档：{document}\n请评分（1-5）：")
])

# 定义节点函数
def retrieve(state):
    question = state["question"]
    documents = retriever.get_relevant_documents(question)
    return {"documents": documents}

def grade_documents(state):
    question = state["question"]
    documents = state["documents"]
    scores = []
    for doc in documents:
        score = llm.invoke(grade_prompt.format(question=question, document=doc))
        scores.append(score)
    return {"scores": scores}

def generate(state):
    question = state["question"]
    documents = state["documents"]
    generation = llm.invoke([HumanMessage(content=f"问题：{question}\n文档：{documents}")])
    return {"generation": generation}

def transform_query(state):
    question = state["question"]
    rewritten_question = llm.invoke([HumanMessage(content=f"请重写以下问题：{question}")])
    return {"question": rewritten_question}

# 构建工作流
workflow = Graph()
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_edge("retrieve", "grade_documents")
workflow.add_edge("grade_documents", "generate")
workflow.add_edge("generate", "transform_query")
workflow.add_edge("transform_query", "retrieve")

app = workflow.compile()

# 执行工作流
question = "智能体系统的关键组件是什么？"
for output in app.stream({"question": question}):
    for key, value in output.items():
        print(f"节点：{key}, 输出：{value}")

总结

本教程实现了一个 Self-RAG 系统，结合了检索、生成、评分和反思功能。你可以根据实际需求调整工作流和提示词，以优化系统性能。

目录