代码评估器(Code evaluators)
代码评估器在所有计划(Hobby、Core、Pro、Enterprise)和自托管版本中可用。
自托管: 代码评估器需要一个配置好的代码评估器 dispatcher。当没有设置 dispatcher 时,它们被禁用。
代码评估器需要使用基于 OpenTelemetry 的 SDK 接入的 observations:Python SDK v3+ 或 JS/TS SDK v4+。如果需要,参见 Python v2 → v3 迁移指南或 JS/TS v3 → v4 迁移指南。
代码评估器在 Langfuse 中运行自定义的 Python 或 TypeScript 逻辑,并返回一个或多个分数。将它们用于确定性的、客观的检查——在这些场景中代码比基于模型的判断更可靠。
常见示例包括精确匹配检查、正则校验、JSON 可解析性、schema 校验、关键词检查、工具调用检查和自定义业务规则。
当评估需要语义判断、基于评分标准的推理,或主观评估(如有用性、语气或答案质量)时,请改用 LLM-as-a-Judge。
如何使用代码评估器?
代码评估器可以在两类数据上运行:Observations(来自实时生产流量的单个操作)或 Experiments(受控测试数据集)。你的选择取决于你是在开发阶段测试还是在监控生产。
决策树
哪类数据需要确定性评估?
- 实时生产数据(监控实时流量)→ Observations(单个操作:LLM 调用、检索、工具调用)
- 离线实验数据(在受控环境中测试)→ Experiments(带数据集的受控测试用例)
生产模式:团队通常在开发阶段使用 Experiments 验证确定性检查,然后在生产环境部署 Observation 级评估器进行可扩展监控。
理解每个评估目标
实时生产数据
对你 traces 中的单个 observations 运行评估器,例如 LLM 调用、检索操作、嵌入生成或工具调用。
为什么以 observations 为目标
- 操作级精度:按 observation 类型过滤,只评估重要的操作,而非完整 traces。
- 确定性生产监控:在实时流量上检查 JSON 有效性、schema 合规性、精确匹配或业务规则。
- 组合式评估:在一个 trace 内的不同操作上运行不同的代码评估器。
- 组合过滤:将 observation 过滤器与 trace 过滤器(如
userId、sessionId、tags、version 和 metadata)叠加。
数据流
在接入时,每个 observation 都会与你的过滤条件进行匹配评估。匹配的 observations 被加入评估队列。评估任务异步处理,分数附加到具体的 observation。
示例用例
- 验证最终的 LLM 响应是可解析的 JSON
- 检查工具调用是否包含必需参数
- 对选定的模型调用强制执行自定义业务规则
离线实验数据
在受控测试数据集上运行评估器,在可复现的环境中对比模型版本、提示词变体或系统配置。
为什么以 experiments 为目标
- 你需要用于开发工作流的确定性通过/失败检查
- 你想对比多个提示词版本或模型配置
- 你有带预期输出或元数据的数据集,你的评估器应检查它们
数据流
每次实验运行都会生成 traces 和 observations,可由你选择的评估器打分。评估器接收 observation 数据加上实验条目上下文,如预期输出和条目元数据。
- 创建包含测试输入和(可选)预期输出的数据集。
- 通过 UI 或 SDK 运行实验。参见通过 UI 做实验或通过 SDK 做实验。
- 选择代码评估器为生成的 observations 打分。
- 跨实验运行对比结果,做出数据驱动的决策。
示例用例
- 在支持问题数据集上对比两个提示词版本,并检查每个响应是否包含必需的 JSON 字段
逐步配置
函数契约
每个评估器暴露一个 evaluate 函数。Langfuse 传入一个 EvaluationContext,并期望一个带一个或多个分数的 EvaluationResult。
from dataclasses import dataclass, field
from typing import Any
@dataclass
class ToolCall:
id: str = ""
name: str = ""
arguments: Any = None
type: str = ""
index: int = 0
@dataclass
class ObservationContext:
input: Any = None
output: Any = None
metadata: Any = None
tool_calls: list[ToolCall] = field(default_factory=list)
@dataclass
class ExperimentContext:
item_expected_output: Any = None
item_metadata: Any = None
@dataclass
class EvaluationContext:
observation: ObservationContext
experiment: ExperimentContext | None = None
@dataclass
class Score:
name: str
value: int | float | str | bool
data_type: str
comment: str | None = None
config_id: str | None = None
metadata: dict[str, Any] | None = None
@dataclass
class EvaluationResult:
scores: list[Score]
def evaluate(ctx: EvaluationContext) -> EvaluationResult:
output_present = ctx.observation.output is not None
return EvaluationResult(
scores=[
Score(
name="Output present",
value=output_present,
data_type="BOOLEAN",
comment=(
"Observation output is present."
if output_present
else "Observation output is missing."
),
metadata={"rule": "output_present"},
)
]
)
type ToolCall = {
id: string;
name: string;
arguments: unknown;
type: string;
index: number;
};
type EvaluationContext = {
observation: {
input: any;
output: any;
metadata: any;
toolCalls: ToolCall[];
};
experiment:
| {
itemExpectedOutput: any;
itemMetadata: any;
}
| undefined;
};
type ScoreBase = {
name: string;
comment?: string;
configId?: string | null;
metadata?: Record<string, unknown>;
};
type NumericScore = ScoreBase & {
dataType: "NUMERIC";
value: number;
};
type BooleanScore = ScoreBase & {
dataType: "BOOLEAN";
value: boolean;
};
type CategoricalScore = ScoreBase & {
dataType: "CATEGORICAL";
value: string;
};
type TextScore = ScoreBase & {
dataType: "TEXT";
value: string;
};
type Score = NumericScore | BooleanScore | CategoricalScore | TextScore;
type EvaluationResult = {
scores: Score[];
};
function evaluate({
observation: { input, output, metadata, toolCalls },
experiment,
}: EvaluationContext): EvaluationResult {
const itemExpectedOutput = experiment?.itemExpectedOutput;
const itemMetadata = experiment?.itemMetadata;
const outputPresent = output != null;
return {
scores: [
{
name: "Output present",
value: outputPresent,
dataType: "BOOLEAN",
comment: outputPresent
? "Observation output is present."
: "Observation output is missing.",
metadata: {
rule: "output_present",
hasInput: input != null,
hasObservationMetadata: metadata != null,
toolCallCount: toolCalls.length,
hasExpectedOutput: itemExpectedOutput != null,
hasExperimentMetadata: itemMetadata != null,
},
},
],
};
}
上下文字段
| 字段 | 描述 |
|---|---|
ctx.observation.input |
记录在评估器目标所选 observation 上的输入。 |
ctx.observation.output |
记录在评估器目标所选 observation 上的输出。 |
ctx.observation.metadata |
记录在该 observation 上的元数据。 |
ctx.observation.tool_calls(Python)/ ctx.observation.toolCalls(TypeScript) |
带 id、name、arguments、type 和 index 的有序调用。有效的 JSON 参数会被解析。 |
ctx.experiment |
仅当评估器在实验上运行时存在。 |
ctx.experiment.item_expected_output(Python)/ ctx.experiment.itemExpectedOutput(TypeScript) |
来自实验条目的预期输出。 |
ctx.experiment.item_metadata(Python)/ ctx.experiment.itemMetadata(TypeScript) |
来自实验条目的元数据。 |
分数字段
| 字段 | 描述 |
|---|---|
name |
必需的分数名称。 |
value |
必需的分数值。 |
data_type / dataType |
必需的分数数据类型。支持的值为 NUMERIC、CATEGORICAL、BOOLEAN 和 TEXT。 |
comment |
与分数一起存储的可选推理或解释。 |
config_id / configId |
可选的分数配置 ID。提供时,分数必须满足所引用的分数配置。 |
metadata |
与分数一起存储的可选元数据。 |
示例:精确匹配
本示例返回一个布尔分数,当 observation 输出精确匹配实验条目的预期输出时通过。
def evaluate(ctx: EvaluationContext) -> EvaluationResult:
"""Evaluates one observation and returns one or more Langfuse scores."""
expected_output = (
ctx.experiment.item_expected_output if ctx.experiment is not None else None
)
matches_expected_output = (
expected_output is not None and ctx.observation.output == expected_output
)
return EvaluationResult(
scores=[
Score(
name="Exact match",
value=matches_expected_output,
data_type="BOOLEAN",
comment=(
"Output exactly matches the expected output."
if matches_expected_output
else "Output does not match the expected output."
),
)
]
)
/**
* Evaluates one observation and returns one or more Langfuse scores.
*/
function evaluate({
observation: { input, output, metadata },
experiment,
}: EvaluationContext): EvaluationResult {
const itemExpectedOutput = experiment?.itemExpectedOutput;
const itemMetadata = experiment?.itemMetadata;
const matchesExpectedOutput =
itemExpectedOutput != null && output === itemExpectedOutput;
return {
scores: [
{
name: "Exact match",
value: matchesExpectedOutput,
dataType: "BOOLEAN",
comment: matchesExpectedOutput
? "Output exactly matches the expected output."
: "Output does not match the expected output.",
metadata: {
hasInput: input != null,
hasObservationMetadata: metadata != null,
hasExperimentMetadata: itemMetadata != null,
},
},
],
};
}
调试代码评估器执行
每次代码评估器执行都会创建一个 trace,让你对评估过程拥有完整的可见性。这让你能够检查所选的输入和输出、实验上下文、运行时延迟、返回的分数、日志和错误。
你可以通过在追踪表中按环境 langfuse-code-eval 过滤来显示代码评估器执行 traces:

代码评估器执行状态
- Completed:评估成功完成并返回有效分数。
- Error:评估失败(点击执行 trace ID 查看输入、输出、延迟、日志和错误详情)。
- Pending:评估已排队,等待运行。
在启用新评估器之前使用评估器测试运行。这是验证所选 observation 数据、实验上下文、分数名称、分数值和分数数据类型的最快方式。
运行时约束
代码评估器旨在用于紧凑、确定性的检查,能够快速且安全地为许多 observations 运行。
代码评估器需要特定的第三方库或网络访问?请在 GitHub Discussions 中分享你的用例。你的反馈帮助我们理解更广泛的运行时支持在何处有用。
| 约束 | 限制/指导 |
|---|---|
| 语言 | 用 Python 或 TypeScript 编写评估器。在自托管部署上,Python 需要 aws-lambda dispatcher;insecure-local 仅支持 TypeScript/JavaScript。 |
| TypeScript 语法 | 使用可擦除的 TypeScript 语法。类型注解和接口可以;避免 enums、namespaces、decorators 和 parameter properties。 |
| 依赖 | 使用语言标准库(Python 和 TS/JS)。第三方包在评估器运行时中不可用。 |
| 网络访问 | 评估器在无网络出口的情况下运行。将所有必需数据保存在 observation 或实验上下文中。 |
| 运行时限制 | 评估器必须在 2 秒内完成。 |
| 结果形状 | 从 evaluate 返回至少一个分数。 |
| 源码大小 | 评估器源代码保持在 256 KB 以下。 |
| 输入大小 | 调度负载(包括源代码和所选变量)保持在 5.5 MB 以下。 |
| 结果大小 | 评估器结果保持在 256 KB 以下。 |
常见问题
如何调试超时错误?
超时通常意味着评估器对 2 秒运行时限制做了太多工作,或试图访问网络。网络请求被运行时阻止,可能表现为超时错误。
要调试此问题,在一个小样本 observation 上运行评估器,移除网络调用,避免大循环或昂贵的解析,并减少为评估器选择的输入、输出、元数据或实验上下文的量。
我可以使用第三方包吗?
不可以。代码评估器目前仅支持标准库。如果你的评估需要第三方包,请在你自己的基础设施中运行该逻辑,并用通过 API/SDK 打分将结果接入 Langfuse。
为什么实验上下文有时不存在?
ctx.experiment 仅当评估器在实验上运行时存在。对于实时 observation 评估器,编写你的代码使其处理 Python 中 ctx.experiment 为 None 或 TypeScript 中为 undefined 的情况。
我可以通过 API 或 SDK 创建代码评估器吗?
可以。除了 Langfuse UI,不稳定的公共评估器端点接受 type: "code" 以创建代码评估器并从评估规则引用它们。参见 Evaluators API 参考——注意这些端点不稳定,可能会变化。
如果你想在自己的应用或 CI 流水线中运行确定性评估逻辑,请使用通过 API/SDK 打分将产生的分数接入 Langfuse。
为什么我找不到代码评估器执行 traces?
代码评估器执行使用内部环境 langfuse-code-eval。内部环境在默认追踪视图中隐藏,因此请按 environment = langfuse-code-eval 过滤追踪表,或从相关分数或评估器日志打开执行 trace。
如何在自托管 Langfuse 上配置代码评估器?
对于自托管部署,在 Code evaluators 中配置代码评估器 dispatcher 和执行 worker。
唯一的 SDK 要求是基于 OpenTelemetry 的接入:
- Python SDK v3+(基于 OTel)。如果你使用 Python SDK v2,参见 Python v2 → v3 迁移指南。
- JS/TS SDK v4+(基于 OTel)。如果你使用 JS/TS SDK v3,参见 JS/TS v3 → v4 迁移指南。
如果你在某个运行时约束上遇到问题,或某个约束阻止了重要的评估用例,请在 GitHub Discussions 中贡献细节。