online-evals

配置在线评估

将评审器关联到配置变体，使用 LLM-as-a-judge 方法自动进行质量评分。评审器会评估响应并返回 0.0 到 1.0 之间的分数。

前提条件

• 已启用 AgentControl 的 LaunchDarkly 账户
• 已在环境中配置托管在远程服务上的 LaunchDarkly MCP 服务
• 已有包含变体的配置（使用 configs-create skill）
• 若要从应用中运行评审器：launchdarkly-ai-server 0.3.0+（Python）或 @launchdarkly/ai-node 0.3.0+（Node.js），以及每个配置和评审器所用提供商对应的处理程序包

API 访问

在工作流的每个步骤中都使用 LaunchDarkly MCP 服务工具：这些工具通过 OAuth 进行身份验证，无需令牌。REST API 仅用于无法使用 MCP 的脚本和 CI（请参阅不使用 MCP 的脚本）。在这种情况下，请用户粘贴本次会话的 API 访问令牌，并将以下示例中的 {api_token} 替换为该令牌。不要从环境变量、.env 文件或 Agent/MCP 配置文件中搜索凭据。

核心概念

什么是评审器？

评审器是处于评审器模式的专用配置，用于评估其他配置的响应。它们使用大模型为输出评分，并返回结构化结果：

{
  "score": 0.85,
  "reasoning": "Answered correctly with one minor omission"
}

内置评审器

LaunchDarkly 提供三个预配置的评审器：

评审器	指标键	衡量内容
准确性	$ld:ai:judge:accuracy	响应是否正确且有依据
相关性	$ld:ai:judge:relevance	响应与用户请求的契合程度
毒性	$ld:ai:judge:toxicity	有害或不安全的措辞（分数越低越安全）

仅限补全模式

在 UI 中，评审器只能关联到补全模式配置。通过 API（步骤 2）关联评审器则适用于任何配置，而 SDK 会运行所提供变体上的所有评审器。

限制

• 不能将评审器关联到其他评审器（不支持递归）
• 不能将具有相同指标键的多个评审器关联到同一个变体
• 不能查看或编辑评审器变体上的模型参数或工具

工作流

步骤 1：创建自定义评审器（可选）

如需进行特定领域的评估，请使用 create-agentcontrol-config 创建评审器配置：

{
  "projectKey": "{projectKey}",
  "key": "security-judge",
  "name": "Security Judge",
  "mode": "judge",
  "evaluationMetricKey": "$ld:ai:judge:security",
  "isInverted": false
}

注意： evaluationMetricKey 必须以 $ld:ai:judge: 开头。对于毒性这类分数越低越好的指标，请设置 isInverted: true。

然后使用 create-agentcontrol-config-variation 添加一个包含评估提示词的变体。通过 list-agentcontrol-model-configs 获取 modelConfigKey：

{
  "projectKey": "{projectKey}",
  "configKey": "security-judge",
  "key": "default",
  "name": "Default",
  "modelConfigKey": "OpenAI.gpt-4o-mini",
  "modelName": "gpt-4o-mini",
  "parameters": {"temperature": 0.3},
  "messages": [
    {
      "role": "system",
      "content": "You are a security auditor. Score from 0.0 to 1.0:\n- 1.0: No security issues\n- 0.7-0.9: Minor issues\n- 0.4-0.6: Moderate issues\n- 0.1-0.3: Serious vulnerabilities\n- 0.0: Critical vulnerabilities\n\nCheck for: SQL injection, XSS, hardcoded secrets, command injection.\n\n{{message_history}}"
    }
  ]
}

重要： 评审器提示词必须以 {{message_history}} 结尾。SDK 会将其替换为待评审的对话，因此不包含该变量的提示词永远无法看到其评分的响应。SDK 会自行添加分数和推理过程的输出格式。

步骤 2：将评审器关联到变体

使用 update-agentcontrol-config-variation 和 judgeConfiguration。通过 get-agentcontrol-config 获取 variationKey：

{
  "projectKey": "{projectKey}",
  "configKey": "{configKey}",
  "variationKey": "{variationKey}",
  "judgeConfiguration": {
    "judges": [
      {"judgeConfigKey": "security-judge", "samplingRate": 1.0},
      {"judgeConfigKey": "api-contract-judge", "samplingRate": 0.5}
    ]
  }
}

重要： judges 数组会替换所有现有的评审器关联。空数组会移除所有评审器。

步骤 3：为评审器设置回退变体

每个评审器配置都需要在运行它的每个环境中，将回退变体设置为已启用的变体。配置默认使用“已禁用”变体。请使用 update-agentcontrol-config-rollout：

{
  "projectKey": "{projectKey}",
  "configKey": "security-judge",
  "env": "production",
  "rolloutType": "variation",
  "variationKey": "default"
}

然后对评审器运行 get-agentcontrol-config-health，确认它可作为目标。

不使用 MCP 的脚本

对于无法使用 MCP 服务的 CI 或脚本，可通过 REST API 执行相同的三个步骤。turnTargetingOn 不适用于配置，因此 set_fallthrough 会发送 updateFallthroughVariationOrRollout：

import requests
import os
from typing import Optional

class AIConfigJudges:
    """Manager for config judge attachments"""

    def __init__(self, api_token: str, project_key: str):
        self.api_token = api_token
        self.project_key = project_key
        self.base_url = "https://app.launchdarkly.com/api/v2"
        self.headers = {
            "Authorization": api_token,
            "Content-Type": "application/json",
            "LD-API-Version": "beta"
        }

    def attach_judges(self, config_key: str, variation_key: str,
                      judges: list[dict]) -> dict:
        """
        Attach judges to a variation.

        Args:
            config_key: config key
            variation_key: Variation key
            judges: List of {"judgeConfigKey": str, "samplingRate": float}
        """
        url = f"{self.base_url}/projects/{self.project_key}/ai-configs/{config_key}/variations/{variation_key}"

        response = requests.patch(url, headers=self.headers, json={
            "judgeConfiguration": {"judges": judges}
        })

        if response.status_code == 200:
            print(f"[OK] Attached {len(judges)} judges to {config_key}/{variation_key}")
            return response.json()
        print(f"[ERROR] {response.status_code}: {response.text}")
        return {}

    def create_judge(self, key: str, name: str, metric_key: str,
                     system_prompt: str, model: str = "OpenAI.gpt-4o-mini",
                     is_inverted: bool = False) -> dict:
        """
        Create a judge config.

        Args:
            key: Judge config key
            name: Display name
            metric_key: Metric key for scoring; "security" is sent as "$ld:ai:judge:security"
            system_prompt: Evaluation instructions; {{message_history}} is appended if missing
            is_inverted: True if lower scores are better (e.g., toxicity)
        """
        # The API rejects a bare metric key
        if not metric_key.startswith("$ld:ai:judge:"):
            metric_key = f"$ld:ai:judge:{metric_key}"

        # Create config
        config_url = f"{self.base_url}/projects/{self.project_key}/ai-configs"
        response = requests.post(config_url, headers=self.headers, json={
            "key": key,
            "name": name,
            "mode": "judge",
            "evaluationMetricKey": metric_key,
            "isInverted": is_inverted
        })

        if response.status_code not in [200, 201]:
            print(f"[ERROR] Creating config: {response.text}")
            return {}

        # Without {{message_history}} the judge never sees the conversation it scores
        if "{{message_history}}" not in system_prompt:
            system_prompt = f"{system_prompt}\n\n{{{{message_history}}}}"

        # Create variation
        var_url = f"{self.base_url}/projects/{self.project_key}/ai-configs/{key}/variations"
        response = requests.post(var_url, headers=self.headers, json={
            "key": "default",
            "name": "Default",
            "messages": [{"role": "system", "content": system_prompt}],
            "modelConfigKey": model,
            "model": {"parameters": {"temperature": 0.3}}
        })

        if response.status_code in [200, 201]:
            print(f"[OK] Created judge: {key}")
            return response.json()
        print(f"[ERROR] Creating variation: {response.text}")
        return {}

    def set_fallthrough(self, config_key: str, environment: str,
                        variation_key: str = "default") -> bool:
        """
        Set fallthrough to enable a judge config.

        Note: turnTargetingOn doesn't work for configs. Instead, set the
        fallthrough from disabled (index 0) to the enabled variation.
        """
        # Get variation ID
        url = f"{self.base_url}/projects/{self.project_key}/ai-configs/{config_key}/targeting"
        response = requests.get(url, headers=self.headers)

        if response.status_code != 200:
            print(f"[ERROR] {response.status_code}: {response.text}")
            return False

        targeting = response.json()
        variation_id = None
        for var in targeting.get("variations", []):
            # The variation key is in value._ldMeta.variationKey, not on the item itself
            meta_key = ((var.get("value") or {}).get("_ldMeta") or {}).get("variationKey")
            if meta_key == variation_key or var.get("name") == variation_key:
                variation_id = var.get("_id")
                break

        if not variation_id:
            print(f"[ERROR] Variation '{variation_key}' not found")
            return False

        # Set fallthrough
        response = requests.patch(url, headers={
            **self.headers,
            "Content-Type": "application/json; domain-model=launchdarkly.semanticpatch"
        }, json={
            "environmentKey": environment,
            "instructions": [{
                "kind": "updateFallthroughVariationOrRollout",
                "variationId": variation_id
            }]
        })

        if response.status_code == 200:
            print(f"[OK] Fallthrough set for {config_key}")
            return True
        print(f"[ERROR] {response.status_code}: {response.text}")
        return False

SDK：内联评估

invoke() 会运行与所提供变体关联的所有评审器，按各自的采样率抽样，并将评分记录为指标。为配置及其评审器使用的每个提供商传入一个处理器。配置必须处于补全模式：消息处理器与 Agent 模式的配置不匹配。

import asyncio
from launchdarkly_ai_server import config, shutdown
from launchdarkly_ai_openai_messages import create_openai_messages_handler

config_key = "support-agent"  # completion mode

async def main():
    # Reads LD_SDK_KEY and OPENAI_API_KEY from the environment
    result = await config(
        key=config_key,
        handler=create_openai_messages_handler(),
    ).invoke("How can LaunchDarkly help me?", {"kind": "user", "key": "user-123"})

    print(result.response)
    for judge_key, judge in (result.judge_results or {}).items():
        print(judge_key, judge.score)

    # Flushes trailing events, including judge scores
    await shutdown()

asyncio.run(main())

import { config, shutdown } from '@launchdarkly/ai-node';
import { createOpenAIHandler } from '@launchdarkly/ai-openai-messages';

const result = await config({
  key: 'support-agent',
  handler: createOpenAIHandler(),
}).invoke('How can LaunchDarkly help me?', { kind: 'user', key: 'user-123' });

console.log(result.response);
for (const [judgeKey, judge] of Object.entries(result.judgeResults ?? {})) {
  console.log(judgeKey, judge.score);
}

await shutdown();

内联评审器会增加请求的延迟。若要避免它们影响请求路径，请在后台运行。

SDK：后台评估

传入 skip_judges=True（skipJudges: true），invoke() 就会将抽样选中的评审器作为 judge_tasks（judgeTasks）返回，而不是运行它们。将每个任务交给工作进程，使用 run_judge 运行，并自行记录评分，因为 run_judge 不会执行此操作：

from ldclient import Context
from launchdarkly_ai_server import config, get_client, run_judge
from launchdarkly_ai_openai_messages import create_openai_messages_handler

handler = create_openai_messages_handler()

config_key = "support-agent"  # completion mode

result = await config(key=config_key, handler=handler, skip_judges=True).invoke(
    "How can LaunchDarkly help me?", {"kind": "user", "key": "user-123"}
)

# In your worker:
for task in result.judge_tasks or []:
    judged = await run_judge(task, [handler])
    if judged and task.evaluation_metric_key:
        get_client().track(
            task.evaluation_metric_key,
            Context.from_dict(task.user_context),
            judged.track_data,
            judged.score,
        )

import { config, getClient, runJudge } from '@launchdarkly/ai-node';
import { createOpenAIHandler } from '@launchdarkly/ai-openai-messages';

const handler = createOpenAIHandler();

const result = await config({ key: 'support-agent', handler, skipJudges: true }).invoke(
  'How can LaunchDarkly help me?',
  { kind: 'user', key: 'user-123' },
);

// In your worker:
for (const task of result.judgeTasks ?? []) {
  const judged = await runJudge(task, [handler]);
  if (judged && task.evaluationMetricKey) {
    getClient().track(task.evaluationMetricKey, task.userContext, judged.trackData, judged.score);
  }
}

**注意：**在 Python 中，task.user_context 是普通 dict，因此调用 track() 前需使用 Context.from_dict() 将其转换。track_data（trackData）已包含 judgeConfigKey。

SDK：自定义管道

如果应用自行调用模型，请使用 create_handler（createHandler）将该调用封装在处理器中。这样，invoke() 就会运行你的代码和关联的评审器，与使用内置处理器时相同。评审器仍需要一个用于其自身提供商的处理器：

from launchdarkly_ai_server import config, create_handler, parse_template
from launchdarkly_ai_openai_messages import create_openai_messages_handler

config_key = "support-agent"  # completion mode

# run_my_agent is your existing pipeline; it returns the text and the provider's usage object
async def my_pipeline(ld_config, user_input, tool_handlers, variables, history):
    # Custom handlers fill {{name}} placeholders themselves
    messages = [{**m, "content": parse_template(m["content"], variables)} for m in ld_config.get("messages", [])]
    output, usage = await run_my_agent(messages, user_input)
    return {"output": output, "usage": {"input": usage.input_tokens, "output": usage.output_tokens}}

handlers = [
    create_handler(("Anthropic", "messages"), my_pipeline),  # the config's provider and mode
    create_openai_messages_handler(),                       # the judge's provider
]

result = await config(key=config_key, handler=handlers).invoke(user_input, context)

create_handler 中的提供商名称必须与配置中的模型提供商完全匹配，包括大小写；第一个参数必须是元组。完整的处理器约定请参阅 tools 技能。

采样率

每个经过评估的响应都会向模型提供商发送一个额外请求，从而增加 token 用量和成本。请从较低的采样百分比开始，仅在需要扩大评估覆盖范围时再提高。

你可以随时在变体的“评审器”部分调整采样率，也可以将某个评审器的采样率设为 0% 来停用它。

查看结果

1. 前往 configs > 选择你的配置
2. 点击 Monitoring 选项卡
3. 从下拉菜单中选择 Evaluator metrics
4. 按变体和时间范围查看评分

评估完成后，结果会在 1–2 分钟内显示。

在护栏和实验中使用

评估指标可用于：

• 受保护的发布：当评分低于阈值时暂停或回滚
• 实验：使用评估指标作为目标来比较变体

错误处理

状态	原因	解决方案
404	未找到配置或变体	确认键存在
400	评审器配置无效	检查 judgeConfigKey 是否存在
403	权限不足	重新连接 MCP 服务，或检查 API 令牌的权限
422	指标键重复	无法关联多个使用相同指标键的评审器

后续步骤

关联评审器后：

1. 设置回退，将评审器配置指向已启用的变体（必需）
2. 在 Monitoring 选项卡中监控结果
3. 根据成本和覆盖范围需求调整采样率
4. 设置受保护的发布，以自动检测回归

相关技能

• configs-create - 创建配置和评审器
• configs-targeting - 配置定向规则
• configs-variations - 管理变体

参考资料

• 在线评估

• 自定义评审器

• Python 的 AI SDK 参考

• Node.js 的 AI SDK 参考