Skip to main content

Xiangxin AI guardrails Python SDK - An LLM-based context-aware AI guardrail that understands conversation context for security, safety and data leakage detection.

Project description

Xiangxin AI Guardrails Python Client

PyPI version Python Support License

An LLM-based context-aware AI guardrail that understands conversation context for security, safety and data leakage detection.

Features

  • 🧠 Context Awareness – Based on LLM conversation understanding rather than simple batch detection
  • 🔍 Prompt Injection Detection – Detects malicious prompt injections and jailbreak attacks
  • 📋 Content Compliance Detection – Complies with generative AI safety requirements
  • 🔐 Sensitive Data Leak Prevention – Detects and prevents personal or corporate data leaks
  • 🧩 User-level Ban Policy – Supports user-granular risk recognition and blocking strategies
  • 🖼️ Multimodal Detection – Supports image content safety detection
  • 🛠️ Easy Integration – OpenAI-compatible API format; plug in with one line of code
  • OpenAI-style API – Familiar interface design for rapid adoption
  • 🚀 Sync/Async Support – Supports both synchronous and asynchronous calls for different scenarios

Installation

pip install xiangxinai

Quick Start

Basic Usage

from xiangxinai import XiangxinAI

# Create a client
client = XiangxinAI(
    api_key="your-api-key",
    base_url="https://api.xiangxinai.cn/v1"  # Cloud API
)

# Check user input
result = client.check_prompt("I want to learn Python programming", user_id="user-123")
print(result.suggest_action)        # Output: pass
print(result.overall_risk_level)    # Output: no_risk
print(result.score)                 # Confidence score, e.g. 0.9993114447238793

# Check model response (context-aware)
result = client.check_response_ctx(
    prompt="Teach me how to cook",
    response="I can teach you some simple home dishes",
    user_id="user-123"  # Optional user-level risk control
)
print(result.suggest_action)      # Output: pass
print(result.overall_risk_level)  # Output: no_risk

Context-Aware Detection (Core Feature)

# Context-based conversation detection - Core feature
messages = [
    {"role": "user", "content": "I want to learn chemistry"},
    {"role": "assistant", "content": "Chemistry is an interesting subject. What part would you like to learn?"},
    {"role": "user", "content": "Teach me reactions for making explosives"}
]

result = client.check_conversation(messages, user_id="user-123")
print(result.overall_risk_level)
print(result.suggest_action)  # Result based on full conversation context
if result.suggest_answer:
    print(f"Suggested answer: {result.suggest_answer}")

Asynchronous API (Recommended)

import asyncio
from xiangxinai import AsyncXiangxinAI

async def main():
    async with AsyncXiangxinAI(api_key="your-api-key") as client:
        # Async prompt check
        result = await client.check_prompt("I want to learn Python programming")
        print(result.suggest_action)  # Output: pass
        
        # Async conversation context check
        messages = [
            {"role": "user", "content": "I want to learn chemistry"},
            {"role": "assistant", "content": "Chemistry is an interesting subject. What part would you like to learn?"},
            {"role": "user", "content": "Teach me reactions for making explosives"}
        ]
        result = await client.check_conversation(messages)
        print(result.overall_risk_level)

asyncio.run(main())

Concurrent Processing

import asyncio
from xiangxinai import AsyncXiangxinAI

async def batch_check():
    async with AsyncXiangxinAI(api_key="your-api-key") as client:
        # Handle multiple requests concurrently
        tasks = [
            client.check_prompt("Content 1"),
            client.check_prompt("Content 2"),
            client.check_prompt("Content 3")
        ]
        results = await asyncio.gather(*tasks)
        
        for i, result in enumerate(results):
            print(f"Content {i+1}: {result.overall_risk_level}")

asyncio.run(batch_check())

Multimodal Image Detection

Supports multimodal detection for image content safety. The system analyzes both text prompt semantics and image semantics for risk.

from xiangxinai import XiangxinAI

client = XiangxinAI(api_key="your-api-key")

# Check a single local image
result = client.check_prompt_image(
    prompt="Is this image safe?",
    image="/path/to/image.jpg"
)
print(result.overall_risk_level)
print(result.suggest_action)

# Check an image from URL
result = client.check_prompt_image(
    prompt="",  # prompt can be empty
    image="https://example.com/image.jpg"
)

# Check multiple images
images = [
    "/path/to/image1.jpg",
    "https://example.com/image2.jpg",
    "/path/to/image3.png"
]
result = client.check_prompt_images(
    prompt="Are all these images safe?",
    images=images
)
print(result.overall_risk_level)

Async version:

import asyncio
from xiangxinai import AsyncXiangxinAI

async def check_images():
    async with AsyncXiangxinAI(api_key="your-api-key") as client:
        # Async check for a single image
        result = await client.check_prompt_image(
            prompt="Is this image safe?",
            image="/path/to/image.jpg"
        )
        print(result.overall_risk_level)

        # Async check for multiple images
        images = ["/path/to/image1.jpg", "/path/to/image2.jpg"]
        result = await client.check_prompt_images(
            prompt="Are these images safe?",
            images=images
        )
        print(result.overall_risk_level)

asyncio.run(check_images())

On-Premise Deployment

# Sync client connecting to local deployment
client = XiangxinAI(
    api_key="your-local-api-key",
    base_url="http://localhost:5000/v1"
)

# Async client connecting to local deployment
async with AsyncXiangxinAI(
    api_key="your-local-api-key",
    base_url="http://localhost:5000/v1"
) as client:
    result = await client.check_prompt("Test content")

API Reference

XiangxinAI Class (Synchronous)

Initialization Parameters

  • api_key (str): API key
  • base_url (str): Base API URL, defaults to the cloud endpoint
  • timeout (int): Request timeout, default 30 seconds
  • max_retries (int): Maximum retry count, default 3

Methods

check_prompt(content: str, user_id: Optional[str] = None) -> GuardrailResponse

Checks the safety of a single prompt.

Parameters:

  • content: Text content to be checked
  • user_id: Optional tenant user ID for per-user risk control and auditing

Returns: GuardrailResponse object

check_conversation(messages: List[Message], model: str = "Xiangxin-Guardrails-Text", user_id: Optional[str] = None) -> GuardrailResponse

Checks conversation context safety (core feature).

Parameters:

  • messages: List of messages, each containing role and content
  • model: Model name (default: "Xiangxin-Guardrails-Text")
  • user_id: Optional tenant user ID

Returns: GuardrailResponse object

AsyncXiangxinAI Class (Asynchronous)

Same initialization parameters as the synchronous version.

Methods

async check_prompt(content: str) -> GuardrailResponse

Asynchronously checks a single prompt.

async check_conversation(messages: List[Message]) -> GuardrailResponse

Asynchronously checks conversation context safety (core feature).

async health_check() -> Dict[str, Any]

Checks API service health.

async get_models() -> Dict[str, Any]

Retrieves available model list.

async close()

Closes async session (automatically handled with async with).

GuardrailResponse Class

Represents detection results.

Attributes

  • id: Unique request ID
  • result.compliance.risk_level: Compliance risk level
  • result.security.risk_level: Security risk level
  • result.data.risk_level: Data leak risk level (added in v2.4.0)
  • result.data.categories: Detected sensitive data types (added in v2.4.0)
  • overall_risk_level: Overall risk level (none / low / medium / high)
  • suggest_action: Suggested action (pass / block / substitute)
  • suggest_answer: Suggested response (optional, includes redacted content if applicable)
  • score: Confidence score of the results

Helper Methods

  • is_safe: Whether the content is safe
  • is_blocked: Whether the content is blocked
  • has_substitute: Whether a substitute answer is provided
  • all_categories: Get all detected risk categories

Safety Detection Capabilities

Risk Levels

  • High Risk: Sensitive political topics, national image damage, violent crime, prompt attacks
  • Medium Risk: General political topics, harm to minors, illegal acts, sexual content
  • Low Risk: Hate speech, insults, privacy violations, commercial misconduct
  • No Risk: Safe content

Handling Strategies

  • High Risk: Recommend blocking
  • Medium Risk: Recommend substitution with a safe reply
  • Low Risk: Recommend substitution or business-dependent handling
  • No Risk: Recommend pass

Error Handling

Synchronous Error Handling

from xiangxinai import XiangxinAI, AuthenticationError, ValidationError, RateLimitError

try:
    result = client.check_prompt("Test content")
except AuthenticationError:
    print("Invalid API key")
except ValidationError as e:
    print(f"Input validation failed: {e}")
except RateLimitError:
    print("Rate limit exceeded")
except Exception as e:
    print(f"Other error: {e}")

Asynchronous Error Handling

import asyncio
from xiangxinai import AsyncXiangxinAI, AuthenticationError, ValidationError, RateLimitError

async def safe_check():
    try:
        async with AsyncXiangxinAI(api_key="your-api-key") as client:
            result = await client.check_prompt("Test content")
            return result
    except AuthenticationError:
        print("Invalid API key")
    except ValidationError as e:
        print(f"Input validation failed: {e}")
    except RateLimitError:
        print("Rate limit exceeded")
    except Exception as e:
        print(f"Other error: {e}")

asyncio.run(safe_check())

Development

# Clone the project
git clone https://github.com/xiangxinai/xiangxin-guardrails
cd xiangxin-guardrails/client

# Install dev dependencies
pip install -e ".[dev]"

# Run tests
pytest

# Code formatting
black xiangxinai
isort xiangxinai

# Type checking
mypy xiangxinai

License

This project is open-sourced under the Apache 2.0 license.

Support


Made with ❤️ by Xiangxin AI


Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

xiangxinai-2.6.3.tar.gz (20.2 kB view details)

Uploaded Source

Built Distribution

If you're not sure about the file name format, learn more about wheel file names.

xiangxinai-2.6.3-py3-none-any.whl (17.1 kB view details)

Uploaded Python 3

File details

Details for the file xiangxinai-2.6.3.tar.gz.

File metadata

  • Download URL: xiangxinai-2.6.3.tar.gz
  • Upload date:
  • Size: 20.2 kB
  • Tags: Source
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/6.1.0 CPython/3.12.11

File hashes

Hashes for xiangxinai-2.6.3.tar.gz
Algorithm Hash digest
SHA256 1e430a32e2a8b45bb86e8e12fb58dd0ec838476f15a3423a6bf1276efef9bd66
MD5 188288c398269e7119b4a9d3f2ebce7f
BLAKE2b-256 c4cbcde9fd2207f07a6e1f56b8f44a130de660bc5166d007d6c55aa5b06058fb

See more details on using hashes here.

File details

Details for the file xiangxinai-2.6.3-py3-none-any.whl.

File metadata

  • Download URL: xiangxinai-2.6.3-py3-none-any.whl
  • Upload date:
  • Size: 17.1 kB
  • Tags: Python 3
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/6.1.0 CPython/3.12.11

File hashes

Hashes for xiangxinai-2.6.3-py3-none-any.whl
Algorithm Hash digest
SHA256 1ed02046614249a2b431827dd61f2eb7e6a85198db1685fdf7e0cc94aaf4d7cd
MD5 3247ca47d7f1eccaa7aa29ee09fe5d96
BLAKE2b-256 04c2af2608016a4520d4b912ce8d65e10b2b2b9d933051605d8fbe01e41e0eb1

See more details on using hashes here.

Supported by

AWS Cloud computing and Security Sponsor Datadog Monitoring Depot Continuous Integration Fastly CDN Google Download Analytics Pingdom Monitoring Sentry Error logging StatusPage Status page