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
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
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
        key = str(edge.id)
        if key not in snapshot:
            snapshot[key] = edge.uses
        new_weight = self.decayed_weight(edge.weight, snapshot[key])
        try:
            await self._graph.upsert_edge_props(
                scope,
                edge.id,
                {"weight": new_weight, "provenance": "correction"},
            )
        except Exception:
            # Fallback: delete and re-add with updated weight
            try:
                from symfonic.memory.models.edge import MemoryEdge

                updated_edge = MemoryEdge(
                    id=edge.id,
                    source=edge.source,
                    target=edge.target,
                    relationship=edge.relationship,
                    tenant_id=edge.tenant_id,
                    weight=new_weight,
                    provenance="correction",
                    uses=edge.uses,
                    source_episodic_ids=edge.source_episodic_ids,
                )
                await self._graph.delete_edge(scope, edge.id)
                await self._graph.add_edge(scope, updated_edge)
            except Exception:
                logger.debug(
                    "Correction decay failed for edge %s", edge.id,
                )