langsmith-evaluator

LangSmith 评估器

创建评估器，以衡量 Agent 在数据集上的表现。LangSmith 支持两种类型：LLM 作为评审（使用 LLM 为输出评分）和 自定义代码（确定性逻辑）。

设置

环境变量

LANGSMITH_API_KEY=lsv2_pt_your_api_key_here          # Required
LANGSMITH_WORKSPACE_ID=your-workspace-id              # Optional: for org-scoped keys
OPENAI_API_KEY=your_openai_key                        # For LLM as Judge

依赖项

pip install langsmith langchain-openai python-dotenv

评估器格式

评估器支持两种函数签名：

方法 1：字典参数（用于在本地运行评估）：

def evaluator_name(inputs: dict, outputs: dict, reference_outputs: dict = None) -> dict:
    """Evaluate a single prediction."""
    user_query = inputs.get("query", "")
    agent_response = outputs.get("expected_response", "")
    expected = reference_outputs.get("expected_response", "") if reference_outputs else None

    return {
        "key": "metric_name",    # Metric identifier
        "score": 0.85,           # Number or boolean
        "comment": "Reason..."   # Optional explanation
    }

方法 2：Run/Example 参数（用于上传到 LangSmith）：

def evaluator_name(run, example):
    """Evaluate using run/example dicts.

    Args:
        run: Dict with run["outputs"] containing agent outputs
        example: Dict with example["outputs"] containing expected outputs
    """
    agent_response = run["outputs"].get("expected_response", "")
    expected = example["outputs"].get("expected_response", "")

    return {
        "metric_name": 0.85,      # Metric name as key directly
        "comment": "Reason..."    # Optional explanation
    }

LLM 作为评审的评估器

使用结构化输出，以确保评分可靠：

from typing import TypedDict, Annotated
from langchain_openai import ChatOpenAI

class AccuracyGrade(TypedDict):
    """Structured evaluation output."""
    reasoning: Annotated[str, ..., "Explain your reasoning"]
    is_accurate: Annotated[bool, ..., "True if response is accurate"]
    confidence: Annotated[float, ..., "Confidence 0.0-1.0"]

# Configure model with structured output
judge = ChatOpenAI(model="gpt-4o-mini", temperature=0).with_structured_output(
    AccuracyGrade, method="json_schema", strict=True
)

async def accuracy_evaluator(run, example):
    """Evaluate factual accuracy for LangSmith upload."""
    expected = example["outputs"].get('expected_response', '')
    agent_output = run["outputs"].get('expected_response', '')

    prompt = f"""Expected: {expected}
Agent Output: {agent_output}
Evaluate accuracy:"""

    grade = await judge.ainvoke([{"role": "user", "content": prompt}])

    return {
        "accuracy": 1 if grade["is_accurate"] else 0,
        "comment": f"{grade['reasoning']} (confidence: {grade['confidence']})"
    }

常见指标： 完整性、正确性、有用性、专业性

自定义代码评估器

精确匹配

def exact_match_evaluator(run, example):
    """Check if output exactly matches expected."""
    output = run["outputs"].get("expected_response", "").strip().lower()
    expected = example["outputs"].get("expected_response", "").strip().lower()

    match = output == expected
    return {
        "exact_match": 1 if match else 0,
        "comment": f"Match: {match}"
    }

轨迹验证

def trajectory_evaluator(run, example):
    """Evaluate tool call sequence."""
    trajectory = run["outputs"].get("expected_trajectory", [])
    expected = example["outputs"].get("expected_trajectory", [])

    # Exact sequence match
    exact = trajectory == expected

    # All required tools used (order-agnostic)
    all_tools = set(expected).issubset(set(trajectory))

    # Efficiency: count extra steps
    extra_steps = len(trajectory) - len(expected)

    return {
        "trajectory_match": 1 if exact else 0,
        "comment": f"Exact: {exact}, All tools: {all_tools}, Extra: {extra_steps}"
    }

单步验证

def single_step_evaluator(run, example):
    """Evaluate single node output."""
    output = run["outputs"].get("output", {})
    expected = example["outputs"].get("expected_output", {})
    node_name = run["outputs"].get("node_name", "")

    # For classification nodes
    if "classification" in node_name:
        classification = output.get("classification", "")
        expected_class = expected.get("classification", "")
        match = classification.lower() == expected_class.lower()

        return {
            "classification_correct": 1 if match else 0,
            "comment": f"Output: {classification}, Expected: {expected_class}"
        }

    # For other nodes
    match = output == expected
    return {
        "output_match": 1 if match else 0,
        "comment": f"Match: {match}"
    }

运行评估

from langsmith import Client

client = Client()

# Define your agent function
def run_agent(inputs: dict) -> dict:
    """Your agent invocation logic."""
    result = your_agent.invoke(inputs)
    return {"expected_response": result}

# Run evaluation
results = await client.aevaluate(
    run_agent,
    data="Skills: Final Response",              # Dataset name
    evaluators=[
        exact_match_evaluator,
        accuracy_evaluator,
        trajectory_evaluator
    ],
    experiment_prefix="skills-eval-v1",
    max_concurrency=4
)

将评估器上传到 LangSmith

上传脚本是一个实用工具，可将自定义评估器部署到 LangSmith。编写适用于你的用例的评估器，然后上传。

前往 skills/langsmith-evaluator/scripts/ 上传评估器。

重要提示： LangSmith API 要求评估器使用 (run, example) 签名，其中：

• run：包含 Agent 输出的字典，位于 run["outputs"] 中
• example：包含预期输出的字典，位于 example["outputs"] 中

创建评估器文件

# my_project/evaluators/custom_evals.py

def my_custom_evaluator(run, example):
    """Your custom evaluation logic.

    Args:
        run: Dict with run["outputs"] - agent outputs
        example: Dict with example["outputs"] - expected outputs

    Returns:
        Dict with metric_name as key, score as value, optional comment
    """
    # Extract relevant data
    agent_output = run["outputs"].get("expected_trajectory", [])
    expected = example["outputs"].get("expected_trajectory", [])

    # Your custom logic here
    match = agent_output == expected

    return {
        "my_metric": 1 if match else 0,
        "comment": "Custom reasoning here"
    }

上传

# List existing evaluators
python upload_evaluators.py list

# Upload evaluator
python upload_evaluators.py upload my_evaluators.py \
  --name "Trajectory Match" \
  --function trajectory_match \
  --dataset "Skills: Trajectory" \
  --replace

# Delete evaluator (will prompt for confirmation)
python upload_evaluators.py delete "Trajectory Match"

# Skip confirmation prompts (use with caution)
python upload_evaluators.py delete "Trajectory Match" --yes
python upload_evaluators.py upload my_evaluators.py \
  --name "Trajectory Match" \
  --function trajectory_match \
  --replace --yes

选项：

• --name - LangSmith 中显示的名称
• --function - 要提取的函数名称
• --dataset - 目标数据集名称
• --project - 目标项目名称
• --sample-rate - 抽样率（0.0-1.0）
• --replace - 如果已存在则替换（将提示确认）
• --yes - 跳过替换/删除操作的确认提示

重要提示 - 安全确认：

• 执行任何破坏性操作（删除、替换）前，脚本都会提示确认
• 始终遵守这些提示——等待用户输入后再继续
• 除非用户明确要求，否则绝不要使用 --yes 标志
• --yes 标志会跳过所有安全确认提示，仅应在用户明确授权的自动化工作流中使用

最佳实践

1. 为 LLM 评审器使用结构化输出 - 比解析自由文本更可靠
2. 让评估器与数据集类型匹配
   • 最终回复 → 使用 LLM 作为评审来评估质量，使用自定义代码来检查格式
   • 单步 → 使用自定义代码进行精确匹配
   • 轨迹 → 使用自定义代码检查序列/效率
3. 组合多个评估器 - 同时运行主观评估（LLM）和客观评估（代码）
4. 为 LLM 评审器使用异步模式 - 支持并行评估，速度快得多
5. 独立测试评估器 - 首先使用已知的优质/劣质示例进行验证
6. 上传到 LangSmith - 自动评估新的运行结果

示例工作流

# 1. Create evaluators file
cat > evaluators.py <<'EOF'
def exact_match(run, example):
    """Check if output exactly matches expected."""
    output = run["outputs"].get("expected_response", "").strip().lower()
    expected = example["outputs"].get("expected_response", "").strip().lower()
    match = output == expected
    return {
        "exact_match": 1 if match else 0,
        "comment": f"Match: {match}"
    }
EOF

# 2. Upload to LangSmith
python upload_evaluators.py upload evaluators.py \
  --name "Exact Match" \
  --function exact_match \
  --dataset "Skills: Final Response" \
  --replace

# 3. Evaluator runs automatically on new dataset runs

资源

• LangSmith 评估概念
• 自定义代码评估器
• OpenEvals - 现成的评估器

相关技能

• 使用 langsmith-trace 技能查询和导出轨迹
• 使用 langsmith-dataset 技能根据轨迹生成评估数据集