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
PIIScrubberbefore sending - [ ] Run prompts through
SecretsDetectorbefore sending - [ ] Keep an
AuditLoggertrail of interactions - [ ] Use encrypted storage (SQLite encryption or PostgreSQL + encryption at rest)
- [ ] Rotate API keys regularly
- [ ] Monitor audit logs
- [ ] Review access patterns periodically