Skip to content

Core

Core data structures and state management for the Empathy Framework.

Overview

The core module provides fundamental data structures used throughout the framework:

  • CollaborationState: Tracks trust level, current empathy level, and interaction history
  • EmpathyResponse: Container for responses with metadata (level, confidence, predictions)
  • EmpathyLevel: Enumeration of the five empathy levels
  • InteractionHistory: Tracks past interactions for pattern learning

Class Reference

CollaborationState

Tracks the state of collaboration between the AI and user.

Stock & Flow model of AI-human collaboration

Tracks: - Trust level (stock that accumulates/erodes) - Shared context (accumulated understanding) - Success/failure rates (quality metrics) - Flow rates (how fast trust builds/erodes)

Source code in empathy_os/core.py 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
@dataclass
class CollaborationState:
    """
    Stock & Flow model of AI-human collaboration

    Tracks:
    - Trust level (stock that accumulates/erodes)
    - Shared context (accumulated understanding)
    - Success/failure rates (quality metrics)
    - Flow rates (how fast trust builds/erodes)
    """

    # Stocks (accumulate over time)
    trust_level: float = 0.5  # 0.0 to 1.0, start neutral
    shared_context: dict = field(default_factory=dict)
    successful_interventions: int = 0
    failed_interventions: int = 0

    # Flow rates (change stocks per interaction)
    trust_building_rate: float = 0.05  # Per successful interaction
    trust_erosion_rate: float = 0.10  # Per failed interaction (erosion faster)
    context_accumulation_rate: float = 0.1

    # Metadata
    session_start: datetime = field(default_factory=datetime.now)
    total_interactions: int = 0
    trust_trajectory: list[float] = field(default_factory=list)  # Historical trust levels

    def update_trust(self, outcome: str):
        """Update trust stock based on interaction outcome"""
        if outcome == "success":
            self.trust_level += self.trust_building_rate
            self.successful_interventions += 1
        elif outcome == "failure":
            self.trust_level -= self.trust_erosion_rate
            self.failed_interventions += 1

        # Clamp to [0, 1]
        self.trust_level = max(0.0, min(1.0, self.trust_level))
        self.total_interactions += 1

        # Track trajectory
        self.trust_trajectory.append(self.trust_level)

update_trust(outcome)

Update trust stock based on interaction outcome

Source code in empathy_os/core.py 
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
def update_trust(self, outcome: str):
    """Update trust stock based on interaction outcome"""
    if outcome == "success":
        self.trust_level += self.trust_building_rate
        self.successful_interventions += 1
    elif outcome == "failure":
        self.trust_level -= self.trust_erosion_rate
        self.failed_interventions += 1

    # Clamp to [0, 1]
    self.trust_level = max(0.0, min(1.0, self.trust_level))
    self.total_interactions += 1

    # Track trajectory
    self.trust_trajectory.append(self.trust_level)

Attributes: - trust_level (float): Current trust level (0.0-1.0) - current_level (int): Active empathy level (1-5) - target_level (int): Target empathy level to progress toward - interaction_count (int): Total number of interactions - success_count (int): Number of successful interactions - failure_count (int): Number of failed interactions

Example: 

from empathy_os.core import CollaborationState

state = CollaborationState(
    user_id="user_123",
    target_level=4
)

# Track interactions
state.record_interaction(success=True)
print(f"Trust: {state.trust_level:.0%}")
print(f"Current level: {state.current_level}")

# Trust increases with successful interactions
for _ in range(10):
    state.record_interaction(success=True)

print(f"New trust: {state.trust_level:.0%}")  # Higher
print(f"New level: {state.current_level}")    # Advanced

Trust-Level Mapping: - 0% - 20%: Level 1 (Reactive) - 20% - 40%: Level 2 (Guided) - 40% - 60%: Level 3 (Proactive) - 60% - 80%: Level 4 (Anticipatory) - 80% - 100%: Level 5 (Transformative)

EmpathyResponse

Container for AI responses with empathy metadata.

Note: EmpathyOS methods currently return dictionaries. A dedicated EmpathyResponse class will be added in a future version.

Attributes: - response (str): The actual response text - level (int): Empathy level of the response (1-5) - confidence (float): Confidence score (0.0-1.0) - predictions (List[str]): List of predictions (Level 4+) - suggestions (List[str]): List of suggestions (Level 3+) - clarifying_questions (List[str]): Clarifying questions (Level 2+) - metadata (dict): Additional metadata

Example: 

from empathy_os import EmpathyOS

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

response = empathy.interact(
    user_id="user_123",
    user_input="I'm deploying to production on Friday afternoon",
    context={"day": "friday", "time": "afternoon"}
)

# Access response data
print(f"Response: {response.response}")
print(f"Level: {response.level}")
print(f"Confidence: {response.confidence:.0%}")

# Level 4 includes predictions
if response.predictions:
    print("\nPredictions:")
    for pred in response.predictions:
        print(f"  • {pred}")

# Level 3+ includes suggestions
if response.suggestions:
    print("\nSuggestions:")
    for suggestion in response.suggestions:
        print(f"  • {suggestion}")

Response by Level:

Level 1 (Reactive): 

EmpathyResponse(
    response="Here's how to deploy to production: ...",
    level=1,
    confidence=0.85,
    predictions=[],
    suggestions=[],
    clarifying_questions=[]
)

Level 2 (Guided): 

EmpathyResponse(
    response="Before I help with deployment, I have some questions...",
    level=2,
    confidence=0.80,
    clarifying_questions=[
        "Have you run all tests?",
        "Is there a rollback plan?",
        "Have you notified the team?"
    ]
)

Level 3 (Proactive): 

EmpathyResponse(
    response="Here's the deployment process with some improvements...",
    level=3,
    confidence=0.82,
    suggestions=[
        "Add automated smoke tests",
        "Use blue-green deployment",
        "Set up monitoring alerts"
    ]
)

Level 4 (Anticipatory): 

EmpathyResponse(
    response="I recommend delaying until Monday morning. Here's why...",
    level=4,
    confidence=0.88,
    predictions=[
        "Friday deployments have 3x higher incident rate",
        "Weekend support team is understaffed",
        "This conflicts with scheduled maintenance window"
    ],
    suggestions=[
        "Schedule for Monday 9am",
        "Prepare detailed runbook",
        "Have rollback plan ready"
    ]
)

EmpathyLevel

Enumeration of empathy levels.

Bases: ABC

Abstract base class for empathy levels

Each level implements specific behaviors appropriate to that level of empathy sophistication.

Source code in empathy_os/levels.py 
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
class EmpathyLevel(ABC):
    """
    Abstract base class for empathy levels

    Each level implements specific behaviors appropriate to that
    level of empathy sophistication.
    """

    level_number: int
    level_name: str

    def __init__(self):
        self.actions_taken: list[EmpathyAction] = []

    @abstractmethod
    def respond(self, context: dict[str, Any]) -> dict[str, Any]:
        """
        Respond to a situation at this empathy level.

        This abstract method defines the core behavior for each empathy level.
        Subclasses must implement level-specific response logic that corresponds
        to their empathy sophistication.

        Args:
            context: dict[str, Any]
                Dictionary containing situation-specific context. The structure
                varies by level but typically includes fields like 'request',
                'observed_need', 'current_state', 'trajectory', or 'problem_class'.

        Returns:
            dict[str, Any]
                A response dictionary containing:
                - 'level': int - The empathy level (1-5)
                - 'level_name': str - Human-readable level name
                - 'action': str - Type of action taken
                - 'description': str - Description of the response
                - 'initiative': str - Level ('none'|'guided'|'proactive'|'anticipatory'|'systems')
                - 'reasoning': str - Explanation of why this level's approach was used
                - Additional fields specific to the level implementation

        Raises:
            KeyError: If required context keys are missing
            ValueError: If context values are invalid or insufficient

        Note:
            - Level 1 (Reactive): Only provide what was explicitly requested
            - Level 2 (Guided): Ask clarifying questions and suggest options
            - Level 3 (Proactive): Identify and offer help for observed needs
            - Level 4 (Anticipatory): Predict future needs and prepare solutions
            - Level 5 (Systems): Design solutions that help at scale

            Implementations should record actions via self.record_action() and
            maintain consistency in the response format across levels.
        """
        pass

    def record_action(
        self,
        action_type: str,
        description: str,
        context: dict[str, Any],
        outcome: str | None = None,
    ):
        """Record an action taken at this level"""
        action = EmpathyAction(
            level=self.level_number,
            action_type=action_type,
            description=description,
            context=context,
            outcome=outcome,
        )
        self.actions_taken.append(action)

    def get_action_history(self) -> list[EmpathyAction]:
        """Get history of actions at this level"""
        return self.actions_taken

get_action_history()

Get history of actions at this level

Source code in empathy_os/levels.py 
113
114
115
def get_action_history(self) -> list[EmpathyAction]:
    """Get history of actions at this level"""
    return self.actions_taken

record_action(action_type, description, context, outcome=None)

Record an action taken at this level

Source code in empathy_os/levels.py 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
def record_action(
    self,
    action_type: str,
    description: str,
    context: dict[str, Any],
    outcome: str | None = None,
):
    """Record an action taken at this level"""
    action = EmpathyAction(
        level=self.level_number,
        action_type=action_type,
        description=description,
        context=context,
        outcome=outcome,
    )
    self.actions_taken.append(action)

respond(context) abstractmethod

Respond to a situation at this empathy level.

This abstract method defines the core behavior for each empathy level. Subclasses must implement level-specific response logic that corresponds to their empathy sophistication.

Parameters:

Name Type Description Default
context dict[str, Any]

dict[str, Any] Dictionary containing situation-specific context. The structure varies by level but typically includes fields like 'request', 'observed_need', 'current_state', 'trajectory', or 'problem_class'.

 required

Returns:

Type Description
dict[str, Any]

dict[str, Any] A response dictionary containing: - 'level': int - The empathy level (1-5) - 'level_name': str - Human-readable level name - 'action': str - Type of action taken - 'description': str - Description of the response - 'initiative': str - Level ('none'|'guided'|'proactive'|'anticipatory'|'systems') - 'reasoning': str - Explanation of why this level's approach was used - Additional fields specific to the level implementation

Raises:

Type Description
KeyError

If required context keys are missing
ValueError

If context values are invalid or insufficient
Note
  • Level 1 (Reactive): Only provide what was explicitly requested
  • Level 2 (Guided): Ask clarifying questions and suggest options
  • Level 3 (Proactive): Identify and offer help for observed needs
  • Level 4 (Anticipatory): Predict future needs and prepare solutions
  • Level 5 (Systems): Design solutions that help at scale

Implementations should record actions via self.record_action() and maintain consistency in the response format across levels.

Source code in empathy_os/levels.py 
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
@abstractmethod
def respond(self, context: dict[str, Any]) -> dict[str, Any]:
    """
    Respond to a situation at this empathy level.

    This abstract method defines the core behavior for each empathy level.
    Subclasses must implement level-specific response logic that corresponds
    to their empathy sophistication.

    Args:
        context: dict[str, Any]
            Dictionary containing situation-specific context. The structure
            varies by level but typically includes fields like 'request',
            'observed_need', 'current_state', 'trajectory', or 'problem_class'.

    Returns:
        dict[str, Any]
            A response dictionary containing:
            - 'level': int - The empathy level (1-5)
            - 'level_name': str - Human-readable level name
            - 'action': str - Type of action taken
            - 'description': str - Description of the response
            - 'initiative': str - Level ('none'|'guided'|'proactive'|'anticipatory'|'systems')
            - 'reasoning': str - Explanation of why this level's approach was used
            - Additional fields specific to the level implementation

    Raises:
        KeyError: If required context keys are missing
        ValueError: If context values are invalid or insufficient

    Note:
        - Level 1 (Reactive): Only provide what was explicitly requested
        - Level 2 (Guided): Ask clarifying questions and suggest options
        - Level 3 (Proactive): Identify and offer help for observed needs
        - Level 4 (Anticipatory): Predict future needs and prepare solutions
        - Level 5 (Systems): Design solutions that help at scale

        Implementations should record actions via self.record_action() and
        maintain consistency in the response format across levels.
    """
    pass

Values: - REACTIVE = 1 - Basic Q&A - GUIDED = 2 - Asks clarifying questions - PROACTIVE = 3 - Suggests improvements - ANTICIPATORY = 4 - Predicts problems - TRANSFORMATIVE = 5 - Reshapes workflows

Example: 

from empathy_os.core import EmpathyLevel

# Use in comparisons
if response.level >= EmpathyLevel.ANTICIPATORY:
    print("Predictions available!")
    for pred in response.predictions:
        print(f"  • {pred}")

# Get level name
level_name = EmpathyLevel(response.level).name
print(f"Current level: {level_name}")

InteractionHistory

Tracks interaction history for pattern learning.

Note: Interaction history is currently tracked within CollaborationState. A dedicated InteractionHistory class may be added in a future version.

Attributes: - interactions (List[dict]): List of past interactions - max_history (int): Maximum interactions to store (default: 100)

Example: 

from empathy_os.core import InteractionHistory

history = InteractionHistory(max_history=100)

# Record interaction
history.add_interaction(
    user_input="How do I deploy?",
    response="Here's the deployment process...",
    level=3,
    success=True,
    metadata={"context": "deployment"}
)

# Retrieve recent interactions
recent = history.get_recent(n=10)
for interaction in recent:
    print(f"Input: {interaction['user_input']}")
    print(f"Level: {interaction['level']}")
    print(f"Success: {interaction['success']}")

Usage Patterns

Trust Management

from empathy_os import EmpathyOS

empathy = EmpathyOS(
    user_id="user_123",
    target_level=4,
    trust_building_rate=0.05,  # +5% on success
    trust_erosion_rate=0.10     # -10% on failure
)

# Interaction cycle with feedback
response = empathy.interact(
    user_id="user_123",
    user_input="How do I fix this bug?",
    context={}
)

# User found it helpful
if user_satisfied:
    empathy.record_success(success=True)
    # Trust increases by 5%
else:
    empathy.record_failure()
    # Trust decreases by 10%

# Check current state
state = empathy.collaboration_state
print(f"Trust: {state.trust_level:.0%}")
print(f"Level: {state.current_level}")
print(f"Success rate: {state.success_count / state.interaction_count:.0%}")

Level Progression

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

# Start at Level 1
print(f"Starting level: {empathy.get_current_level()}")  # 1

# Build trust to progress
for i in range(15):
    response = empathy.interact(
        user_id="user_123",
        user_input=f"Question {i}",
        context={}
    )
    empathy.record_success(success=True)

    # Check for level advancement
    if response.level > prev_level:
        print(f"Advanced to Level {response.level}!")

# Should reach Level 3 or 4
print(f"Final level: {empathy.get_current_level()}")
print(f"Final trust: {empathy.get_trust_level():.0%}")

Response Handling

response = empathy.interact(
    user_id="user_123",
    user_input="I need to refactor this code",
    context={"task": "refactoring"}
)

# Handle by level
if response.level == 1:
    # Basic response
    print(response.response)

elif response.level == 2:
    # Show clarifying questions
    print(response.response)
    if response.clarifying_questions:
        print("\nQuestions:")
        for q in response.clarifying_questions:
            print(f"  ? {q}")

elif response.level == 3:
    # Show suggestions
    print(response.response)
    if response.suggestions:
        print("\nSuggestions:")
        for s in response.suggestions:
            print(f"  💡 {s}")

elif response.level >= 4:
    # Show predictions and suggestions
    print(response.response)

    if response.predictions:
        print("\n🔮 Predictions:")
        for p in response.predictions:
            print(f"  • {p}")

    if response.suggestions:
        print("\n💡 Suggestions:")
        for s in response.suggestions:
            print(f"  • {s}")

See Also