mcpField
C
custom-metrics

custom-metrics

github.com/launchdarklyデータ・データベース281.8k 回インストール1 日前に更新
リポジトリ

custom-metrics について

LaunchDarkly AI SDKを使う開発者向けに、カスタム業務指標の定義、イベント記録、取得、APIでの更新・削除を案内します。

主な機能

  1. 01MCP経由で指標を作成
  2. 02SDKでイベントを記録
  3. 03定義とイベントを取得
  4. 04REST APIで更新・削除

活用シーン

  1. 01エージェントのタスク完了率を測定
  2. 02指標を設定バリエーションに関連付け
  3. 03指標イベントキーの不一致を特定

MCP 対応

リモートのLaunchDarkly MCPサーバーで指標の作成・取得とイベント確認を行います。更新と削除にはREST APIを使います。

インストール

コマンドをコピーしてターミナルで実行します。

$npx skills add https://github.com/launchdarkly/ai-tooling --skill custom-metrics

インストール後、Claude Code で/custom-metricsを入力して読み込まれたか確認します。

詳細

Config 向けカスタムメトリクス

カスタムビジネスメトリクスのライフサイクル全体を管理します。MCP サーバーでメトリクス定義を作成し、SDK でイベントを追跡して、メトリクスデータを取得し、メトリクスをプログラムから管理できます。

前提条件

  • LaunchDarkly AI SDK が初期化されていること(sdk-install を参照)。SDK は環境変数 LD_SDK_KEY を読み込みます。
  • お使いの環境でリモートホスト型の LaunchDarkly MCP サーバーが設定されていること
  • メトリクスの更新または削除を行う場合、あるいは MCP を使用できないスクリプトの場合にのみ、writer ロールを持つ LaunchDarkly API トークン
  • 組み込みエージェントメトリクスに関する知識(built-in-metrics を参照)

API アクセス

対応するツールがある場合は、LaunchDarkly MCP サーバーのツールを使用してください。これらのツールは OAuth で認証されるため、トークンは不要です。create-metric でメトリクスを作成し、list-metric-events でイベントが届いていることを確認し、get-metric と list-metrics で定義を読み取ります。MCP サーバーではメトリクスを更新または削除できないため、これらの手順では REST API を使用します。その場合、または MCP を使用できないスクリプトの場合は、このセッションで使用する API アクセストークンをユーザーに貼り付けてもらい、以下の例の {api_token} に代入してください。認証情報を探すために、環境変数、.env ファイル、エージェント/MCP 設定ファイルを検索してはいけません。

メトリクスのライフサイクル概要

手順方法目的
1. 作成MCP create-metricLaunchDarkly でメトリクスを定義する
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 - 任意のイベントを追跡します(エージェントメトリクスで最も一般的です)
  • 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
    )
Config のバリエーションにメトリクスを関連付ける

Config の組み込みメトリクスの横にカスタムメトリクスを表示し、バリエーション間で比較するには、invoke() の結果から取得した track_data(trackData)をイベントデータとして送信します。このデータには、Config キー、バリエーションキー、バージョン、モデルが含まれます。

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 のメトリクスページ
  • 設定のモニタリングタブ
  • エージェントから、get-metric、list-metrics、list-metric-events MCP ツールを使用
  • スクリプトから、get_metric() または list_metrics() を使用

関連スキル

  • sdk-install - SDK のセットアップ
  • built-in-metrics - 組み込みエージェントメトリクス(トークン、期間、コスト)
  • online-evals - ジャッジを使用した品質メトリクス

参考資料

コメント