Plugin Framework
The Safeguards library includes a flexible plugin framework that allows you to create custom safeguards for your specific use cases, including industry-specific checks and rules.
Plugin Architecture
The plugin system is based on the SafeguardPlugin abstract base class, which all plugins must extend. Plugins are managed through the PluginManager, which handles registration, configuration, and lifecycle management.
Plugin Manager
The PluginManager provides a central registry for all plugins:
from safeguards.plugins import PluginManager
# Create a plugin manager
plugin_manager = PluginManager()
# Register plugins
plugin_manager.register_plugin(my_plugin, config={"key": "value"})
# Get a plugin by name
my_plugin = plugin_manager.get_plugin("my_plugin_name")
# List all registered plugins
plugin_names = plugin_manager.list_plugins()
# Unregister a plugin
plugin_manager.unregister_plugin("my_plugin_name")
# Shutdown all plugins
plugin_manager.shutdown_all()
Creating Custom Plugins
To create a custom plugin, extend the SafeguardPlugin abstract base class:
from safeguards.plugins import SafeguardPlugin
from typing import Dict, Any
class MyCustomPlugin(SafeguardPlugin):
"""Example custom plugin."""
def __init__(self):
"""Initialize the plugin."""
self._config = {}
@property
def name(self) -> str:
"""Return the name of the plugin."""
return "my_custom_plugin"
@property
def version(self) -> str:
"""Return the version of the plugin."""
return "1.0.0"
def initialize(self, config: Dict[str, Any]) -> None:
"""Initialize the plugin with configuration.
Args:
config: Plugin-specific configuration
"""
self._config = config
# Perform any initialization based on config
def shutdown(self) -> None:
"""Clean up resources when shutting down."""
# Clean up any resources used by the plugin
Industry-Specific Safeguards
The Safeguards library includes pre-built industry-specific safeguards for common use cases. These extend the IndustrySafeguard base class, which itself extends SafeguardPlugin.
Available Industry Safeguards
Currently, the library provides:
FinancialServicesSafeguard: For financial services operationsHealthcareSafeguard: For healthcare operations
Using Industry Safeguards
Here's how to use a financial services safeguard:
from decimal import Decimal
from safeguards.plugins.industry import FinancialServicesSafeguard
from safeguards.types.agent import Agent
# Create and configure the safeguard
financial_safeguard = FinancialServicesSafeguard()
financial_safeguard.initialize({
"restricted_actions": ["high_risk_investment", "unauthorized_withdrawal"],
"compliance_rules": {
"kyc_required": True,
"aml_check": True
},
"transaction_limits": {
"agent_1": Decimal("1000.00")
}
})
# Start monitoring an agent
financial_safeguard.monitor_agent("agent_1")
# Validate an action
agent = get_agent("agent_1") # Get your agent instance
action_context = {
"action_type": "transaction",
"amount": Decimal("1500.00"),
"description": "Investment purchase"
}
# Check for violations
alerts = financial_safeguard.validate_agent_action(agent, action_context)
if alerts:
print(f"Action violates safeguards: {alerts}")
else:
# Proceed with action
agent.run(**action_context)
# Stop monitoring when done
financial_safeguard.stop_monitoring_agent("agent_1")
Healthcare Safeguard Example
Similarly, for healthcare operations:
from safeguards.plugins.industry import HealthcareSafeguard
# Create and configure the safeguard
healthcare_safeguard = HealthcareSafeguard()
healthcare_safeguard.initialize({
"phi_patterns": ["SSN", "DOB", "MRN"],
"restricted_operations": ["mass_record_access", "export_all_records"],
"required_approvals": {
"prescription": ["doctor", "pharmacist"]
}
})
# Monitor and validate as with the financial safeguard
Creating Industry-Specific Safeguards
To create a custom industry safeguard, extend the IndustrySafeguard class:
from typing import Dict, Any, List
from safeguards.plugins.industry import IndustrySafeguard
from safeguards.types.agent import Agent
from safeguards.types import SafetyAlert, AlertSeverity
class RetailSafeguard(IndustrySafeguard):
"""Safeguards specific to the retail industry."""
def __init__(self):
"""Initialize the retail safeguard."""
super().__init__("retail")
self._pricing_limits = {}
self._discount_limits = {}
@property
def version(self) -> str:
"""Return the version of the plugin."""
return "1.0.0"
def initialize(self, config: Dict[str, Any]) -> None:
"""Initialize the plugin with configuration."""
self._config = config
self._pricing_limits = config.get("pricing_limits", {})
self._discount_limits = config.get("discount_limits", {})
self._initialized = True
def shutdown(self) -> None:
"""Clean up resources when shutting down."""
self._initialized = False
self._monitored_agents.clear()
def validate_agent_action(self, agent: Agent, action_context: Dict[str, Any]) -> List[SafetyAlert]:
"""Validate a retail agent action."""
alerts = []
# Check pricing within limits
if action_context.get("action_type") == "price_change":
product = action_context.get("product", "")
price = action_context.get("price", 0)
min_price = self._pricing_limits.get(product, {}).get("min", 0)
max_price = self._pricing_limits.get(product, {}).get("max", float("inf"))
if price < min_price:
alerts.append(SafetyAlert(
title="Price Below Minimum",
description=f"Price {price} is below minimum {min_price} for {product}",
severity=AlertSeverity.WARNING
))
if price > max_price:
alerts.append(SafetyAlert(
title="Price Above Maximum",
description=f"Price {price} is above maximum {max_price} for {product}",
severity=AlertSeverity.WARNING
))
return alerts
Integrating with the Safety Controller
You can integrate your custom safeguards with the SafetyController for centralized management:
from safeguards.core.safety_controller import SafetyController
from safeguards.types import SafetyConfig
from safeguards.plugins.industry import FinancialServicesSafeguard
# Create controller
config = SafetyConfig(...)
controller = SafetyController(config)
# Create industry safeguard
financial_safeguard = FinancialServicesSafeguard()
financial_safeguard.initialize({...})
# Register agent with controller and safeguard
agent = MyAgent("agent_1")
controller.register_agent(agent, budget=Decimal("1000.00"))
financial_safeguard.monitor_agent(agent.id)
# When validating actions, combine checks
def execute_safe_action(agent, action_context):
# First check controller rules
validation = controller.validate_action(agent.id, action_context)
if not validation.valid:
return {"success": False, "violations": validation.violations}
# Then check industry safeguards
alerts = financial_safeguard.validate_agent_action(agent, action_context)
if alerts:
return {"success": False, "alerts": alerts}
# If all checks pass, execute the action
return agent.run(**action_context)