在运行评估时，您可能希望在脚本中以编程方式处理结果，而不是在LangSmith UI中查看。这在以下场景中非常有用：

• CI/CD 流水线：实现质量门控，如果评估分数低于阈值，则使构建失败。
• 本地调试：无需进行 API 调用即可检查和分析结果。
• 自定义聚合：使用您自己的逻辑计算指标和统计数据。
• 集成测试：使用评估结果来管控合并或部署。

本指南向您展示如何遍历和处理来自 Client.evaluate() 返回的 ExperimentResults 对象的实验结果。

本页重点介绍以编程方式处理结果，同时仍将它们上传到 LangSmith。

如果您想在本地运行评估而不将任何内容记录到 LangSmith（用于快速测试或验证），请参考在本地运行评估，它使用 upload_results=False。

遍历评估结果

evaluate() 函数返回一个 ExperimentResults 对象，您可以对其进行遍历。blocking 参数控制结果何时可用：

• blocking=False：立即返回一个迭代器，该迭代器在结果产生时生成它们。这允许您在评估运行时实时处理结果。
• blocking=True（默认）：阻塞直到所有评估完成后再返回。当您遍历结果时，所有数据都已可用。

两种模式都返回相同的 ExperimentResults 类型；区别在于函数在返回前是否等待完成。使用 blocking=False 进行流式处理和实时调试，或在需要完整数据集进行批处理时使用 blocking=True。

以下示例演示了 blocking=False。它遍历流式传入的结果，将它们收集到一个列表中，然后在单独的循环中处理它们：

from langsmith import Client
import random

client = Client()

def target(inputs):
    """您的应用程序或 LLM 链"""
    return {"output": "MY OUTPUT"}

def evaluator(run, example):
    """您的评估器函数"""
    return {"key": "randomness", "score": random.randint(0, 1)}

# 使用 blocking=False 运行评估以获取迭代器
streamed_results = client.evaluate(
    target,
    data="MY_DATASET_NAME",
    evaluators=[evaluator],
    blocking=False
)

# 在结果流式传入时收集它们
aggregated_results = []
for result in streamed_results:
    aggregated_results.append(result)

# 单独的循环，以避免与 evaluate() 的日志同时记录
for result in aggregated_results:
    print("Input:", result["run"].inputs)
    print("Output:", result["run"].outputs)
    print("Evaluation Results:", result["evaluation_results"]["results"])
    print("--------------------------------")

这将产生类似以下的输出：

Input: {'input': 'MY INPUT'}
Output: {'output': 'MY OUTPUT'}
Evaluation Results: [EvaluationResult(key='randomness', score=1, value=None, comment=None, correction=None, evaluator_info={}, feedback_config=None, source_run_id=UUID('7ebb4900-91c0-40b0-bb10-f2f6a451fd3c'), target_run_id=None, extra=None)]
--------------------------------

理解结果结构

迭代器中的每个结果包含：

• result["run"]：您的目标函数的执行。

  • result["run"].inputs：来自您的数据集示例的输入。
  • result["run"].outputs：您的目标函数产生的输出。
  • result["run"].id：此运行的唯一 ID。

• result["evaluation_results"]["results"]：一个 EvaluationResult 对象列表，每个评估器一个。

  • key：指标名称（来自您的评估器的返回值）。
  • score：数值分数（通常是 0-1 或布尔值）。
  • comment：可选的解释性文本。
  • source_run_id：评估器运行的 ID。

• result["example"]：被评估的数据集示例。

  • result["example"].inputs：输入值。
  • result["example"].outputs：参考输出（如果有）。

示例

实现质量门控

此示例使用评估结果，根据质量阈值自动通过或使 CI/CD 构建失败。脚本遍历结果，计算平均准确率分数，如果准确率低于 85%，则以非零状态码退出。这确保您可以部署符合质量标准的代码更改。

from langsmith import Client
import sys

client = Client()

def my_application(inputs):
    # 您的应用程序逻辑
    return {"response": "..."}

def accuracy_evaluator(run, example):
    # 您的评估逻辑
    is_correct = run.outputs["response"] == example.outputs["expected"]
    return {"key": "accuracy", "score": 1 if is_correct else 0}

# 运行评估
results = client.evaluate(
    my_application,
    data="my_test_dataset",
    evaluators=[accuracy_evaluator],
    blocking=False
)

# 计算聚合指标
total_score = 0
count = 0

for result in results:
    eval_result = result["evaluation_results"]["results"][0]
    total_score += eval_result.score
    count += 1

average_accuracy = total_score / count

print(f"Average accuracy: {average_accuracy:.2%}")

# 如果准确率太低，则使构建失败
if average_accuracy < 0.85:
    print("❌ Evaluation failed! Accuracy below 85% threshold.")
    sys.exit(1)

print("✅ Evaluation passed!")

使用 blocking=True 进行批处理

当您需要执行需要完整数据集的操作（如计算百分位数、按分数排序或生成摘要报告）时，使用 blocking=True 等待所有评估完成后再进行处理：

# 运行评估并等待所有结果
results = client.evaluate(
    target,
    data=dataset,
    evaluators=[evaluator],
    blocking=True  # 等待所有评估完成
)

# 评估完成后处理所有结果
for result in results:
    print("Input:", result["run"].inputs)
    print("Output:", result["run"].outputs)

    # 访问单个评估结果
    for eval_result in result["evaluation_results"]["results"]:
        print(f"  {eval_result.key}: {eval_result.score}")

使用 blocking=True，您的处理代码仅在所有评估完成后运行，避免了与评估日志的输出混合。

有关在不上传结果的情况下运行评估的更多信息，请参考在本地运行评估。

相关

• 评估您的 LLM 应用程序
• 在本地运行评估
• 从实验中获取性能指标