如何在本地读取实验结果 - 以编程方式使用文档

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

CI/CD pipelines：实现质量门控，如果评估分数低于阈值则使构建失败。
本地调试：在不调用API的情况下检查和分析结果。
自定义聚合：使用您自己的逻辑计算指标和统计信息。
集成测试：使用评估结果来控制合并或部署。

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

遍历评估结果

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

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

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

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

from langsmith import Client


client = Client()

def target(inputs):
    """Your application or LLM chain"""
    return {"output": "MY OUTPUT"}

def evaluator(run, example):
    """Your evaluator function"""
    return {"key": "randomness", "score": random.randint(0, 1)}

# Run evaluation with blocking=False to get an iterator
streamed_results = client.evaluate(
    target,
    data="MY_DATASET_NAME",
    evaluators=[evaluator],
    blocking=False
)

# Collect results as they stream in
aggregated_results = []
for result in streamed_results:
    aggregated_results.append(result)

# Separate loop to avoid logging at the same time as logs from 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：来自您的数据集 example.
- 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: 参考输出（如有）。

示例

实现质量门控

This example uses evaluation results to pass or fail a CI/CD build automatically based on quality thresholds. The script iterates through results, calculates an average accuracy score, and exits with a non-zero status code if the accuracy falls below 85%. This ensures that you can deploy code changes that meet quality standards.

from langsmith import Client


client = Client()

def my_application(inputs):
    # Your application logic
    return {"response": "..."}

def accuracy_evaluator(run, example):
    # Your evaluation logic
    is_correct = run.outputs["response"] == example.outputs["expected"]
    return {"key": "accuracy", "score": 1 if is_correct else 0}

# Run evaluation
results = client.evaluate(
    my_application,
    data="my_test_dataset",
    evaluators=[accuracy_evaluator],
    blocking=False
)

# Calculate aggregate metrics
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%}")

# Fail the build if accuracy is too low
if average_accuracy < 0.85:
    print("❌ Evaluation failed! Accuracy below 85% threshold.")
    sys.exit(1)

print("✅ Evaluation passed!")

使用 blocking=True 进行批处理

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

# Run evaluation and wait for all results
results = client.evaluate(
    target,
    data=dataset,
    evaluators=[evaluator],
    blocking=True  # Wait for all evaluations to complete
)

# Process all results after evaluation completes
for result in results:
    print("Input:", result["run"].inputs)
    print("Output:", result["run"].outputs)

    # Access individual evaluation results
    for eval_result in result["evaluation_results"]["results"]:
        print(f"  {eval_result.key}: {eval_result.score}")

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

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

遍历评估结果

了解结果结构

示例

实现质量门控

使用 blocking=True 进行批处理

相关