评估 RAG 应用

import os
os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_API_KEY"] = "您的 LANGSMITH API 密钥"
os.environ["OPENAI_API_KEY"] = "您的 OPENAI API 密钥"

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

# 要加载文档的 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]

# 使用指定的块大小和重叠度初始化文本分割器
text_splitter = RecursiveCharacterTextSplitter.from_tiktoken_encoder(
    chunk_size=250, chunk_overlap=0
)

# 将文档分割成块
doc_splits = text_splitter.split_documents(docs_list)

# 使用 OpenAIEmbeddings 将文档块添加到“向量存储”中
vectorstore = InMemoryVectorStore.from_documents(
    documents=doc_splits,
    embedding=OpenAIEmbeddings(),
)

# 使用 langchain，我们可以轻松地将任何向量存储转换为检索组件：
retriever = vectorstore.as_retriever(k=6)

from langchain_openai import ChatOpenAI
from langsmith import traceable

llm = ChatOpenAI(model="gpt-4.1", 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 将被自动追踪
    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) 常见词元偏差。"},
    },
    {
        "inputs": {"question": "对抗性攻击有哪五种类型？"},
        "outputs": {"answer": "五种对抗性攻击类型是 (1) 词元操纵，(2) 基于梯度的攻击，(3) 越狱提示，(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 = """您是一位给测验评分的老师。您将收到一个问题、真实（正确）答案和学生答案。请遵循以下评分标准：
(1) 仅根据学生答案相对于真实答案的事实准确性来评分。(2) 确保学生答案不包含任何相互矛盾的陈述。
(3) 如果学生答案包含的信息多于真实答案，只要其相对于真实答案在事实上是准确的，也是可以的。

正确性：
正确性值为 True 表示学生答案符合所有标准。
正确性值为 False 表示学生答案不符合所有标准。

以逐步的方式解释您的推理，以确保您的推理和结论是正确的。避免在一开始就直接陈述正确答案。"""

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

def correctness(inputs: dict, outputs: dict, reference_outputs: dict) -> bool:
    """用于 RAG 答案准确性的评估器"""
    answers = f"""\
问题：{inputs['question']}
真实答案：{reference_outputs['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 = """您是一位给测验评分的老师。您将收到一个问题和一个学生答案。请遵循以下评分标准：
(1) 确保学生答案简洁且与问题相关
(2) 确保学生答案有助于回答问题

相关性：
相关性值为 True 表示学生答案符合所有标准。
相关性值为 False 表示学生答案不符合所有标准。

以逐步的方式解释您的推理，以确保您的推理和结论是正确的。避免在一开始就直接陈述正确答案。"""

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

# 评估器
def relevance(inputs: dict, outputs: dict) -> bool:
    """用于 RAG 答案有用性的简单评估器。"""
    answer = f"问题：{inputs['question']}\n学生答案：{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 = """您是一位给测验评分的老师。您将收到一些事实和一个学生答案。请遵循以下评分标准：
(1) 确保学生答案基于这些事实。(2) 确保学生答案不包含超出事实范围的“幻觉”信息。

Grounded（基于事实）：
Grounded 值为 True 表示学生答案符合所有标准。
Grounded 值为 False 表示学生答案不符合所有标准。

以逐步的方式解释您的推理，以确保您的推理和结论是正确的。避免在一开始就直接陈述正确答案。"""

# 评分 LLM
grounded_llm = ChatOpenAI(model="gpt-4.1", 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"事实：{doc_string}\n学生答案：{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",
    ]

# 评分提示
retrieval_relevance_instructions = """您是一位给测验评分的老师。您将收到一个问题以及学生提供的一组事实。请遵循以下评分标准：
(1) 您的目标是识别与问题完全无关的事实
(2) 如果事实包含与问题相关的任何关键词或语义含义，则认为它们相关
(3) 只要满足 (2)，事实中包含一些与问题无关的信息也是可以的

相关性：
相关性值为 True 表示事实包含与问题相关的任何关键词或语义含义，因此是相关的。
相关性值为 False 表示事实与问题完全无关。

以逐步的方式解释您的推理，以确保您的推理和结论是正确的。避免在一开始就直接陈述正确答案。"""

# 评分 LLM
retrieval_relevance_llm = ChatOpenAI(
    model="gpt-4.1", 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"事实：{doc_string}\n问题：{inputs['question']}"
    # 运行评估器
    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"},
)

# 如果您安装了 pandas，可以在本地将结果作为数据框查看
# experiment_results.to_pandas()