Skip to content

symfonic.core.callbacks.correction

correction

NegativeReinforcementHandler -- logarithmic decay on user correction.

Default CallbackHandler helper that walks the turn's inference_paths and applies a logarithmic negative decay to each traversed edge weight.

Stateless except for the injected graph store. Safe to reuse across turns.

NegativeReinforcementHandler 

NegativeReinforcementHandler(graph_store: Any)

Walks traversed edges from the corrected turn and decays their weights.

Uses a logarithmic decay formula

w_new = w_old * (1 - 1/log(uses + e))

with a floor of 0.01 to prevent total zeroing.

Source code in src/symfonic/core/callbacks/correction.py 
28
29
def __init__(self, graph_store: Any) -> None:
    self._graph = graph_store

decayed_weight staticmethod 

decayed_weight(old_weight: float, uses: int) -> float

Pure function -- exposed for unit testing.

Formula: w_new = w_old * (1 - 1/log(uses + e)) where e = math.e so uses=0 gives factor (1 - 1/1.0) = 0.0 (hard reset).

Floor at 0.01 to prevent complete zeroing / division issues.

Source code in src/symfonic/core/callbacks/correction.py 
135
136
137
138
139
140
141
142
143
144
145
146
@staticmethod
def decayed_weight(old_weight: float, uses: int) -> float:
    """Pure function -- exposed for unit testing.

    Formula: w_new = w_old * (1 - 1/log(uses + e))
    where e = math.e so uses=0 gives factor (1 - 1/1.0) = 0.0 (hard reset).

    Floor at 0.01 to prevent complete zeroing / division issues.
    """
    factor = 1.0 - (1.0 / math.log(uses + math.e))
    new = old_weight * factor
    return max(new, 0.01)

on_user_correction async 

on_user_correction(event: UserCorrectionEvent) -> None

Public entrypoint. Iterates every edge in activation_log.traversed_edges (falls back to inference_paths label-pair lookup when absent) and calls _decay_edge() once per edge.

Idempotence: per-turn uses snapshot per edge_id. Repeated application of the same correction with the same snapshot yields the same result.

Source code in src/symfonic/core/callbacks/correction.py 
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
async def on_user_correction(self, event: UserCorrectionEvent) -> None:
    """Public entrypoint. Iterates every edge in activation_log.traversed_edges
    (falls back to inference_paths label-pair lookup when absent) and
    calls _decay_edge() once per edge.

    Idempotence: per-turn uses snapshot per edge_id. Repeated application
    of the same correction with the same snapshot yields the same result.
    """
    from symfonic.memory.types import TenantScope

    # T-7.21.8 (Slice B site #7): prefer event.scope when set so
    # sub_tenant_id + namespace survive the correction-decay path.
    # Pre-7.21 the bare reconstruction from event.tenant_id silently
    # dropped both -- a multi-brand tenant correction would decay
    # edges in the tenant-root partition instead of the brand
    # partition where the original turn was scoped.
    scope = (
        event.scope
        if event.scope is not None
        else TenantScope(tenant_id=event.tenant_id)
    )
    log = event.activation_log or {}

    # Prefer explicit traversed_edges when present
    edges_raw = log.get("traversed_edges") or []
    if not edges_raw:
        # Fallback: walk inference_paths as label pairs
        paths = log.get("inference_paths") or []
        edges_raw = [
            {
                "source_label": p[0],
                "target_label": p[-1],
                "relationship": "ASSOCIATED",
            }
            for p in paths
            if len(p) >= 2
        ]

    if not edges_raw:
        return

    # Snapshot uses for idempotence within this turn
    snapshot: dict[str, int] = {}

    for e_info in edges_raw:
        edge = await self._resolve_edge(scope, e_info)
        if edge is None:
            continue
        owner_path = getattr(edge, "_scope_path", None)
        if not isinstance(owner_path, str):
            owner_path = scope.ancestor_prefix_paths()[0]
        if owner_path != scope.scope_path:
            continue
        key = str(edge.id)
        if key not in snapshot:
            snapshot[key] = edge.uses
        new_weight = self.decayed_weight(edge.weight, snapshot[key])
        try:
            props = {"weight": new_weight, "provenance": "correction"}
            if callable(getattr(type(self._graph), "_upsert_resolved_edge_props", None)):
                await self._graph._upsert_resolved_edge_props(scope, edge, props)
            else:
                await self._graph.upsert_edge_props(scope, edge.id, props)
        except Exception:
            logger.debug("Correction decay failed for edge %s", edge.id,
                         exc_info=True)