Integrate with CI/CD

Automating Snowglobe testing in your development pipeline

Snowglobe SDK Integration Guide

Still on snowglobe<=0.4.x? See the documentation for previous versions here.

Overview

Snowglobe is a simulation engine designed for testing and evaluating AI agents and chatbots through automated conversation generation and analysis. This guide demonstrates how to integrate Snowglobe into your continuous integration (CI) pipeline and programmatic workflows for comprehensive agent testing.

Table of Contents

• Prerequisites
• Authentication Setup
• Core Components
• CI Integration Workflow
• Troubleshooting

Prerequisites

Before integrating Snowglobe, ensure you have:

• Python 3.11+ installed
• Snowglobe SDK package (pip install snowglobe-sdk)
• Valid API credentials (API key and Organization ID) from here.
• OpenAI API key (or other supported LLM provider credentials)
• Access to a Snowglobe control plane instance

Required Dependencies

import asyncio
from snowglobe.sdk import SnowglobeClient
from snowglobe.sdk.types import AgentCreateSchema, SimulationCreateSchema

Authentication Setup

Environment Variables

Set up your credentials as environment variables or configuration constants:

X_API_KEY = "eyJhbM..."  # Your Snowglobe API key
X_SNOWGLOBE_ORG_ID = "orZu..."  # Your organization ID
OPENAI_API_KEY = "sk-prnZS..."  # Your LLM provider API key

Client Configuration

Initialize the Snowglobe client with proper authentication headers:

control_plane_url = "https://api.snowglobe.guardrailsai.com"  # Or your hosted instance

client = SnowglobeClient(
    api_key=X_API_KEY,
    organization_id=X_SNOWGLOBE_ORG_ID,
    base_url=control_plane_url,
    follow_redirects=True
)

Core Components

1. Agent Creation

Agents represent the AI systems you want to test. Each agent requires a name, icon, and connection info describing how Snowglobe should reach your LLM.

agent_config: AgentCreateSchema = {
    "name": "customer_support_agent",
    "description": "Customer support agent for Amatto's pizza restaurant",
    "icon": "pizza",
    "connection_info": {
        "provider": "OpenAI",
        "endpoint": "",
        "model_name": "openai/gpt-4o",
        "api_key_ref": "OPENAI_API_KEY",
        "system_prompt": "You are a helpful expert customer support agent for Amatto's pizza.",
        "extra_body": [],
        "extra_headers": [],
    },
}

Note: Agents using a custom code integration via snowglobe-connect require a two-step setup:

1. Create the agent with the API as shown above.
2. Map the agent’s ID in your snowglobe-connect deployment’s agents.json file.
3. Run snowglobe-connect start.

2. Simulation Configuration

Simulations define how conversations are generated and evaluated against your agent.

simulation_config: SimulationCreateSchema = {
    "name": "continuous integration simulation",
    "role": "Customer support agent for Amatto's pizza restaurant",
    "user_description": "",
    "use_cases": "",
    "generation_status": "pending",
    "evaluation_status": "pending",
    "validation_status": "pending",
    "source_data": {
        "docs": {
            "misc": [],
            "knowledge_base": [],
            "historical_data": []
        },
        "evaluation_configuration": {
            "No Financial Advice": {
                "id": "e5af8dee-6d8d-4144-b754-204d24879ec9",
                "name": "No Financial Advice",
                "version": 1,
                "metadata": {},
            },
        },
        "generation_configuration": {
            "max_topics": 1,
            "max_personas": 4,
            "branching_factor": 25,
            "max_conversations": 500,
            "max_conversation_length": 4,
            "continue_conversations_from_adapted_messages": False,
            "data_gen_mode": "coverage_focused_v3",
            "intent": "",
            "persona_topic_generators": [
                {
                    "name": "app_description_system_prompt",
                    "settings": {
                        "max_personas": 4
                    }
                }
            ],
            "min_conversation_length": 1
        }
    },
    "is_template": False
}

---

CI Integration Workflow

Step 1: Create and Configure Agent

async def create_agent(client, agent_config):
    """Create a new agent for testing."""
    agent = await client.agents.create_agent(agent_config)
    print(f"Agent created successfully: {agent.id}")
    return agent.id

Step 2: Launch Simulation

async def launch_simulation(client, agent_id, simulation_config):
    """Create and launch a new simulation."""
    config = {
        **simulation_config,
        "application_id": agent_id,
        "app_id": agent_id,
    }
    simulation = await client.simulations.create_simulation(config)
    simulation_id = simulation.id

    # Auto-approve personas for CI automation
    await client.simulations.update_simulation_settings(
        simulation_id, {"autoApprovePersonas": True}
    )
    return simulation_id

Step 3: Monitor Simulation Progress

import asyncio

async def wait_for_completion(client, simulation_id, timeout_minutes=20):
    """Poll simulation until completion or timeout."""
    max_attempts = timeout_minutes * 6  # poll every 10 seconds
    poll_interval = 10

    for _ in range(max_attempts):
        sim = await client.simulations.get_simulation(simulation_id)
        state_num = sim.state_num
        print(f"Simulation state: {state_num}")

        # state_num >= 17 indicates completion (see Simulation States below)
        if state_num is not None and state_num >= 17:
            print("Simulation completed successfully")
            return True

        await asyncio.sleep(poll_interval)

    raise TimeoutError("Simulation timed out before completion")

Step 4: Retrieve Results

import json

async def download_results(client, simulation_id):
    """Download simulation results for analysis."""
    data = await client.simulations.download_simulation_data(simulation_id)

    filename = f"{simulation_id}_results.json"
    with open(filename, "w") as f:
        json.dump([d.model_dump() for d in data], f, indent=2)

    print(f"Results saved to {filename}")
    return filename

---

Complete CI Integration Example

import asyncio
import json
import os

from snowglobe.sdk import SnowglobeClient

async def run_agent_simulation(agent_config, simulation_config):
    """Complete workflow for running agent simulations in CI."""
    client = SnowglobeClient(
        api_key=os.environ["SNOWGLOBE_API_KEY"],
        organization_id=os.environ.get("SNOWGLOBE_ORG_ID"),
    )

    try:
        agent_id = await create_agent(client, agent_config)
        simulation_id = await launch_simulation(client, agent_id, simulation_config)
        await wait_for_completion(client, simulation_id)
        results_file = await download_results(client, simulation_id)

        return {
            "success": True,
            "agent_id": agent_id,
            "simulation_id": simulation_id,
            "results_file": results_file,
        }

    except Exception as e:
        print(f"Simulation failed: {e}")
        return {"success": False, "error": str(e)}


asyncio.run(
    run_agent_simulation(agent_config, simulation_config)
)

Error Handling

async def robust_simulation_run(agent_config, simulation_config):
    """Simulation run with comprehensive error handling."""
    import httpx

    try:
        return await run_agent_simulation(agent_config, simulation_config)
    except ValidationError as e:
        print(f"Configuration validation failed: {e}")
        return {"success": False, "error": "validation", "details": str(e)}
    except httpx.HTTPStatusError as e:
        print(f"API error {e.response.status_code}: {e.response.text}")
        return {"success": False, "error": "api_error", "details": str(e)}
    except TimeoutError as e:
        print(f"Simulation timed out: {e}")
        return {"success": False, "error": "timeout", "details": str(e)}
    except Exception as e:
        print(f"Unexpected error: {e}")
        return {"success": False, "error": "unexpected", "details": str(e)}

CI Pipeline Integration

# Example GitHub Actions workflow
name: Agent Testing
on: [push, pull_request]

jobs:
  test-agent:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.11"
      - name: Install dependencies
        run: pip install snowglobe-sdk
      - name: Run agent simulation
        env:
          SNOWGLOBE_API_KEY: ${{ secrets.SNOWGLOBE_API_KEY }}
          SNOWGLOBE_ORG_ID: ${{ secrets.SNOWGLOBE_ORG_ID }}
          OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
        run: python test_agent_simulation.py

---

Simulation States

The state_num field on a simulation indicates its current phase:

state_num	State name	Description
0–2	Draft / Queued	Simulation created, waiting to start
3–5	Experiment started	Initialization and setup
6–8	Generation in progress	Persona, topic, and conversation generation
9–11	Evaluation in progress	Agent responses being judged against risks
12–14	Validation in progress	Results validated
15–16	Adaptation in progress	Adapted (adversarial) conversations being generated
17+	Experiment completed	Results available; download_simulation_data can be called

Poll sim.state_num and wait for >= 17 before downloading results.

---

Troubleshooting

Common Issues

Authentication errors

• Verify your API key and organization ID are correct.
• Confirm the x-api-key header is being sent (the SDK sets this automatically from api_key).
• Check network connectivity to your control plane URL.

Simulation failures

• Review the agent’s connection_info — all required fields (api_key_ref, model_name, system_prompt) must be present.
• Verify the LLM provider API key referenced by api_key_ref is valid and has sufficient quota.
• Check that source_data.generation_configuration values are within acceptable ranges.

Timeout issues

• Increase timeout_minutes in wait_for_completion for large simulations.
• Reduce max_personas, max_topics, or branching_factor for faster runs.

Debugging Tips

async def debug_simulation(client, simulation_id):
    """Print detailed simulation status for debugging."""
    sim = await client.simulations.get_simulation(simulation_id)

    print(f"State:             {sim.state} (num={sim.state_num})")
    print(f"Generation status: {sim.generation_status}")
    print(f"Evaluation status: {sim.evaluation_status}")
    print(f"Validation status: {sim.validation_status}")
    print(f"Status reason:     {sim.status_reason}")
    print(f"Statistics:        {sim.statistics}")