如何评估应用程序的中间步骤

pip install -U langsmith langchain[openai] wikipedia

import wikipedia as wp
from openai import OpenAI
from langsmith import traceable, wrappers

oai_client = wrappers.wrap_openai(OpenAI())

@traceable
def generate_wiki_search(question: str) -> str:
    """生成用于在维基百科中搜索的查询。"""
    instructions = (
        "生成一个搜索查询以传入维基百科来回答用户的问题。"
        "仅返回搜索查询，不要返回其他内容。"
        "这将直接传递给维基百科搜索引擎。"
    )
    messages = [
        {"role": "system", "content": instructions},
        {"role": "user", "content": question}
    ]
    result = oai_client.chat.completions.create(
        messages=messages,
        model="gpt-4.1-mini",
        temperature=0,
    )
    return result.choices[0].message.content

@traceable(run_type="retriever")
def retrieve(query: str) -> list:
    """获取最多两个维基百科搜索结果。"""
    results = []
    for term in wp.search(query, results = 10):
        try:
            page = wp.page(term, auto_suggest=False)
            results.append({
                "page_content": page.summary,
                "type": "Document",
                "metadata": {"url": page.url}
            })
        except wp.DisambiguationError:
            pass
        if len(results) >= 2:
            return results

@traceable
def generate_answer(question: str, context: str) -> str:
    """根据检索到的信息回答问题。"""
    instructions = f"仅根据以下内容回答用户的问题：\n\n{context}"
    messages = [
        {"role": "system", "content": instructions},
        {"role": "user", "content": question}
    ]
    result = oai_client.chat.completions.create(
        messages=messages,
        model="gpt-4.1-mini",
        temperature=0
    )
    return result.choices[0].message.content

@traceable
def qa_pipeline(question: str) -> str:
    """完整的流水线。"""
    query = generate_wiki_search(question)
    context = "\n\n".join([doc["page_content"] for doc in retrieve(query)])
    return generate_answer(question, context)

from langsmith import Client

ls_client = Client()
dataset_name = "Wikipedia RAG"

if not ls_client.has_dataset(dataset_name=dataset_name):
    dataset = ls_client.create_dataset(dataset_name=dataset_name)
    examples = [
      {"inputs": {"question": "What is LangChain?"}},
      {"inputs": {"question": "What is LangSmith?"}},
    ]
    ls_client.create_examples(
      dataset_id=dataset.id,
      examples=examples,
    )

from langchain.chat_models import init_chat_model
from langsmith.schemas import Run
from pydantic import BaseModel, Field

def document_relevance(run: Run) -> bool:
    """检查检索器输入是否存在于检索到的文档中。"""
    qa_pipeline_run = next(
        r for run in run.child_runs if r.name == "qa_pipeline"
    )
    retrieve_run = next(
        r for run in qa_pipeline_run.child_runs if r.name == "retrieve"
    )
    page_contents = "\n\n".join(
        doc["page_content"] for doc in retrieve_run.outputs["output"]
    )
    return retrieve_run.inputs["query"] in page_contents

# 数据模型
class GradeHallucinations(BaseModel):
    """生成答案中是否存在幻觉的二元分数。"""
    is_grounded: bool = Field(..., description="如果答案基于事实，则为True，否则为False。")

# 用于分级幻觉的具有结构化输出的LLM
# 更多信息请参见：https://docs.langchain.com/oss/python/langchain/structured-output
grader_llm= init_chat_model("gpt-4.1-mini", temperature=0).with_structured_output(
    GradeHallucinations,
    method="json_schema",
    strict=True,
)

def no_hallucination(run: Run) -> bool:
    """检查答案是否基于文档。
    如果没有幻觉，则返回True，否则返回False。
    """
    # 获取文档和答案
    qa_pipeline_run = next(
        r for r in run.child_runs if r.name == "qa_pipeline"
    )
    retrieve_run = next(
        r for r in qa_pipeline_run.child_runs if r.name == "retrieve"
    )
    retrieved_content = "\n\n".join(
        doc["page_content"] for doc in retrieve_run.outputs["output"]
    )

    # 构建提示
    instructions = (
        "您是一个评估者，负责评估LLM生成是否基于/支持一组检索到的事实。"
        "给出一个二元分数1或0，其中1表示答案基于/支持这组事实。"
    )
    messages = [
        {"role": "system", "content": instructions},
        {"role": "user", "content": f"事实集：\n{retrieved_content}\n\nLLM生成：{run.outputs['answer']}"},
    ]
    grade = grader_llm.invoke(messages)
    return grade.is_grounded

def qa_wrapper(inputs: dict) -> dict:
  """包装qa_pipeline，使其可以接受Example.inputs字典作为输入。"""
  return {"answer": qa_pipeline(inputs["question"])}

experiment_results = ls_client.evaluate(
    qa_wrapper,
    data=dataset_name,
    evaluators=[document_relevance, no_hallucination],
    experiment_prefix="rag-wiki-oai"
)