1
import opik
2
from opik.anonymizer import create_anonymizer
3
4
# Dictionary format
5
email_rule = {"regex": r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b", "replace": "[EMAIL]"}
6
7
# Tuple format
8
phone_rule = (r"\b\d{3}-\d{3}-\d{4}\b", "[PHONE]")
9
10
# Create anonymizer with multiple rules
11
anonymizer = create_anonymizer([email_rule, phone_rule])
12
13
# Register globally
14
opik.hooks.add_anonymizer(anonymizer)

1
import opik
2
from opik.anonymizer import create_anonymizer
3
4
def mask_api_keys(text: str) -> str:
5
    """Custom function to anonymize API keys"""
6
    import re
7
    # Match common API key patterns
8
    api_key_pattern = r'\b(sk-[a-zA-Z0-9]{32,}|pk_[a-zA-Z0-9]{24,})\b'
9
    return re.sub(api_key_pattern, '[API_KEY]', text)
10
11
def uppercase_names(text: str) -> str:
12
    """Transform names to uppercase for consistency"""
13
    return text.upper()
14
15
# Create anonymizer with function rules
16
anonymizer = create_anonymizer([mask_api_keys, uppercase_names])
17
opik.hooks.add_anonymizer(anonymizer)

1
import opik
2
import opik.hooks
3
from opik.anonymizer import create_anonymizer
4
5
# Mix of dictionary, tuple, and function rules
6
mixed_rules = [
7
    {"regex": r"\b\d{3}-\d{2}-\d{4}\b", "replace": "[SSN]"},  # Social Security Numbers
8
    (r"\b\d{4}[- ]?\d{4}[- ]?\d{4}[- ]?\d{4}\b", "[CARD]"),  # Credit Cards
9
    lambda text: text.replace("CONFIDENTIAL", "[REDACTED]"),  # Custom replacements
10
]
11
12
anonymizer = create_anonymizer(mixed_rules)
13
opik.hooks.add_anonymizer(anonymizer)

1
import opik
2
import opik.hooks
3
from opik.anonymizer import Anonymizer
4
5
class AdvancedPIIAnonymizer(Anonymizer):
6
    def anonymize(self, data, **kwargs):
7
        """
8
        Custom anonymizer with access to metadata about the field being processed
9
        """
10
        field_name = kwargs.get("field_name")
11
        object_type = kwargs.get("object_type")
12
13
        # Handle different data types
14
        if isinstance(data, dict):
15
            # Remove sensitive keys entirely
16
            if "api_key" in data:
17
                del data["api_key"]
18
            if "password" in data:
19
                del data["password"]
20
21
            # Anonymize specific fields
22
            for key, value in data.items():
23
                if key.lower() in ["email", "user_email"]:
24
                    data[key] = "[EMAIL_REDACTED]"
25
                elif key.lower() in ["phone", "telephone", "mobile"]:
26
                    data[key] = "[PHONE_REDACTED]"
27
28
        elif isinstance(data, str):
29
            # Apply string-based anonymization
30
            import re
31
            # Names (simple heuristic)
32
            data = re.sub(r'\b[A-Z][a-z]+ [A-Z][a-z]+\b', '[NAME]', data)
33
            # Addresses
34
            data = re.sub(r'\d+\s+\w+\s+(Street|St|Avenue|Ave|Road|Rd|Drive|Dr)\b', '[ADDRESS]', data)
35
36
        return data
37
38
# Register the custom anonymizer
39
opik.hooks.add_anonymizer(AdvancedPIIAnonymizer())

1
import opik
2
import opik.hooks
3
from opik.anonymizer import create_anonymizer
4
5
# Define PII anonymization rules
6
pii_rules = [
7
    # Email addresses
8
    {"regex": r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b", "replace": "[EMAIL]"},
9
    # Phone numbers (US format)
10
    {"regex": r"\b\d{3}-\d{3}-\d{4}\b", "replace": "[PHONE]"},
11
    # Social Security Numbers
12
    {"regex": r"\b\d{3}-\d{2}-\d{4}\b", "replace": "[SSN]"},
13
    # Credit card numbers
14
    {"regex": r"\b\d{4}[- ]?\d{4}[- ]?\d{4}[- ]?\d{4}\b", "replace": "[CARD]"},
15
]
16
17
# Create and register anonymizer
18
anonymizer = create_anonymizer(pii_rules)
19
opik.hooks.add_anonymizer(anonymizer)
20
21
# Now all traced functions will automatically anonymize PII
22
@opik.track
23
def process_customer_data(customer_info):
24
    """This function processes customer data with automatic PII anonymization"""
25
    # The input and output will be automatically anonymized
26
    return f"Processed customer: {customer_info}"
27
28
# Example usage - PII will be automatically anonymized in traces
29
result = process_customer_data("John Doe, email: john@example.com, phone: 555-123-4567")

1
import opik
2
import opik.hooks
3
from opik.anonymizer import create_anonymizer, Anonymizer
4
5
class ComplianceAnonymizer(Anonymizer):
6
    """Enterprise-grade anonymizer for compliance requirements"""
7
8
    def __init__(self, compliance_level="standard"):
9
        self.compliance_level = compliance_level
10
        self.sensitive_fields = {
11
            "standard": ["email", "phone", "ssn"],
12
            "strict": ["email", "phone", "ssn", "name", "address", "dob"],
13
            "minimal": ["ssn", "password"]
14
        }
15
16
    def anonymize(self, data, **kwargs):
17
        field_name = kwargs.get("field_name", "")
18
19
        if isinstance(data, dict):
20
            # Process dictionary fields
21
            for key, value in list(data.items()):
22
                if key.lower() in self.sensitive_fields[self.compliance_level]:
23
                    data[key] = f"[{key.upper()}_REDACTED]"
24
25
        elif isinstance(data, str):
26
            # Apply string-level anonymization based on the compliance level
27
            if self.compliance_level == "strict":
28
                # More aggressive anonymization
29
                import re
30
                data = re.sub(r'\b[A-Z][a-z]+ [A-Z][a-z]+\b', '[NAME]', data)
31
                data = re.sub(r'\b\d{1,4}\s+\w+\s+\w+\b', '[ADDRESS]', data)
32
33
        return data
34
35
# Set up multi-layer anonymization
36
opik.hooks.clear_anonymizers()  # Clear any existing anonymizers
37
38
# Layer 1: Basic PII patterns
39
basic_rules = [
40
    (r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b", "[EMAIL]"),
41
    (r"\b\d{3}-\d{3}-\d{4}\b", "[PHONE]"),
42
]
43
opik.hooks.add_anonymizer(create_anonymizer(basic_rules))
44
45
# Layer 2: Compliance-specific anonymization
46
opik.hooks.add_anonymizer(ComplianceAnonymizer(compliance_level="standard"))
47
48
# Layer 3: Custom business logic
49
def remove_internal_identifiers(text):
50
    """Remove company-specific internal identifiers"""
51
    import re
52
    return re.sub(r'\bEMP-\d{6}\b', '[EMPLOYEE_ID]', text)
53
54
opik.hooks.add_anonymizer(create_anonymizer(remove_internal_identifiers))

1
import opik
2
import opik.hooks
3
from opik.anonymizer import create_anonymizer
4
from opik.integrations.openai import track_openai
5
import openai
6
7
# Set up anonymization
8
pii_rules = [
9
    {"regex": r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b", "replace": "[EMAIL]"},
10
    {"regex": r"\b\d{3}-\d{3}-\d{4}\b", "replace": "[PHONE]"},
11
]
12
opik.hooks.add_anonymizer(create_anonymizer(pii_rules))
13
14
# Enable OpenAI tracking with automatic anonymization
15
client = track_openai(openai.OpenAI())
16
17
# PII in prompts will be automatically anonymized in traces
18
response = client.chat.completions.create(
19
    model="gpt-3.5-turbo",
20
    messages=[{
21
        "role": "user",
22
        "content": "Help me draft an email to john.doe@company.com about his phone number 555-123-4567"
23
    }]
24
)

1
import opik
2
import opik.hooks
3
from opik.anonymizer import create_anonymizer
4
from opik.integrations.langchain import OpikTracer
5
from langchain_openai import ChatOpenAI
6
from langchain.schema import HumanMessage
7
8
# Configure anonymization
9
anonymizer_rules = [
10
    # Email pattern
11
    (r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b", "[EMAIL]"),
12
    # Custom function to handle API keys
13
    lambda text: text.replace("sk-", "[API_KEY]"),
14
]
15
opik.hooks.add_anonymizer(create_anonymizer(anonymizer_rules))
16
17
# Set up LangChain with Opik tracing
18
llm = ChatOpenAI(callbacks=[OpikTracer()])
19
20
# All inputs and outputs will be automatically anonymized
21
messages = [HumanMessage(content="Contact sarah@example.com about the API key sk-1234567890")]
22
result = llm.invoke(messages)

1
from opik.anonymizer import create_anonymizer
2
3
rules = [{"regex": r"\b\d{3}-\d{3}-\d{4}\b", "replace": "[PHONE]"}]
4
5
# Default max_depth is 10
6
anonymizer = create_anonymizer(rules, max_depth=5)

1
import opik
2
import opik.hooks
3
from opik.anonymizer import create_anonymizer
4
5
# Clear existing anonymizers
6
opik.hooks.clear_anonymizers()
7
8
# Add multiple anonymizers in order
9
opik.hooks.add_anonymizer(create_anonymizer([
10
    {"regex": r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b", "replace": "[EMAIL]"}
11
]))
12
13
opik.hooks.add_anonymizer(create_anonymizer([
14
    {"regex": r"\b\d{3}-\d{3}-\d{4}\b", "replace": "[PHONE]"}
15
]))
16
17
# Check if any anonymizers are registered
18
if opik.hooks.has_anonymizers():
19
    print(f"Active anonymizers: {len(opik.hooks.get_anonymizers())}")

1
rules = [
2
    # Specific: Credit cards (more specific pattern first)
3
    {"regex": r"\b4\d{3}[- ]?\d{4}[- ]?\d{4}[- ]?\d{4}\b", "replace": "[VISA_CARD]"},
4
    # General: Any credit card
5
    {"regex": r"\b\d{4}[- ]?\d{4}[- ]?\d{4}[- ]?\d{4}\b", "replace": "[CARD]"},
6
    # General: Any number sequence
7
    {"regex": r"\b\d{4,}\b", "replace": "[NUMBER]"},
8
]

1
import re
2
from opik.anonymizer import create_anonymizer
3
4
# Pre-compile regex for better performance
5
EMAIL_PATTERN = re.compile(r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b")
6
7
def efficient_email_anonymizer(text):
8
    return EMAIL_PATTERN.sub("[EMAIL]", text)
9
10
anonymizer = create_anonymizer(efficient_email_anonymizer)

1
from opik.anonymizer import create_anonymizer
2
3
# Define your rules
4
rules = [
5
    {"regex": r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b", "replace": "[EMAIL]"},
6
    {"regex": r"\b\d{3}-\d{3}-\d{4}\b", "replace": "[PHONE]"},
7
]
8
9
anonymizer = create_anonymizer(rules)
10
11
# Test with sample data
12
test_data = "Contact John at john.doe@company.com or call 555-123-4567"
13
anonymized = anonymizer.anonymize(test_data)
14
print(anonymized)  # Should output: "Contact John at [EMAIL] or call [PHONE]"
15
16
# Test with nested data
17
test_nested = {
18
    "user": {
19
        "email": "user@example.com",
20
        "phone": "555-987-6543",
21
        "notes": "Called regarding john@company.com"
22
    }
23
}
24
anonymized_nested = anonymizer.anonymize(test_nested)
25
print(anonymized_nested)