Examples

This section provides comprehensive examples demonstrating Aisert usage patterns from basic configuration to production deployment.

Configuration Examples

Learn different ways to configure Aisert for various providers and use cases.

Download: 01_configuration_examples.py

Configuration Options and Patterns

"""
Configuration Examples - Different ways to configure Aisert

Shows all configuration options and patterns.
"""

import sys
import os
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

from aisert import AisertConfig


def default_configuration():
    """Using default configuration - no setup required."""
    print("=== Default Configuration ===")
    
    config = AisertConfig.get_default_config()
    print(f"Result: {config}")
    # Expected: AisertConfig(token(openai, gpt-3.5-turbo), semantic(openai, text-embedding-3-small))


def constructor_configuration():
    """Using constructor parameters for custom configuration."""
    print("\n=== Constructor Configuration ===")
    
    # Token counting only
    config1 = AisertConfig(token_provider="openai", token_model="gpt-4")
    print(f"Token only: {config1}")
    # Expected: AisertConfig(token(openai, gpt-4))
    
    # Semantic validation only
    config2 = AisertConfig(semantic_provider="sentence_transformers", semantic_model="all-MiniLM-L6-v2")
    print(f"Semantic only: {config2}")
    # Expected: AisertConfig(semantic(sentence_transformers, all-MiniLM-L6-v2))
    
    # Full configuration
    config3 = AisertConfig(
        token_provider="anthropic",
        token_model="claude-3",
        semantic_provider="tfidf"
    )
    print(f"Full config: {config3}")
    # Expected: AisertConfig(token(anthropic, claude-3), semantic(tfidf, None))


def global_defaults_configuration():
    """Setting global defaults for all future configurations."""
    print("\n=== Global Defaults Configuration ===")
    
    # Set custom global defaults
    AisertConfig.set_defaults(
        token_provider="anthropic",
        token_model="claude-3",
        semantic_provider="tfidf"
    )
    
    config = AisertConfig.get_default_config()
    print(f"Custom defaults: {config}")
    # Expected: AisertConfig(token(anthropic, claude-3), semantic(tfidf, text-embedding-3-small))
    
    # Reset to original defaults
    AisertConfig.set_defaults(
        token_provider="openai",
        token_model="gpt-3.5-turbo",
        semantic_provider="openai",
        semantic_model="text-embedding-3-small"
    )


def provider_specific_configurations():
    """Examples for different AI providers."""
    print("\n=== Provider-Specific Configurations ===")
    
    # OpenAI configuration
    openai_config = AisertConfig(
        token_provider="openai",
        token_model="gpt-4",
        token_encoding="cl100k_base",
        semantic_provider="openai",
        semantic_model="text-embedding-3-large"
    )
    print(f"OpenAI: {openai_config}")
    
    # Anthropic configuration
    anthropic_config = AisertConfig(
        token_provider="anthropic",
        token_model="claude-3-sonnet",
        semantic_provider="sentence_transformers",
        semantic_model="all-MiniLM-L6-v2"
    )
    print(f"Anthropic: {anthropic_config}")
    
    # Lightweight configuration (no heavy dependencies)
    lightweight_config = AisertConfig(
        semantic_provider="tfidf"  # No torch/transformers needed
    )
    print(f"Lightweight: {lightweight_config}")


if __name__ == "__main__":
    print("🔧 Aisert Configuration Examples")
    print("=" * 50)
    
    default_configuration()
    constructor_configuration()
    global_defaults_configuration()
    provider_specific_configurations()
    
    print("\n✨ Configuration examples completed!")

Key topics covered:

• Default configuration usage

• Constructor-based configuration

• Global defaults management

• Provider-specific configurations

• Lightweight vs full-featured setups

Functions in this example:

• default_configuration() - Using built-in defaults

• constructor_configuration() - Custom configuration setup

• global_defaults_configuration() - Setting application-wide defaults

• provider_specific_configurations() - Provider-specific examples

Usage Patterns

Master direct validation and fluent interface patterns.

Download: 02_usage_patterns.py

Direct Usage and Fluent Interface

"""
Usage Patterns - Direct validation and fluent interface examples

Shows different ways to use Aisert for validation.
"""

import sys
import os
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

from aisert import Aisert, AisertConfig
from pydantic import BaseModel


class UserModel(BaseModel):
    name: str
    age: int


def direct_validation():
    """Direct validation - single validator per call."""
    print("=== Direct Validation ===")
    
    content = "Hello world, this is a test message."
    
    # Content validation
    result1 = Aisert(content).assert_contains(["Hello", "test"]).collect()
    print(f"Contains validation: {result1.status}")
    # Expected: True
    
    # Schema validation
    json_content = '{"name": "John", "age": 30}'
    result2 = Aisert(json_content).assert_schema(UserModel).collect()
    print(f"Schema validation: {result2.status}")
    # Expected: True
    
    # Token validation (requires config)
    config = AisertConfig(token_provider="openai", token_model="gpt-3.5-turbo")
    result3 = Aisert(content, config).assert_tokens(max_tokens=20).collect()
    print(f"Token validation: {result3.status}")
    # Expected: True (content is under 20 tokens)


def fluent_interface():
    """Fluent interface - chain multiple validations."""
    print("\n=== Fluent Interface ===")
    
    content = "AI and machine learning are transforming technology."
    config = AisertConfig(
        token_provider="openai",
        token_model="gpt-3.5-turbo",
        semantic_provider="openai",
        semantic_model="text-embedding-3-small"
    )
    
    # Chain multiple validations
    result = (Aisert(content, config)
             .assert_contains(["AI", "technology"])
             .assert_not_contains(["spam", "inappropriate"])
             .assert_tokens(max_tokens=50)
             .assert_semantic_matches("artificial intelligence technology", threshold=0.6)
             .collect())
    
    print(f"Chained validation: {result.status}")
    print(f"Validation count: {len(result.rules)}")
    # Expected: True, 4 validations


def strict_vs_non_strict():
    """Strict mode vs non-strict mode behavior."""
    print("\n=== Strict vs Non-Strict Modes ===")
    
    content = "Hello world"
    
    # Non-strict mode - collects all errors
    result = (Aisert(content)
             .assert_contains(["Hello"], strict=False)  # Pass
             .assert_contains(["missing"], strict=False)  # Fail but continue
             .assert_not_contains(["world"], strict=False)  # Fail but continue
             .collect())
    
    print(f"Non-strict mode - Overall: {result.status}")
    print(f"Individual results: {[v['status'] for v in result.rules.values()]}")
    # Expected: False, [True, False, False]
    
    # Strict mode - stops at first error
    try:
        result = (Aisert(content)
                 .assert_contains(["Hello"])  # Pass
                 .assert_contains(["missing"])  # Fail and raise exception
                 .collect())
        print("Strict mode completed")
    except Exception as e:
        print(f"Strict mode exception: {type(e).__name__}")
        # Expected: ContainsValidationError


def error_handling():
    """Proper error handling patterns."""
    print("\n=== Error Handling ===")
    
    from aisert.exception import AisertError, ContainsValidationError
    
    content = "Test content"
    
    # Specific exception handling
    try:
        Aisert(content).assert_contains(["missing"]).collect()
    except ContainsValidationError as e:
        print(f"Caught specific error: {type(e).__name__}")
        # Expected: ContainsValidationError
    
    # General exception handling
    try:
        Aisert("invalid json").assert_schema(UserModel).collect()
    except AisertError as e:
        print(f"Caught general error: {type(e).__name__}")
        # Expected: SchemaValidationError
    
    # Graceful handling with non-strict mode
    result = (Aisert(content)
             .assert_contains(["Test"], strict=False)
             .assert_contains(["missing"], strict=False)
             .collect())
    
    if result.status:
        print("All validations passed")
    else:
        failed = [v for v in result.rules.values() if not v['status']]
        print(f"Failed validations: {len(failed)}")
        # Expected: 1 failed validation


if __name__ == "__main__":
    print("🚀 Aisert Usage Patterns")
    print("=" * 50)
    
    direct_validation()
    fluent_interface()
    strict_vs_non_strict()
    error_handling()
    
    print("\n✨ Usage patterns completed!")

Key topics covered:

• Direct validation (single validators)

• Fluent interface (chained validations)

• Strict vs non-strict modes

• Error handling patterns

• Exception hierarchy usage

Functions in this example:

• direct_validation() - Single validator usage

• fluent_interface() - Chaining multiple validations

• strict_vs_non_strict() - Error handling modes

• error_handling() - Exception management

Production Use Cases

Real-world scenarios and pipeline integration examples.

Download: 03_production_use_cases.py

Real-World Production Scenarios

"""
Production Use Cases - Real-world scenarios and pipeline integration

Shows practical applications in production environments.
"""

import sys
import os
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

from aisert import Aisert, AisertConfig
from typing import List, Dict, Any


def content_moderation_pipeline():
    """Content moderation for user-generated content."""
    print("=== Content Moderation Pipeline ===")
    
    user_comments = [
        "Great product, really helpful!",
        "This spam content contains inappropriate language",
        "Excellent customer service, highly recommend!"
    ]
    
    flagged_terms = ["spam", "inappropriate", "offensive", "scam"]
    approved_count = 0
    
    for i, comment in enumerate(user_comments, 1):
        result = (Aisert(comment)
                 .assert_not_contains(flagged_terms, strict=False)
                 .collect())
        
        if result.status:
            print(f"Comment {i}: ✅ APPROVED")
            approved_count += 1
        else:
            print(f"Comment {i}: ❌ FLAGGED - {result.rules[1]['reason']}")
    
    print(f"Approval rate: {approved_count}/{len(user_comments)} ({approved_count/len(user_comments)*100:.1f}%)")
    # Expected: 2/3 approved (66.7%)


def api_response_validation():
    """Validate LLM API responses in production."""
    print("\n=== API Response Validation ===")
    
    config = AisertConfig(
        token_provider="openai",
        token_model="gpt-4",
        semantic_provider="openai",
        semantic_model="text-embedding-3-small"
    )
    
    # Simulate LLM responses
    responses = [
        "Thank you for your inquiry. I can help you with account setup.",
        "I cannot process this request due to insufficient information.",
        "Here's a comprehensive guide to getting started with our platform."
    ]
    
    valid_responses = 0
    for i, response in enumerate(responses, 1):
        result = (Aisert(response, config)
                 .assert_contains(["help", "information", "guide"], strict=False)  # Helpful content
                 .assert_not_contains(["error", "failed", "unavailable"], strict=False)  # No errors
                 .assert_tokens(max_tokens=100, strict=False)  # Reasonable length
                 .assert_semantic_matches("helpful customer service", threshold=0.6, strict=False)
                 .collect())
        
        if result.status:
            print(f"Response {i}: ✅ VALID")
            valid_responses += 1
        else:
            failed_checks = sum(1 for v in result.rules.values() if not v['status'])
            print(f"Response {i}: ❌ INVALID ({failed_checks} failed checks)")
    
    print(f"Valid responses: {valid_responses}/{len(responses)}")
    # Expected: Varies based on semantic similarity


def ci_cd_integration():
    """Integration with CI/CD pipelines for automated testing."""
    print("\n=== CI/CD Integration ===")
    
    def validate_generated_docs(content: str) -> bool:
        """Validate auto-generated documentation."""
        result = (Aisert(content)
                 .assert_contains(["API", "endpoint", "parameter"], strict=False)
                 .assert_not_contains(["TODO", "FIXME", "placeholder"], strict=False)
                 .assert_tokens(max_tokens=500, strict=False)
                 .collect())
        return result.status
    
    # Test documentation samples
    docs = [
        "API endpoint /users accepts GET requests with optional id parameter",
        "TODO: Add documentation for this endpoint",
        "This endpoint returns user data in JSON format with id, name, and email fields"
    ]
    
    passed_tests = 0
    for i, doc in enumerate(docs, 1):
        if validate_generated_docs(doc):
            print(f"Doc {i}: ✅ PASS")
            passed_tests += 1
        else:
            print(f"Doc {i}: ❌ FAIL")
    
    print(f"CI/CD Result: {passed_tests}/{len(docs)} tests passed")
    # Expected: 2/3 tests pass (doc with TODO fails)
    
    # Simulate CI/CD exit code
    exit_code = 0 if passed_tests == len(docs) else 1
    print(f"Exit code: {exit_code}")


def batch_processing():
    """High-volume batch processing with performance optimization."""
    print("\n=== Batch Processing ===")
    
    # Lightweight config for high-volume processing
    config = AisertConfig(semantic_provider="tfidf")  # Fast, no model loading
    
    # Simulate large batch of content
    batch_size = 100
    content_batch = [f"User message {i} with some content to validate" for i in range(batch_size)]
    
    import time
    start_time = time.time()
    
    processed = 0
    valid_count = 0
    
    for content in content_batch:
        result = (Aisert(content, config)
                 .assert_not_contains(["spam", "inappropriate"], strict=False)
                 .assert_semantic_matches("user message", threshold=0.3, strict=False)
                 .collect())
        
        processed += 1
        if result.status:
            valid_count += 1
    
    processing_time = time.time() - start_time
    
    print(f"Processed: {processed} items")
    print(f"Valid: {valid_count} ({valid_count/processed*100:.1f}%)")
    print(f"Time: {processing_time:.2f}s ({processing_time/processed*1000:.1f}ms per item)")
    print(f"Throughput: {processed/processing_time:.1f} items/second")
    # Expected: High throughput with TFIDF (no model loading)


def quality_monitoring():
    """Monitor content quality in production with metrics."""
    print("\n=== Quality Monitoring ===")
    
    class QualityMonitor:
        def __init__(self):
            self.metrics = {"total": 0, "passed": 0, "failed": 0}
            self.config = AisertConfig(
                token_provider="openai",
                token_model="gpt-3.5-turbo"
            )
        
        def validate_content(self, content: str, requirements: Dict[str, Any]) -> bool:
            self.metrics["total"] += 1
            
            aisert = Aisert(content, self.config)
            
            if "required_terms" in requirements:
                aisert.assert_contains(requirements["required_terms"], strict=False)
            
            if "forbidden_terms" in requirements:
                aisert.assert_not_contains(requirements["forbidden_terms"], strict=False)
            
            if "max_tokens" in requirements:
                aisert.assert_tokens(requirements["max_tokens"], strict=False)
            
            result = aisert.collect()
            
            if result.status:
                self.metrics["passed"] += 1
                return True
            else:
                self.metrics["failed"] += 1
                return False
        
        def get_quality_score(self) -> float:
            if self.metrics["total"] == 0:
                return 0.0
            return self.metrics["passed"] / self.metrics["total"]
    
    # Usage example
    monitor = QualityMonitor()
    
    # Validate different content types
    test_cases = [
        ("Customer service response with helpful information", {"required_terms": ["helpful"], "max_tokens": 50}),
        ("Product description with spam content", {"forbidden_terms": ["spam"], "max_tokens": 100}),
        ("Technical documentation explaining the API", {"required_terms": ["API"], "max_tokens": 200})
    ]
    
    for content, requirements in test_cases:
        is_valid = monitor.validate_content(content, requirements)
        print(f"Content valid: {is_valid}")
    
    quality_score = monitor.get_quality_score()
    print(f"Overall quality score: {quality_score:.2f} ({quality_score*100:.1f}%)")
    print(f"Metrics: {monitor.metrics}")
    # Expected: Quality score based on validation results


if __name__ == "__main__":
    print("🏭 Aisert Production Use Cases")
    print("=" * 50)
    
    content_moderation_pipeline()
    api_response_validation()
    ci_cd_integration()
    batch_processing()
    quality_monitoring()
    
    print("\n✨ Production use cases completed!")
    print("💡 These patterns can be adapted for your specific production needs.")

Key topics covered:

• Content moderation pipelines

• API response validation

• CI/CD integration

• Batch processing optimization

• Quality monitoring with metrics

Functions in this example:

• content_moderation_pipeline() - User content filtering

• api_response_validation() - LLM response validation

• ci_cd_integration() - Automated testing integration

• batch_processing() - High-volume processing

• quality_monitoring() - Production metrics

Running Examples

To run these examples locally:

# Install Aisert with optional dependencies
pip install aisert[all]

# Set environment variables (for API-based validation)
export OPENAI_API_KEY="your-openai-key"
export ANTHROPIC_API_KEY="your-anthropic-key"

# Run examples in order
python examples/01_configuration_examples.py
python examples/02_usage_patterns.py
python examples/03_production_use_cases.py

Prerequisites (for running examples)

Basic Installation:

pip install aisert

Optional Dependencies:

# For semantic validation with sentence-transformers
pip install aisert[sentence-transformers]

# For HuggingFace models
pip install aisert[huggingface]

# For all features
pip install aisert[all]

Custom Validators

Aisert supports bringing your own token and semantic validators by extending the base classes.

Custom Token Validator

Create custom token validators by extending TokenValidatorBase:

from aisert.validators.token_validator.token_validator_base import TokenValidatorBase
from aisert.validators.token_validator.token_validator_factory import TokenValidatorFactory
from aisert.exception import TokenValidationError
import threading

class CustomTokenValidator(TokenValidatorBase):
    """Custom token validator example."""
    _instances = {}
    _lock = threading.RLock()

    def __init__(self, model_name: str):
        super().__init__(model_provider="custom")
        self.model_name = model_name

    @classmethod
    def get_instance(cls, token_model: str = None, **kwargs):
        with cls._lock:
            if token_model not in cls._instances:
                cls._instances[token_model] = cls(token_model)
            return cls._instances[token_model]

    def count(self, text: str) -> int:
        """Implement your custom token counting logic."""
        # Example: simple word count
        return len(text.split())

# Register the custom validator
TokenValidatorFactory.register_token_validator("custom", CustomTokenValidator)

# Usage
config = AisertConfig(token_provider="custom", token_model="my-model")
result = Aisert("Hello world", config).assert_tokens(max_tokens=5).collect()

Custom Semantic Validator

Create custom semantic validators by extending SemanticValidatorBase:

from aisert.validators.semantic_validator.semantic_validator_base import SemanticValidatorBase
from aisert.validators.semantic_validator.semantic_validator_factory import SemanticValidatorFactory
from aisert.exception import SemanticValidationError
from aisert.models.result import Result
import threading

class CustomSemanticValidator(SemanticValidatorBase):
    """Custom semantic validator example."""
    _instances = {}
    _lock = threading.RLock()

    def __init__(self, model_name: str = None):
        super().__init__()
        self.model_name = model_name or "custom-model"

    @classmethod
    def get_instance(cls, model_name: str = None, **kwargs):
        key = model_name or "default"
        with cls._lock:
            if key not in cls._instances:
                cls._instances[key] = cls(model_name)
            return cls._instances[key]

    def validate(self, text1: str, text2: str, threshold: float = 0.8) -> Result:
        """Implement your custom semantic similarity logic."""
        if not (0 <= threshold <= 1):
            raise SemanticValidationError("Threshold must be between 0 and 1")

        # Example: simple keyword overlap similarity
        words1 = set(text1.lower().split())
        words2 = set(text2.lower().split())
        similarity = len(words1 & words2) / len(words1 | words2) if words1 | words2 else 0

        if similarity < threshold:
            raise SemanticValidationError(
                f"Custom similarity score: {similarity:.3f} is less than threshold: {threshold}"
            )

        return Result(self.validator_name, True,
                     f"Custom similarity score: {similarity:.3f}, Threshold: {threshold}")

# Register the custom validator
SemanticValidatorFactory.register_semantic_validator("custom", CustomSemanticValidator)

# Usage
config = AisertConfig(semantic_provider="custom", semantic_model="my-model")
result = Aisert("Hello world", config).assert_semantic_matches("Hi world", 0.5).collect()

Key Implementation Points

Token Validators:

• Extend TokenValidatorBase

• Implement count(text: str) -> int method

• Use singleton pattern with thread-safe get_instance()

• Register with TokenValidatorFactory.register_token_validator()

Semantic Validators:

• Extend SemanticValidatorBase

• Implement validate(text1: str, text2: str, threshold: float) -> Result method

• Use singleton pattern with thread-safe get_instance()

• Register with SemanticValidatorFactory.register_semantic_validator()

Best Practices:

• Use thread-safe singleton pattern for performance

• Handle errors gracefully with appropriate exceptions

• Validate input parameters (e.g., threshold range)

• Return descriptive error messages

• Cache expensive operations (models, connections)

Complete Custom Validator Example

Download: custom_validators_example.py

# Complete working example
from aisert import Aisert, AisertConfig
from aisert.validators.token_validator.token_validator_base import TokenValidatorBase
from aisert.validators.token_validator.token_validator_factory import TokenValidatorFactory
from aisert.validators.semantic_validator.semantic_validator_base import SemanticValidatorBase
from aisert.validators.semantic_validator.semantic_validator_factory import SemanticValidatorFactory
from aisert.models.result import Result
import threading

# Custom implementations here...
# (See full example in downloadable file)

if __name__ == "__main__":
    # Test custom validators
    config = AisertConfig(
        token_provider="custom",
        token_model="word-counter",
        semantic_provider="custom",
        semantic_model="keyword-overlap"
    )

    result = (
        Aisert("Hello world from custom validators", config)
        .assert_tokens(max_tokens=10)
        .assert_semantic_matches("Hi world custom", threshold=0.3)
        .collect()
    )

    print(f"Custom validation result: {result.status}")