评估 RAG 应用程序

import os
os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_API_KEY"] = "YOUR LANGSMITH API KEY"
os.environ["OPENAI_API_KEY"] = "YOUR OPENAI API KEY"

pip install -U langsmith langchain[openai] langchain-community

from langchain_community.document_loaders import WebBaseLoader
from langchain_core.vectorstores import InMemoryVectorStore
from langchain_openai import OpenAIEmbeddings
from langchain_text_splitters import RecursiveCharacterTextSplitter

# List of URLs to load documents from
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/",
]

# Load documents from the URLs
docs = [WebBaseLoader(url).load() for url in urls]
docs_list = [item for sublist in docs for item in sublist]

# Initialize a text splitter with specified chunk size and overlap
text_splitter = RecursiveCharacterTextSplitter.from_tiktoken_encoder(
    chunk_size=250, chunk_overlap=0
)

# Split the documents into chunks
doc_splits = text_splitter.split_documents(docs_list)

# Add the document chunks to the "vector store" using OpenAIEmbeddings
vectorstore = InMemoryVectorStore.from_documents(
    documents=doc_splits,
    embedding=OpenAIEmbeddings(),
)

# With langchain we can easily turn any vector store into a retrieval component:
retriever = vectorstore.as_retriever(k=6)

from langchain_openai import ChatOpenAI
from langsmith import traceable

llm = ChatOpenAI(model="gpt-4o", temperature=1)

# 添加装饰器以便在 LangSmith 中跟踪此函数
@traceable()
def rag_bot(question: str) -> dict:
    # LangChain 检索器会自动跟踪
    docs = retriever.invoke(question)
    docs_string = "".join(doc.page_content for doc in docs)
    instructions = f"""你是一个擅长分析参考资料并回答问题的智能助理。
       使用以下源文档回答用户的问题。
       如果你不知道答案，就直接说明不知道。
       最多使用三句话，并保持回答简洁。

文档：
{docs_string}"""
    # LangChain ChatModel 会自动被 traced
    ai_msg = llm.invoke([
            {"role": "system", "content": instructions},
            {"role": "user", "content": question},
        ],
    )
    return {"answer": ai_msg.content, "documents": docs}

from langsmith import Client

client = Client()

# 为数据集定义示例
examples = [
    {
        "inputs": {"question": "ReAct 智能体如何利用自我反思？"},
        "outputs": {"answer": "ReAct 将推理与行动融合在一起：先调用诸如 Wikipedia 搜索 API 等工具执行动作，再对工具输出进行观察与推理。"},
    },
    {
        "inputs": {"question": "少样本提示会产生哪些偏差类型？"},
        "outputs": {"answer": "少样本提示可能出现的偏差包括： (1) 多数标签偏差、(2) 近期偏差、(3) 常见 token 偏差。"},
    },
    {
        "inputs": {"question": "常见的五种对抗攻击类型是什么？"},
        "outputs": {"answer": "常见的五种对抗攻击包括： (1) Token 篡改、(2) 基于梯度的攻击、(3) Jailbreak 提示、(4) 人工红队测试、(5) 模型红队测试。"},
    },
]

# 在 LangSmith 中创建数据集和示例
dataset_name = "Lilian Weng 博客问答"
dataset = client.create_dataset(dataset_name=dataset_name)
client.create_examples(
    dataset_id=dataset.id,
    examples=examples
)

from typing_extensions import Annotated, TypedDict

# 评分输出模式
class CorrectnessGrade(TypedDict):
    # 字段定义顺序决定了模型生成字段的顺序。
    # 先让模型给出解释，再输出最终判断，有助于它先思考再作答：
    explanation: Annotated[str, ..., "请说明你给出该分数的理由"]
    correct: Annotated[bool, ..., "如果答案正确则为 True，否则为 False。"]

# 评分提示词
correctness_instructions = """你是一名正在批改测验的老师。你会获得 QUESTION、GROUND TRUTH（正确答案）和 STUDENT ANSWER。请遵循以下评分标准：
(1) 仅根据学生答案相对于正确答案的事实准确性评分。
(2) 确保学生答案中不存在与正确答案冲突的陈述。
(3) 如果学生答案包含比正确答案更多的信息，只要这些信息与正确答案一致且准确，也算正确。

Correctness（正确性）：
当正确性为 True 时，表示学生答案满足全部评分标准。
当正确性为 False 时，表示学生答案未满足全部评分标准。

请按步骤解释你的推理，以确保结论准确。避免在开头直接给出正确答案。"""

# 评分器 LLM
grader_llm = ChatOpenAI(model="gpt-4o", temperature=0).with_structured_output(
    CorrectnessGrade, method="json_schema", strict=True
)

def correctness(inputs: dict, outputs: dict, reference_outputs: dict) -> bool:
    """RAG 答案准确性的评估器"""
    answers = f"""\
QUESTION: {inputs['question']}
GROUND TRUTH ANSWER: {reference_outputs['answer']}
STUDENT ANSWER: {outputs['answer']}"""
    # 运行评估器
    grade = grader_llm.invoke([
        {"role": "system", "content": correctness_instructions},
        {"role": "user", "content": answers}
    ])
    return grade["correct"]

# 评分输出模式
class RelevanceGrade(TypedDict):
    explanation: Annotated[str, ..., "请说明你给出该分数的理由"]
    relevant: Annotated[
        bool, ..., "判断该答案是否真正回应了问题的评分结果"
    ]

# 评分提示
relevance_instructions = """你是一名正在批改测验的老师。你会获得 QUESTION 和 STUDENT ANSWER。请遵循以下评分标准：
(1) 确保学生答案简洁且与问题相关。
(2) 确保学生答案确实有助于回答该问题。

Relevance（相关性）：
当相关性为 True 时，表示学生答案满足全部评分标准。
当相关性为 False 时，表示学生答案未满足全部评分标准。

请按步骤解释你的推理，以确保推理过程与结论正确。避免在开头直接给出正确答案。"""

# 评分器 LLM
relevance_llm = ChatOpenAI(model="gpt-4o", temperature=0).with_structured_output(
    RelevanceGrade, method="json_schema", strict=True
)

# 评估函数
def relevance(inputs: dict, outputs: dict) -> bool:
    """用于评估 RAG 答案是否有帮助的简单评估器。"""
    answer = f"QUESTION: {inputs['question']}\nSTUDENT ANSWER: {outputs['answer']}"
    grade = relevance_llm.invoke([
        {"role": "system", "content": relevance_instructions},
        {"role": "user", "content": answer}
    ])
    return grade["relevant"]

# 评分输出模式
class GroundedGrade(TypedDict):
    explanation: Annotated[str, ..., "请说明你给出该分数的理由"]
    grounded: Annotated[
        bool, ..., "判断答案是否脱离文档产生幻觉的评分结果"
    ]

# 评分提示
grounded_instructions = """你是一名正在批改测验的老师。你会获得 FACTS 和 STUDENT ANSWER。请遵循以下评分标准：
(1) 确保学生答案基于 FACTS。
(2) 确保学生答案不包含超出 FACTS 范围的“幻觉”信息。

Grounded（基于文档）：
当 grounded 为 True 时，表示学生答案满足全部评分标准。
当 grounded 为 False 时，表示学生答案未满足全部评分标准。

请按步骤解释你的推理，以确保推理过程与结论正确。避免在开头直接给出正确答案。"""

# 评分器 LLM
grounded_llm = ChatOpenAI(model="gpt-4o", temperature=0).with_structured_output(
    GroundedGrade, method="json_schema", strict=True
)

# 评估函数
def groundedness(inputs: dict, outputs: dict) -> bool:
    """用于检查 RAG 答案是否基于检索文档的简单评估器。"""
    doc_string = "\n\n".join(doc.page_content for doc in outputs["documents"])
    answer = f"FACTS: {doc_string}\nSTUDENT ANSWER: {outputs['answer']}"
    grade = grounded_llm.invoke([
        {"role": "system", "content": grounded_instructions},
        {"role": "user", "content": answer}
    ])
    return grade["grounded"]

# 评分输出模式
class RetrievalRelevanceGrade(TypedDict):
    explanation: Annotated[str, ..., "请说明你给出该分数的理由"]
    relevant: Annotated[
        bool,
        ...,
        "如果检索到的文档与问题相关则为 True，否则为 False",
    ]

# Grade prompt
retrieval_relevance_instructions = """你是一名正在批改测验的老师。你会获得一个 QUESTION 和一组由学生提供的 FACTS。请遵循以下评分标准：
(1) 你的目标是识别与 QUESTION 完全无关的 FACTS。
(2) 如果 FACTS 包含任何与问题相关的关键词或语义，请考虑它们相关。
(3) 即使 FACTS 包含一些与问题无关的信息，只要 (2) 满足，也可以接受。

Relevance（相关性）：
当相关性为 True 时，表示 FACTS 包含与 QUESTION 相关的任何关键词或语义，因此相关。
当相关性为 False 时，表示 FACTS 与 QUESTION 完全无关。

请按步骤解释你的推理，以确保推理过程与结论正确。避免在开头直接给出正确答案。"""

# Grader LLM
retrieval_relevance_llm = ChatOpenAI(
    model="gpt-4o", temperature=0
).with_structured_output(RetrievalRelevanceGrade, method="json_schema", strict=True)

def retrieval_relevance(inputs: dict, outputs: dict) -> bool:
    """用于评估检索文档相关性的评估器"""
    doc_string = "\n\n".join(doc.page_content for doc in outputs["documents"])
    answer = f"FACTS: {doc_string}\nQUESTION: {inputs['question']}"
    # Run evaluator
    grade = retrieval_relevance_llm.invoke([
        {"role": "system", "content": retrieval_relevance_instructions},
        {"role": "user", "content": answer}
    ])
    return grade["relevant"]

def target(inputs: dict) -> dict:
    return rag_bot(inputs["question"])

experiment_results = client.evaluate(
    target,
    data=dataset_name,
    evaluators=[correctness, groundedness, relevance, retrieval_relevance],
    experiment_prefix="rag-doc-relevance",
    metadata={"version": "LCEL context, gpt-4-0125-preview"},
)

# Explore results locally as a dataframe if you have pandas installed
# experiment_results.to_pandas()