online-evals

curl -X POST "https://app.launchdarkly.com/api/v2/projects/{projectKey}/ai-configs/security-judge/variations" \
  -H "Authorization: {api_token}" \
  -H "Content-Type: application/json" \
  -H "LD-API-Version: beta" \
  -d '{
    "key": "default",
    "name": "Default",
    "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."
      }
    ],
    "modelConfigKey": "OpenAI.gpt-4o-mini",
    "model": {
      "parameters": {
        "temperature": 0.3
      }
    }
  }'

curl -X PATCH "https://app.launchdarkly.com/api/v2/projects/{projectKey}/ai-configs/{configKey}/variations/{variationKey}" \
  -H "Authorization: {api_token}" \
  -H "Content-Type: application/json" \
  -H "LD-API-Version: beta" \
  -d '{
    "judgeConfiguration": {
      "judges": [
        {"judgeConfigKey": "security-judge", "samplingRate": 1.0},
        {"judgeConfigKey": "api-contract-judge", "samplingRate": 0.5}
      ]
    }
  }'

# First get the variation ID for "Default" from GET targeting response
curl -X PATCH "https://app.launchdarkly.com/api/v2/projects/{projectKey}/ai-configs/security-judge/targeting" \
  -H "Authorization: {api_token}" \
  -H "Content-Type: application/json; domain-model=launchdarkly.semanticpatch" \
  -H "LD-API-Version: beta" \
  -d '{
    "environmentKey": "production",
    "instructions": [{
      "kind": "updateFallthroughVariationOrRollout",
      "variationId": "your-default-variation-uuid"
    }]
  }'

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 (appears as $ld:ai:judge:{metric_key})
            system_prompt: Evaluation instructions
            is_inverted: True if lower scores are better (e.g., toxicity)
        """
        # 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 {}
 
        # 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", []):
            if var.get("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

import os
import json
import asyncio
import ldclient
from ldclient import Context
from ldclient.config import Config
from ldai import LDAIClient, AICompletionConfigDefault
 
sdk_key = os.getenv('LAUNCHDARKLY_SDK_KEY')
ai_config_key = os.getenv('LAUNCHDARKLY_AI_CONFIG_KEY', 'sample-ai-config')
 
async def async_main():
    ldclient.set_config(Config(sdk_key))
    aiclient = LDAIClient(ldclient.get())
 
    context = (
        Context.builder('example-user-key')
        .kind('user')
        .name('Sandy')
        .build()
    )
 
    default_value = AICompletionConfigDefault(enabled=False)
 
    # create_model() initializes with judges from Config
    model = await aiclient.create_model(ai_config_key, context, default_value, {})
 
    if not model:
        print(f"agent configuration not enabled for: {ai_config_key}")
        return
 
    user_input = 'How can LaunchDarkly help me?'
 
    # run() automatically evaluates with attached judges
    result = await model.run(user_input)
    print("Response:", result.content)
 
    # Await evaluation results
    if result.evaluations and len(result.evaluations) > 0:
        eval_results = await asyncio.gather(*result.evaluations)
        results_to_display = [
            r.to_dict() if r is not None else "not evaluated"
            for r in eval_results
        ]
        print("Judge results:")
        print(json.dumps(results_to_display, indent=2, default=str))
 
    # Always flush events before closing — trailing events are at risk of being
    # lost otherwise, in short-lived scripts and long-running services alike.
    ldclient.get().flush()
    ldclient.get().close()

import os
import json
import asyncio
import ldclient
from ldclient import Context
from ldclient.config import Config
from ldai import LDAIClient, AIJudgeConfigDefault
 
sdk_key = os.getenv('LAUNCHDARKLY_SDK_KEY')
judge_key = os.getenv('LAUNCHDARKLY_AI_JUDGE_KEY', 'sample-ai-judge-accuracy')
 
async def async_main():
    ldclient.set_config(Config(sdk_key))
    aiclient = LDAIClient(ldclient.get())
 
    context = (
        Context.builder('example-user-key')
        .kind('user')
        .name('Sandy')
        .build()
    )
 
    judge_default_value = AIJudgeConfigDefault(enabled=False)
 
    # Get judge configuration from LaunchDarkly
    judge = aiclient.create_judge(judge_key, context, judge_default_value)
 
    if not judge:
        print(f"agent judge configuration not enabled for key: {judge_key}")
        return
 
    input_text = 'You are a helpful assistant. How can you help me?'
    output_text = 'I can answer any question you have.'
 
    # Evaluate the input/output pair — returns a JudgeResult.
    judge_result = await judge.evaluate(input_text, output_text)
 
    if not judge_result.sampled:
        print("Judge evaluation was skipped (sample rate or configuration issue)")
        return
 
    # Track the consolidated result on the Config tracker if needed:
    # tracker = ai_config.create_tracker()
    # tracker.track_judge_result(judge_result)
 
    print("Judge Result:")
    print(json.dumps(judge_result.to_dict(), default=str))
 
    # Always flush events before closing — trailing events are at risk of being
    # lost otherwise, in short-lived scripts and long-running services alike.
    ldclient.get().flush()
    ldclient.get().close()