memory-path-engine/src/memory_engine/activation.py at master · ly85206559/memory-path-engine · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
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
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
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
from __future__ import annotations

from dataclasses import dataclass
from typing import Protocol

from memory_engine.schema import MemoryEdge, MemoryNode
from memory_engine.semantics import (
    ContradictionCandidate,
    contradiction_bonus,
    semantic_activation_bonus,
)


@dataclass(frozen=True, slots=True)
class ActivationSignal:
    node_id: str
    activation: float
    hop: int = 0
    source_node_id: str | None = None
    via_edge_type: str | None = None


@dataclass(frozen=True, slots=True)
class ActivatedNode:
    node_id: str
    activation: float
    score: float
    hop: int
    source_node_id: str | None = None
    via_edge_type: str | None = None


@dataclass(frozen=True, slots=True)
class PropagationStep:
    from_node_id: str
    to_node_id: str
    edge_type: str
    hop: int
    incoming_activation: float
    propagated_activation: float
    stopped_reason: str | None = None


class PropagationPolicy(Protocol):
    def seed_activation(self, *, seed_score: float) -> float:
        """Return the initial activation for a seed node."""

    def propagate(
        self,
        *,
        signal: ActivationSignal,
        edge: MemoryEdge,
    ) -> PropagationStep:
        """Return the propagation result for a single edge traversal."""

    def adjust_propagated_activation(
        self,
        *,
        propagated_activation: float,
        edge: MemoryEdge,
        destination_node: MemoryNode,
        source_node_id: str | None = None,
        contradiction_candidates: list[ContradictionCandidate] | None = None,
    ) -> float:
        """Apply domain-aware propagation adjustments after a successful edge traversal."""


class DefaultPropagationPolicy:
    def __init__(
        self,
        *,
        activation_decay: float = 0.75,
        activation_threshold: float = 0.15,
        allowed_edge_types: set[str] | None = None,
    ) -> None:
        self.activation_decay = activation_decay
        self.activation_threshold = activation_threshold
        self.allowed_edge_types = allowed_edge_types

    def seed_activation(self, *, seed_score: float) -> float:
        return max(0.0, seed_score)

    def propagate(
        self,
        *,
        signal: ActivationSignal,
        edge: MemoryEdge,
    ) -> PropagationStep:
        if self.allowed_edge_types is not None and edge.edge_type not in self.allowed_edge_types:
            return PropagationStep(
                from_node_id=signal.node_id,
                to_node_id=edge.to_id,
                edge_type=edge.edge_type,
                hop=signal.hop + 1,
                incoming_activation=signal.activation,
                propagated_activation=0.0,
                stopped_reason="disallowed_edge_type",
            )

        propagated_activation = signal.activation * self.activation_decay * edge.weight
        if propagated_activation < self.activation_threshold:
            return PropagationStep(
                from_node_id=signal.node_id,
                to_node_id=edge.to_id,
                edge_type=edge.edge_type,
                hop=signal.hop + 1,
                incoming_activation=signal.activation,
                propagated_activation=propagated_activation,
                stopped_reason="below_threshold",
            )

        return PropagationStep(
            from_node_id=signal.node_id,
            to_node_id=edge.to_id,
            edge_type=edge.edge_type,
            hop=signal.hop + 1,
            incoming_activation=signal.activation,
            propagated_activation=propagated_activation,
        )

    def adjust_propagated_activation(
        self,
        *,
        propagated_activation: float,
        edge: MemoryEdge,
        destination_node: MemoryNode,
        source_node_id: str | None = None,
        contradiction_candidates: list[ContradictionCandidate] | None = None,
    ) -> float:
        adjusted_activation = propagated_activation
        if edge.edge_type == "exception_to":
            adjusted_activation += 0.12
        adjusted_activation += semantic_activation_bonus(destination_node)
        adjusted_activation += contradiction_bonus(
            node_id=destination_node.id,
            candidates=contradiction_candidates or [],
            source_node_id=source_node_id,
        )
        return min(adjusted_activation, 1.0)