LLM Toolkit¶

Enterprise-grade LLM integration with security controls and compliance features.

Overview¶

The LLM Toolkit provides:

Unified LLM Interface: Single API for multiple providers (Anthropic, OpenAI, Ollama)
Security Controls: PII scrubbing, secrets detection, content filtering
Compliance: HIPAA, GDPR, SOC2 audit logging
Claude Memory Integration: CLAUDE.md support with Long-Term Memory pattern storage
Healthcare Wizards: FHIR, HL7, clinical protocol support

Key Features¶

Multi-Provider Support¶

from empathy_llm_toolkit import EmpathyLLM

# Anthropic Claude (recommended)
claude = EmpathyLLM(
    provider="anthropic",
    api_key=os.getenv("ANTHROPIC_API_KEY"),
    model="claude-sonnet-4"
)

# OpenAI GPT
openai = EmpathyLLM(
    provider="openai",
    api_key=os.getenv("OPENAI_API_KEY"),
    model="gpt-4"
)

# Local Ollama
local = EmpathyLLM(
    provider="ollama",
    model="llama2"
)

Automatic Security Controls¶

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

Class Reference¶

EmpathyLLM¶

Wraps any LLM provider with Empathy Framework 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=False, security_config=None, enable_model_routing=False, **kwargs)` ¶

Initialize EmpathyLLM.

Parameters:

Name	Type	Description	Default
`provider`	`str`	"anthropic", "openai", or "local"	`'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`	Enable Phase 2 security controls (default: False)	`False`
`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`		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 with empathy integration.

Example:

from empathy_llm_toolkit import EmpathyLLM
from empathy_os import EmpathyOS

# Initialize with security controls
llm = EmpathyLLM(
    provider="anthropic",
    api_key=os.getenv("ANTHROPIC_API_KEY"),
    enable_pii_scrubbing=True,
    enable_secrets_detection=True,
    enable_audit_logging=True
)

# Integrate with EmpathyOS
empathy = EmpathyOS(
    user_id="user_123",
    target_level=4,
    llm_provider=llm
)

# Secure interaction
response = empathy.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 configured) - Healthcare identifiers (MRN, Patient ID)

Example:

from empathy_llm_toolkit.security import PIIScrubber

scrubber = PIIScrubber()

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

# Scrub PII
scrubbed = scrubber.scrub(text)
print(scrubbed)
# Output:
# Patient [NAME_REDACTED] (SSN: [SSN_REDACTED])
# called from [PHONE_REDACTED] about his
# credit card ending in [CREDIT_CARD_REDACTED].

# Get scrubbed items
items = scrubber.get_scrubbed_items(text)
for item in items:
    print(f"Found {item['type']}: {item['value']}")

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 empathy_llm_toolkit.security 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"  {secret['type']}: {secret['value'][:20]}...")
        print(f"  Line {secret['line']}, position {secret['position']}")
else:
    print("✓ No secrets detected")

AuditLogger¶

Compliance audit logging (HIPAA, GDPR, SOC2).

Logs: - All LLM interactions - PII scrubbing events - Secrets detection events - Security policy violations - User access patterns

Example:

from empathy_llm_toolkit.security import AuditLogger

logger = AuditLogger(
    log_path="logs/audit.jsonl",
    include_phi=False  # HIPAA: Don't log PHI
)

# Log LLM interaction
logger.log_llm_request(
    user_id="user_123",
    prompt="Help with deployment",
    model="claude-sonnet-4",
    tokens=1500
)

# Log security event
logger.log_pii_scrubbed(
    user_id="user_123",
    items_scrubbed=["ssn", "phone"],
    count=2
)

# Log access event
logger.log_access(
    user_id="user_123",
    resource="patient_records",
    action="read",
    success=True
)

Security Features¶

PII Scrubbing Patterns¶

from empathy_llm_toolkit.security import PIIScrubber

# Default patterns
scrubber = PIIScrubber()

# Add custom patterns
scrubber.add_pattern(
    name="employee_id",
    pattern=r'\bEMP\d{6}\b',
    replacement="[EMP_ID_REDACTED]"
)

# Healthcare-specific patterns
scrubber.add_pattern(
    name="mrn",
    pattern=r'\bMRN:?\s*\d{6,10}\b',
    replacement="[MRN_REDACTED]"
)

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

Secrets Detection Configuration¶

from empathy_llm_toolkit.security import SecretsDetector

detector = SecretsDetector(
    entropy_threshold=4.5,  # Lower = more sensitive
    allow_test_keys=True    # Allow obvious test keys
)

# Custom secret patterns
detector.add_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"  Line {secret['line']}: {secret['type']}")
        exit(1)

Audit Logging Format¶

{
  "timestamp": "2025-01-20T15:30:00Z",
  "event_id": "evt_abc123",
  "event_type": "llm_request",
  "user_id": "user_123",
  "action": "interact",

  "request": {
    "provider": "anthropic",
    "model": "claude-sonnet-4",
    "prompt_length": 245,
    "tokens_used": 1500
  },

  "security": {
    "pii_scrubbed": 2,
    "secrets_detected": 0,
    "classification": "INTERNAL"
  },

  "empathy": {
    "level": 4,
    "confidence": 0.88,
    "predictions_count": 3
  },

  "performance": {
    "duration_ms": 1234,
    "trust_level": 0.72
  }
}

Claude Memory Integration¶

CLAUDE.md Support¶

from empathy_llm_toolkit import EmpathyLLM
from empathy_llm_toolkit.claude_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",
    prompt="Help with deployment",
    context={}
)

# Memory instructions from CLAUDE.md are automatically followed

Long-Term Memory Pattern Storage¶

from empathy_llm_toolkit.secure_pattern-storage import SecureLong-Term MemoryIntegration

# Initialize with classification
pattern-storage = SecureLong-Term MemoryIntegration(
    claude_memory_config=memory_config,
    classification_mode="auto"  # or "PUBLIC", "INTERNAL", "SENSITIVE"
)

# Store pattern with automatic classification
pattern_data = """
# Deployment Best Practice

Always deploy on Monday mornings:
- Full team available
- Time to fix issues
- Avoid weekend emergencies
"""

result = pattern-storage.store_pattern(
    pattern_content=pattern_data,
    pattern_type="best_practice",
    user_id="user_123",
    auto_classify=True
)

print(f"Pattern stored: {result['pattern_id']}")
print(f"Classification: {result['classification']}")
# Output: Classification: PUBLIC

Healthcare Wizards¶

Clinical Protocol Monitor¶

from empathy_llm_toolkit.wizards import ClinicalProtocolMonitor

# Monitor clinical handoffs
monitor = ClinicalProtocolMonitor(
    protocol="SBAR",  # Situation, Background, Assessment, Recommendation
    enable_hipaa_audit=True
)

# Process handoff
handoff_text = """
Situation: 65yo male, chest pain x2h
Background: Hx of MI, on aspirin
Assessment: STEMI suspected, vitals stable
Recommendation: Activate cath lab
"""

result = monitor.process_handoff(handoff_text)

if result.complete:
    print("✓ SBAR protocol complete")
else:
    print("⚠️  Missing components:")
    for component in result.missing:
        print(f"  - {component}")

if result.safety_flags:
    print("🚨 Safety flags:")
    for flag in result.safety_flags:
        print(f"  - {flag}")

Healthcare Compliance Wizard¶

from empathy_llm_toolkit.wizards import HealthcareComplianceWizard

wizard = HealthcareComplianceWizard(
    frameworks=["HIPAA", "HITECH", "FDA_21CFR11"]
)

# Check compliance of a system
result = wizard.check_compliance(
    system_description="Patient portal with EHR integration",
    features=[
        "patient_authentication",
        "data_encryption",
        "audit_logging",
        "access_controls"
    ]
)

print(f"Compliance score: {result.score:.0%}")

if result.violations:
    print("\n⚠️  Violations:")
    for violation in result.violations:
        print(f"  {violation.framework}: {violation.description}")
        print(f"  Severity: {violation.severity}")
        print(f"  Remediation: {violation.remediation}")

Usage Patterns¶

Complete Security Setup¶

from empathy_llm_toolkit import EmpathyLLM
from empathy_llm_toolkit.security import (
    PIIScrubber,
    SecretsDetector,
    AuditLogger
)

# Initialize security components
pii_scrubber = PIIScrubber()
secrets_detector = SecretsDetector()
audit_logger = AuditLogger(log_path="logs/audit.jsonl")

# Configure LLM with all security features
llm = EmpathyLLM(
    provider="anthropic",
    api_key=os.getenv("ANTHROPIC_API_KEY"),
    enable_pii_scrubbing=True,
    enable_secrets_detection=True,
    enable_audit_logging=True,
    pii_scrubber=pii_scrubber,
    secrets_detector=secrets_detector,
    audit_logger=audit_logger
)

# All interactions are automatically secured
response = llm.interact(
    user_id="user_123",
    prompt="Help debug this error",
    context={}
)

# Security audit trail is automatically created

Multi-Provider Fallback¶

from empathy_llm_toolkit import EmpathyLLM

providers = [
    {"provider": "anthropic", "api_key": os.getenv("ANTHROPIC_API_KEY")},
    {"provider": "openai", "api_key": os.getenv("OPENAI_API_KEY")},
    {"provider": "ollama", "model": "llama2"}  # Local fallback
]

def interact_with_fallback(prompt, context):
    """Try providers in order until one succeeds"""
    for config in providers:
        try:
            llm = EmpathyLLM(**config)
            return llm.interact(
                user_id="user_123",
                prompt=prompt,
                context=context
            )
        except Exception as e:
            print(f"Provider {config['provider']} failed: {e}")
            continue

    raise Exception("All providers failed")

Best Practices¶

HIPAA-Compliant Setup¶

# Healthcare application with HIPAA compliance
llm = EmpathyLLM(
    provider="anthropic",
    api_key=os.getenv("ANTHROPIC_API_KEY"),

    # Security controls
    enable_pii_scrubbing=True,
    enable_secrets_detection=True,
    enable_audit_logging=True,

    # Healthcare-specific
    healthcare_mode=True,
    phi_protection=True,

    # Audit configuration
    audit_config={
        "include_phi": False,  # Never log PHI
        "retention_days": 90,   # HIPAA minimum
        "encryption": "AES-256-GCM"
    }
)

Production Security Checklist¶

[ ] Enable PII scrubbing
[ ] Enable secrets detection
[ ] Enable audit logging
[ ] Use encrypted storage (SQLite encryption or PostgreSQL + encryption at rest)
[ ] Rotate API keys regularly
[ ] Monitor audit logs daily
[ ] Set up alerts for security events
[ ] Test security controls monthly
[ ] Review access patterns weekly

LLM Toolkit¶

Overview¶

Key Features¶

Multi-Provider Support¶

Automatic Security Controls¶

Class Reference¶

EmpathyLLM¶

PII automatically scrubbed, request logged¶

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

add_pattern(user_id, pattern) ¶

get_statistics(user_id) ¶

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

reload_memory() ¶

reset_state(user_id) ¶

update_trust(user_id, outcome, magnitude=1.0) ¶

PIIScrubber¶

SecretsDetector¶

AuditLogger¶

Security Features¶

PII Scrubbing Patterns¶

Secrets Detection Configuration¶

Audit Logging Format¶

Claude Memory Integration¶

CLAUDE.md Support¶

Long-Term Memory Pattern Storage¶

Healthcare Wizards¶

Clinical Protocol Monitor¶

Healthcare Compliance Wizard¶

Usage Patterns¶

Complete Security Setup¶

Multi-Provider Fallback¶

Best Practices¶

HIPAA-Compliant Setup¶

Production Security Checklist¶

See Also¶

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

`add_pattern(user_id, pattern)` ¶

`get_statistics(user_id)` ¶

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

`reload_memory()` ¶

`reset_state(user_id)` ¶

`update_trust(user_id, outcome, magnitude=1.0)` ¶