ZENETICS API - Guard

POST https://api.zenetics.io/guard/v1/check

{
    "guard_result_id": 12345,
    "status": "queued",
    "message": "Guard evaluation queued successfully",
    "estimated_completion": "2025-06-02T10:35:00Z"
}

{
    "error": "validation_error",
    "message": "Invalid guard group ID",
    "details": {
        "field": "guardGroupId",
        "code": "invalid"
    }
}

{
    "error": "authentication_error",
    "message": "Invalid or missing API key"
}

{
    "error": "resource_not_found",
    "message": "Guard group not found"
}

{
    "refId": "ref-1234",
    "guardResult_id": 12345,
    "guardStatus": "passed",
    "guardResults": [
        {
            "id": 67890,
            "accountId": 123,
            "applicationId": 456,
            "guard_groupId": 4,
            "guardGroupSeqId": 1,
            "name": "Toxicity Check",
            "status": "passed"
        },
        {
            "id": 67891,
            "accountId": 123,
            "applicationId": 456,
            "guardGroupId": 4,
            "guardGroupSeqId": 2,
            "name": "Bias Detection",
            "status": "failed"
        }
    ]
}

curl -X POST https://api.zenetics.io/guard/v1/check \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer your-api-key-here" \
  -d '{
    "generationContext": {
      "input": "What are the benefits of our premium subscription plan?",
      "output": "Our premium subscription offers unlimited access to all features, priority customer support, advanced analytics, and exclusive content. You'\''ll also get early access to new features and a dedicated account manager.",
      "retrievalContext": [
        "Premium Plan Features: Unlimited access, priority support, advanced analytics",
        "Customer Benefits: Early feature access, dedicated account manager, exclusive content"
      ],
      "promptTemplate": "You are a helpful sales assistant. Answer the customer'\''s question about our subscription plans: {input}",
      "metadata": {
        "provider": "openai",
        "model": "chatgpt-4o"
      }
    },
    "guardGroupId": 4,
    "callbackConfig": {
      "refId": "sales-query-001",
      "webhookUrl": "https://api.yourcompany.com/zenetics/guards/callback",
      "webhookHeaders": {
        "Authorization": "Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9..."
      }
    },
    "options": {
      "notification": true
    }
  }'

const response = await fetch('https://api.zenetics.io/guard/v1/check', {
    method: 'POST',
    headers: {
        'Content-Type': 'application/json',
        Authorization: 'Bearer your-api-key-here',
    },
    body: JSON.stringify({
        generationContext: {
            input: 'What are the benefits of our premium subscription plan?',
            output: "Our premium subscription offers unlimited access to all features, priority customer support, advanced analytics, and exclusive content. You'll also get early access to new features and a dedicated account manager.",
            retrievalContext: [
                'Premium Plan Features: Unlimited access, priority support, advanced analytics',
                'Customer Benefits: Early feature access, dedicated account manager, exclusive content',
            ],
            promptTemplate:
                "You are a helpful sales assistant. Answer the customer's question about our subscription plans: {input}",
            metadata: {
                provider: 'openai',
                model: 'chatgpt-4o',
            },
        },
        guardGroupId: 4,
        callbackConfig: {
            ref_id: 'sales-query-001',
            webhookUrl: 'https://api.yourcompany.com/zenetics/guards/callback',
            webhookHeaders: {
                Authorization: 'Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...',
            },
        },
        options: {
            notification: true,
        },
    }),
});
 
const result = await response.json();
console.log('Guard check queued:', result.jobId);

import requests
import json
 
url = "https://api.zenetics.io/guard/v1/check"
headers = {
    "Content-Type": "application/json",
    "Authorization": "Bearer your-api-key-here"
}
 
data = {
    "generationContext": {
        "input": "What are the benefits of our premium subscription plan?",
        "output": "Our premium subscription offers unlimited access to all features, priority customer support, advanced analytics, and exclusive content. You'll also get early access to new features and a dedicated account manager.",
        "retrievalContext": [
            "Premium Plan Features: Unlimited access, priority support, advanced analytics",
            "Customer Benefits: Early feature access, dedicated account manager, exclusive content"
        ],
        "promptTemplate": "You are a helpful sales assistant. Answer the customer's question about our subscription plans: {input}",
        "metadata": {
            "provider": "openai",
            "model": "chatgpt-4o"
        }
    },
    "guardGroupId": 4,
    "callbackConfig": {
        "refId": "sales-query-001",
        "webhookUrl": "https://api.yourcompany.com/zenetics/guards/callback",
        "webhookHeaders": {
            "Authorization": "Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9..."
        }
    },
    "options": {
        "notification": True
    }
}
 
response = requests.post(url, headers=headers, json=data)
result = response.json()
print(f"Guard check queued: {result['jobId']}")

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import List, Optional
from enum import Enum
 
app = FastAPI()
 
class EvaluationResultState(str, Enum):
    PASSED = "passed"
    FAILED = "failed"
    SKIPPED = "skipped"
 
class GuardResult(BaseModel):
    id: int
    accountId: int
    applicationId: int
    guardGroupId: int
    guardGroupSeqId: int
    name: str
    status: EvaluationResultState
 
class CallbackRequest(BaseModel):
    jobId: str
    refId: Optional[str] = None
    guardResultId: int
    guardStatus: EvaluationResultState
    guardResults: List[GuardResult]
 
@app.post("/zenetics/guards/callback")
async def handle_guard_callback(callback: CallbackRequest):
    """Handle guard evaluation results from ZENETICS"""
 
    print(f"Received guard results for ref_id: {callback.ref_id}")
    print(f"Overall status: {callback.guard_status}")
 
    # Process individual guard results
    for guard_result in callback.guard_results:
        if guard_result.status == EvaluationResultState.FAILED:
            print(f"⚠️  Guard '{guard_result.name}' failed!")
            # Handle failed guard (e.g., log, alert, take corrective action)
            await handle_failed_guard(guard_result)
        else:
            print(f"✅ Guard '{guard_result.name}' passed")
 
    # Acknowledge receipt
    return {"status": "received", "guardResultId": callback.guardResultId}
 
async def handle_failed_guard(guard_result: GuardResult):
    """Handle a failed guard evaluation"""
    # Implement your business logic for handling failed guards
    # Examples: logging, alerting, blocking content, triggering reviews
    pass

@app.post("/zenetics/guards/callback")
async def handleGuardCallback(callback: CallbackRequest):
    try:
        # Process the callback
        await processGuardResults(callback)
        return {"status": "success"}
    except Exception as e:
        # Log the error but still return 200 to prevent retries
        logger.error(f"Error processing guard callback: {e}")
        return {"status": "error", "message": str(e)}

async def generate_and_guard_content(user_input: str):
    # Generate content with your AI model
    ai_response = await generate_ai_response(user_input)
 
    # Submit for guard evaluation
    guard_response = await submit_to_guards(user_input, ai_response)
 
    # Return content immediately, handle guards asynchronously
    return {
        "content": ai_response,
        "guardCheckId": guard_response["jobId"]
    }

async def process_content_batch(content_items: List[dict]):
    guard_checks = []
 
    for item in content_items:
        guard_response = await submit_to_guards(
            item["input"],
            item["output"]
        )
        guard_checks.append(guard_response["jobId"])
 
    return guard_checks