EmpathyOS¶

The main entry point for the Empathy Framework. EmpathyOS orchestrates empathy level progression, trust management, and interaction handling.

Overview¶

EmpathyOS is the primary class you'll interact with when building empathy-aware AI systems. It handles:

Level Progression: Automatically advances through empathy levels 1-5 based on trust
Trust Management: Tracks collaboration trust with built-in erosion and building rates
Interaction Logic: Routes requests through appropriate empathy level handlers
Pattern Learning: Discovers and applies patterns for improved responses
State Persistence: Saves and restores user collaboration states

Basic Usage¶

from empathy_os import EmpathyOS

# Initialize with Level 4 target
empathy = EmpathyOS(
    user_id="user_123",
    target_level=4,
    confidence_threshold=0.75,
    persistence_enabled=True
)

# Single interaction
response = empathy.interact(
    user_id="user_123",
    user_input="How do I fix this bug?",
    context={"task": "debugging"}
)

print(response.response)  # AI response
print(response.level)     # Current empathy level
print(response.confidence)  # Confidence score

Class Reference¶

Empathy Operating System for AI-Human Collaboration

Integrates: - 5-level Empathy Maturity Model - Systems Thinking (feedback loops, emergence, leverage points) - Tactical Empathy (Voss) - Emotional Intelligence (Goleman) - Clear Thinking (Naval)

Goal: Enable AI to operate at Levels 3-4 (Proactive/Anticipatory)

Example

empathy = EmpathyOS(user_id="developer_123", target_level=4) result = await empathy.level_4_anticipatory(system_trajectory) print(result["bottlenecks_predicted"])

`memory` `property` ¶

Unified memory interface for both short-term and long-term storage.

Lazily initializes on first access with environment auto-detection.

Usage

empathy = EmpathyOS(user_id="agent_1")

Store working data (short-term)¶

empathy.memory.stash("analysis", {"results": [...]})

Persist pattern (long-term)¶

result = empathy.memory.persist_pattern( content="Algorithm for X", pattern_type="algorithm", )

Retrieve pattern¶

pattern = empathy.memory.recall_pattern(result["pattern_id"])

`session_id` `property` ¶

Get or generate a unique session ID for this agent instance.

`aenter()` `async` ¶

Enter async context manager

Enables usage: async with EmpathyOS(...) as empathy:

Returns:

Name	Type	Description
`self`		The EmpathyOS instance

`aexit(exc_type, exc_val, exc_tb)` `async` ¶

Exit async context manager

Performs cleanup when exiting the context: - Saves patterns if persistence is enabled - Closes any open connections - Logs final collaboration state

Parameters:

Name	Description	Default
`exc_type`	Exception type if an exception occurred	required
`exc_val`	Exception value if an exception occurred	required
`exc_tb`	Exception traceback if an exception occurred	required

Returns:

Type	Description
	False to propagate exceptions (standard behavior)

`init(user_id, target_level=3, confidence_threshold=0.75, logger=None, shared_library=None, short_term_memory=None, access_tier=AccessTier.CONTRIBUTOR)` ¶

Initialize EmpathyOS

Parameters:

Name	Type	Description	Default
`user_id`	`str`	Unique identifier for user/team	required
`target_level`	`int`	Target empathy level (1-5), default 3 (Proactive)	`3`
`confidence_threshold`	`float`	Minimum confidence for anticipatory actions (0.0-1.0)	`0.75`
`logger`	`Logger \| None`	Optional logger instance for structured logging	`None`
`shared_library`	`PatternLibrary \| None`	Optional shared PatternLibrary for multi-agent collaboration. When provided, enables agents to share discovered patterns, supporting Level 5 (Systems Empathy) distributed memory networks.	`None`
`short_term_memory`	`RedisShortTermMemory \| None`	Optional RedisShortTermMemory for fast, TTL-based working memory. Enables real-time multi-agent coordination, pattern staging, and conflict resolution.	`None`
`access_tier`	`AccessTier`	Access tier for this agent (Observer, Contributor, Validator, Steward). Determines what operations the agent can perform on shared memory.	`CONTRIBUTOR`

`contribute_pattern(pattern)` ¶

Contribute a discovered pattern to the shared library.

Enables Level 5 Systems Empathy: patterns discovered by this agent become available to all other agents sharing the same library.

Parameters:

Name	Type	Description	Default
`pattern`		Pattern object to contribute	required

Raises:

Type	Description
`RuntimeError`	If no shared library is configured

Example

from empathy_os import Pattern, PatternLibrary library = PatternLibrary() agent = EmpathyOS(user_id="code_reviewer", shared_library=library) pattern = Pattern( ... id="pat_001", ... agent_id="code_reviewer", ... pattern_type="best_practice", ... name="Test pattern", ... description="A discovered pattern", ... ) agent.contribute_pattern(pattern)

`get_collaboration_state()` ¶

Get current collaboration state

`get_memory_stats()` ¶

Get statistics about the short-term memory system.

Returns:

Type	Description
`dict \| None`	Dict with memory usage, key counts, mode, or None if not configured

`get_staged_patterns()` ¶

Get all patterns currently in staging.

Returns patterns staged by any agent that are awaiting validation. Validators use this to review and promote/reject patterns.

Returns:

Type	Description
`list[StagedPattern]`	List of StagedPattern objects

Raises:

Type	Description
`RuntimeError`	If no short-term memory configured

`has_shared_library()` ¶

Check if this agent has a shared pattern library configured.

`has_short_term_memory()` ¶

Check if this agent has short-term memory configured.

`level_1_reactive(user_request)` `async` ¶

Level 1: Reactive Empathy

Respond to explicit request accurately and helpfully. No anticipation, no proactive action.

Parameters:

Name	Type	Description	Default
`user_request`	`str`	User's explicit request	required

Returns:

Type	Description
`dict`	Dict with result and reasoning

Raises:

Type	Description
`ValueError`	If user_request is empty or not a string

`level_2_guided(user_request)` `async` ¶

Level 2: Guided Empathy

Use calibrated questions (Voss) to clarify intent before acting. Collaborative exploration to uncover hidden needs.

Parameters:

Name	Type	Description	Default
`user_request`	`str`	User's request (potentially ambiguous)	required

Returns:

Type	Description
`dict`	Dict with clarification questions or refined result

Raises:

Type	Description
`ValueError`	If user_request is empty or not a string

`level_3_proactive(context)` `async` ¶

Level 3: Proactive Empathy

Detect patterns, act on leading indicators. Take initiative without being asked.

Parameters:

Name	Type	Description	Default
`context`	`dict`	Current context (user activity, system state, etc.)	required

Returns:

Type	Description
`dict`	Dict with proactive actions taken

Raises:

Type	Description
`ValueError`	If context is not a dict or is empty

`level_4_anticipatory(system_trajectory)` `async` ¶

Level 4: Anticipatory Empathy (THE INNOVATION)

Predict future bottlenecks, design relief in advance.

This is STRATEGIC CARE: - Timing + Prediction + Initiative - Solve tomorrow's pain today - Act without being told (but without overstepping)

Parameters:

Name	Type	Description	Default
`system_trajectory`	`dict`	System state + growth trends + constraints	required

Returns:

Type	Description
`dict`	Dict with predicted bottlenecks and interventions

Raises:

Type	Description
`ValueError`	If system_trajectory is not a dict or is empty

`level_5_systems(domain_context)` `async` ¶

Level 5: Systems Empathy

Build structures that help at scale. Design leverage points, frameworks, self-sustaining systems.

This is ARCHITECTURAL CARE: - One framework → infinite applications - Solve entire problem class, not individual instances - Design for emergence of desired properties

Parameters:

Name	Type	Description	Default
`domain_context`	`dict`	Domain information, recurring problems, patterns	required

Returns:

Type	Description
`dict`	Dict with designed frameworks and leverage points

Raises:

Type	Description
`ValueError`	If domain_context is not a dict or is empty

`monitor_feedback_loops(session_history)` ¶

Detect and manage feedback loops in collaboration

`persist_collaboration_state()` ¶

Persist current collaboration state to short-term memory.

Call periodically to save state that can be recovered if the agent restarts. State expires after 30 minutes by default.

Returns:

Type	Description
`bool`	True if persisted successfully

Raises:

Type	Description
`RuntimeError`	If no short-term memory configured

`persist_pattern(content, pattern_type, classification=None, auto_classify=True)` ¶

Store a pattern in long-term memory with security controls.

This is a convenience method that delegates to memory.persist_pattern().

Parameters:

Name	Type	Description	Default
`content`	`str`	Pattern content	required
`pattern_type`	`str`	Type (algorithm, protocol, config, etc.)	required
`classification`	`Classification \| str \| None`	Security classification (or auto-detect)	`None`
`auto_classify`	`bool`	Auto-detect classification from content	`True`

Returns:

Type	Description
`dict \| None`	Storage result with pattern_id and classification

Example

empathy = EmpathyOS(user_id="dev@company.com") result = empathy.persist_pattern( ... content="Our proprietary algorithm for...", ... pattern_type="algorithm", ... ) print(result["classification"]) # "INTERNAL"

`query_patterns(context, **kwargs)` ¶

Query the shared library for patterns relevant to the current context.

Enables agents to benefit from patterns discovered by other agents in the distributed memory network.

Parameters:

Name	Type	Description	Default
`context`	`dict`	Dictionary describing the current context	required
`**kwargs`		Additional arguments passed to PatternLibrary.query_patterns() (e.g., pattern_type, min_confidence, limit)	`{}`

Returns:

Type	Description
	List of PatternMatch objects sorted by relevance

Raises:

Type	Description
`RuntimeError`	If no shared library is configured

Example

matches = agent.query_patterns( ... context={"language": "python", "task": "code_review"}, ... min_confidence=0.7 ... ) for match in matches: ... print(f"{match.pattern.name}: {match.relevance_score:.0%}")

`recall_pattern(pattern_id)` ¶

Retrieve a pattern from long-term memory.

This is a convenience method that delegates to memory.recall_pattern().

Parameters:

Name	Type	Description	Default
`pattern_id`	`str`	ID of pattern to retrieve	required

Returns:

Type	Description
`dict \| None`	Pattern data with content and metadata

Example

pattern = empathy.recall_pattern("pat_123") print(pattern["content"])

`receive_signals(signal_type=None)` ¶

Receive coordination signals from other agents.

Returns signals targeted at this agent or broadcast signals. Signals expire after 5 minutes (TTL).

Parameters:

Name	Type	Description	Default
`signal_type`	`str \| None`	Filter by signal type, or None for all	`None`

Returns:

Type	Description
`list[dict]`	List of signal dicts with sender, type, data, timestamp

Raises:

Type	Description
`RuntimeError`	If no short-term memory configured

Example

signals = empathy.receive_signals("analysis_complete") for sig in signals: ... print(f"From {sig['sender']}: {sig['data']}")

`reset_collaboration_state()` ¶

Reset collaboration state (new session)

`restore_collaboration_state(session_id=None)` ¶

Restore collaboration state from short-term memory.

Use to recover state after agent restart or to continue a previous session.

Parameters:

Name	Type	Description	Default
`session_id`	`str \| None`	Session to restore, or None for current session	`None`

Returns:

Type	Description
`bool`	True if state was found and restored

Raises:

Type	Description
`RuntimeError`	If no short-term memory configured

`retrieve(key)` ¶

Retrieve data from short-term memory.

This is a convenience method that delegates to memory.retrieve().

Parameters:

Name	Type	Description	Default
`key`	`str`	Storage key	required

Returns:

Type	Description
`Any`	Stored data or None

`send_signal(signal_type, data, target_agent=None)` ¶

Send a coordination signal to other agents.

Use signals for real-time coordination: - Notify completion of tasks - Request assistance - Broadcast status updates

Parameters:

Name	Type	Description	Default
`signal_type`	`str`	Type of signal (e.g., "task_complete", "need_review")	required
`data`	`dict`	Signal payload	required
`target_agent`	`str \| None`	Specific agent to target, or None for broadcast	`None`

Returns:

Type	Description
`bool`	True if sent successfully

Raises:

Type	Description
`RuntimeError`	If no short-term memory configured

Example

Notify specific agent¶

empathy.send_signal( ... "analysis_complete", ... {"files": 10, "issues_found": 3}, ... target_agent="lead_reviewer" ... )

Broadcast to all¶

empathy.send_signal("status_update", {"phase": "testing"})

`stage_pattern(pattern)` ¶

Stage a discovered pattern for validation.

Patterns are held in a staging area until a Validator promotes them to the active pattern library. This implements the trust-but-verify approach to multi-agent knowledge building.

Parameters:

Name	Type	Description	Default
`pattern`	`StagedPattern`	StagedPattern with discovery details	required

Returns:

Type	Description
`bool`	True if staged successfully

Raises:

Type	Description
`RuntimeError`	If no short-term memory configured
`PermissionError`	If agent lacks Contributor+ access

Example

from empathy_os import StagedPattern pattern = StagedPattern( ... pattern_id="pat_auth_001", ... agent_id=empathy.user_id, ... pattern_type="security", ... name="JWT Token Refresh Pattern", ... description="Refresh tokens before expiry to prevent auth failures", ... confidence=0.85, ... ) empathy.stage_pattern(pattern)

`stash(key, value, ttl_seconds=3600)` ¶

Store data in short-term memory with TTL.

This is a convenience method that delegates to memory.stash().

Parameters:

Name	Type	Description	Default
`key`	`str`	Storage key	required
`value`	`Any`	Data to store	required
`ttl_seconds`	`int`	Time-to-live (default 1 hour)	`3600`

Returns:

Type	Description
`bool`	True if stored successfully

Key Methods¶

`init()`¶

Initialize a new EmpathyOS instance with configuration.

Parameters: - user_id (str): Unique identifier for the user - target_level (int): Target empathy level (1-5, default: 4) - confidence_threshold (float): Minimum confidence for level advancement (0.0-1.0, default: 0.75) - persistence_enabled (bool): Enable state/pattern persistence (default: True) - trust_building_rate (float): Rate of trust increase on success (default: 0.05) - trust_erosion_rate (float): Rate of trust decrease on failure (default: 0.10)

`interact()`¶

Process a user interaction and return an empathy-aware response.

Parameters: - user_id (str): User identifier - user_input (str): User's input message - context (dict): Additional context for the interaction

Returns: - EmpathyResponse: Response object with message, level, confidence, and predictions

Example:

response = empathy.interact(
    user_id="user_123",
    user_input="I'm deploying to production",
    context={"environment": "production", "time": "friday_afternoon"}
)

if response.level >= 4 and response.predictions:
    print("⚠️  Predictions:")
    for prediction in response.predictions:
        print(f"  • {prediction}")

`record_success()` / `record_failure()`¶

Provide feedback to improve trust tracking and pattern learning.

Parameters: - success (bool): Whether the interaction was successful

Example:

response = empathy.interact(user_id="user_123", user_input="Help me debug this")

# User found the response helpful
empathy.record_success(success=True)
print(f"Trust level: {empathy.get_trust_level():.0%}")

`save_state()` / `load_state()`¶

Persist and restore user collaboration state.

Example:

# Save state after session
empathy.save_state(user_id="user_123", filepath=".empathy/user_123.json")

# Restore state in next session
empathy.load_state(user_id="user_123", filepath=".empathy/user_123.json")

Empathy Levels¶

Level 1: Reactive¶

Basic Q&A responses without proactivity.

Trust Required: 0% - 20%

Characteristics: - Answers direct questions only - No suggestions or predictions - Minimal context awareness

Level 2: Guided¶

Asks clarifying questions to understand intent.

Trust Required: 20% - 40%

Characteristics: - Clarifying questions - Better context understanding - More thorough responses

Level 3: Proactive¶

Suggests improvements and best practices.

Trust Required: 40% - 60%

Characteristics: - Proactive suggestions - Best practice recommendations - Code improvements

Level 4: Anticipatory 🎯¶

Predicts problems before they occur (30-90 day horizon).

Trust Required: 60% - 80%

Characteristics: - Problem prediction - Risk assessment - Anticipatory guidance - "What if" scenarios

Example:

response = empathy.interact(
    user_id="user_123",
    user_input="I'm adding this new API endpoint",
    context={"api_version": "v2", "breaking_change": False}
)

# Level 4 response includes predictions
if response.predictions:
    print(response.predictions)
    # ["This may conflict with v1 authentication flow",
    #  "Consider rate limiting for this endpoint",
    #  "Mobile app may need updates"]

Level 5: Transformative 🚀¶

Reshapes workflows and system architecture (90+ day horizon).

Trust Required: 80% - 100%

Characteristics: - Workflow transformation - Architectural recommendations - Long-term strategic guidance - Cross-system optimization

Trust Management¶

Trust level affects which empathy level is active:

empathy = EmpathyOS(user_id="user_123", target_level=4)

# Start at Level 1 (trust = 0%)
print(empathy.get_current_level())  # 1

# Build trust through successful interactions
for _ in range(10):
    response = empathy.interact(user_id="user_123", user_input="...")
    empathy.record_success(success=True)

print(empathy.get_current_level())  # 3 or 4 (depending on trust)
print(f"Trust: {empathy.get_trust_level():.0%}")  # ~50%

Trust Dynamics: - Starts at 0% - Increases on record_success(True) by trust_building_rate (default: +5%) - Decreases on record_failure() by trust_erosion_rate (default: -10%) - Capped at 100%

Configuration¶

See Configuration API for detailed configuration options.

EmpathyOS¶

Overview¶

Basic Usage¶

Class Reference¶

memory property ¶

Store working data (short-term)¶

Persist pattern (long-term)¶

Retrieve pattern¶

session_id property ¶

__aenter__() async ¶

__aexit__(exc_type, exc_val, exc_tb) async ¶

__init__(user_id, target_level=3, confidence_threshold=0.75, logger=None, shared_library=None, short_term_memory=None, access_tier=AccessTier.CONTRIBUTOR) ¶

contribute_pattern(pattern) ¶

get_collaboration_state() ¶

get_memory_stats() ¶

get_staged_patterns() ¶

has_shared_library() ¶

has_short_term_memory() ¶

level_1_reactive(user_request) async ¶

level_2_guided(user_request) async ¶

level_3_proactive(context) async ¶

level_4_anticipatory(system_trajectory) async ¶

level_5_systems(domain_context) async ¶

monitor_feedback_loops(session_history) ¶

persist_collaboration_state() ¶

persist_pattern(content, pattern_type, classification=None, auto_classify=True) ¶

query_patterns(context, **kwargs) ¶

recall_pattern(pattern_id) ¶

receive_signals(signal_type=None) ¶

reset_collaboration_state() ¶

restore_collaboration_state(session_id=None) ¶

retrieve(key) ¶

send_signal(signal_type, data, target_agent=None) ¶

Notify specific agent¶

Broadcast to all¶

stage_pattern(pattern) ¶

stash(key, value, ttl_seconds=3600) ¶

Key Methods¶

__init__()¶

interact()¶

record_success() / record_failure()¶

save_state() / load_state()¶

Empathy Levels¶

Level 1: Reactive¶

Level 2: Guided¶

Level 3: Proactive¶

Level 4: Anticipatory 🎯¶

Level 5: Transformative 🚀¶

Trust Management¶

Configuration¶

See Also¶

`memory` `property` ¶

`session_id` `property` ¶

`aenter()` `async` ¶

`aexit(exc_type, exc_val, exc_tb)` `async` ¶

`init(user_id, target_level=3, confidence_threshold=0.75, logger=None, shared_library=None, short_term_memory=None, access_tier=AccessTier.CONTRIBUTOR)` ¶

`contribute_pattern(pattern)` ¶

`get_collaboration_state()` ¶

`get_memory_stats()` ¶

`get_staged_patterns()` ¶

`has_shared_library()` ¶

`has_short_term_memory()` ¶

`level_1_reactive(user_request)` `async` ¶

`level_2_guided(user_request)` `async` ¶

`level_3_proactive(context)` `async` ¶

`level_4_anticipatory(system_trajectory)` `async` ¶

`level_5_systems(domain_context)` `async` ¶

`monitor_feedback_loops(session_history)` ¶

`persist_collaboration_state()` ¶

`persist_pattern(content, pattern_type, classification=None, auto_classify=True)` ¶

`query_patterns(context, **kwargs)` ¶

`recall_pattern(pattern_id)` ¶

`receive_signals(signal_type=None)` ¶

`reset_collaboration_state()` ¶

`restore_collaboration_state(session_id=None)` ¶

`retrieve(key)` ¶

`send_signal(signal_type, data, target_agent=None)` ¶

`stage_pattern(pattern)` ¶

`stash(key, value, ttl_seconds=3600)` ¶

`init()`¶

`interact()`¶

`record_success()` / `record_failure()`¶

`save_state()` / `load_state()`¶