Skip to content

LLM Toolkit

Claude integration with security controls: PII scrubbing, secrets detection, and audit logging.

Overview

The LLM Toolkit provides:

  • Anthropic Integration: Claude API via EmpathyLLM
  • Security Controls: PII scrubbing, secrets detection
  • Audit Logging: JSONL audit trail of LLM interactions
  • Claude Memory Integration: CLAUDE.md support for persistent context

Key Features

Anthropic Integration

import os
from attune.llm import EmpathyLLM

# Anthropic Claude (the only supported provider)
llm = EmpathyLLM(
    provider="anthropic",
    api_key=os.getenv("ANTHROPIC_API_KEY"),
    model="claude-sonnet-4-5",
    target_level=4,
)

Automatic Security Controls

  • PII Scrubbing: Removes SSN, credit cards, phone numbers, addresses
  • Secrets Detection: Flags API keys, tokens, passwords
  • Audit Logging: JSONL audit trail of interactions

Class Reference

EmpathyLLM

Bases: SecurityMixin, InteractionMixin

Wraps any LLM provider with Attune AI levels.

Automatically progresses from Level 1 (reactive) to Level 4 (anticipatory) based on user collaboration state.

Security Features (Phase 3): - PII Scrubbing: Automatically detect and redact PII from user inputs - Secrets Detection: Block requests containing API keys, passwords, etc. - Audit Logging: Comprehensive compliance logging (SOC2, HIPAA, GDPR) - Backward Compatible: Security disabled by default

Example

llm = EmpathyLLM(provider="anthropic", target_level=4) response = await llm.interact( ... user_id="developer_123", ... user_input="Help me optimize my code", ... context={"code_snippet": "..."} ... ) print(response["content"])

Example with Security

llm = EmpathyLLM( ... provider="anthropic", ... target_level=4, ... enable_security=True, ... security_config={ ... "audit_log_dir": "/var/log/empathy", ... "block_on_secrets": True, ... "enable_pii_scrubbing": True ... } ... ) response = await llm.interact( ... user_id="user@company.com", ... user_input="My email is john@example.com" ... )

PII automatically scrubbed, request logged

Example with Model Routing (Cost Optimization): >>> llm = EmpathyLLM( ... provider="anthropic", ... enable_model_routing=True # Enable smart model selection ... ) >>> # Simple task -> uses Haiku (cheap) >>> response = await llm.interact( ... user_id="dev", ... user_input="Summarize this function", ... task_type="summarize" ... ) >>> # Complex task -> uses Opus (premium) >>> response = await llm.interact( ... user_id="dev", ... user_input="Design the architecture", ... task_type="architectural_decision" ... )

__init__(provider='anthropic', target_level=3, api_key=None, model=None, pattern_library=None, claude_memory_config=None, project_root=None, enable_security=None, security_config=None, enable_model_routing=False, **kwargs)

Initialize EmpathyLLM.

Parameters:

Name Type Description Default
provider str

"anthropic"

'anthropic'
target_level int

Target empathy level (1-5)

3
api_key str | None

API key for provider (if needed)

None
model str | None

Specific model to use (overrides routing if set)

None
pattern_library dict | None

Shared pattern library (Level 5)

None
claude_memory_config ClaudeMemoryConfig | None

Configuration for Claude memory integration (v1.8.0+)

None
project_root str | None

Project root directory for loading .claude/CLAUDE.md

None
enable_security bool | None

Enable Phase 2 security controls. - If None (default): Check ATTUNE_ENABLE_SECURITY env var - If env var not set: Defaults to False (disabled) - In production environments, a warning is logged if security is disabled

None
security_config dict | None

Security configuration dictionary with options: - audit_log_dir: Directory for audit logs (default: "./logs") - block_on_secrets: Block requests with detected secrets (default: True) - enable_pii_scrubbing: Enable PII detection/scrubbing (default: True) - enable_name_detection: Enable name PII detection (default: False) - enable_audit_logging: Enable audit logging (default: True) - enable_console_logging: Log to console for debugging (default: False)

None
enable_model_routing bool

Enable smart model routing for cost optimization. When enabled, uses ModelRouter to select appropriate model tier: - CHEAP (Haiku): summarize, classify, triage tasks - CAPABLE (Sonnet): code generation, bug fixes, security review - PREMIUM (Opus): coordination, synthesis, architectural decisions

False
**kwargs Any

Provider-specific options

{}

add_pattern(user_id, pattern)

Manually add a detected pattern.

Parameters:

Name Type Description Default
user_id str

User identifier

required
pattern UserPattern

UserPattern instance

required

get_statistics(user_id)

Get collaboration statistics for user.

Parameters:

Name Type Description Default
user_id str

User identifier

required

Returns:

Type Description
dict[str, Any]

Dictionary with stats

interact(user_id, user_input, context=None, force_level=None, task_type=None) async

Main interaction method.

Automatically selects appropriate empathy level and responds.

Phase 3 Security Pipeline (if enabled): 1. PII Scrubbing: Detect and redact PII from user input 2. Secrets Detection: Block requests containing secrets 3. LLM Interaction: Process sanitized input 4. Audit Logging: Log request details for compliance

Model Routing (if enable_model_routing=True): Routes to appropriate model based on task_type: - CHEAP (Haiku): summarize, classify, triage, match_pattern - CAPABLE (Sonnet): generate_code, fix_bug, review_security, write_tests - PREMIUM (Opus): coordinate, synthesize_results, architectural_decision

Parameters:

Name Type Description Default
user_id str

Unique user identifier

required
user_input str

User's input/question

required
context dict[str, Any] | None

Optional context dictionary

None
force_level int | None

Force specific level (for testing/demos)

None
task_type str | None

Type of task for model routing (e.g., "summarize", "fix_bug"). If not provided with routing enabled, defaults to "capable" tier.

None

Returns:

Type Description
dict[str, Any]

Dictionary with: - content: LLM response - level_used: Which empathy level was used - proactive: Whether action was proactive - metadata: Additional information (includes routed_model if routing enabled) - security: Security details (if enabled)

Raises:

Type Description
SecurityError

If secrets detected and block_on_secrets=True

reload_memory()

Reload Claude memory files.

Useful if CLAUDE.md files have been updated during runtime. Call this to pick up changes without restarting.

reset_state(user_id)

Reset collaboration state for user.

update_trust(user_id, outcome, magnitude=1.0)

Update trust level based on interaction outcome.

Parameters:

Name Type Description Default
user_id str

User identifier

required
outcome str

"success" or "failure"

required
magnitude float

How much to adjust (0.0 to 1.0)

1.0

Main LLM interface. Enable the built-in security pipeline with enable_security=True.

Example:

import os
from attune.llm import EmpathyLLM

# Initialize with the built-in security pipeline
llm = EmpathyLLM(
    provider="anthropic",
    api_key=os.getenv("ANTHROPIC_API_KEY"),
    enable_security=True,
)

# Secure interaction
response = llm.interact(
    user_id="user_123",
    user_input="Help me debug this API issue",
    context={},
)

PIIScrubber

Detect and scrub personally identifiable information.

Detects:

  • SSN (Social Security Numbers)
  • Credit card numbers
  • Phone numbers (US and international)
  • Email addresses
  • Physical addresses
  • Names (when enabled)
  • Healthcare identifiers (MRN, Patient ID)

Example:

from attune.memory import PIIScrubber

scrubber = PIIScrubber()

# Text with PII
text = """
John Doe (SSN: 123-45-6789)
called from 555-123-4567 about his
credit card ending in 4532.
"""

# scrub() returns (scrubbed_text, detections)
scrubbed, detections = scrubber.scrub(text)
print(scrubbed)
# Output:
# John Doe (SSN: [SSN])
# called from [PHONE] about his
# credit card ending in 4532.

# Inspect what was detected
for item in detections:
    print(f"Confidence: {item.confidence}")

SecretsDetector

Detect API keys, tokens, and credentials.

Detects:

  • API keys (AWS, Stripe, GitHub, etc.)
  • OAuth tokens
  • Private keys
  • Database connection strings
  • JWT tokens

Example:

from attune.memory import SecretsDetector

detector = SecretsDetector()

# Code with secrets
code = """
# Config
STRIPE_KEY = "sk_live_51HxJ..."
AWS_SECRET = "wJalrXUtnFEMI/K7MDENG..."
DB_CONN = "postgresql://user:pass@localhost/db"
"""

# Check for secrets
secrets = detector.detect(code)
if secrets:
    print("Secrets detected!")
    for secret in secrets:
        print(f"  confidence: {secret.confidence}")
        print(f"  context: {secret.context_snippet}")
else:
    print("No secrets detected")

AuditLogger

JSONL audit logging of LLM interactions and security events.

Logs:

  • LLM interactions
  • PII scrubbing and secrets counts
  • Security policy violations
  • Pattern store/retrieve events

Example:

from attune.memory.security import AuditLogger

logger = AuditLogger(log_dir="logs")

# Log an LLM interaction
logger.log_llm_request(
    user_id="user_123",
    empathy_level=4,
    provider="anthropic",
    model="claude-sonnet-4-5",
    memory_sources=[],
    pii_count=2,
    secrets_count=0,
)

# Log a security policy violation
logger.log_security_violation(
    user_id="user_123",
    violation_type="blocked_secret",
    severity="high",
    details={"reason": "API key in prompt"},
)

Security Features

PII Scrubbing Patterns

from attune.memory import PIIScrubber

# Default patterns (includes MRN and Patient ID)
scrubber = PIIScrubber()

# Add a custom pattern
scrubber.add_custom_pattern(
    name="employee_id",
    pattern=r"\bEMP\d{6}\b",
    replacement="[EMP_ID]",
)

text = "Employee EMP123456 accessed MRN: 987654"
scrubbed, _ = scrubber.scrub(text)
print(scrubbed)
# Output: Employee [EMP_ID] accessed [MRN]

Secrets Detection Configuration

from attune.memory import SecretsDetector

detector = SecretsDetector(
    entropy_threshold=4.5,  # Lower = more sensitive
)

# Custom secret pattern
detector.add_custom_pattern(
    name="internal_api_key",
    pattern=r"INTERNAL_[A-Za-z0-9]{32}",
    severity="high",
)

# Check code before committing
with open("config.py") as f:
    code = f.read()

secrets = detector.detect(code)
if secrets:
    print("Do not commit! Secrets detected:")
    for secret in secrets:
        print(f"  confidence {secret.confidence}")

Audit Logging Format

{
  "timestamp": "2025-01-20T15:30:00Z",
  "event_type": "llm_request",
  "user_id": "user_123",
  "provider": "anthropic",
  "model": "claude-sonnet-4-5",
  "empathy_level": 4,
  "pii_count": 2,
  "secrets_count": 0,
  "duration_ms": 1234
}

Claude Memory Integration

CLAUDE.md Support

import os
from attune.llm import EmpathyLLM
from attune.memory import ClaudeMemoryConfig

# Configure Claude Memory
memory_config = ClaudeMemoryConfig(
    enabled=True,
    load_enterprise=True,  # /etc/claude/CLAUDE.md
    load_user=True,        # ~/.claude/CLAUDE.md
    load_project=True,     # ./.claude/CLAUDE.md
)

# Initialize with memory
llm = EmpathyLLM(
    provider="anthropic",
    api_key=os.getenv("ANTHROPIC_API_KEY"),
    claude_memory_config=memory_config,
)

# Memory is automatically loaded and included in context
response = llm.interact(
    user_id="user_123",
    user_input="Help with deployment",
    context={},
)

# Memory instructions from CLAUDE.md are followed

Usage Patterns

Complete Security Setup

import os
from attune.llm import EmpathyLLM
from attune.memory import PIIScrubber, SecretsDetector
from attune.memory.security import AuditLogger

# Initialize security components
pii_scrubber = PIIScrubber()
secrets_detector = SecretsDetector()
audit_logger = AuditLogger(log_dir="logs")

# Configure the LLM with the built-in security pipeline
llm = EmpathyLLM(
    provider="anthropic",
    api_key=os.getenv("ANTHROPIC_API_KEY"),
    enable_security=True,
)

# Interactions run through the security pipeline
response = llm.interact(
    user_id="user_123",
    user_input="Help debug this error",
    context={},
)

Best Practices

Production Security Checklist

  • [ ] Enable the security pipeline (enable_security=True)
  • [ ] Run prompts through PIIScrubber before sending
  • [ ] Run prompts through SecretsDetector before sending
  • [ ] Keep an AuditLogger trail of interactions
  • [ ] Use encrypted storage (SQLite encryption or PostgreSQL + encryption at rest)
  • [ ] Rotate API keys regularly
  • [ ] Monitor audit logs
  • [ ] Review access patterns periodically

See Also