custom-metrics

配置的自定义指标

对自定义业务指标进行完整生命周期管理：使用 MCP 服务创建指标定义，通过 SDK 跟踪事件，检索指标数据，并以编程方式管理指标。

前提条件

• 已初始化 LaunchDarkly AI SDK（参见 sdk-install）。SDK 从环境变量中读取 LD_SDK_KEY。
• 已在环境中配置远程托管的 LaunchDarkly MCP 服务
• 具有 writer 角色的 LaunchDarkly API 令牌，仅用于更新或删除指标，或供无法使用 MCP 的脚本使用
• 了解内置 Agent 指标（参见 built-in-metrics）

API 访问

只要有对应工具，就使用 LaunchDarkly MCP 服务工具：它们通过 OAuth 进行身份验证，无需令牌。使用 create-metric 创建指标，使用 list-metric-events 检查事件是否已到达，并使用 get-metric 和 list-metrics 读取定义。MCP 服务无法更新或删除指标，因此这些步骤需使用 REST API。对于这些操作，或者对于无法使用 MCP 的脚本，请让用户粘贴本次会话使用的 API 访问令牌，并在以下示例中用它替换 {api_token}。不要在环境变量、.env 文件或 Agent/MCP 配置文件中搜索凭据。

指标生命周期概览

步骤	方法	用途
1. 创建	MCP create-metric	在 LaunchDarkly 中定义指标
2. 跟踪	SDK	向指标发送事件
3. 获取	MCP get-metric、list-metrics	检索指标定义/数据
4. 更新	API	修改指标属性
5. 删除	API	移除指标

1. 创建指标

调用 create-metric。eventKey 必须与代码传递给 track() 的键一致：

{
  "projectKey": "support-ai",
  "key": "ai.task.completion",
  "name": "Agent Task Completion Rate",
  "kind": "custom",
  "eventKey": "ai.task.completion",
  "measureType": "value",
  "valueAggregation": "average",
  "successCriteria": "HigherThanBaseline",
  "unit": "count",
  "description": "Tracks successful agent task completions"
}

measureType 可以是 count（事件触发的次数）、occurrence（每位用户是否至少触发过一次事件）或 value（随事件发送的数值）。然后使用 get-metric 确认。

不使用 MCP 的脚本

对应的 REST API：

数值型自定义指标的必填字段：

• successCriteria - 必须为 "HigherThanBaseline" 或 "LowerThanBaseline"
• unit - 例如 "count"、"percent"、"milliseconds"

如果数值型指标缺少这些字段，API 将返回 400 Bad Request。

import requests
import os

def create_metric(
    project_key: str,
    metric_key: str,
    name: str,
    kind: str = "custom",
    is_numeric: bool = True,
    unit: str = "count",
    success_criteria: str = "HigherThanBaseline",
    event_key: str = None,
    description: str = None
):
    """Create a new metric definition in LaunchDarkly."""
    API_TOKEN = "{api_token}"  # token the user provided for this session

    url = f"https://app.launchdarkly.com/api/v2/metrics/{project_key}"

    payload = {
        "key": metric_key,
        "name": name,
        "kind": kind,
        "isNumeric": is_numeric,
        "eventKey": event_key or metric_key
    }

    # Unit and successCriteria are required for numeric custom metrics
    if is_numeric and kind == "custom":
        payload["unit"] = unit
        payload["successCriteria"] = success_criteria

    if description:
        payload["description"] = description

    headers = {
        "Authorization": API_TOKEN,
        "Content-Type": "application/json"
    }

    response = requests.post(url, json=payload, headers=headers)

    if response.status_code == 201:
        print(f"[OK] Created metric: {metric_key}")
        return response.json()
    elif response.status_code == 409:
        print(f"[INFO] Metric already exists: {metric_key}")
        return None
    else:
        print(f"[ERROR] Failed to create metric: {response.status_code}")
        print(f"        {response.text}")
        return None

指标类型：

• custom - 跟踪任意事件（Agent 指标中最常见）
• pageview - 跟踪页面浏览
• click - 跟踪点击事件

成功标准（适用于数值型指标）：

• HigherThanBaseline - 数值越高越好（例如收入、满意度）
• LowerThanBaseline - 数值越低越好（例如错误、延迟）

常见单位：

• count - 通用计数
• milliseconds - 持续时间
• percent - 百分比数值
• dollars - 货币

2. 跟踪事件（SDK）

创建指标后，使用 AI SDK 管理的客户端跟踪事件。get_client() 会返回底层的 LaunchDarkly 服务端 SDK 客户端，因此 track() 接受一个 Context、可选数据和一个数值：

from ldclient import Context
from launchdarkly_ai_server import get_client, init_client

await init_client()  # reads LD_SDK_KEY
ld_client = get_client()

def track_metric(ld_client, user_id: str, metric_key: str, value: float, data: dict = None):
    """Track an event to a metric."""
    context = Context.builder(user_id).build()

    ld_client.track(
        metric_key,
        context,
        data=data,
        metric_value=value
    )

将指标归因于配置变体

若要在配置的内置指标旁查看自定义指标，并按变体进行比较，请将 invoke() 结果中的 track_data（trackData）作为事件数据发送。它包含配置键、变体键、版本和模型：

from ldclient import Context
from launchdarkly_ai_server import config, get_client

context = {"kind": "user", "key": "user-123"}
result = await config(key="support-agent", handler=handler).invoke(user_input, context)

resolved = customer_confirmed_resolution(result.response)
get_client().track(
    "ai.task.completion",
    Context.from_dict(context),
    result.track_data,
    1.0 if resolved else 0.0,
)

import { config, getClient } from '@launchdarkly/ai-node';

const context = { kind: 'user', key: 'user-123' };
const result = await config({ key: 'support-agent', handler }).invoke(userInput, context);

const resolved = customerConfirmedResolution(result.response);
getClient().track('ai.task.completion', context, result.trackData, resolved ? 1 : 0);

注意： invoke() 接受普通 dict 形式的上下文，但 Python 的 track() 需要 ldclient Context，因此请使用 Context.from_dict() 转换。Node.js 在两处都接受相同的对象。

常见跟踪模式

def track_conversion(ld_client, user_id: str, amount: float, config_key: str):
    """Track a conversion event with revenue."""
    context = Context.builder(user_id).build()

    ld_client.track(
        "business.conversion",
        context,
        data={"configKey": config_key, "category": "electronics"},
        metric_value=amount
    )

def track_task_success(ld_client, user_id: str, task_type: str, success: bool):
    """Track task completion success/failure."""
    context = Context.builder(user_id).build()

    ld_client.track(
        "task.success_rate",
        context,
        data={"taskType": task_type},
        metric_value=1.0 if success else 0.0
    )

def track_satisfaction(ld_client, user_id: str, score: float, feedback_type: str):
    """Track user satisfaction (0-100 scale)."""
    context = Context.builder(user_id).build()

    ld_client.track(
        "user.satisfaction",
        context,
        data={"feedbackType": feedback_type},
        metric_value=score
    )

    # Track negative feedback separately for alerts
    if score < 50:
        ld_client.track(
            "user.negative_feedback",
            context,
            metric_value=1.0
        )

def track_revenue(ld_client, user_id: str, revenue: float, source: str):
    """Track revenue generated after agent interaction."""
    context = Context.builder(user_id).set("tier", "premium").build()

    if revenue > 0:
        ld_client.track(
            "revenue.impact",
            context,
            data={"source": source},
            metric_value=revenue
        )

3. 获取指标

使用 get-metric 获取单个指标，使用 list-metrics 获取全部指标。开始跟踪后，list-metric-events 会显示已到达的事件键及其到达时间，这是最快发现 eventKey 与 track() 不匹配的方法。供脚本使用的 REST API 对应操作：

获取单个指标

def get_metric(project_key: str, metric_key: str):
    """Get a single metric definition."""
    API_TOKEN = "{api_token}"  # token the user provided for this session

    url = f"https://app.launchdarkly.com/api/v2/metrics/{project_key}/{metric_key}"

    headers = {"Authorization": API_TOKEN}

    response = requests.get(url, headers=headers)

    if response.status_code == 200:
        metric = response.json()
        print(f"[OK] Metric: {metric['key']}")
        print(f"     Name: {metric.get('name', 'N/A')}")
        print(f"     Kind: {metric.get('kind', 'N/A')}")
        print(f"     Numeric: {metric.get('isNumeric', False)}")
        print(f"     Event Key: {metric.get('eventKey', 'N/A')}")
        return metric
    elif response.status_code == 404:
        print(f"[INFO] Metric not found: {metric_key}")
        return None
    else:
        print(f"[ERROR] Failed to get metric: {response.status_code}")
        return None

列出所有指标

def list_metrics(project_key: str, limit: int = 20):
    """List all metrics in a project."""
    API_TOKEN = "{api_token}"  # token the user provided for this session

    url = f"https://app.launchdarkly.com/api/v2/metrics/{project_key}"

    headers = {"Authorization": API_TOKEN}
    params = {"limit": limit}

    response = requests.get(url, headers=headers, params=params)

    if response.status_code == 200:
        data = response.json()
        metrics = data.get("items", [])
        print(f"[OK] Found {len(metrics)} metrics:")
        for metric in metrics:
            numeric = "numeric" if metric.get("isNumeric") else "non-numeric"
            print(f"     - {metric['key']} ({metric.get('kind', 'custom')}, {numeric})")
        return metrics
    else:
        print(f"[ERROR] Failed to list metrics: {response.status_code}")
        return None

4. 更新指标（API）

def update_metric(project_key: str, metric_key: str, updates: list):
    """
    Update a metric using JSON Patch operations.

    Args:
        updates: List of patch operations, e.g.:
            [{"op": "replace", "path": "/name", "value": "New Name"}]
    """
    API_TOKEN = "{api_token}"  # token the user provided for this session

    url = f"https://app.launchdarkly.com/api/v2/metrics/{project_key}/{metric_key}"

    headers = {
        "Authorization": API_TOKEN,
        "Content-Type": "application/json"
    }

    response = requests.patch(url, json=updates, headers=headers)

    if response.status_code == 200:
        print(f"[OK] Updated metric: {metric_key}")
        return response.json()
    elif response.status_code == 404:
        print(f"[ERROR] Metric not found: {metric_key}")
        return None
    else:
        print(f"[ERROR] Failed to update metric: {response.status_code}")
        print(f"        {response.text}")
        return None

# Example: Update metric name and description
def rename_metric(project_key: str, metric_key: str, new_name: str, new_description: str = None):
    """Rename a metric and optionally update description."""
    updates = [
        {"op": "replace", "path": "/name", "value": new_name}
    ]
    if new_description:
        updates.append({"op": "replace", "path": "/description", "value": new_description})

    return update_metric(project_key, metric_key, updates)

5. 删除指标（API）

def delete_metric(project_key: str, metric_key: str):
    """Delete a metric from the project."""
    API_TOKEN = "{api_token}"  # token the user provided for this session

    url = f"https://app.launchdarkly.com/api/v2/metrics/{project_key}/{metric_key}"

    headers = {"Authorization": API_TOKEN}

    response = requests.delete(url, headers=headers)

    if response.status_code == 204:
        print(f"[OK] Deleted metric: {metric_key}")
        return True
    elif response.status_code == 404:
        print(f"[INFO] Metric not found: {metric_key}")
        return False
    else:
        print(f"[ERROR] Failed to delete metric: {response.status_code}")
        return False

完整工作流示例

import asyncio
from ldclient import Context
from launchdarkly_ai_server import get_client, init_client, shutdown

API_TOKEN = "{api_token}"  # token the user provided for this session
PROJECT_KEY = "support-ai"

async def main():
    # 1. Create metric
    create_metric(
        PROJECT_KEY,
        "ai.task.completion",
        name="Agent Task Completion Rate",
        kind="custom",
        is_numeric=True,
        description="Tracks successful agent task completions"
    )

    # 2. Track events
    await init_client()  # reads LD_SDK_KEY
    ld_client = get_client()
    context = Context.builder("user-123").build()
    ld_client.track("ai.task.completion", context, metric_value=1.0)
    ld_client.track("ai.task.completion", context, metric_value=1.0)
    ld_client.track("ai.task.completion", context, metric_value=0.0)  # failure
    await shutdown()  # flushes events, then closes the client

    # 3. Get metric definition
    metric = get_metric(PROJECT_KEY, "ai.task.completion")

    # 4. Update metric name
    rename_metric(PROJECT_KEY, "ai.task.completion", "Agent Task Success Rate")

    # 5. List all metrics
    list_metrics(PROJECT_KEY)

    # 6. Delete metric (when no longer needed)
    # delete_metric(PROJECT_KEY, "ai.task.completion")

asyncio.run(main())

会话指标跟踪器

import time
from ldclient import Context

class SessionMetricsTracker:
    """Track metrics across an entire user session."""

    def __init__(self, ld_client):
        self.ld_client = ld_client
        self.session_data = {}

    def start_session(self, user_id: str, session_id: str):
        """Initialize session tracking."""
        self.session_data[session_id] = {
            "user_id": user_id,
            "start_time": time.time(),
            "interactions": 0,
            "successful_tasks": 0
        }

    def track_interaction(self, session_id: str, success: bool):
        """Track individual interaction within session."""
        if session_id not in self.session_data:
            return
        session = self.session_data[session_id]
        session["interactions"] += 1
        if success:
            session["successful_tasks"] += 1

    def end_session(self, session_id: str):
        """Finalize and track session metrics."""
        if session_id not in self.session_data:
            return None

        session = self.session_data[session_id]
        duration = time.time() - session["start_time"]

        context = Context.builder(session["user_id"]).build()

        # Track session duration
        self.ld_client.track(
            "session.duration",
            context,
            data={"interactions": session["interactions"]},
            metric_value=duration
        )

        # Track session success rate
        if session["interactions"] > 0:
            success_rate = session["successful_tasks"] / session["interactions"]
            self.ld_client.track(
                "session.success_rate",
                context,
                metric_value=success_rate * 100
            )

        result = dict(session)
        result["duration"] = duration
        del self.session_data[session_id]
        return result

命名约定

# Use dot notation for hierarchy
"quality.accuracy"
"quality.relevance"
"user.satisfaction"
"user.engagement"
"revenue.conversion"
"task.success_rate"
"session.duration"
"ai.task.completion"
"ai.recommendation.conversion"

最佳实践

1. 先创建，再跟踪 - 跟踪事件前，指标必须已存在
2. 使用数值指标 - 设置 isNumeric=True 以进行聚合
3. 使用一致的键 - 在 create_metric() 和 ld_client.track() 中使用相同的键
4. 始终妥善关闭 - 在进程退出前调用 await shutdown()。此操作会刷新待处理事件，然后关闭客户端。否则，无论是短时运行的脚本还是长时间运行的服务，末尾的事件都有丢失的风险。这并非仅适用于无服务器环境的规则；凡是会退出的进程都适用。
5. 限制频率 - 不要在每次击键时都跟踪

查看指标

自定义指标会显示在：

• LaunchDarkly UI 中的 Metrics 页面
• 配置的 Monitoring tab
• 通过 Agent 使用 get-metric、list-metrics 和 list-metric-events MCP 工具
• 通过脚本使用 get_metric() 或 list_metrics()

相关技能

• sdk-install - SDK 设置
• built-in-metrics - 内置 Agent 指标（令牌、时长、成本）
• online-evals - 通过评审器衡量质量指标

参考资料

• 指标 API 文档
• 自定义事件文档
• Python 的 AI SDK 参考
• Node.js 的 AI SDK 参考