深入 LangSmith 的 ‘Custom Evaluators’：如何针对业务指标（如代码正确率）编写自动化的评估逻辑？

深入LangSmith的’Custom Evaluators’：如何针对业务指标编写自动化评估逻辑

尊敬的各位开发者，各位对大型语言模型（LLM）充满热情的同行们：

欢迎来到今天的讲座。在LLM技术飞速发展的今天，我们正面临着一个核心挑战：如何高效、准确、客观地评估我们所构建的LLM应用？当模型从实验室走向生产环境，其性能不再仅仅是传统NLP指标（如BLEU、ROUGE）所能全面衡量的，更多时候，我们需要将其与实际业务场景深度结合，用业务指标来衡量其真正的价值。

LangSmith作为一个强大的LLMOps平台，为LLM应用的开发、调试、测试和部署提供了全面的支持。其中，其“评估器”（Evaluators）模块是确保模型质量和迭代效率的关键。虽然LangSmith提供了多种内置评估器，但面对千变万化的业务需求，这些通用评估器往往力有不逮。此时，“自定义评估器”（Custom Evaluators）便成为了我们手中的利器，它允许我们将任何复杂的业务逻辑，例如代码正确率、SQL查询有效性、API调用成功率等，转化为可量化的自动化评估指标。

今天，我们将深入探讨LangSmith的自定义评估器，并以一个极具挑战性且普遍的业务指标——“代码正确率”为例，手把手教您如何从零开始，设计并实现一个能够自动化评估LLM生成代码的评估器。

---

第一章：自动化评估的必要性与LangSmith的定位

在LLM驱动的应用开发中，我们不断迭代模型，尝试不同的提示词、模型参数、检索策略乃至不同的基础模型。每一次迭代都需要快速且可靠的反馈，以判断改动是进步还是退步。

传统评估的局限性：

1. 手动评估： 耗时、成本高昂，尤其在数据量大时不可扩展。
2. 主观性强： 依赖人工判断，评估结果易受个人偏好、理解偏差影响，一致性差。
3. 覆盖不足： 人工评估往往只能覆盖少量边缘案例，难以发现模型在复杂场景下的潜在问题。
4. 脱离业务： 传统的NLP指标（如语义相似度、流畅度）可能无法直接反映模型在特定业务场景下的实际价值。例如，一个生成流畅但语法错误的SQL查询，其业务价值为零。

LLMOps与LangSmith的崛起：

为了应对这些挑战，LLMOps（Large Language Model Operations）的概念应运而生。它旨在将DevOps的实践引入LLM开发，实现模型的持续集成、持续部署和持续评估。

LangSmith正是这一生态系统中的核心工具之一。它提供了：

• 可观测性（Observability）： 记录并可视化每一次LLM调用的输入、输出、中间步骤、延迟、成本等，便于调试和理解模型行为。
• 调试（Debugging）： 基于可观测数据，快速定位问题根源，优化提示词或链结构。
• 数据集管理（Dataset Management）： 组织和管理用于测试和评估模型的数据集。
• 评估（Evaluation）： 自动化或半自动化地衡量模型性能，是本次讲座的重点。
• 代理（Agents）和链（Chains）的开发： 支持复杂LLM应用的构建和管理。

LangSmith的内置评估器覆盖了一些通用场景，例如基于LLM的反馈（CriteriaEvaluator）、语义相似度（EmbeddingDistanceEvaluator）或正则表达式匹配（RegexEvaluator）。然而，当我们的业务指标深入到代码的实际运行结果、数据库查询的有效性或复杂业务规则的满足程度时，内置评估器便显得力不从心。

这就是自定义评估器发挥作用的舞台。

---

第二章：LangSmith评估器基础：工作原理与内置选项

在深入自定义评估器之前，我们先快速回顾一下LangSmith评估器的基本概念。

评估器的核心思想：

一个评估器本质上是一个函数或一个类，它接收关于LLM模型运行的上下文信息（例如，模型的输入、模型生成的输出、预期的正确输出、模型的内部执行轨迹等），然后输出一个量化的分数、文字反馈以及可选的评论或元数据，以描述模型在该特定示例上的表现。

评估器的输入：

• run 对象：代表LangChain（或其他LLM框架）的一次完整执行轨迹。它包含了模型的输入、输出、中间步骤（如果是链或代理）、错误信息等所有运行时数据。
• example 对象：代表数据集中的一个特定示例。它通常包含原始输入、期望的正确输出（如果有），以及其他与该示例相关的元数据。

评估器的输出：

• EvaluationResult 对象：这是一个结构化的数据容器，包含：
  • key (str): 评估器的唯一标识符。
  • score (float, Optional): 一个介于0到1之间的分数，代表评估结果。1表示完美，0表示完全失败。
  • feedback (str, Optional): 简短的文字反馈，概括评估结果（例如，“通过所有测试”，“输出格式错误”）。
  • comment (str, Optional): 更详细的文字说明，包含评估细节、错误信息等。
  • metadata (Dict, Optional): 任意额外的键值对数据，可以存储评估过程中产生的中间结果或更详细的诊断信息。

LangSmith内置评估器概览：

LangSmith提供了多种开箱即用的评估器，它们可以帮助我们快速设置基础评估流程：

评估器名称	描述	典型用途
CriteriaEvaluator	使用另一个LLM（或用户定义的标准）来评估给定输出是否满足特定标准。	评估答案的相关性、安全性、礼貌性、遵循指令的能力等主观或复杂文本属性。
LlmFeedbackEvaluator	类似于 CriteriaEvaluator，但通常用于更通用的LLM反馈收集，可以指定多个标准。	收集模型输出的整体质量、有用性、准确性等综合反馈。
EmbeddingDistanceEvaluator	计算模型输出与期望输出之间嵌入向量的距离，以衡量语义相似度。	评估问答系统、文本摘要、语义搜索等任务中输出的语义准确性。
ExactMatchEvaluator	检查模型输出是否与期望输出完全一致。	评估事实性问答、关键字提取、固定格式生成等需要精确匹配的任务。
RegexEvaluator	使用正则表达式匹配模型输出，检查是否符合特定模式。	评估输出的格式、是否包含特定关键词、是否符合日期/邮箱/电话等模式。
StringDistanceEvaluator	计算模型输出与期望输出之间的字符串编辑距离（如Levenshtein距离）。	评估文本生成任务中字符级别的相似度，对拼写错误或微小差异敏感。
JsonValidityEvaluator	检查模型输出是否是有效的JSON格式。	评估需要生成结构化JSON数据的模型，例如API调用参数、数据传输对象等。
ToolCodeEvaluator	评估Agent在执行工具代码时是否成功（例如，Python REPL工具）。	评估Agent是否能正确调用和使用工具来解决问题。
ComparisonEvaluator	比较两个模型运行的输出，通常用于A/B测试或评估不同模型/提示词变体。	对比不同模型版本或提示词策略的性能，找出最优解。

尽管这些内置评估器功能强大，但它们无法直接判断一段代码是否能正确编译、运行，并产生预期的结果。例如，一个LLM可能生成一段语法正确的Python代码，但这段代码在逻辑上是错误的，或者在特定输入下会抛出运行时异常。此时，我们就需要自定义评估器。

---

第三章：深入理解LangSmith Custom Evaluators

LangSmith自定义评估器的核心是实现一个Python类，该类继承自langsmith.evaluation.evaluator.BaseEvaluator。通过重写其关键方法，我们可以注入任何我们需要的评估逻辑。

BaseEvaluator的关键方法：

1. `init(self, name: str, kwargs: Any)`：**

   • 这是评估器的构造函数。您可以在这里初始化评估器所需的任何资源，例如加载模型、配置沙箱环境参数、设置数据库连接等。
   • name 参数用于标识评估器。
   • **kwargs 允许您传递额外的配置参数。

2. _evaluate(self, run: Run, example: Optional[Example] = None) -> EvaluationResult：

   • 这是评估器实际执行评估逻辑的核心方法。
   • 它接收两个主要参数：
     • run：一个langsmith.schemas.Run对象，包含了LLM模型本次执行的所有详细信息，包括模型的输入 (run.inputs)、输出 (run.outputs)、名称 (run.name)、类型 (run.run_type)、开始时间 (run.start_time)、结束时间 (run.end_time)，以及可能的错误信息 (run.error) 和中间步骤 (run.trace)。
     • example：一个langsmith.schemas.Example对象，代表了当前评估的数据集示例。它包含了原始的输入 (example.inputs) 和期望的输出 (example.outputs)。并非所有评估都必须有example，例如，有些评估可能只关注run自身的属性（如延迟）。
   • 此方法必须返回一个EvaluationResult对象。

3. evaluate(self, run: Run, example: Optional[Example] = None) -> EvaluationResult：

   • 这是一个包装方法，它负责调用 _evaluate，并处理一些通用的逻辑，如异常捕获、记录元数据等。
   • 通常情况下，您不需要重写此方法。 只需要专注于实现 _evaluate 即可。

Run对象和Example对象的重要属性：

理解这两个对象的结构对于编写自定义评估器至关重要，因为它们提供了评估所需的所有上下文信息。

• Run对象（langsmith.schemas.Run）：

  • id (UUID): 运行的唯一标识符。
  • name (str): 运行的名称（通常是链或模型的名称）。
  • run_type (str): 运行的类型（如“llm”、“chain”、“agent”、“tool”等）。
  • inputs (Dict): 模型的输入数据。
  • outputs (Dict, Optional): 模型的输出数据。对于LLM类型，通常在outputs["generations"]中。对于链，可能是outputs["text"]或其他键。
  • error (str, Optional): 如果运行失败，则为错误信息。
  • parent_run_id (UUID, Optional): 如果是嵌套运行，则为父运行的ID。
  • start_time, end_time (datetime): 运行开始和结束时间。
  • child_runs (List[Run], Optional): 嵌套运行的子运行。
  • events (List[Dict], Optional): 运行过程中的事件日志。

• Example对象（langsmith.schemas.Example）：

  • id (UUID): 示例的唯一标识符。
  • inputs (Dict): 用于生成模型的输入的原始数据。
  • outputs (Dict, Optional): 期望的模型输出。这通常是“黄金标准”答案，用于与模型实际输出进行比较。
  • dataset_id (UUID): 所属数据集的ID。
  • metadata (Dict, Optional): 示例相关的额外元数据。

在我们的“代码正确率”评估器中，我们将从run.outputs中提取LLM生成的代码，并从example.inputs或example.outputs中获取用于测试该代码的测试用例。

---

第四章：场景示例：基于代码正确率的自动化评估器设计

现在，让我们聚焦于本次讲座的核心示例：如何自动化评估LLM生成的Python代码片段是否能在给定测试用例下正确运行。

问题定义：

假设我们的LLM应用旨在帮助开发者生成特定功能的Python代码。例如，给定一个自然语言描述：“编写一个函数，接收一个整数列表，返回其中所有偶数的和”，LLM需要生成一个Python函数。我们的目标是自动化验证LLM生成的函数是否：

1. 语法正确。
2. 逻辑正确，即在各种输入下都能给出预期的输出。
3. 没有运行时错误（如ZeroDivisionError, IndexError等）。
4. 在合理的时间内完成执行。

挑战分析：

实现这样一个评估器并非易事，需要考虑以下关键挑战：

1. 沙箱环境（Sandbox Environment）： 直接执行LLM生成的代码存在巨大的安全风险。生成的代码可能是恶意的（例如，删除文件、访问敏感信息、发起网络请求）。因此，必须在一个高度隔离且受限的沙箱环境中执行代码。
2. 动态执行与结果捕获： 需要能够程序化地执行Python代码，捕获其标准输出（stdout）、标准错误（stderr）以及任何异常。
3. 测试用例管理： 如何将预设的测试用例（输入-期望输出对）与LLM生成的代码关联起来？这些测试用例应该存储在哪里？
4. 语言多样性（Language Agnostic）： 虽然我们以Python为例，但理想的评估器设计应能扩展到其他语言（Java, JavaScript, Go等），这意味着沙箱执行机制需要一定的灵活性。
5. 超时机制： 恶意或无限循环的代码可能导致评估过程永久挂起。必须设置执行时间限制。
6. 资源限制： 除了时间，还应考虑内存、CPU等资源限制（更高级的沙箱特性）。

设计思路：

我们的自定义评估器 CodeCorrectnessEvaluator 将遵循以下设计：

1. 代码获取： 从run.outputs中提取LLM生成的Python代码字符串。
2. 测试用例获取： 从example.inputs或example.outputs中获取一个包含多个输入-期望输出对的列表。
3. 沙箱执行： 调用一个辅助函数 _execute_python_code_in_sandbox。这个函数负责：
   • 创建一个临时目录，将LLM生成的代码写入临时文件。
   • 构造一个“测试运行器”脚本，该脚本将导入用户代码，并针对每个测试用例调用它。
   • 使用subprocess模块在一个独立的Python进程中执行测试运行器。
   • 通过stdin将测试用例传递给运行器，并通过stdout捕获运行结果（最好是JSON格式）。
   • 设置执行超时。
   • 捕获任何运行时错误或超时。
   • 执行完成后，清理临时文件。
4. 结果解析与比较： 解析沙箱执行返回的结果，逐个测试用例地比较实际输出与期望输出。
5. 分数计算： 根据通过的测试用例数量计算总分（例如，通过率）。
6. 反馈与评论： 提供详细的反馈，说明哪些测试用例通过，哪些失败，以及失败的原因（错误消息、堆栈跟踪等）。

---

第五章：实现一个Python代码正确率评估器

现在，让我们一步步实现CodeCorrectnessEvaluator。

5.1. 核心辅助函数：沙箱执行Python代码

我们首先编写一个独立的辅助函数_execute_python_code_in_sandbox，它将处理代码的实际执行和结果捕获。为了简化，我们在这里使用subprocess模块模拟沙箱，并利用临时文件进行隔离。

import os
import subprocess
import tempfile
import json
import shutil
from typing import Any, Dict, List, Optional, Union

# --- 辅助函数：沙箱执行Python代码 ---
def _execute_python_code_in_sandbox(
    code: str,
    test_cases: List[Dict[str, Any]],
    timeout: int = 10,
    func_name: str = "solution" # 假设LLM生成的是一个名为'solution'的函数
) -> Dict[str, Any]:
    """
    在沙箱环境中执行Python代码，并针对给定的测试用例运行。
    每个测试用例包含 'input' 和 'expected_output'。

    Args:
        code (str): LLM生成的Python代码字符串。
        test_cases (List[Dict[str, Any]]): 包含测试用例的列表。每个字典应包含
                                            'input' (作为函数参数) 和 'expected_output'。
        timeout (int): 代码执行的超时时间（秒）。
        func_name (str): 期望用户代码中定义的函数名称。

    Returns:
        Dict[str, Any]: 包含执行结果的字典，包括：
                        'success' (bool): 整体执行是否成功。
                        'output' (str, Optional): 沙箱进程的标准输出。
                        'error_message' (str, Optional): 如果失败，则为错误消息。
                        'details' (List[Dict[str, Any]]): 每个测试用例的详细结果。
    """
    temp_dir = tempfile.mkdtemp() # 创建一个临时目录用于隔离
    code_file_path = os.path.join(temp_dir, "user_code.py")
    test_runner_path = os.path.join(temp_dir, "test_runner.py")

    try:
        # 1. 将LLM生成的代码写入临时文件
        with open(code_file_path, "w", encoding="utf-8") as f:
            f.write(code)

        # 2. 构造一个测试运行器脚本
        # 这个脚本将导入用户代码，并针对每个测试用例调用指定的函数
        # 它将测试结果以JSON格式打印到stdout
        runner_script_template = f"""
import sys
import json
import traceback
import os

# 将临时目录添加到PYTHONPATH，以便可以导入 user_code
sys.path.insert(0, os.path.dirname(__file__))

try:
    from user_code import {func_name} # 尝试从用户代码中导入指定的函数
except ImportError:
    # 如果函数不存在，则尝试执行整个脚本（如果用户代码是可执行脚本）
    # 对于本评估器，我们假设生成的是一个函数
    print(json.dumps([{{
        "status": "error", 
        "error_message": f"Could not import function '{func_name}' from user_code.py. "
                         "Please ensure your generated code defines this function.",
        "traceback": ""
    }}]))
    sys.exit(1)
except Exception as e:
    print(json.dumps([{{
        "status": "error", 
        "error_message": f"Error importing user_code.py: {{e}}",
        "traceback": traceback.format_exc()
    }}]))
    sys.exit(1)

def run_single_test_case(input_data):
    try:
        # 调用LLM生成的函数
        result = {func_name}(input_data)
        return {{"status": "success", "output": result}}
    except Exception as e:
        return {{"status": "error", "error_message": str(e), "traceback": traceback.format_exc()}}

if __name__ == "__main__":
    test_cases_str = sys.stdin.read()

    try:
        test_cases = json.loads(test_cases_str)
    except json.JSONDecodeError:
        print(json.dumps([{{
            "status": "error",
            "error_message": "Invalid JSON for test cases received via stdin.",
            "traceback": ""
        }}]))
        sys.exit(1)

    results = []
    for tc in test_cases:
        input_data = tc["input"] # 假设测试用例的输入字段为 "input"

        # 针对每个测试用例运行并收集结果
        case_result = run_single_test_case(input_data)
        results.append(case_result)

    print(json.dumps(results))
        """

        # 写入测试运行器脚本
        with open(test_runner_path, "w", encoding="utf-8") as f:
            f.write(runner_script_template)

        # 将测试用例作为JSON字符串通过stdin传递给测试运行器
        test_cases_json = json.dumps(test_cases)

        # 3. 使用subprocess执行测试运行器
        # 注意：使用text=True，stdout和stderr将是字符串
        # env={"PYTHONPATH": temp_dir} 确保runner能找到user_code
        process = subprocess.run(
            ["python", test_runner_path],
            input=test_cases_json.encode("utf-8"), # 输入需编码为字节
            capture_output=True,
            text=True,
            timeout=timeout,
            check=False, # 不抛出CalledProcessError，而是捕获stderr
            env={"PYTHONPATH": temp_dir, **os.environ} # 确保runner能找到user_code，并继承当前环境
        )

        # 4. 处理subprocess执行结果
        if process.returncode != 0:
            # Python解释器本身或运行器脚本有错误
            return {
                "success": False,
                "output": process.stdout,
                "error_message": f"Runner script failed with exit code {process.returncode}. "
                                 f"Stderr: {process.stderr}",
                "details": []
            }

        try:
            # 尝试解析测试运行器打印的JSON结果
            execution_results = json.loads(process.stdout)

            overall_success = True
            detailed_results = []
            for i, res in enumerate(execution_results):
                # 检查单个测试用例的执行状态
                is_case_success = res["status"] == "success"

                # 将期望输出也加入详细结果，方便后续比较
                expected_output_for_case = test_cases[i]["expected_output"]

                detailed_results.append({
                    "test_case_index": i,
                    "input": test_cases[i]["input"],
                    "expected_output": expected_output_for_case,
                    "actual_output": res.get("output"),
                    "execution_success": is_case_success, # 表示代码执行无异常
                    "error_message": res.get("error_message"),
                    "traceback": res.get("traceback")
                })

                # 如果单个测试用例执行失败 (有异常) 或输出不匹配，则整体不成功
                if not is_case_success or res.get("output") != expected_output_for_case:
                    overall_success = False

            return {
                "success": overall_success, # 整体成功意味着所有测试用例执行无异常且输出匹配
                "output": process.stdout, # 原始输出，可能是JSON
                "error_message": None,
                "details": detailed_results
            }

        except json.JSONDecodeError:
            # 如果运行器没有打印有效的JSON
            return {
                "success": False,
                "output": process.stdout,
                "error_message": f"Failed to parse JSON output from runner. "
                                 f"Raw stdout: {process.stdout}. Stderr: {process.stderr}",
                "details": []
            }

    except subprocess.TimeoutExpired:
        # 代码执行超时
        return {
            "success": False,
            "output": None,
            "error_message": f"Code execution timed out after {timeout} seconds.",
            "details": []
        }
    except Exception as e:
        # 其他意外错误（如文件操作失败）
        return {
            "success": False,
            "output": None,
            "error_message": f"An unexpected error occurred during execution setup: {str(e)}",
            "details": []
        }
    finally:
        # 无论成功失败，都清理临时文件
        if os.path.exists(temp_dir):
            shutil.rmtree(temp_dir)

5.2. CodeCorrectnessEvaluator类定义

现在，我们将集成上述沙箱执行逻辑到我们的LangSmith自定义评估器类中。

from langsmith.evaluation import EvaluationResult
from langsmith.schemas import Run, Example
from langsmith.evaluation.evaluator import BaseEvaluator

# 确保导入上面定义的 _execute_python_code_in_sandbox
# from .your_module import _execute_python_code_in_sandbox 
# 如果在同一个文件，则无需导入

class CodeCorrectnessEvaluator(BaseEvaluator):
    """
    一个自定义LangSmith评估器，用于评估LLM生成的Python代码的正确性。
    它通过在安全的沙箱中执行代码并与预定义的测试用例进行比较来判断。
    """

    def __init__(self, name: str = "CodeCorrectnessEvaluator", **kwargs: Any):
        """
        初始化评估器。
        Args:
            name (str): 评估器的名称。
            **kwargs: 可以传递额外的配置参数，如执行超时时间、预期的函数名称。
        """
        super().__init__(name=name, **kwargs)
        self.execution_timeout = kwargs.get("execution_timeout", 10) # 默认10秒
        self.expected_func_name = kwargs.get("func_name", "solution") # 默认评估名为'solution'的函数

    def _evaluate(
        self, run: Run, example: Optional[Example] = None
    ) -> EvaluationResult:
        """
        执行评估逻辑。

        Args:
            run (Run): LLM模型的一次执行轨迹。
            example (Optional[Example]): 数据集中的一个示例，包含期望的输入/输出。

        Returns:
            EvaluationResult: 评估结果对象。
        """
        # 1. 检查LLM是否产生了输出
        if not run.outputs:
            return EvaluationResult(
                key=self.name,
                score=0,
                comment="LLM did not produce any output.",
                feedback="No code generated."
            )

        # 2. 从run.outputs中获取生成的代码
        # 假设LLM的输出被存储在 'text' 键下
        generated_code = run.outputs.get("text") 

        if not generated_code or not isinstance(generated_code, str):
            return EvaluationResult(
                key=self.name,
                score=0,
                comment="LLM output is not a valid code string or is empty.",
                feedback="Invalid or empty code generated."
            )

        # 3. 从example中获取测试用例
        # 假设测试用例存储在 example.inputs["test_cases"] 中
        # 或者 example.outputs["expected_test_cases"] 中
        test_cases = []
        if example and example.inputs and "test_cases" in example.inputs:
            test_cases = example.inputs["test_cases"]
        elif example and example.outputs and "expected_test_cases" in example.outputs:
            # 另一种情况：如果期望输出中带有测试用例
            test_cases = example.outputs["expected_test_cases"]

        if not test_cases:
            return EvaluationResult(
                key=self.name,
                score=0,
                comment="No test cases provided in the example for evaluation. "
                        "Please ensure 'test_cases' is in example.inputs or 'expected_test_cases' in example.outputs.",
                feedback="Missing test cases."
            )

        # 4. 执行代码并获取结果
        execution_report = _execute_python_code_in_sandbox(
            code=generated_code,
            test_cases=test_cases,
            timeout=self.execution_timeout,
            func_name=self.expected_func_name
        )

        # 5. 根据执行报告计算分数、生成反馈和评论
        score = 0
        feedback = "Code execution failed."
        comment_parts = []

        if execution_report["success"]:
            # 代码在沙箱中成功执行，并且所有测试用例的执行都没有抛出异常且输出匹配
            score = 1.0 # 如果整体成功，直接给满分
            feedback = "All test cases passed successfully."
            comment_parts.append("All test cases passed.")
        else:
            # 整体不成功，可能是因为：
            # 1. 沙箱执行本身失败 (超时, 解释器错误等)
            # 2. 代码执行时抛出异常 (某个测试用例失败)
            # 3. 代码执行成功但输出不匹配

            total_tests = len(test_cases)
            passed_tests_count = 0

            if execution_report["details"]:
                # 如果有详细的测试用例结果
                for i, detail in enumerate(execution_report["details"]):
                    is_case_correct = detail["execution_success"] and (detail["actual_output"] == detail["expected_output"])
                    if is_case_correct:
                        passed_tests_count += 1

                    status_str = "Passed" if is_case_correct else "Failed"
                    detail_str = (
                        f"Test Case {i+1} (Input: {json.dumps(detail['input'])}): "
                        f"Status: {status_str}. "
                        f"Expected Output: {json.dumps(detail['expected_output'])}, "
                        f"Actual Output: {json.dumps(detail['actual_output'])}. "
                    )
                    if not detail["execution_success"]:
                        detail_str += f"Error: {detail.get('error_message', 'Unknown error')}. "
                        if detail.get('traceback'):
                             detail_str += f"Traceback: {detail['traceback'].splitlines()[-1].strip()}." # 只显示最后一行
                    elif detail["actual_output"] != detail["expected_output"]:
                        detail_str += "Output mismatch."

                    comment_parts.append(detail_str)

                if total_tests > 0:
                    score = passed_tests_count / total_tests

                if score == 1:
                    feedback = "All test cases passed successfully."
                elif score > 0:
                    feedback = f"Passed {passed_tests_count}/{total_tests} test cases. Some tests failed or had incorrect output."
                else:
                    feedback = "All test cases failed or produced incorrect output."
            else:
                # 没有详细的测试用例结果，说明是沙箱环境或运行器脚本本身的问题
                score = 0
                feedback = f"Code execution environment setup or runner script failed: {execution_report['error_message']}"
                comment_parts.append(f"Execution environment error: {execution_report['error_message']}")

        # 构建最终的评论
        final_comment = "n".join(comment_parts)
        if execution_report["error_message"] and not execution_report["details"]:
            final_comment = f"Overall Execution Error: {execution_report['error_message']}n" + final_comment

        # 返回评估结果
        return EvaluationResult(
            key=self.name,
            score=score,
            comment=final_comment,
            feedback=feedback,
            metadata={
                "overall_execution_success": execution_report["success"], # 总体沙箱执行是否成功
                "test_cases_details": execution_report["details"], # 每个测试用例的详细结果
                "raw_sandbox_stdout": execution_report["output"], # 沙箱进程的原始stdout
                "sandbox_error_message": execution_report["error_message"] # 沙箱层面的错误信息
            }
        )

5.3. 如何在LangSmith平台使用

要让LangSmith使用您的自定义评估器，您需要：

1. 打包评估器： 将您的CodeCorrectnessEvaluator类及其辅助函数放在一个Python模块中（例如，my_evaluators.py）。
2. 部署评估器： LangSmith需要在可访问的环境中找到并执行您的评估器。常见的部署方式包括：
   • LangSmith Cloud Functions： 这是最推荐的方式。您可以将评估器代码打包成一个Lambda函数或其他无服务函数，并在LangSmith UI中配置其调用。LangSmith会负责在评估时触发这些函数。
   • Docker镜像： 对于复杂的依赖或特殊的运行时环境，您可以将评估器及其所有依赖打包成一个Docker镜像，并提供给LangSmith。
   • 本地运行： 在开发和调试阶段，您可以在本地Python环境中直接使用langsmith.run_evaluator函数或通过LangChain client.run_on_dataset来集成评估器。

示例：本地使用CodeCorrectnessEvaluator

首先，我们需要一个LangSmith客户端和一个数据集。

import os
from langsmith import Client
from langsmith import Dataset
from langsmith import traceable
from langsmith.evaluation import run_evaluator

# 配置LangSmith环境变量
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "YOUR_LANGSMITH_API_KEY" # 替换为您的LangSmith API Key
os.environ["LANGCHAIN_PROJECT"] = "Code Correctness Evaluation Demo" # 您的项目名称

client = Client()

# 1. 准备一个数据集
# 每个example包含原始输入 (例如，问题描述) 和测试用例
# 以及期望输出 (这里是测试用例的期望结果，而非LLM生成的代码本身)
dataset_name = "Python Code Generation Test Cases"
examples = [
    {
        "input": "编写一个Python函数，名为`solution`，接收一个整数`n`，返回`n`的平方。",
        "test_cases": [
            {"input": 2, "expected_output": 4},
            {"input": 0, "expected_output": 0},
            {"input": -3, "expected_output": 9}
        ]
    },
    {
        "input": "请实现一个Python函数`solution`，计算给定整数列表`numbers`中所有偶数的和。",
        "test_cases": [
            {"input": [1, 2, 3, 4, 5, 6], "expected_output": 12}, # 2+4+6
            {"input": [7, 9, 11], "expected_output": 0},
            {"input": [], "expected_output": 0},
            {"input": [2, -4, 6], "expected_output": 4} # 2 + (-4) + 6
        ]
    },
    {
        "input": "编写一个名为`solution`的Python函数，接收一个字符串`s`，返回其反转字符串。",
        "test_cases": [
            {"input": "hello", "expected_output": "olleh"},
            {"input": "Python", "expected_output": "nohtyP"},
            {"input": "", "expected_output": ""},
            {"input": "a", "expected_output": "a"}
        ]
    }
]

# 检查数据集是否存在，如果不存在则创建
try:
    dataset = client.read_dataset(dataset_name=dataset_name)
    print(f"Dataset '{dataset_name}' already exists.")
except Exception:
    print(f"Creating dataset '{dataset_name}'...")
    dataset = client.create_dataset(
        dataset_name=dataset_name,
        description="Dataset for evaluating Python code generation models."
    )
    for i, ex in enumerate(examples):
        client.create_example(
            dataset_id=dataset.id,
            inputs={"question": ex["input"], "test_cases": ex["test_cases"]},
            # 这里的outputs可以为空，因为评估器主要从run.outputs中获取代码，
            # 并从example.inputs中获取test_cases。
            # 但如果测试用例是期望输出的一部分，也可以放在outputs里。
            outputs={"expected_test_cases": ex["test_cases"]} 
        )
    print(f"Dataset '{dataset_name}' created with {len(examples)} examples.")

# 2. 定义一个模拟LLM函数
# 这个函数将模拟LLM生成代码的过程
@traceable(run_type="llm") # 标记为LLM运行，LangSmith会自动追踪
def mock_llm_code_generator(question: str) -> str:
    """
    模拟LLM根据问题生成Python代码。
    """
    if "平方" in question:
        return "def solution(n):n    return n * n"
    elif "偶数" in question and "和" in question:
        return "def solution(numbers):n    total = 0n    for num in numbers:n        if num % 2 == 0:n            total += numn    return total"
    elif "反转字符串" in question:
        return "def solution(s):n    return s[::-1]"
    else:
        return "def solution(x):n    return 'Hello from default solution!'"

# 3. 定义一个函数，用于将LLM输出包装成评估器期望的格式
# 这里的runnable可以是LangChain的Runnable，也可以是自定义函数
def code_generation_runnable(example_inputs: Dict[str, Any]) -> Dict[str, Any]:
    question = example_inputs["question"]
    generated_code = mock_llm_code_generator(question)
    return {"text": generated_code} # LLM输出的key必须是 'text'，与评估器中获取的保持一致

# 4. 运行评估
print(f"nRunning evaluation on dataset '{dataset_name}' with custom evaluator...")

# 实例化我们的自定义评估器
custom_evaluator = CodeCorrectnessEvaluator(
    name="PythonCodeCorrectness",
    execution_timeout=5, # 设置5秒超时
    func_name="solution" # 期望LLM生成的函数名为 'solution'
)

# 使用 client.run_on_dataset 触发评估
# 这里我们直接传入评估器实例
# LangSmith会遍历数据集中的每个example，为每个example调用runnable，
# 然后用评估器评估runnable的输出。
evaluation_results = client.run_on_dataset(
    dataset_name=dataset_name,
    llm_or_chain_factory=code_generation_runnable,
    evaluators=[
        custom_evaluator,
        "cot_qa" # 可以同时运行内置评估器，例如基于LLM的QA评估
    ],
    project_name="Code Correctness Evaluation Demo - Run",
    concurrency_level=1, # 演示目的，设为1
    verbose=True
)

print("nEvaluation completed. Check LangSmith UI for detailed results.")
print(f"Evaluation Run ID: {evaluation_results.id}")

# 您可以通过访问 LangSmith UI 来查看详细的评估结果
# 访问 URL: https://smith.langchain.com/o/<your_organization_id>/p/Code%20Correctness%20Evaluation%20Demo%20-%20Run/

在上述示例中，我们首先创建了一个LangSmith数据集，其中每个示例包含一个问题描述和一组测试用例。然后，我们模拟了一个LLM代码生成器，并将其包装在一个可运行的函数中。最后，我们实例化了CodeCorrectnessEvaluator并使用client.run_on_dataset在数据集上运行评估。

在LangSmith UI中查看结果：

当评估运行完成后，您可以在LangSmith UI中找到名为“Code Correctness Evaluation Demo – Run”的项目。点击进入后，您将看到每次LLM调用的详细轨迹，以及我们自定义评估器（PythonCodeCorrectness）和内置评估器（cot_qa）给出的分数、反馈和评论。metadata字段中的test_cases_details将提供每个测试用例的详细结果，包括输入、期望输出、实际输出、执行是否成功以及任何错误信息。

---

第六章：高级定制与最佳实践

6.1. 数据源与测试用例管理：

• LangSmith Dataset的利用： 强烈建议将所有测试用例存储在LangSmith数据集中。这有助于版本控制、共享和复用。您可以将测试用例直接作为example.inputs或example.outputs的一部分存储。
  • example.inputs: {"question": "...", "test_cases": [...]}
  • example.outputs: {"expected_answer": "...", "test_cases": [...]}
• 动态生成测试用例： 对于某些场景，测试用例可能不是预设的。评估器可以在运行时调用另一个LLM或专用工具来根据问题描述生成测试用例。这增加了评估的灵活性，但也会增加复杂性和成本。

6.2. 沙箱环境的强化：

我们示例中的subprocess沙箱相对基础。在生产环境中，应考虑更强大的沙箱方案：

• Docker容器： 将代码执行封装在Docker容器中，提供更强的隔离性、资源限制（CPU、内存）和更可控的环境。您可以使用docker run命令来执行代码。
• 专用沙箱服务： 对于极高的安全性要求，可以考虑使用如gVisor、Firecracker等轻量级虚拟机技术，或云服务商提供的沙箱环境（如Google Cloud Run、AWS Lambda、Azure Container Apps）。
• 安全性考量： 永远不要在生产环境中直接执行来自LLM的、未经严格沙箱隔离的代码。确保沙箱环境没有网络访问权限，没有文件系统写入权限（除了临时文件），并且对可执行文件有严格限制。

6.3. 多语言支持：

如果您的LLM生成多种编程语言的代码，评估器可以根据example中的语言类型字段，动态调用不同的解释器或编译器。

# 假设example.inputs中有一个 'language' 字段
language = example.inputs.get("language", "python")
if language == "python":
    execution_report = _execute_python_code_in_sandbox(...)
elif language == "javascript":
    # 调用一个执行JavaScript代码的沙箱函数
    execution_report = _execute_javascript_code_in_sandbox(...)
# ...

6.4. 评估指标的多元化：

除了代码正确率，您还可以结合其他指标来全面评估代码生成质量：

• 性能（Execution Time）： 在沙箱执行时记录代码的运行时间，作为额外的元数据。
• 代码质量（Code Quality）： 调用静态代码分析工具（如Pylint, Flake8, SonarQube）来评估可读性、复杂度、遵循编码规范等。
• 安全性（Security Vulnerabilities）： 使用SAST（Static Application Security Testing）工具检查生成的代码是否存在已知的安全漏洞。
• 代码风格（Code Style）： 使用black, isort等工具格式化代码，并比较格式化前后的差异，或直接检查代码是否符合特定风格。

组合评估器： LangSmith允许您在一次评估运行中指定多个评估器。您也可以编写一个CompositeEvaluator，它内部调用多个子评估器，并根据业务逻辑组合它们的结果。

6.5. 异步评估：

如果您的评估逻辑非常耗时（例如，运行大量测试用例，或涉及复杂的模型推理），可以考虑异步执行评估器。LangSmith提供了对异步评估的支持，允许评估过程不阻塞主线程。

6.6. 错误处理与日志记录：

在评估器内部，务必实现健壮的错误处理，捕获所有可能的异常。使用详细的日志记录（例如，Python的logging模块）来记录评估过程中的关键步骤、遇到的问题和中间结果，这对于调试至关重要。将这些信息包含在EvaluationResult的comment或metadata中，以便在LangSmith UI中查看。

---

第七章：部署与集成

将自定义评估器从开发环境推向生产环境通常涉及以下几种方式：

1. 本地运行评估：
   • 如前所示，直接在本地Python脚本中导入并实例化您的评估器，然后使用client.run_on_dataset()。这适用于开发、测试和小型评估。
2. 远程LangSmith评估 (推荐)：
   • 将您的评估器代码部署为LangSmith Cloud Function（例如，AWS Lambda）。
   • 在LangSmith UI中，导航到“Evaluators”页面，点击“Create Evaluator”，选择“Cloud Function”类型，并填写您的函数信息（ARN、环境变量等）。
   • 在配置评估运行时，选择您创建的自定义评估器。LangSmith会在需要时调用您的云函数来执行评估。
   • 优势： 无需管理服务器，高度可伸缩，与LangSmith平台无缝集成。
3. Docker容器部署：
   • 如果您的评估器有复杂的依赖或需要特定的运行时环境，可以将其打包成Docker镜像。
   • 将Docker镜像推送到一个可访问的容器注册表（如ECR、Docker Hub）。
   • 在LangSmith UI中配置评估器时，选择“Docker Image”类型，并提供镜像信息。
   • 优势： 环境一致性，适用于复杂依赖。

---

第八章：实际业务中的应用场景拓展

自定义评估器的应用远不止代码正确率：

• SQL生成模型：
  • 评估逻辑： 连接到目标数据库，执行LLM生成的SQL查询。检查查询是否语法正确、是否返回预期结果集、是否触发错误。
  • 业务指标： 查询执行成功率、结果集匹配度、查询性能（执行时间）、是否有潜在的SQL注入风险。
• 单元测试生成模型：
  • 评估逻辑： 将LLM生成的单元测试代码与原始代码一起运行。检查测试是否通过、是否能发现已知缺陷、代码覆盖率是否提高。
  • 业务指标： 测试通过率、测试覆盖率、缺陷发现能力。
• 数据转换脚本生成：
  • 评估逻辑： 在沙箱中执行LLM生成的数据转换脚本。提供原始输入数据，检查输出数据是否符合期望的格式和内容。
  • 业务指标： 数据转换准确率、输出数据格式合规性、脚本执行效率。
• 特定API调用代码生成：
  • 评估逻辑： 模拟或实际调用LLM生成的代码所涉及的外部API。检查API调用是否成功、返回数据是否符合预期模式、错误处理是否得当。
  • 业务指标： API调用成功率、响应数据有效性、错误恢复能力。

---

总结：Custom Evaluators的价值与未来展望

LangSmith的Custom Evaluators为开发者提供了一座桥梁，将特定的业务逻辑和高质量的评估标准无缝集成到LLM开发流程中。它们是实现LLM应用生产级质量和可靠性的关键，使得我们能够从主观、低效的人工评估转向客观、可扩展的自动化评估。

随着LLM能力边界的不断拓展，以及其在各行各业应用的深入，对评估复杂性和准确性的要求将持续提高。自定义评估器将扮演越来越重要的角色，帮助我们构建更智能、更可靠、更贴近业务需求的LLM应用。通过拥抱自动化评估，我们能更快地迭代，更高质量地交付，最终释放LLM技术的真正潜力。